From 12c7e4ad3c36a2f909a0fa0fa3a29dc2ca975d84 Mon Sep 17 00:00:00 2001 From: PoAn Yang Date: Thu, 23 Jul 2026 09:57:31 +0900 Subject: [PATCH] Detect stale metrics registry entries in the metrics sync prek hook Signed-off-by: PoAn Yang --- .pre-commit-config.yaml | 3 +- .../check_metrics_synced_with_the_registry.py | 162 +++++++++++++++--- ..._check_metrics_synced_with_the_registry.py | 129 ++++++++++++++ .../metrics/metrics_template.yaml | 20 --- .../observability/metrics/validators.py | 1 - .../observability/metrics/test_otel_logger.py | 4 +- 6 files changed, 269 insertions(+), 50 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 736f3fff51c4f..adcc9c11c0f8c 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -950,9 +950,10 @@ repos: name: Check that metrics in the codebase are in sync with the metrics registry YAML file. entry: ./scripts/ci/prek/check_metrics_synced_with_the_registry.py language: python - files: \.py$ + files: \.py$|^shared/observability/src/airflow_shared/observability/metrics/metrics_template\.yaml$ exclude: ^(tests/|.*/tests/|airflow-core/src/airflow/observability/stats\.py$|task-sdk/src/airflow/sdk/observability/stats\.py$) pass_filenames: true + require_serial: true additional_dependencies: ["PyYAML>=6.0", "rich>=13.6.0"] - id: check-boring-cyborg-configuration name: Checks for Boring Cyborg configuration consistency diff --git a/scripts/ci/prek/check_metrics_synced_with_the_registry.py b/scripts/ci/prek/check_metrics_synced_with_the_registry.py index 3d42939b6ce57..140e6ab9e91ed 100644 --- a/scripts/ci/prek/check_metrics_synced_with_the_registry.py +++ b/scripts/ci/prek/check_metrics_synced_with_the_registry.py @@ -32,6 +32,7 @@ import argparse import ast import re +import subprocess import sys from dataclasses import dataclass from pathlib import Path @@ -67,6 +68,16 @@ AIRFLOW_ROOT_PATH / "shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml" ) +EXCLUDED_TESTS_PATTERN = re.compile(r"(^|/)tests/") + +# Registry metrics emitted through a variable that the AST scan cannot resolve back to a +# string literal, e.g. names built by BaseExecutor._get_metric_name. +INDIRECTLY_EMITTED_METRICS = { + "executor.open_slots", + "executor.queued_tasks", + "executor.running_tasks", +} + def load_metrics_registry_yaml() -> dict[str, dict[str, Any]]: """Load the metrics registry YAML and return a dict keyed by metric name.""" @@ -92,6 +103,12 @@ def normalize_metric_name(registry_metric_name: str) -> str: _PREFIX_MATCHED = "__prefix_matched__" +def find_prefix_matched_registry_entries(metric_name: str, metrics_registry: dict[str, dict]) -> list[str]: + """Return the registry entry names whose name matches the static prefix of a dynamic metric name.""" + base = metric_name.split("{")[0].rstrip(".") + return [name for name in metrics_registry if name == base or name.startswith(base + ".")] + + def find_registry_match(metric_name: str, metrics_registry: dict[str, dict]) -> str | None: """Return the registry entry name that best matches the given metric name, or None.""" if metric_name in metrics_registry: @@ -109,16 +126,13 @@ def find_registry_match(metric_name: str, metrics_registry: dict[str, dict]) -> return registry_metric_name # Dynamic metric name. - if "{" in metric_name: - base = metric_name.split("{")[0].rstrip(".") - for registry_metric_name in metrics_registry: - if registry_metric_name == base or registry_metric_name.startswith(base + "."): - # Metric prefix matches the prefix of a dynamic registry entry. - # If the static part before the first variable, matches an exact registry entry name, - # or a dotted-prefix of one, then τηε name is considered covered and - # _PREFIX_MATCHED is returned. The type check must be skipped because - # the resulting metric name with all variables expanded, cannot be determined. - return _PREFIX_MATCHED + if "{" in metric_name and find_prefix_matched_registry_entries(metric_name, metrics_registry): + # Metric prefix matches the prefix of a dynamic registry entry. + # If the static part before the first variable, matches an exact registry entry name, + # or a dotted-prefix of one, then the name is considered covered and + # _PREFIX_MATCHED is returned. The type check must be skipped because + # the resulting metric name with all variables expanded, cannot be determined. + return _PREFIX_MATCHED # All checks for matching failed. return None @@ -177,6 +191,22 @@ def extract_metric_name_from_ast_node(name_node: ast.expr) -> str | None: return None +def extract_metric_names_from_ast_node(name_node: ast.expr) -> list[str]: + """Resolve a metric name AST node to all names it can produce. + + A conditional expression like ``"a.success" if ok else "a.failed"`` yields both names. + Returns an empty list when the expression is too dynamic to resolve. + """ + if isinstance(name_node, ast.IfExp): + return [ + name + for branch in (name_node.body, name_node.orelse) + for name in extract_metric_names_from_ast_node(branch) + ] + metric_name = extract_metric_name_from_ast_node(name_node) + return [metric_name] if metric_name is not None else [] + + @dataclass class MetricCall: file_path: str @@ -232,8 +262,16 @@ def scan_file_for_metrics(file_path: Path) -> list[MetricCall]: """Return all Stats metric calls found in the provided file_path.""" try: source = file_path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + return [] + + # Cheap text pre-filter to avoid parsing the vast majority of files that emit no metrics. + if "Stats." not in source and "stats." not in source: + return [] + + try: tree = ast.parse(source, filename=str(file_path)) - except (OSError, UnicodeDecodeError, SyntaxError): + except SyntaxError: return [] metrics_found: list[MetricCall] = [] @@ -251,22 +289,22 @@ def scan_file_for_metrics(file_path: Path) -> list[MetricCall]: if name_node is None: continue - metric_name = extract_metric_name_from_ast_node(name_node) - if metric_name is None: + if not (metric_names := extract_metric_names_from_ast_node(name_node)): # Metric name is unresolvable. Probably has too many variables. continue stats_obj = get_stats_obj_name(node.func.value) - metrics_found.append( - MetricCall( - file_path=str(file_path), - line_num=node.lineno, - metric_name=metric_name, - method=method, - stats_obj=stats_obj or "", - is_dynamic="{" in metric_name, + for metric_name in metric_names: + metrics_found.append( + MetricCall( + file_path=str(file_path), + line_num=node.lineno, + metric_name=metric_name, + method=method, + stats_obj=stats_obj or "", + is_dynamic="{" in metric_name, + ) ) - ) return metrics_found @@ -318,6 +356,43 @@ def scan_file_for_direct_stats_imports(file_path: Path) -> list[DirectStatsImpor return violations +def list_repository_python_files() -> list[Path]: + """Return all git-tracked, non-test Python files in the repository.""" + result = subprocess.run( + ["git", "ls-files", "*.py"], + cwd=AIRFLOW_ROOT_PATH, + capture_output=True, + text=True, + check=True, + ) + return [ + AIRFLOW_ROOT_PATH / line + for line in result.stdout.splitlines() + if not EXCLUDED_TESTS_PATTERN.search(line) + ] + + +def compute_unused_registry_entries( + code_metric_names: set[str], metrics_registry: dict[str, dict] +) -> list[str]: + """Return the registry entry names that no code metric name matches.""" + used_entries = set(INDIRECTLY_EMITTED_METRICS) + for metric_name in code_metric_names: + registry_metric_name = find_registry_match(metric_name, metrics_registry) + if registry_metric_name is None: + continue + if registry_metric_name is _PREFIX_MATCHED: + used_entries.update(find_prefix_matched_registry_entries(metric_name, metrics_registry)) + else: + used_entries.add(registry_metric_name) + return sorted(set(metrics_registry) - used_entries) + + +def find_stale_indirectly_emitted_metrics(metrics_registry: dict[str, dict]) -> list[str]: + """Return the ``INDIRECTLY_EMITTED_METRICS`` names that no longer exist in the registry.""" + return sorted(INDIRECTLY_EMITTED_METRICS - set(metrics_registry)) + + def main() -> None: parser = argparse.ArgumentParser( description="Check that metrics in the codebase are in sync with the metrics registry YAML file." @@ -377,12 +452,24 @@ def main() -> None: if mismatched: metrics_with_type_mismatch[name] = mismatched - # There is no point in checking whether the metrics exist in the YAML but not in the code, - # because the script is comparing the entire YAML against certain files at a time. - # For that to work, the script would have to run against all project files EVERY TIME. + # Violation 3: the metric exists in the registry but nowhere in the code. The hook only + # receives the changed files, so this check scans all git-tracked Python files itself. + all_code_metric_names = { + call.metric_name + for repo_file_path in list_repository_python_files() + for call in scan_file_for_metrics(repo_file_path) + } + unused_registry_entries = compute_unused_registry_entries(all_code_metric_names, metrics_registry) + + # Violation 4: this script exempts a metric from the check above, but the registry no longer has it. + stale_indirectly_emitted_metrics = find_stale_indirectly_emitted_metrics(metrics_registry) total_violations = ( - len(metrics_not_in_registry) + len(metrics_with_type_mismatch) + len(direct_stats_imports) + len(metrics_not_in_registry) + + len(metrics_with_type_mismatch) + + len(direct_stats_imports) + + len(stale_indirectly_emitted_metrics) + + len(unused_registry_entries) ) if total_violations: @@ -436,6 +523,29 @@ def main() -> None: ) console.print() + if unused_registry_entries: + console.print( + f" [red]-> {len(unused_registry_entries)} metric(s) found in the registry YAML but not emitted anywhere in the code:[/red]" + ) + for metric_name in unused_registry_entries: + console.print(f" [green]{metric_name}[/green]") + console.print( + " [yellow]Remove them from the registry, or if they are emitted through a variable " + "the scan cannot resolve, add them to INDIRECTLY_EMITTED_METRICS in this script.[/yellow]" + ) + console.print() + + if stale_indirectly_emitted_metrics: + console.print( + f" [red]-> {len(stale_indirectly_emitted_metrics)} metric(s) in INDIRECTLY_EMITTED_METRICS that no longer exist in the registry YAML:[/red]" + ) + for metric_name in stale_indirectly_emitted_metrics: + console.print(f" [green]{metric_name}[/green]") + console.print( + " [yellow]Update INDIRECTLY_EMITTED_METRICS in this script to match the registry.[/yellow]" + ) + console.print() + sys.exit(1) diff --git a/scripts/tests/ci/prek/test_check_metrics_synced_with_the_registry.py b/scripts/tests/ci/prek/test_check_metrics_synced_with_the_registry.py index fdd91fb4dcef5..e9b5e2ebe2979 100644 --- a/scripts/tests/ci/prek/test_check_metrics_synced_with_the_registry.py +++ b/scripts/tests/ci/prek/test_check_metrics_synced_with_the_registry.py @@ -19,14 +19,20 @@ import ast import textwrap from pathlib import Path +from unittest import mock import pytest +from ci.prek import check_metrics_synced_with_the_registry from ci.prek.check_metrics_synced_with_the_registry import ( _PREFIX_MATCHED, _except_handler_catches_expected_error, _is_stats_module_path, + compute_unused_registry_entries, extract_metric_name_from_ast_node, + extract_metric_names_from_ast_node, + find_prefix_matched_registry_entries, find_registry_match, + find_stale_indirectly_emitted_metrics, get_stats_obj_name, normalize_metric_name, scan_file_for_direct_stats_imports, @@ -183,6 +189,121 @@ def test_extract_metric_name_from_ast_node(code: str, expected_result): assert extract_metric_name_from_ast_node(node) == expected_result +@pytest.mark.parametrize( + "code, expected_result", + [ + pytest.param('"scheduler_heartbeat"', ["scheduler_heartbeat"], id="static_string_single_name"), + pytest.param( + '"connection_test.success" if success else "connection_test.failed"', + ["connection_test.success", "connection_test.failed"], + id="conditional_expression_both_branches", + ), + pytest.param( + '"connection_test.success" if success else get_name()', + ["connection_test.success"], + id="conditional_expression_unresolvable_branch_dropped", + ), + pytest.param( + '"a" if x else ("b" if y else "c")', + ["a", "b", "c"], + id="nested_conditional_expression", + ), + pytest.param("some_variable", [], id="unresolvable_name_returns_empty_list"), + ], +) +def test_extract_metric_names_from_ast_node(code: str, expected_result): + node = ast.parse(code, mode="eval").body + assert extract_metric_names_from_ast_node(node) == expected_result + + +@pytest.mark.parametrize( + "metric_name, expected_result", + [ + pytest.param( + "ti.{state}", + ["ti.scheduled", "ti.queued", "ti.start.{dag_id}.{task_id}"], + id="base_prefix_matches_multiple_entries", + ), + pytest.param("dagrun.duration.{state}", ["dagrun.duration.success"], id="dotted_base_prefix"), + pytest.param("non.existent.{var}", [], id="no_prefix_match_returns_empty_list"), + ], +) +def test_find_prefix_matched_registry_entries(metric_name, expected_result): + assert find_prefix_matched_registry_entries(metric_name, METRICS_REGISTRY) == expected_result + + +# 'executor.open_slots' is in INDIRECTLY_EMITTED_METRICS, so it is never reported as unused. +@pytest.mark.parametrize( + "code_metric_names, expected_unused", + [ + pytest.param( + set(), + [ + "dagrun.duration.success", + "pool.open_slots", + "scheduler.heartbeat", + "task.duration", + "ti.queued", + "ti.scheduled", + "ti.start.{dag_id}.{task_id}", + ], + id="no_code_metrics_reports_all_but_indirectly_emitted", + ), + pytest.param( + {"scheduler.heartbeat", "dag.{x}.{y}.duration", "unknown.metric"}, + [ + "dagrun.duration.success", + "pool.open_slots", + "ti.queued", + "ti.scheduled", + "ti.start.{dag_id}.{task_id}", + ], + id="exact_and_legacy_matches_mark_entries_used", + ), + pytest.param( + {"ti.{state}"}, + ["dagrun.duration.success", "pool.open_slots", "scheduler.heartbeat", "task.duration"], + id="prefix_match_marks_all_prefix_entries_used", + ), + pytest.param( + {"pool.open_slots.{my_pool}"}, + [ + "dagrun.duration.success", + "scheduler.heartbeat", + "task.duration", + "ti.queued", + "ti.scheduled", + "ti.start.{dag_id}.{task_id}", + ], + id="legacy_name_structure_match_marks_entry_used", + ), + ], +) +def test_compute_unused_registry_entries(code_metric_names, expected_unused): + assert compute_unused_registry_entries(code_metric_names, METRICS_REGISTRY) == expected_unused + + +@pytest.mark.parametrize( + "indirectly_emitted_metrics, expected_stale", + [ + pytest.param({"executor.open_slots"}, [], id="entry_present_in_registry"), + pytest.param(set(), [], id="empty_allowlist"), + pytest.param( + {"executor.open_slots", "executor.renamed_away", "executor.deleted"}, + ["executor.deleted", "executor.renamed_away"], + id="renamed_and_deleted_entries_reported", + ), + ], +) +def test_find_stale_indirectly_emitted_metrics(indirectly_emitted_metrics, expected_stale): + with mock.patch.object( + check_metrics_synced_with_the_registry, + "INDIRECTLY_EMITTED_METRICS", + indirectly_emitted_metrics, + ): + assert find_stale_indirectly_emitted_metrics(METRICS_REGISTRY) == expected_stale + + @pytest.fixture def code_to_py_file(tmp_path): """Write python source code to a tmp file and return its path.""" @@ -240,6 +361,14 @@ def _write(code: str) -> Path: [{"stats_obj": "stats"}], id="self_stats_attribute", ), + pytest.param( + 'stats.incr("connection_test.success" if success else "connection_test.failed")', + [ + {"metric_name": "connection_test.success", "method": "incr"}, + {"metric_name": "connection_test.failed", "method": "incr"}, + ], + id="conditional_expression_yields_call_per_branch", + ), pytest.param('metrics.incr("triggerer_heartbeat")', [], id="unknown_stats_object_ignored"), pytest.param("Stats.incr(get_metric_name())", [], id="unresolvable_metric_name_skipped"), pytest.param("def foo(:\n pass\n", [], id="syntax_error_returns_empty"), diff --git a/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml b/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml index c0d43641559d6..a561db2d7aef6 100644 --- a/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml +++ b/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml @@ -39,14 +39,6 @@ metrics: legacy_name: "-" name_variables: ["job_name"] - - name: "local_task_job.task_exit" - description: "Number of ``LocalTaskJob`` terminations with a ``{return_code}`` while - running a task ``{task_id}`` of a Dag ``{dag_id}``. - Metric with job_id, dag_id, task_id and return_code tagging." - type: "counter" - legacy_name: "local_task_job.task_exit.{job_id}.{dag_id}.{task_id}.{return_code}" - name_variables: ["job_id", "dag_id", "task_id", "return_code"] - - name: "operator_failures" description: "Operator ``{operator_name}`` failures." type: "counter" @@ -162,12 +154,6 @@ metrics: legacy_name: "-" name_variables: [] - - name: "dag_file_processor_timeouts" - description: "(DEPRECATED) same behavior as ``dag_processing.processor_timeouts``" - type: "counter" - legacy_name: "-" - name_variables: [] - - name: "scheduler.tasks.killed_externally" description: "Number of tasks killed externally. Metric with dag_id and task_id tagging." type: "counter" @@ -433,12 +419,6 @@ metrics: legacy_name: "dag_processing.last_run.seconds_ago.{file_name}" name_variables: ["file_name"] - - name: "dag_processing.last_num_of_db_queries.{dag_file}" - description: "Number of queries to Airflow database during parsing per ``{dag_file}``" - type: "gauge" - legacy_name: "-" - name_variables: ["dag_file"] - - name: "scheduler.tasks.starving" description: "Number of tasks that cannot be scheduled because of no open slot in pool" type: "gauge" diff --git a/shared/observability/src/airflow_shared/observability/metrics/validators.py b/shared/observability/src/airflow_shared/observability/metrics/validators.py index 3d9d3d3d3e5ab..cab81c7d98be1 100644 --- a/shared/observability/src/airflow_shared/observability/metrics/validators.py +++ b/shared/observability/src/airflow_shared/observability/metrics/validators.py @@ -56,7 +56,6 @@ class MetricNameLengthExemptionWarning(Warning): r"^(?P.*)_start$", r"^(?P.*)_end$", r"^(?P.*)_heartbeat_failure$", - r"^local_task_job.task_exit\.(?P.*)\.(?P.*)\.(?P.*)\.(?P.*)$", r"^operator_failures_(?P.*)$", r"^operator_successes_(?P.*)$", r"^ti.start.(?P.*)\.(?P.*)$", diff --git a/shared/observability/tests/observability/metrics/test_otel_logger.py b/shared/observability/tests/observability/metrics/test_otel_logger.py index 2b5c71193c38c..f1e90e5300311 100644 --- a/shared/observability/tests/observability/metrics/test_otel_logger.py +++ b/shared/observability/tests/observability/metrics/test_otel_logger.py @@ -77,9 +77,9 @@ def test_is_up_down_counter_negative(self): assert not _is_up_down_counter("this_is_not_a_udc") def test_exemption_list_has_not_grown(self): - assert len(BACK_COMPAT_METRIC_NAMES) <= 26, ( + assert len(BACK_COMPAT_METRIC_NAMES) <= 25, ( "This test exists solely to ensure that nobody is adding names to the exemption list. " - "There are 26 names which are potentially too long for OTel and that number should " + "There are 25 names which are potentially too long for OTel and that number should " "only ever go down as these names are deprecated. If this test is failing, please " "adjust your new stat's name; do not add as exemption without a very good reason." )