Fix custom stats metrics not being sent from plugins/listeners - #69270
Fix custom stats metrics not being sent from plugins/listeners#69270Aaryan123456679 wants to merge 3 commits into
Conversation
boring-cyborg
Bot
added
area:API
|
potiuk
added
the
ready for maintainer review
|
Friendly ping! This PR is ready for review whenever someone has time. Thank you! |
pierrejeambrun
left a comment
pierrejeambrun
left a comment
There was a problem hiding this comment.
Thanks for the PR.
Having calls to initialize_sdk_stats_backend all over the core code base really feels off. We're trying to split server / client and limit the dependencies there. We probably should use a _shared lib.
pierrejeambrun
removed
the
ready for maintainer review
Airflow keeps two independent Stats singletons: one internal to airflow-core, and the task-sdk copy that airflow.sdk.observability.stats (and the deprecated airflow.stats shim) expose to plugins and listener hooks. Long-running components only initialized the internal copy at startup, so a plugin's DAG-run listener hook running in the scheduler or API server silently got NoStatsLogger regardless of StatsD config, even though native Airflow metrics worked fine. closes: apache#69172
Plugins and listeners typically reach Stats through the task-sdk import path, a separate singleton from the one each component initializes for its own internal use, so plugin code kept using an unconfigured NoStatsLogger even after the component's own Stats was set up. Propagate the configuration to any sibling copy of the stats module already loaded under a different distribution path instead of having core import from airflow.sdk directly to reach it. Metrics initialization failures no longer prevent the API server, scheduler, Dag processor, triggerer, or executor from starting.
0de55d0 to
396cb29
Compare
|
Hi @pierrejeambrun , I have done the necessary changes, kindly let me know in case any other change is required. Thanks. |
pierrejeambrun
left a comment
pierrejeambrun
left a comment
There was a problem hiding this comment.
I would love to avoid _propagate_to_sibling_modules. Can't we instead let both Stats module (core and sdk loaded) self configure on load lazily?
shared has access to the configuration values, either read from config or move get_stats_factory to shared, so the module can self configure.
This will avoid the use case mentioned bellow that could still break even with this fix.
| def _propagate_to_sibling_modules( | ||
| *, | ||
| factory: Callable[[], StatsLogger | NoStatsLogger], | ||
| export_legacy_names: bool, | ||
| ) -> None: | ||
| """ | ||
| Apply the same configuration to other loaded copies of this module. | ||
|
|
||
| This source file is symlinked into multiple distributions (e.g. ``airflow-core`` and | ||
| ``task-sdk``), each importing it under a different module name (``airflow._shared...`` vs | ||
| ``airflow.sdk._shared...``). Python treats each as a distinct module object with its own | ||
| module-level globals, so a process that has both loaded (e.g. the scheduler, which also runs |
There was a problem hiding this comment.
I think this can still be a problem if there is a local import, or somehow the plugin Stat is imported after core is initialized.
There was a problem hiding this comment.
And to be honest this piece of code seems quite hacky.
4 participants
Airflow keeps two independent
Statssingletons because of the shared/symlinked library architecture: one internal copy that airflow-core initializes at startup (airflow._shared.observability.metrics.stats), and the task-sdk copy exposed by the officially recommendedairflow.sdk.observability.statsimport (and the deprecatedairflow.statsshim, which re-exports it). Long-running components — scheduler, DAG file processor, API server, executors, triggerer — only initialized the internal copy at startup. The task-sdk copy was only ever initialized inside the task-execution subprocess.Plugin/listener hooks such as
on_dag_run_success/on_dag_run_failedrun in the scheduler or API server process, never in the task subprocess, so a plugin callingStats.gauge(...)via the recommended import path silently fell back toNoStatsLogger— no error, no custom metrics — regardless of StatsD/Datadog/OTel configuration, even though Airflow's own native metrics worked fine.This adds
initialize_sdk_stats_backend()toairflow.observability.metrics.stats_utils, which configures the task-sdkStatssingleton with the same factory/config each process already uses for its own internal singleton, and calls it alongside the existingstats.initialize(...)call in the scheduler, DAG processor, API server, base executor, and triggerer.closes: #69172
Test plan
TestInitializeSdkStatsBackendinairflow-core/tests/unit/observability/metrics/test_stats.pyverifying the sdk singleton is configured with this process's backend and the same factory/legacy-names flag.test_app.py,test_manager.py,test_base_executor.py,test_scheduler_job.py, andtest_triggerer_job.pyto assert the new call happens alongside the existing one.Stats.gauge(...)silently resolved toNoStatsLogger; after callinginitialize_sdk_stats_backend(), it resolved toSafeStatsdLogger.Was generative AI tooling used to co-author this PR?
Generated-by: Claude Code (Sonnet 5) following the guidelines