You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "202201142 (via GitHub)" <gi...@apache.org> on 2026/07/25 17:12:53 UTC
[PR] Add airflow dags stability check CLI command [airflow]
202201142 opened a new pull request, #70446:
URL: https://github.com/apache/airflow/pull/70446
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: https://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
For user-facing UI changes, please attach before/after screenshots (or a short
screen recording) so reviewers can assess the visual impact.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add airflow dags stability check CLI command [airflow]
Posted by "justinpakzad (via GitHub)" <gi...@apache.org>.
justinpakzad commented on code in PR #70446:
URL: https://github.com/apache/airflow/pull/70446#discussion_r4115803217
##########
airflow-core/src/airflow/cli/commands/dag_command.py:
##########
@@ -80,6 +82,95 @@
_RUN_CHUNK_SIZE = 500
+def _normalize_serialized_dag_for_stability_check(serialized_dag: dict[str, Any]) -> dict[str, Any]:
+ normalized = SerializedDagModel._sort_serialized_dag_dict(serialized_dag)
+ normalized["dag"].pop("fileloc", None)
+ normalized["dag"].pop("bundle_name", None)
+ return normalized
+
+
+def _serialize_dag_for_stability_check(dag: DAG) -> tuple[str, dict[str, Any]]:
+ serialized_dag = DagSerialization.to_dict(dag)
+ return SerializedDagModel.hash(serialized_dag), _normalize_serialized_dag_for_stability_check(
+ serialized_dag
+ )
+
+
+def _format_stability_diff(
+ dag_id: str,
+ first_serialized_dag: dict[str, Any],
+ second_serialized_dag: dict[str, Any],
+) -> str:
+ before = json.dumps(first_serialized_dag, indent=2, sort_keys=True).splitlines()
+ after = json.dumps(second_serialized_dag, indent=2, sort_keys=True).splitlines()
+ diff = difflib.unified_diff(
+ before,
+ after,
+ fromfile=f"parse 1: {dag_id} ",
+ tofile=f"parse 2: {dag_id} ",
+ lineterm="",
+ )
+ return "\n".join(diff)
+
+
+def _parse_dags_for_stability_check(dag_folder: str | None) -> DagBag:
+ return DagBag(dag_folder=dag_folder, load_op_links=False)
+
+
+@cli_utils.action_cli
+@providers_configuration_loaded
+def dag_stability_check(args) -> None:
+ dag_hashes_by_id: dict[str, list[str]] = {}
+ serialized_dags_by_id: dict[str, list[dict[str, Any]]] = {}
+ seen_dag_ids: set[str] = set()
+ nParse = (
+ 2 # NOTE: Set parsing number 2, in most of the case twice parse should catch the stability issues.
Review Comment:
On second thought, if we did make this configurable, we would need to figure out which diffs to show since there could be many different versions. So maybe keeping this fixed at 2 is actually a better approach for now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add airflow dags stability check CLI command [airflow]
Posted by "justinpakzad (via GitHub)" <gi...@apache.org>.
justinpakzad commented on code in PR #70446:
URL: https://github.com/apache/airflow/pull/70446#discussion_r4112057546
##########
airflow-core/src/airflow/cli/commands/dag_command.py:
##########
@@ -80,6 +82,95 @@
_RUN_CHUNK_SIZE = 500
+def _normalize_serialized_dag_for_stability_check(serialized_dag: dict[str, Any]) -> dict[str, Any]:
+ normalized = SerializedDagModel._sort_serialized_dag_dict(serialized_dag)
+ normalized["dag"].pop("fileloc", None)
+ normalized["dag"].pop("bundle_name", None)
+ return normalized
+
+
+def _serialize_dag_for_stability_check(dag: DAG) -> tuple[str, dict[str, Any]]:
+ serialized_dag = DagSerialization.to_dict(dag)
+ return SerializedDagModel.hash(serialized_dag), _normalize_serialized_dag_for_stability_check(
+ serialized_dag
+ )
+
+
+def _format_stability_diff(
+ dag_id: str,
+ first_serialized_dag: dict[str, Any],
+ second_serialized_dag: dict[str, Any],
+) -> str:
+ before = json.dumps(first_serialized_dag, indent=2, sort_keys=True).splitlines()
+ after = json.dumps(second_serialized_dag, indent=2, sort_keys=True).splitlines()
+ diff = difflib.unified_diff(
+ before,
+ after,
+ fromfile=f"parse 1: {dag_id} ",
+ tofile=f"parse 2: {dag_id} ",
+ lineterm="",
+ )
+ return "\n".join(diff)
+
+
+def _parse_dags_for_stability_check(dag_folder: str | None) -> DagBag:
+ return DagBag(dag_folder=dag_folder, load_op_links=False)
+
+
+@cli_utils.action_cli
+@providers_configuration_loaded
+def dag_stability_check(args) -> None:
+ dag_hashes_by_id: dict[str, list[str]] = {}
+ serialized_dags_by_id: dict[str, list[dict[str, Any]]] = {}
+ seen_dag_ids: set[str] = set()
+ nParse = (
Review Comment:
Nit: I think co-pilot noted this already but we should be using snake case.
```suggestion
n_parse = (
```
##########
airflow-core/src/airflow/cli/commands/dag_command.py:
##########
@@ -80,6 +82,95 @@
_RUN_CHUNK_SIZE = 500
+def _normalize_serialized_dag_for_stability_check(serialized_dag: dict[str, Any]) -> dict[str, Any]:
+ normalized = SerializedDagModel._sort_serialized_dag_dict(serialized_dag)
+ normalized["dag"].pop("fileloc", None)
+ normalized["dag"].pop("bundle_name", None)
+ return normalized
+
+
+def _serialize_dag_for_stability_check(dag: DAG) -> tuple[str, dict[str, Any]]:
+ serialized_dag = DagSerialization.to_dict(dag)
+ return SerializedDagModel.hash(serialized_dag), _normalize_serialized_dag_for_stability_check(
+ serialized_dag
+ )
+
+
+def _format_stability_diff(
+ dag_id: str,
+ first_serialized_dag: dict[str, Any],
+ second_serialized_dag: dict[str, Any],
+) -> str:
+ before = json.dumps(first_serialized_dag, indent=2, sort_keys=True).splitlines()
+ after = json.dumps(second_serialized_dag, indent=2, sort_keys=True).splitlines()
+ diff = difflib.unified_diff(
+ before,
+ after,
+ fromfile=f"parse 1: {dag_id} ",
+ tofile=f"parse 2: {dag_id} ",
+ lineterm="",
+ )
+ return "\n".join(diff)
+
+
+def _parse_dags_for_stability_check(dag_folder: str | None) -> DagBag:
+ return DagBag(dag_folder=dag_folder, load_op_links=False)
+
+
+@cli_utils.action_cli
+@providers_configuration_loaded
+def dag_stability_check(args) -> None:
+ dag_hashes_by_id: dict[str, list[str]] = {}
+ serialized_dags_by_id: dict[str, list[dict[str, Any]]] = {}
+ seen_dag_ids: set[str] = set()
+ nParse = (
+ 2 # NOTE: Set parsing number 2, in most of the case twice parse should catch the stability issues.
+ )
+
+ for _iteration in range(1, nParse + 1):
Review Comment:
Nit: we should use `_` for placeholder variables that aren't used inside the loop at all.
```suggestion
for _ in range(1, nParse + 1):
```
##########
airflow-core/src/airflow/cli/commands/dag_command.py:
##########
@@ -80,6 +82,95 @@
_RUN_CHUNK_SIZE = 500
+def _normalize_serialized_dag_for_stability_check(serialized_dag: dict[str, Any]) -> dict[str, Any]:
+ normalized = SerializedDagModel._sort_serialized_dag_dict(serialized_dag)
+ normalized["dag"].pop("fileloc", None)
+ normalized["dag"].pop("bundle_name", None)
+ return normalized
+
+
+def _serialize_dag_for_stability_check(dag: DAG) -> tuple[str, dict[str, Any]]:
+ serialized_dag = DagSerialization.to_dict(dag)
+ return SerializedDagModel.hash(serialized_dag), _normalize_serialized_dag_for_stability_check(
+ serialized_dag
+ )
+
+
+def _format_stability_diff(
+ dag_id: str,
+ first_serialized_dag: dict[str, Any],
+ second_serialized_dag: dict[str, Any],
+) -> str:
+ before = json.dumps(first_serialized_dag, indent=2, sort_keys=True).splitlines()
+ after = json.dumps(second_serialized_dag, indent=2, sort_keys=True).splitlines()
+ diff = difflib.unified_diff(
+ before,
+ after,
+ fromfile=f"parse 1: {dag_id} ",
+ tofile=f"parse 2: {dag_id} ",
+ lineterm="",
+ )
+ return "\n".join(diff)
+
+
+def _parse_dags_for_stability_check(dag_folder: str | None) -> DagBag:
+ return DagBag(dag_folder=dag_folder, load_op_links=False)
+
+
+@cli_utils.action_cli
+@providers_configuration_loaded
+def dag_stability_check(args) -> None:
+ dag_hashes_by_id: dict[str, list[str]] = {}
+ serialized_dags_by_id: dict[str, list[dict[str, Any]]] = {}
+ seen_dag_ids: set[str] = set()
+ nParse = (
+ 2 # NOTE: Set parsing number 2, in most of the case twice parse should catch the stability issues.
+ )
+
+ for _iteration in range(1, nParse + 1):
+ dagbag = _parse_dags_for_stability_check(args.dag_folder)
+
+ dags = dagbag.dags
+ if args.dag_id is not None:
+ dags = {args.dag_id: dagbag.dags[args.dag_id]} if args.dag_id in dagbag.dags else {}
+
+ for _dag_id, dag in sorted(dags.items()):
+ dag_hash, serialized_dag = _serialize_dag_for_stability_check(dag)
+ dag_hashes_by_id.setdefault(_dag_id, []).append(dag_hash)
+ serialized_dags_by_id.setdefault(_dag_id, []).append(serialized_dag)
+ seen_dag_ids.add(_dag_id)
+
+ if args.fail_fast:
+ for _dag_id, dag_hashes in dag_hashes_by_id.items():
+ if len(set(dag_hashes)) > 1:
+ break
+ else:
+ continue
+ break
Review Comment:
I think this is a no-op if the number of parses is fixed at 2. On the first iteration, each Dag will only have 1 hash, `if len(set(dag_hashes)) > 1:` can only be True on the second iteration, which is already the last iteration of `range(1, nParse + 1)` so the loop ends regardless.
Related to my other comment, if we made the number of parses something a user can pass in (e.g. 5), then I think this would be useful.
##########
airflow-core/src/airflow/cli/commands/dag_command.py:
##########
@@ -80,6 +82,95 @@
_RUN_CHUNK_SIZE = 500
+def _normalize_serialized_dag_for_stability_check(serialized_dag: dict[str, Any]) -> dict[str, Any]:
+ normalized = SerializedDagModel._sort_serialized_dag_dict(serialized_dag)
+ normalized["dag"].pop("fileloc", None)
+ normalized["dag"].pop("bundle_name", None)
+ return normalized
+
+
+def _serialize_dag_for_stability_check(dag: DAG) -> tuple[str, dict[str, Any]]:
+ serialized_dag = DagSerialization.to_dict(dag)
+ return SerializedDagModel.hash(serialized_dag), _normalize_serialized_dag_for_stability_check(
+ serialized_dag
+ )
+
+
+def _format_stability_diff(
+ dag_id: str,
+ first_serialized_dag: dict[str, Any],
+ second_serialized_dag: dict[str, Any],
+) -> str:
+ before = json.dumps(first_serialized_dag, indent=2, sort_keys=True).splitlines()
+ after = json.dumps(second_serialized_dag, indent=2, sort_keys=True).splitlines()
+ diff = difflib.unified_diff(
+ before,
+ after,
+ fromfile=f"parse 1: {dag_id} ",
+ tofile=f"parse 2: {dag_id} ",
+ lineterm="",
+ )
+ return "\n".join(diff)
+
+
+def _parse_dags_for_stability_check(dag_folder: str | None) -> DagBag:
+ return DagBag(dag_folder=dag_folder, load_op_links=False)
+
+
+@cli_utils.action_cli
+@providers_configuration_loaded
+def dag_stability_check(args) -> None:
+ dag_hashes_by_id: dict[str, list[str]] = {}
+ serialized_dags_by_id: dict[str, list[dict[str, Any]]] = {}
+ seen_dag_ids: set[str] = set()
+ nParse = (
+ 2 # NOTE: Set parsing number 2, in most of the case twice parse should catch the stability issues.
+ )
+
+ for _iteration in range(1, nParse + 1):
+ dagbag = _parse_dags_for_stability_check(args.dag_folder)
Review Comment:
If the folder provided by the args doesn't exist the command would succeed and return "Dag stability check passed for 0 Dags(s)". Should a non-existent path fail here, given a missing --dag-id does?
##########
airflow-core/src/airflow/cli/commands/dag_command.py:
##########
@@ -80,6 +82,95 @@
_RUN_CHUNK_SIZE = 500
+def _normalize_serialized_dag_for_stability_check(serialized_dag: dict[str, Any]) -> dict[str, Any]:
+ normalized = SerializedDagModel._sort_serialized_dag_dict(serialized_dag)
+ normalized["dag"].pop("fileloc", None)
+ normalized["dag"].pop("bundle_name", None)
+ return normalized
+
+
+def _serialize_dag_for_stability_check(dag: DAG) -> tuple[str, dict[str, Any]]:
+ serialized_dag = DagSerialization.to_dict(dag)
+ return SerializedDagModel.hash(serialized_dag), _normalize_serialized_dag_for_stability_check(
+ serialized_dag
+ )
+
+
+def _format_stability_diff(
+ dag_id: str,
+ first_serialized_dag: dict[str, Any],
+ second_serialized_dag: dict[str, Any],
+) -> str:
+ before = json.dumps(first_serialized_dag, indent=2, sort_keys=True).splitlines()
+ after = json.dumps(second_serialized_dag, indent=2, sort_keys=True).splitlines()
+ diff = difflib.unified_diff(
+ before,
+ after,
+ fromfile=f"parse 1: {dag_id} ",
+ tofile=f"parse 2: {dag_id} ",
+ lineterm="",
+ )
+ return "\n".join(diff)
+
+
+def _parse_dags_for_stability_check(dag_folder: str | None) -> DagBag:
+ return DagBag(dag_folder=dag_folder, load_op_links=False)
+
+
+@cli_utils.action_cli
+@providers_configuration_loaded
+def dag_stability_check(args) -> None:
+ dag_hashes_by_id: dict[str, list[str]] = {}
+ serialized_dags_by_id: dict[str, list[dict[str, Any]]] = {}
+ seen_dag_ids: set[str] = set()
+ nParse = (
+ 2 # NOTE: Set parsing number 2, in most of the case twice parse should catch the stability issues.
Review Comment:
Do you think we should make this a configurable value with a default of two (e.g., an optional flag --n_parses)?
##########
airflow-core/src/airflow/cli/commands/dag_command.py:
##########
@@ -80,6 +82,95 @@
_RUN_CHUNK_SIZE = 500
+def _normalize_serialized_dag_for_stability_check(serialized_dag: dict[str, Any]) -> dict[str, Any]:
+ normalized = SerializedDagModel._sort_serialized_dag_dict(serialized_dag)
+ normalized["dag"].pop("fileloc", None)
+ normalized["dag"].pop("bundle_name", None)
Review Comment:
Any reason these are popped here but still included in the `hash()` used for comparison?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add airflow dags stability check CLI command [airflow]
Posted by "dipb-sl (via GitHub)" <gi...@apache.org>.
dipb-sl commented on PR #70446:
URL: https://github.com/apache/airflow/pull/70446#issuecomment-5161442949
> Is a CLI command really necessary?
>
> I feel this belongs to lint rules. Eg., Ruff [AIR304](https://docs.astral.sh/ruff/rules/airflow3-dag-dynamic-value/) captures this exactly
AIR304 is AST based checking so it will not cover all the cases. Simple example is:
If you store `datetime.now()` in variable and then assign it to `start_time` Ruff check will not catch this issue if assign `datetime.now` directly then only it can catch you can verify it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add airflow dags stability check CLI command [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #70446:
URL: https://github.com/apache/airflow/pull/70446#discussion_r3653122727
##########
airflow-core/docs/faq.rst:
##########
@@ -418,7 +418,8 @@ configuration option:
Additionally, you can catch these issues earlier in your development workflow by using the
`AIR304 <https://docs.astral.sh/ruff/rules/airflow3-dag-dynamic-value/>`_ ruff rule, which detects
dynamic values in Dag and Task constructors as part of static linting. See
-:ref:`best_practices/code_quality_and_linting` for how to set up ruff with Airflow-specific rules.
+:ref:`best_practices/code_quality_and_linting` for how to set up ruff with Airflow-specific rules.
+Alternatively, you can use the airflow dags stability CLI command to detect non-deterministic DAG serialization.
Review Comment:
This adds trailing whitespace at the end of the line (can break some linting) and uses "DAG" in prose. In this repo's prose, prefer "Dag"; also consider formatting the CLI command as an inline literal.
##########
airflow-core/src/airflow/cli/commands/dag_command.py:
##########
@@ -80,6 +82,96 @@
_RUN_CHUNK_SIZE = 500
+def _normalize_serialized_dag_for_stability_check(serialized_dag: dict[str, Any]) -> dict[str, Any]:
+ normalized = SerializedDagModel._sort_serialized_dag_dict(serialized_dag)
+ normalized["dag"].pop("fileloc", None)
+ normalized["dag"].pop("bundle_name", None)
+ return normalized
+
+
+def _serialize_dag_for_stability_check(dag: DAG) -> tuple[str, dict[str, Any]]:
+ serialized_dag = DagSerialization.to_dict(dag)
+ return SerializedDagModel.hash(serialized_dag), _normalize_serialized_dag_for_stability_check(
+ serialized_dag
+ )
+
+
+def _format_stability_diff(
+ dag_id: str,
+ first_serialized_dag: dict[str, Any],
+ second_serialized_dag: dict[str, Any],
+) -> str:
+ before = json.dumps(first_serialized_dag, indent=2, sort_keys=True).splitlines()
+ after = json.dumps(second_serialized_dag, indent=2, sort_keys=True).splitlines()
+ diff = difflib.unified_diff(
+ before,
+ after,
+ fromfile=f"parse 1: {dag_id} ",
+ tofile=f"parse 2: {dag_id} ",
+ lineterm="",
+ )
+ return "\n".join(diff)
+
+
+def _parse_dags_for_stability_check(dag_folder: str | None) -> DagBag:
+ return DagBag(dag_folder=dag_folder, load_op_links=False)
+
+
+@cli_utils.action_cli
+@providers_configuration_loaded
+def dag_stability_check(args) -> None:
+ dag_hashes_by_id: dict[str, list[str]] = {}
+ serialized_dags_by_id: dict[str, list[dict[str, Any]]] = {}
+ seen_dag_ids: set[str] = set()
+ nParse = (
+ 2 # NOTE: Set parsing number 2, in most of the case twice parse should catch the stability issues.
+ )
+
+ for _iteration in range(1, nParse + 1):
+ dagbag = _parse_dags_for_stability_check(args.dag_folder)
+
+ dags = dagbag.dags
+ if args.dag_id is not None:
+ dags = {args.dag_id: dagbag.dags[args.dag_id]} if args.dag_id in dagbag.dags else {}
+
+ for _dag_id, dag in sorted(dags.items()):
+ dag_hash, serialized_dag = _serialize_dag_for_stability_check(dag)
+ dag_hashes_by_id.setdefault(_dag_id, []).append(dag_hash)
+ serialized_dags_by_id.setdefault(_dag_id, []).append(serialized_dag)
+ seen_dag_ids.add(_dag_id)
+
+ if args.fail_fast:
+ for _dag_id, dag_hashes in dag_hashes_by_id.items():
+ if len(set(dag_hashes)) > 1:
+ break
+ else:
+ continue
+ break
+
+ if args.dag_id is not None and args.dag_id not in seen_dag_ids:
+ raise SystemExit(f"Dag {args.dag_id!r} was not found.")
+
+ unstable_dag_ids = [
+ dag_id for dag_id, dag_hashes in sorted(dag_hashes_by_id.items()) if len(set(dag_hashes)) > 1
+ ]
+ if unstable_dag_ids:
+ print("Dag stability check failed. The following Dags produced different serialized output:")
+ for dag_id in unstable_dag_ids:
+ dag_hashes = dag_hashes_by_id[dag_id]
Review Comment:
`dag_hashes = dag_hashes_by_id[dag_id]` is assigned but never used, which will fail linting (unused local). Remove the assignment.
##########
airflow-core/src/airflow/cli/commands/dag_command.py:
##########
@@ -80,6 +82,96 @@
_RUN_CHUNK_SIZE = 500
+def _normalize_serialized_dag_for_stability_check(serialized_dag: dict[str, Any]) -> dict[str, Any]:
+ normalized = SerializedDagModel._sort_serialized_dag_dict(serialized_dag)
+ normalized["dag"].pop("fileloc", None)
+ normalized["dag"].pop("bundle_name", None)
+ return normalized
+
+
+def _serialize_dag_for_stability_check(dag: DAG) -> tuple[str, dict[str, Any]]:
+ serialized_dag = DagSerialization.to_dict(dag)
+ return SerializedDagModel.hash(serialized_dag), _normalize_serialized_dag_for_stability_check(
+ serialized_dag
+ )
+
+
+def _format_stability_diff(
+ dag_id: str,
+ first_serialized_dag: dict[str, Any],
+ second_serialized_dag: dict[str, Any],
+) -> str:
+ before = json.dumps(first_serialized_dag, indent=2, sort_keys=True).splitlines()
+ after = json.dumps(second_serialized_dag, indent=2, sort_keys=True).splitlines()
+ diff = difflib.unified_diff(
+ before,
+ after,
+ fromfile=f"parse 1: {dag_id} ",
+ tofile=f"parse 2: {dag_id} ",
+ lineterm="",
+ )
+ return "\n".join(diff)
+
+
+def _parse_dags_for_stability_check(dag_folder: str | None) -> DagBag:
+ return DagBag(dag_folder=dag_folder, load_op_links=False)
+
+
+@cli_utils.action_cli
+@providers_configuration_loaded
+def dag_stability_check(args) -> None:
+ dag_hashes_by_id: dict[str, list[str]] = {}
+ serialized_dags_by_id: dict[str, list[dict[str, Any]]] = {}
+ seen_dag_ids: set[str] = set()
+ nParse = (
+ 2 # NOTE: Set parsing number 2, in most of the case twice parse should catch the stability issues.
+ )
+
+ for _iteration in range(1, nParse + 1):
+ dagbag = _parse_dags_for_stability_check(args.dag_folder)
+
+ dags = dagbag.dags
+ if args.dag_id is not None:
+ dags = {args.dag_id: dagbag.dags[args.dag_id]} if args.dag_id in dagbag.dags else {}
+
+ for _dag_id, dag in sorted(dags.items()):
+ dag_hash, serialized_dag = _serialize_dag_for_stability_check(dag)
+ dag_hashes_by_id.setdefault(_dag_id, []).append(dag_hash)
+ serialized_dags_by_id.setdefault(_dag_id, []).append(serialized_dag)
+ seen_dag_ids.add(_dag_id)
+
+ if args.fail_fast:
+ for _dag_id, dag_hashes in dag_hashes_by_id.items():
+ if len(set(dag_hashes)) > 1:
+ break
+ else:
+ continue
+ break
Review Comment:
The new `--fail-fast` flag changes control flow (can stop after detecting the first unstable Dag), but there is no unit test covering this behavior. Please add a test that demonstrates `--fail-fast` exits after the first detected instability (and would fail if the early-exit logic is removed).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add airflow dags stability check CLI command [airflow]
Posted by "bhavaniravi (via GitHub)" <gi...@apache.org>.
bhavaniravi commented on PR #70446:
URL: https://github.com/apache/airflow/pull/70446#issuecomment-5159354204
Is a CLI command really necessary?
I feel this belongs to lint rules. Eg., Ruff [AIR304](https://docs.astral.sh/ruff/rules/airflow3-dag-dynamic-value/) captures this exactly
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add airflow dags stability check CLI command [airflow]
Posted by "r-richmond (via GitHub)" <gi...@apache.org>.
r-richmond commented on PR #70446:
URL: https://github.com/apache/airflow/pull/70446#issuecomment-5956186135
> Does this need a new CLI command? The Dag processor already detects version inflation (`[dag_processor] dag_version_inflation_check_level`), the FAQ describes diffing `serialized_dag` rows, and ruff's AIR304 catches the common causes statically. An `airflow dags stability` command is permanent CLI surface, and `--fail-fast` can only trigger on the last of the two parses, so it has no effect. Is there an issue or a user asking for this? Converting to draft until there is.
I would be a user asking for this, although I did jump on after it was submitted.
See my comment https://github.com/apache/airflow/pull/70446#issuecomment-5159593952 here for my motivation.
Note: We are currently running 3.3.1 and `dag_version_inflation_check_level` did not catch the issue that caused our version number to explode and as mentioned before neither did AIR304. TLDR this command would fit perfectly in our CI/CD and seems to be the most rubout solution yet.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add airflow dags stability check CLI command [airflow]
Posted by "r-richmond (via GitHub)" <gi...@apache.org>.
r-richmond commented on PR #70446:
URL: https://github.com/apache/airflow/pull/70446#issuecomment-5159593952
This CLI command is much more robust. And would be a welcome addition.
As an anecdote we just got burned by a subtle issue that caused DAG version inflation. This was missed by the ruff rule but this CLI command would have caught it.
Personally I plan on using it in our CICD. Ty for the contribution
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add airflow dags stability check CLI command [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #70446:
URL: https://github.com/apache/airflow/pull/70446#discussion_r3650615499
##########
airflow-core/src/airflow/cli/commands/dag_command.py:
##########
@@ -79,6 +81,89 @@
# Chunk size for bulk delete.
_RUN_CHUNK_SIZE = 500
+def _normalize_serialized_dag_for_stability_check(serialized_dag: dict[str, Any]) -> dict[str, Any]:
Review Comment:
PEP 8 / ruff will flag this block because top-level function definitions must be preceded by two blank lines. Add an extra blank line between the module-level constant and the first helper.
##########
airflow-core/src/airflow/cli/commands/dag_command.py:
##########
@@ -79,6 +81,89 @@
# Chunk size for bulk delete.
_RUN_CHUNK_SIZE = 500
+def _normalize_serialized_dag_for_stability_check(serialized_dag: dict[str, Any]) -> dict[str, Any]:
+ normalized = SerializedDagModel._sort_serialized_dag_dict(serialized_dag)
+ normalized["dag"].pop("fileloc", None)
+ normalized["dag"].pop("bundle_name", None)
+ return normalized
+
+def _serialize_dag_for_stability_check(dag: DAG) -> tuple[str, dict[str, Any]]:
+ serialized_dag = DagSerialization.to_dict(dag)
+ return SerializedDagModel.hash(serialized_dag), _normalize_serialized_dag_for_stability_check(serialized_dag)
+
+def _format_stability_diff(
+ dag_id: str,
+ first_serialized_dag: dict[str, Any],
+ second_serialized_dag: dict[str, Any],
+) -> str:
+ before = json.dumps(first_serialized_dag, indent=2, sort_keys=True).splitlines()
+ after = json.dumps(second_serialized_dag, indent=2, sort_keys=True).splitlines()
+ diff = difflib.unified_diff(
+ before,
+ after,
+ fromfile=f"parse 1: {dag_id} ",
+ tofile=f"parse 2: {dag_id} ",
+ lineterm="",
+ )
+ return "\n".join(diff)
+
+def _parse_dags_for_stability_check(dag_folder: str | None) -> DagBag:
+ return DagBag(dag_folder=dag_folder, safe_mode=True, load_op_links=False)
Review Comment:
The stability check currently forces DagBag safe_mode=True, which overrides the configured DAG_DISCOVERY_SAFE_MODE and can silently skip Dag files. This makes the CLI check incomplete in deployments that disable safe_mode (or want it honored).
##########
airflow-core/src/airflow/cli/commands/dag_command.py:
##########
@@ -79,6 +81,89 @@
# Chunk size for bulk delete.
_RUN_CHUNK_SIZE = 500
+def _normalize_serialized_dag_for_stability_check(serialized_dag: dict[str, Any]) -> dict[str, Any]:
+ normalized = SerializedDagModel._sort_serialized_dag_dict(serialized_dag)
+ normalized["dag"].pop("fileloc", None)
+ normalized["dag"].pop("bundle_name", None)
+ return normalized
+
+def _serialize_dag_for_stability_check(dag: DAG) -> tuple[str, dict[str, Any]]:
+ serialized_dag = DagSerialization.to_dict(dag)
+ return SerializedDagModel.hash(serialized_dag), _normalize_serialized_dag_for_stability_check(serialized_dag)
+
+def _format_stability_diff(
+ dag_id: str,
+ first_serialized_dag: dict[str, Any],
+ second_serialized_dag: dict[str, Any],
+) -> str:
+ before = json.dumps(first_serialized_dag, indent=2, sort_keys=True).splitlines()
+ after = json.dumps(second_serialized_dag, indent=2, sort_keys=True).splitlines()
+ diff = difflib.unified_diff(
+ before,
+ after,
+ fromfile=f"parse 1: {dag_id} ",
+ tofile=f"parse 2: {dag_id} ",
+ lineterm="",
+ )
+ return "\n".join(diff)
+
+def _parse_dags_for_stability_check(dag_folder: str | None) -> DagBag:
+ return DagBag(dag_folder=dag_folder, safe_mode=True, load_op_links=False)
+
+@cli_utils.action_cli
+@providers_configuration_loaded
+def dag_stability_check(args) -> None:
+ dag_hashes_by_id: dict[str, list[str]] = {}
+ serialized_dags_by_id: dict[str, list[dict[str, Any]]] = {}
+ seen_dag_ids: set[str] = set()
+ nParse = 2 #NOTE: Set parsing number 2, in most of the case twice parse should catch the stability issues.
+
+ for iteration in range(1, nParse+1):
+ dagbag = _parse_dags_for_stability_check(args.dag_folder)
+
+ dags = dagbag.dags
+ if args.dag_id is not None:
+ dags = {args.dag_id: dagbag.dags[args.dag_id]} if args.dag_id in dagbag.dags else {}
+
+ for dag_id, dag in sorted(dags.items()):
+ dag_hash, serialized_dag = _serialize_dag_for_stability_check(dag)
+ dag_hashes_by_id.setdefault(dag_id, []).append(dag_hash)
+ serialized_dags_by_id.setdefault(dag_id, []).append(serialized_dag)
+ seen_dag_ids.add(dag_id)
+
+ if args.fail_fast:
+ for dag_id, dag_hashes in dag_hashes_by_id.items():
+ if len(set(dag_hashes)) > 1:
+ break
+ else:
+ continue
+ break
+
+ if args.dag_id is not None and args.dag_id not in seen_dag_ids:
+ raise SystemExit(f"Dag {args.dag_id!r} was not found.")
+
+ unstable_dag_ids = [
+ dag_id for dag_id, dag_hashes in sorted(dag_hashes_by_id.items()) if len(set(dag_hashes)) > 1
+ ]
+ if unstable_dag_ids:
+ print("Dag stability check failed. The following Dags produced different serialized output:")
+ for dag_id in unstable_dag_ids:
+ dag_hashes = dag_hashes_by_id[dag_id]
+ print(f"\ndag_id: {dag_id}:")
+ for iteration, dag_hash in enumerate(dag_hashes, start=1):
Review Comment:
The per-Dag header line prints as "dag_id: <id>:" (duplicated label + trailing colon), which is confusing in the failure output and makes grepping harder. Consider printing just "<dag_id>:" (or "dag_id: <id>") instead.
##########
airflow-core/src/airflow/cli/cli_config.py:
##########
@@ -1242,6 +1252,21 @@ class GroupCommand(NamedTuple):
func=lazy_load_command("airflow.cli.commands.dag_command.dag_report"),
args=(ARG_BUNDLE_NAME, ARG_OUTPUT, ARG_VERBOSE),
),
+ ActionCommand(
+ name="stability",
+ help="Check that serialized Dags are stable across repeated parses to prevent dag_version inflection",
+ description=(
Review Comment:
Typo in help text: this should be "dag_version inflation" (the feature this command is intended to help diagnose), not "inflection".
##########
airflow-core/docs/faq.rst:
##########
@@ -417,7 +417,7 @@ configuration option:
Additionally, you can catch these issues earlier in your development workflow by using the
`AIR304 <https://docs.astral.sh/ruff/rules/airflow3-dag-dynamic-value/>`_ ruff rule, which detects
-dynamic values in Dag and Task constructors as part of static linting. See
+dynamic values in Dag and Task constructors as part of static linting or ``airflow dags stability`` CLI command. See
:ref:`best_practices/code_quality_and_linting` for how to set up ruff with Airflow-specific rules.
Review Comment:
This sentence is grammatically ambiguous: it reads as if the ruff rule detects issues "as part of ... CLI command". Consider splitting into two sentences so it's clear the CLI command is an alternative way to catch the same issues.
##########
airflow-core/src/airflow/cli/commands/dag_command.py:
##########
@@ -79,6 +81,89 @@
# Chunk size for bulk delete.
_RUN_CHUNK_SIZE = 500
+def _normalize_serialized_dag_for_stability_check(serialized_dag: dict[str, Any]) -> dict[str, Any]:
+ normalized = SerializedDagModel._sort_serialized_dag_dict(serialized_dag)
+ normalized["dag"].pop("fileloc", None)
+ normalized["dag"].pop("bundle_name", None)
+ return normalized
+
+def _serialize_dag_for_stability_check(dag: DAG) -> tuple[str, dict[str, Any]]:
+ serialized_dag = DagSerialization.to_dict(dag)
+ return SerializedDagModel.hash(serialized_dag), _normalize_serialized_dag_for_stability_check(serialized_dag)
+
+def _format_stability_diff(
+ dag_id: str,
+ first_serialized_dag: dict[str, Any],
+ second_serialized_dag: dict[str, Any],
+) -> str:
+ before = json.dumps(first_serialized_dag, indent=2, sort_keys=True).splitlines()
+ after = json.dumps(second_serialized_dag, indent=2, sort_keys=True).splitlines()
+ diff = difflib.unified_diff(
+ before,
+ after,
+ fromfile=f"parse 1: {dag_id} ",
+ tofile=f"parse 2: {dag_id} ",
+ lineterm="",
+ )
+ return "\n".join(diff)
+
+def _parse_dags_for_stability_check(dag_folder: str | None) -> DagBag:
+ return DagBag(dag_folder=dag_folder, safe_mode=True, load_op_links=False)
+
+@cli_utils.action_cli
+@providers_configuration_loaded
+def dag_stability_check(args) -> None:
+ dag_hashes_by_id: dict[str, list[str]] = {}
+ serialized_dags_by_id: dict[str, list[dict[str, Any]]] = {}
+ seen_dag_ids: set[str] = set()
+ nParse = 2 #NOTE: Set parsing number 2, in most of the case twice parse should catch the stability issues.
+
+ for iteration in range(1, nParse+1):
Review Comment:
Use snake_case for local variables and clean up the inline comment spacing to match project style (and avoid ruff E261/E262).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add airflow dags stability check CLI command [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on PR #70446:
URL: https://github.com/apache/airflow/pull/70446#issuecomment-5942836372
Does this need a new CLI command? The Dag processor already detects version inflation (`[dag_processor] dag_version_inflation_check_level`), the FAQ describes diffing `serialized_dag` rows, and ruff's AIR304 catches the common causes statically. An `airflow dags stability` command is permanent CLI surface, and `--fail-fast` can only trigger on the last of the two parses, so it has no effect. Is there an issue or a user asking for this? Converting to draft until there is.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org