diff --git a/airflow-core/src/airflow/api_fastapi/app.py b/airflow-core/src/airflow/api_fastapi/app.py index 8931840c8807c..a4dbc3c9b722e 100644 --- a/airflow-core/src/airflow/api_fastapi/app.py +++ b/airflow-core/src/airflow/api_fastapi/app.py @@ -71,8 +71,32 @@ class _AuthManagerState: _lock = threading.Lock() +def _initialize_api_server_stats() -> None: + """ + Initialize the ``Stats`` singleton in the API server process. + + Initialization is guarded so a metrics misconfiguration can never prevent the API server + from starting. + """ + try: + from airflow._shared.observability.metrics import stats + from airflow.observability.metrics import stats_utils + + stats.initialize( + factory=stats_utils.get_stats_factory(), + export_legacy_names=conf.getboolean("metrics", "legacy_names_on"), + ) + except Exception: + log.warning( + "Failed to initialize API server Stats in the API server; metrics emitted through the " + "API server Stats singleton will not be recorded.", + exc_info=True, + ) + + @asynccontextmanager async def lifespan(app: FastAPI): + _initialize_api_server_stats() async with AsyncExitStack() as stack: for route in app.routes: if isinstance(route, Mount) and isinstance(route.app, FastAPI): diff --git a/airflow-core/tests/unit/api_fastapi/test_app.py b/airflow-core/tests/unit/api_fastapi/test_app.py index 1e8817ef2431a..9fd9a3edad1ae 100644 --- a/airflow-core/tests/unit/api_fastapi/test_app.py +++ b/airflow-core/tests/unit/api_fastapi/test_app.py @@ -168,3 +168,47 @@ def call_create_auth_manager(): assert call_count == 1 app_module.purge_cached_app() + + +class TestInitializeApiServerStats: + """ + Ensure that stats subsystem is properly initialized in API server. + """ + + def test_initializes_api_server_stats_with_factory(self): + """It initializes the Stats singleton in the API server using the configured factory.""" + sentinel_factory = object() + with ( + mock.patch("airflow._shared.observability.metrics.stats") as mock_stats, + mock.patch( + "airflow.observability.metrics.stats_utils.get_stats_factory", + return_value=sentinel_factory, + ) as mock_get_factory, + ): + app_module._initialize_api_server_stats() + + mock_get_factory.assert_called_once_with() + mock_stats.initialize.assert_called_once() + _, kwargs = mock_stats.initialize.call_args + assert kwargs["factory"] is sentinel_factory + assert isinstance(kwargs["export_legacy_names"], bool) + + def test_stats_failure_does_not_block_startup(self): + """A metrics misconfiguration must not prevent the API server from starting.""" + with ( + mock.patch("airflow._shared.observability.metrics.stats") as mock_stats, + mock.patch("airflow.observability.metrics.stats_utils.get_stats_factory"), + mock.patch("airflow.api_fastapi.app.log") as mock_logger, + ): + mock_stats.initialize.side_effect = RuntimeError("boom") + + # Must not raise. + app_module._initialize_api_server_stats() + + mock_logger.warning.assert_called_once() + + def test_stats_initialized_during_lifespan(self, client): + """_initialize_api_server_stats must be called as part of the app lifespan, not just defined.""" + with mock.patch.object(app_module, "_initialize_api_server_stats") as mock_init: + with client(): + mock_init.assert_called_once()