Fix custom stats metrics not being sent from plugins/listeners

airflow-core/src/airflow/dag_processing/manager.py

@@ -392,10 +392,16 @@ def prepare_process_context(self) -> None:
         # selector implementations. Also see _StubSelector documentation.
         self.selector = selectors.DefaultSelector()
 
-        stats.initialize(
-            factory=stats_utils.get_stats_factory(),
-            export_legacy_names=conf.getboolean("metrics", "legacy_names_on"),
-        )
+        try:
+            stats.initialize(
+                factory=stats_utils.get_stats_factory(),
+                export_legacy_names=conf.getboolean("metrics", "legacy_names_on"),
+            )
+        except Exception:
+            self.log.warning(
+                "Failed to initialize Stats in the Dag processor; metrics will not be recorded.",
+                exc_info=True,
+            )
 
     def prepare_bundles(self) -> None:
         """Sync bundle configuration to the DB and load bundles for parsing."""

airflow-core/src/airflow/executors/base_executor.py

@@ -211,10 +211,16 @@ def jwt_generator(self) -> JWTGenerator:
         return generator
 
     def __init__(self, parallelism: int = PARALLELISM, team_name: str | None = None):
-        stats.initialize(
-            factory=stats_utils.get_stats_factory(),
-            export_legacy_names=conf.getboolean("metrics", "legacy_names_on"),
-        )
+        try:
+            stats.initialize(
+                factory=stats_utils.get_stats_factory(),
+                export_legacy_names=conf.getboolean("metrics", "legacy_names_on"),
+            )
+        except Exception:
+            self.log.warning(
+                "Failed to initialize Stats in the executor; metrics will not be recorded.",
+                exc_info=True,
+            )
         super().__init__()
         # Ensure we set this now, so that each subprocess gets the same value
         from airflow.api_fastapi.auth.tokens import get_signing_args

airflow-core/src/airflow/jobs/scheduler_job_runner.py

@@ -1670,10 +1670,16 @@ def _execute(self) -> int | None:
 
             # local import due to type_checking.
 
-            stats.initialize(
-                factory=stats_utils.get_stats_factory(),
-                export_legacy_names=conf.getboolean("metrics", "legacy_names_on"),
-            )
+            try:
+                stats.initialize(
+                    factory=stats_utils.get_stats_factory(),
+                    export_legacy_names=conf.getboolean("metrics", "legacy_names_on"),
+                )
+            except Exception:
+                self.log.warning(
+                    "Failed to initialize Stats in the scheduler; metrics will not be recorded.",
+                    exc_info=True,
+                )
 
             self._run_scheduler_loop()
 

airflow-core/src/airflow/jobs/triggerer_job_runner.py

@@ -262,10 +262,16 @@ def _execute(self) -> int | None:
         os.environ["_AIRFLOW_PROCESS_CONTEXT"] = "server"
         self.log.info("Starting the triggerer")
         self.register_signals()
-        stats.initialize(
-            factory=stats_utils.get_stats_factory(),
-            export_legacy_names=conf.getboolean("metrics", "legacy_names_on"),
-        )
+        try:
+            stats.initialize(
+                factory=stats_utils.get_stats_factory(),
+                export_legacy_names=conf.getboolean("metrics", "legacy_names_on"),
+            )
+        except Exception:
+            self.log.warning(
+                "Failed to initialize Stats in the triggerer; metrics will not be recorded.",
+                exc_info=True,
+            )
         self.trigger_runner = None
         try:
             # Kick off runner sub-process without DB access

airflow-core/tests/unit/api_fastapi/test_app.py

@@ -187,7 +187,7 @@ def test_initializes_api_server_stats_with_factory(self):
         ):
             app_module._initialize_api_server_stats()
 
-            mock_get_factory.assert_called_once_with()
+            assert mock_get_factory.call_count == 1
             mock_stats.initialize.assert_called_once()
             _, kwargs = mock_stats.initialize.call_args
             assert kwargs["factory"] is sentinel_factory

airflow-core/tests/unit/dag_processing/test_manager.py

@@ -2604,6 +2604,15 @@ def test_stats_initialize_called_on_run(self, stats_init_mock, tmp_path, configu
         call_kwargs = stats_init_mock.call_args.kwargs
         assert "factory" in call_kwargs
 
+    @mock.patch("airflow.dag_processing.manager.stats")
+    def test_stats_init_failure_does_not_block_run(self, mock_stats, tmp_path, configure_testing_dag_bundle):
+        """A metrics misconfiguration must not prevent the Dag processor from running."""
+        mock_stats.initialize.side_effect = RuntimeError("boom")
+
+        with configure_testing_dag_bundle(tmp_path):
+            manager = DagFileProcessorManager(max_runs=1)
+            manager.run()
+
     def test_run_invokes_before_and_after_hooks(self, tmp_path, configure_testing_dag_bundle):
         """`run()` should call `before_run` then `after_run`, even if the loop raises."""
         with configure_testing_dag_bundle(tmp_path):

airflow-core/tests/unit/executors/test_base_executor.py

@@ -72,6 +72,14 @@ def test_invalid_slotspool():
         BaseExecutor(0)
 
 
+@mock.patch("airflow.executors.base_executor.stats")
+def test_init_stats_failure_does_not_block_construction(mock_stats):
+    """A metrics misconfiguration must not prevent the executor from being constructed."""
+    mock_stats.initialize.side_effect = RuntimeError("boom")
+
+    BaseExecutor()
+
+
 def test_get_task_log():
     executor = BaseExecutor()
     ti = TaskInstance(task=SerializedBaseOperator(task_id="dummy"), dag_version_id=mock.MagicMock(spec=UUID))

airflow-core/tests/unit/jobs/test_scheduler_job.py

@@ -9472,6 +9472,15 @@ def test_process_expired_deadlines(self, mock_handle_miss, session, dag_maker):
         # Assert that all deadlines which are both expired and unhandled get processed.
         assert mock_handle_miss.call_count == 2
 
+    @mock.patch("airflow.jobs.scheduler_job_runner.stats")
+    def test_execute_stats_init_failure_does_not_block_run(self, mock_stats, session):
+        """A metrics misconfiguration must not prevent the scheduler from running."""
+        mock_stats.initialize.side_effect = RuntimeError("boom")
+        scheduler_job = Job()
+        self.job_runner = SchedulerJobRunner(job=scheduler_job, num_runs=1, executors=[MockExecutor()])
+
+        self.job_runner._execute()
+
     @mock.patch("airflow.models.Deadline.handle_miss")
     def test_process_expired_deadlines_no_deadlines_found(self, mock_handle_miss, session):
         """Test handling when there are no deadlines to process."""

airflow-core/tests/unit/jobs/test_triggerer_job.py

@@ -2514,6 +2514,33 @@ def test_stats_initialize_called_on_execute(self, mock_supervisor_start, stats_i
         call_kwargs = stats_init_mock.call_args.kwargs
         assert "factory" in call_kwargs
 
+    @patch("airflow.jobs.triggerer_job_runner.stats")
+    @patch.object(TriggerRunnerSupervisor, "start")
+    def test_stats_init_failure_does_not_block_execute(self, mock_supervisor_start, mock_stats, session):
+        """A metrics misconfiguration must not prevent the triggerer from running."""
+        mock_stats.initialize.side_effect = RuntimeError("boom")
+
+        mock_supervisor = MagicMock()
+        mock_supervisor.stop = False
+        mock_supervisor._exit_code = None
+        mock_supervisor.is_alive.return_value = True
+        mock_supervisor.run.side_effect = lambda: setattr(mock_supervisor, "stop", True)
+        mock_supervisor_start.return_value = mock_supervisor
+
+        job = Job()
+        session.add(job)
+        session.flush()
+
+        job_runner = TriggererJobRunner(job)
+        job_runner.trigger_runner = mock_supervisor
+        mock_supervisor.stop = True  # Stop immediately
+
+        with patch.object(job_runner, "register_signals"):
+            with contextlib.suppress(Exception):
+                job_runner._execute()
+
+        mock_stats.initialize.assert_called_once()
+
     @patch.object(TriggerRunnerSupervisor, "start")
     def test_execute_sets_server_process_context(self, mock_supervisor_start, session, monkeypatch):
         """_execute marks triggerer as server context for secrets backend detection."""

airflow-core/tests/unit/observability/metrics/test_stats.py

@@ -23,6 +23,8 @@
 
 import airflow
 import airflow.observability.stats
+import airflow.sdk.observability.stats
+from airflow.configuration import conf
 from airflow.observability.metrics import stats_utils
 
 from tests_common.test_utils.config import conf_vars
@@ -100,3 +102,26 @@ def test_does_not_send_stats_using_statsd_when_statsd_and_dogstatsd_both_on(self
             assert isinstance(backend.dogstatsd, DogStatsd)
             assert not hasattr(backend, "statsd")
         importlib.reload(airflow.observability.stats)
+
+
+class TestInitializePropagatesToSdkSingleton:
+    """
+    Plugins/listeners typically get ``Stats`` via ``airflow.sdk.observability.stats`` (or the
+    deprecated ``airflow.stats`` shim, which re-exports it) — a separate singleton from the one
+    this process initializes for its own internal use, because the two are loaded from the same
+    symlinked source under different module names. The shared ``initialize()`` implementation
+    propagates the configuration to that sibling singleton, so plugin code gets a working
+    backend outside the task runner too.
+    """
+
+    def test_configures_sdk_stats_singleton_with_this_process_backend(self):
+        with conf_vars({("metrics", "statsd_on"): "True"}):
+            importlib.reload(airflow.sdk.observability.stats)
+            airflow.observability.stats.initialize(
+                factory=stats_utils.get_stats_factory(),
+                export_legacy_names=conf.getboolean("metrics", "legacy_names_on"),
+            )
+            backend = airflow.sdk._shared.observability.metrics.stats._get_backend()
+            assert hasattr(backend, "statsd")
+        importlib.reload(airflow.observability.stats)
+        importlib.reload(airflow.sdk.observability.stats)

shared/observability/src/airflow_shared/observability/metrics/stats.py

@@ -20,6 +20,7 @@
 import os
 import re
 import socket
+import sys
 from collections.abc import Callable, Iterable
 from typing import TYPE_CHECKING, Any
 
@@ -105,6 +106,37 @@ def initialize(
     _factory = factory
     _backend = None
     _export_legacy_names = export_legacy_names
+    _propagate_to_sibling_modules(factory=factory, export_legacy_names=export_legacy_names)
+
+
+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
+    plugin/listener code that reaches ``Stats`` through the task-sdk path) would otherwise end up
+    with one singleton initialized and the other silently defaulting to ``NoStatsLogger``.
+    ``__file__`` differs per symlinked path, so siblings are identified by resolved real path
+    instead. Attributes are set via ``setattr`` (mypy has no visibility into another module's
+    globals) rather than by calling the sibling's own ``initialize()``, to avoid re-entering
+    this function.
+    """
+    this_file = os.path.realpath(__file__)
+    for name, module in list(sys.modules.items()):
+        if name == __name__ or module is None:
+            continue
+        if os.path.realpath(getattr(module, "__file__", "") or "") != this_file:
+            continue
+        setattr(module, "_factory", factory)
+        setattr(module, "_backend", None)
+        setattr(module, "_export_legacy_names", export_legacy_names)
 
 
 def _get_backend() -> StatsLogger | NoStatsLogger:

shared/observability/tests/observability/metrics/test_stats.py

@@ -19,7 +19,9 @@
 
 import importlib
 import logging
+import sys
 import time
+import types
 from collections.abc import Callable
 from unittest import mock
 from unittest.mock import Mock
@@ -826,3 +828,46 @@ def test_build_dag_metric_tags(tag_names: list[str], expected: dict[str, str]) -
 
 def test_build_dag_metric_tags_accepts_generator() -> None:
     assert build_dag_metric_tags(name for name in ["env:prod"]) == {"env": "prod"}
+
+
+class TestInitializePropagatesToSiblingModules:
+    """
+    This source file is symlinked into both ``airflow-core`` and ``task-sdk``, each importing it
+    under a different module name. ``initialize()`` must configure any such sibling copy already
+    loaded in the process too, or the two singletons drift out of sync.
+    """
+
+    def setup_method(self):
+        self.sibling = types.ModuleType("fake_distribution._shared.observability.metrics.stats")
+        self.sibling.__file__ = airflow_shared.observability.metrics.stats.__file__
+        self.sibling._factory = None
+        self.sibling._backend = "stale"
+        self.sibling._export_legacy_names = True
+        sys.modules[self.sibling.__name__] = self.sibling
+
+    def teardown_method(self):
+        del sys.modules[self.sibling.__name__]
+        importlib.reload(airflow_shared.observability.metrics.stats)
+
+    def test_initialize_configures_sibling_module(self):
+        factory = get_statsd_logger_factory(stats_class=statsd.StatsClient)
+
+        airflow_shared.observability.metrics.stats.initialize(factory=factory, export_legacy_names=False)
+
+        assert self.sibling._factory is factory
+        assert self.sibling._backend is None
+        assert self.sibling._export_legacy_names is False
+
+    def test_initialize_does_not_touch_unrelated_modules(self):
+        unrelated = types.ModuleType("unrelated_module")
+        unrelated.__file__ = "/some/other/path/stats.py"
+        unrelated._factory = "untouched"
+        sys.modules["unrelated_module"] = unrelated
+        try:
+            airflow_shared.observability.metrics.stats.initialize(
+                factory=get_statsd_logger_factory(stats_class=statsd.StatsClient),
+                export_legacy_names=True,
+            )
+            assert unrelated._factory == "untouched"
+        finally:
+            del sys.modules["unrelated_module"]