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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
