diff --git a/airflow-core/src/airflow/dag_processing/manager.py b/airflow-core/src/airflow/dag_processing/manager.py index f523ca6c3e1e2..e53624ccd96db 100644 --- a/airflow-core/src/airflow/dag_processing/manager.py +++ b/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.""" diff --git a/airflow-core/src/airflow/executors/base_executor.py b/airflow-core/src/airflow/executors/base_executor.py index ebd850ad944b1..bb269be26d2cf 100644 --- a/airflow-core/src/airflow/executors/base_executor.py +++ b/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 diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py b/airflow-core/src/airflow/jobs/scheduler_job_runner.py index cb4c8f645520c..00560c916bf84 100644 --- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py +++ b/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() diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py b/airflow-core/src/airflow/jobs/triggerer_job_runner.py index 759735242ab1d..69cf4cfcbdc0b 100644 --- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py +++ b/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 diff --git a/airflow-core/tests/unit/api_fastapi/test_app.py b/airflow-core/tests/unit/api_fastapi/test_app.py index 9fd9a3edad1ae..27ce822eac665 100644 --- a/airflow-core/tests/unit/api_fastapi/test_app.py +++ b/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 diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py b/airflow-core/tests/unit/dag_processing/test_manager.py index c61e29c591a2a..81e6f0f94db0d 100644 --- a/airflow-core/tests/unit/dag_processing/test_manager.py +++ b/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): diff --git a/airflow-core/tests/unit/executors/test_base_executor.py b/airflow-core/tests/unit/executors/test_base_executor.py index 8bc7cfd73db23..27dd4e74a1c3b 100644 --- a/airflow-core/tests/unit/executors/test_base_executor.py +++ b/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)) diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py b/airflow-core/tests/unit/jobs/test_scheduler_job.py index 782bcf81c4aa3..76a948c02daec 100644 --- a/airflow-core/tests/unit/jobs/test_scheduler_job.py +++ b/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.""" diff --git a/airflow-core/tests/unit/jobs/test_triggerer_job.py b/airflow-core/tests/unit/jobs/test_triggerer_job.py index fa4d68f5fff6a..398c3477e920a 100644 --- a/airflow-core/tests/unit/jobs/test_triggerer_job.py +++ b/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.""" diff --git a/airflow-core/tests/unit/observability/metrics/test_stats.py b/airflow-core/tests/unit/observability/metrics/test_stats.py index 426e87a11991d..d1488354825a3 100644 --- a/airflow-core/tests/unit/observability/metrics/test_stats.py +++ b/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) diff --git a/shared/observability/src/airflow_shared/observability/metrics/stats.py b/shared/observability/src/airflow_shared/observability/metrics/stats.py index 7b51e580cd036..8acad3f97ebb9 100644 --- a/shared/observability/src/airflow_shared/observability/metrics/stats.py +++ b/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: diff --git a/shared/observability/tests/observability/metrics/test_stats.py b/shared/observability/tests/observability/metrics/test_stats.py index 9bdb4c473121c..c6a163d87513c 100644 --- a/shared/observability/tests/observability/metrics/test_stats.py +++ b/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"]