From 37b78b6e9785aa21c52385db62fc3d7ff0e59a12 Mon Sep 17 00:00:00 2001 From: Diogo Silva Date: Fri, 5 Jun 2026 14:20:20 +0100 Subject: [PATCH 1/7] Initialize Task SDK Stats in the API server so Edge Worker metrics are emitted The API server serves the Edge Worker REST API (/edge_worker/v1/...) whose heartbeat handler records edge_worker.* metrics through the Task SDK Stats singleton (resolved by the Edge provider via airflow.providers.common.compat). Unlike the scheduler, triggerer, dag-processor, executors and task runner, the API server never called Stats.initialize(...). After the auto-initializing Stats was removed in #63932, that singleton stays a NoStatsLogger in the API server process and every Edge Worker metric is silently dropped. Initialize the Task SDK Stats singleton from the FastAPI lifespan (runs once per worker, post-fork), mirroring the existing init in serde/task_runner. The call is guarded so a metrics misconfiguration can never block API server startup. Closes: #68077 Signed-off-by: Diogo Silva --- airflow-core/src/airflow/api_fastapi/app.py | 33 ++++++++++++++++ .../tests/unit/api_fastapi/test_app.py | 39 +++++++++++++++++++ 2 files changed, 72 insertions(+) diff --git a/airflow-core/src/airflow/api_fastapi/app.py b/airflow-core/src/airflow/api_fastapi/app.py index 8931840c8807c..47f9a7d04f60a 100644 --- a/airflow-core/src/airflow/api_fastapi/app.py +++ b/airflow-core/src/airflow/api_fastapi/app.py @@ -71,8 +71,41 @@ class _AuthManagerState: _lock = threading.Lock() +def _initialize_task_sdk_stats() -> None: + """ + Initialize the Task SDK ``Stats`` singleton in the API server process. + + The API server serves the Edge Worker REST API (``/edge_worker/v1/...``). Its heartbeat + handler records ``edge_worker.*`` metrics through the Task SDK ``Stats`` singleton + (resolved by the Edge provider via ``airflow.providers.common.compat``). + + Unlike the scheduler, triggerer, dag-processor, executors and task runner, the API server + never called ``Stats.initialize(...)``. Since the auto-initializing ``Stats`` was removed + (#63932) that singleton stays a ``NoStatsLogger`` in the API server process, so every Edge + Worker metric is silently dropped. + + Initialization is guarded so a metrics misconfiguration can never prevent the API server + from starting. + """ + try: + from airflow.sdk._shared.observability.metrics import stats + from airflow.sdk.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 Task SDK Stats in the API server; " + "Edge Worker metrics will not be emitted.", + exc_info=True, + ) + + @asynccontextmanager async def lifespan(app: FastAPI): + _initialize_task_sdk_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..44d1d11186ab5 100644 --- a/airflow-core/tests/unit/api_fastapi/test_app.py +++ b/airflow-core/tests/unit/api_fastapi/test_app.py @@ -168,3 +168,42 @@ def call_create_auth_manager(): assert call_count == 1 app_module.purge_cached_app() + + +class TestInitializeTaskSdkStats: + """ + The API server serves the Edge Worker REST API whose heartbeat handler records + ``edge_worker.*`` metrics through the Task SDK ``Stats`` singleton. Unlike other + components it never initialized that singleton, so Edge Worker metrics were dropped. + ``_initialize_task_sdk_stats`` (called from ``lifespan``) closes that gap. + """ + + def test_initializes_task_sdk_stats_with_factory(self): + """It initializes the Task SDK Stats singleton using the configured factory.""" + with ( + mock.patch("airflow.sdk._shared.observability.metrics.stats") as mock_stats, + mock.patch("airflow.sdk.observability.metrics.stats_utils") as mock_stats_utils, + ): + sentinel_factory = object() + mock_stats_utils.get_stats_factory.return_value = sentinel_factory + + app_module._initialize_task_sdk_stats() + + mock_stats_utils.get_stats_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, caplog): + """A metrics misconfiguration must not prevent the API server from starting.""" + with ( + mock.patch("airflow.sdk._shared.observability.metrics.stats") as mock_stats, + mock.patch("airflow.sdk.observability.metrics.stats_utils"), + ): + mock_stats.initialize.side_effect = RuntimeError("boom") + + # Must not raise. + app_module._initialize_task_sdk_stats() + + assert any("Failed to initialize Task SDK Stats" in rec.message for rec in caplog.records) From 3aa522fe57dd4e6e78d959fbb3d65611d125121d Mon Sep 17 00:00:00 2001 From: Diogo Silva Date: Fri, 5 Jun 2026 14:21:17 +0100 Subject: [PATCH 2/7] Add newsfragment for #68078 Signed-off-by: Diogo Silva --- airflow-core/newsfragments/68078.bugfix.rst | 1 + 1 file changed, 1 insertion(+) create mode 100644 airflow-core/newsfragments/68078.bugfix.rst diff --git a/airflow-core/newsfragments/68078.bugfix.rst b/airflow-core/newsfragments/68078.bugfix.rst new file mode 100644 index 0000000000000..f0e89a5b89a38 --- /dev/null +++ b/airflow-core/newsfragments/68078.bugfix.rst @@ -0,0 +1 @@ +Initialize the Task SDK ``Stats`` singleton in the API server so Edge Worker metrics are emitted again. The API server serves the Edge Worker REST API whose heartbeat handler records ``edge_worker.*`` metrics through the Task SDK ``Stats`` singleton, but -- unlike the scheduler, triggerer, dag-processor, executors and task runner -- it never called ``Stats.initialize(...)``. After the auto-initializing ``Stats`` was removed, that singleton stayed a ``NoStatsLogger`` and every Edge Worker metric was silently dropped. From a568eb925c41e74b08378a265231c549474c2f2e Mon Sep 17 00:00:00 2001 From: Diogo Silva Date: Fri, 5 Jun 2026 18:11:40 +0100 Subject: [PATCH 3/7] fix: fix ruff errors --- airflow-core/src/airflow/api_fastapi/app.py | 7 +++--- .../tests/unit/api_fastapi/test_app.py | 23 ++++++++++++------- 2 files changed, 18 insertions(+), 12 deletions(-) diff --git a/airflow-core/src/airflow/api_fastapi/app.py b/airflow-core/src/airflow/api_fastapi/app.py index 47f9a7d04f60a..b69d98ece85e3 100644 --- a/airflow-core/src/airflow/api_fastapi/app.py +++ b/airflow-core/src/airflow/api_fastapi/app.py @@ -88,8 +88,8 @@ def _initialize_task_sdk_stats() -> None: from starting. """ try: - from airflow.sdk._shared.observability.metrics import stats - from airflow.sdk.observability.metrics import stats_utils + from airflow._shared.observability.metrics import stats + from airflow.observability.metrics import stats_utils stats.initialize( factory=stats_utils.get_stats_factory(), @@ -97,8 +97,7 @@ def _initialize_task_sdk_stats() -> None: ) except Exception: log.warning( - "Failed to initialize Task SDK Stats in the API server; " - "Edge Worker metrics will not be emitted.", + "Failed to initialize Task SDK Stats in the API server; Edge Worker metrics will not be emitted.", exc_info=True, ) diff --git a/airflow-core/tests/unit/api_fastapi/test_app.py b/airflow-core/tests/unit/api_fastapi/test_app.py index 44d1d11186ab5..97092a729d561 100644 --- a/airflow-core/tests/unit/api_fastapi/test_app.py +++ b/airflow-core/tests/unit/api_fastapi/test_app.py @@ -180,16 +180,17 @@ class TestInitializeTaskSdkStats: def test_initializes_task_sdk_stats_with_factory(self): """It initializes the Task SDK Stats singleton using the configured factory.""" + sentinel_factory = object() with ( - mock.patch("airflow.sdk._shared.observability.metrics.stats") as mock_stats, - mock.patch("airflow.sdk.observability.metrics.stats_utils") as mock_stats_utils, + 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, ): - sentinel_factory = object() - mock_stats_utils.get_stats_factory.return_value = sentinel_factory - app_module._initialize_task_sdk_stats() - mock_stats_utils.get_stats_factory.assert_called_once_with() + 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 @@ -198,8 +199,8 @@ def test_initializes_task_sdk_stats_with_factory(self): def test_stats_failure_does_not_block_startup(self, caplog): """A metrics misconfiguration must not prevent the API server from starting.""" with ( - mock.patch("airflow.sdk._shared.observability.metrics.stats") as mock_stats, - mock.patch("airflow.sdk.observability.metrics.stats_utils"), + mock.patch("airflow._shared.observability.metrics.stats") as mock_stats, + mock.patch("airflow.observability.metrics.stats_utils.get_stats_factory"), ): mock_stats.initialize.side_effect = RuntimeError("boom") @@ -207,3 +208,9 @@ def test_stats_failure_does_not_block_startup(self, caplog): app_module._initialize_task_sdk_stats() assert any("Failed to initialize Task SDK Stats" in rec.message for rec in caplog.records) + + def test_stats_initialized_during_lifespan(self, client): + """_initialize_task_sdk_stats must be called as part of the app lifespan, not just defined.""" + with mock.patch.object(app_module, "_initialize_task_sdk_stats") as mock_init: + with client(): + mock_init.assert_called_once() From 1e7ad0c59c98782981c5d602e4216bebd9ca7d6a Mon Sep 17 00:00:00 2001 From: Diogo Silva Date: Tue, 9 Jun 2026 13:37:30 +0100 Subject: [PATCH 4/7] =?UTF-8?q?Remove=20newsfragment=20=E2=80=94=20bug=20f?= =?UTF-8?q?ix=20does=20not=20require=20a=20user-facing=20changelog=20entry?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- airflow-core/newsfragments/68078.bugfix.rst | 1 - 1 file changed, 1 deletion(-) delete mode 100644 airflow-core/newsfragments/68078.bugfix.rst diff --git a/airflow-core/newsfragments/68078.bugfix.rst b/airflow-core/newsfragments/68078.bugfix.rst deleted file mode 100644 index f0e89a5b89a38..0000000000000 --- a/airflow-core/newsfragments/68078.bugfix.rst +++ /dev/null @@ -1 +0,0 @@ -Initialize the Task SDK ``Stats`` singleton in the API server so Edge Worker metrics are emitted again. The API server serves the Edge Worker REST API whose heartbeat handler records ``edge_worker.*`` metrics through the Task SDK ``Stats`` singleton, but -- unlike the scheduler, triggerer, dag-processor, executors and task runner -- it never called ``Stats.initialize(...)``. After the auto-initializing ``Stats`` was removed, that singleton stayed a ``NoStatsLogger`` and every Edge Worker metric was silently dropped. From df67a185283d4804106c767c80bc44d1f3fe88ef Mon Sep 17 00:00:00 2001 From: Diogo Silva Date: Thu, 11 Jun 2026 14:28:41 +0100 Subject: [PATCH 5/7] Make API server Stats init log and docstring metric-agnostic --- airflow-core/src/airflow/api_fastapi/app.py | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/airflow-core/src/airflow/api_fastapi/app.py b/airflow-core/src/airflow/api_fastapi/app.py index b69d98ece85e3..889cd3a8e08fb 100644 --- a/airflow-core/src/airflow/api_fastapi/app.py +++ b/airflow-core/src/airflow/api_fastapi/app.py @@ -75,14 +75,16 @@ def _initialize_task_sdk_stats() -> None: """ Initialize the Task SDK ``Stats`` singleton in the API server process. - The API server serves the Edge Worker REST API (``/edge_worker/v1/...``). Its heartbeat - handler records ``edge_worker.*`` metrics through the Task SDK ``Stats`` singleton - (resolved by the Edge provider via ``airflow.providers.common.compat``). - Unlike the scheduler, triggerer, dag-processor, executors and task runner, the API server never called ``Stats.initialize(...)``. Since the auto-initializing ``Stats`` was removed - (#63932) that singleton stays a ``NoStatsLogger`` in the API server process, so every Edge - Worker metric is silently dropped. + (#63932) that singleton stays a ``NoStatsLogger`` in the API server process, so any metric + emitted through the Task SDK ``Stats`` singleton from code served by the API server is + silently dropped. + + The case that surfaced this is the Edge Worker REST API (``/edge_worker/v1/...``): its + heartbeat handler records ``edge_worker.*`` metrics through the Task SDK ``Stats`` singleton + (resolved by the Edge provider via ``airflow.providers.common.compat``), but the problem is + not specific to Edge. Initialization is guarded so a metrics misconfiguration can never prevent the API server from starting. @@ -97,7 +99,8 @@ def _initialize_task_sdk_stats() -> None: ) except Exception: log.warning( - "Failed to initialize Task SDK Stats in the API server; Edge Worker metrics will not be emitted.", + "Failed to initialize Task SDK Stats in the API server; metrics emitted through the " + "Task SDK Stats singleton will not be recorded.", exc_info=True, ) From f8159e56eeec7fcf7e85d5967aa8042880212828 Mon Sep 17 00:00:00 2001 From: Diogo Silva <49190578+diogosilva30@users.noreply.github.com> Date: Thu, 11 Jun 2026 22:16:02 +0100 Subject: [PATCH 6/7] Update airflow-core/src/airflow/api_fastapi/app.py Co-authored-by: Jens Scheffler <95105677+jscheffl@users.noreply.github.com> --- airflow-core/src/airflow/api_fastapi/app.py | 11 ----------- 1 file changed, 11 deletions(-) diff --git a/airflow-core/src/airflow/api_fastapi/app.py b/airflow-core/src/airflow/api_fastapi/app.py index 889cd3a8e08fb..45b64dc36f707 100644 --- a/airflow-core/src/airflow/api_fastapi/app.py +++ b/airflow-core/src/airflow/api_fastapi/app.py @@ -75,17 +75,6 @@ def _initialize_task_sdk_stats() -> None: """ Initialize the Task SDK ``Stats`` singleton in the API server process. - Unlike the scheduler, triggerer, dag-processor, executors and task runner, the API server - never called ``Stats.initialize(...)``. Since the auto-initializing ``Stats`` was removed - (#63932) that singleton stays a ``NoStatsLogger`` in the API server process, so any metric - emitted through the Task SDK ``Stats`` singleton from code served by the API server is - silently dropped. - - The case that surfaced this is the Edge Worker REST API (``/edge_worker/v1/...``): its - heartbeat handler records ``edge_worker.*`` metrics through the Task SDK ``Stats`` singleton - (resolved by the Edge provider via ``airflow.providers.common.compat``), but the problem is - not specific to Edge. - Initialization is guarded so a metrics misconfiguration can never prevent the API server from starting. """ From a82116c06ba0dffff0c1f6d68dbded7494f4bb5c Mon Sep 17 00:00:00 2001 From: Diogo Silva <49190578+diogosilva30@users.noreply.github.com> Date: Thu, 11 Jun 2026 22:16:13 +0100 Subject: [PATCH 7/7] Update airflow-core/tests/unit/api_fastapi/test_app.py Co-authored-by: Jens Scheffler <95105677+jscheffl@users.noreply.github.com> --- airflow-core/tests/unit/api_fastapi/test_app.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/airflow-core/tests/unit/api_fastapi/test_app.py b/airflow-core/tests/unit/api_fastapi/test_app.py index 97092a729d561..1633e11de4ee2 100644 --- a/airflow-core/tests/unit/api_fastapi/test_app.py +++ b/airflow-core/tests/unit/api_fastapi/test_app.py @@ -172,10 +172,7 @@ def call_create_auth_manager(): class TestInitializeTaskSdkStats: """ - The API server serves the Edge Worker REST API whose heartbeat handler records - ``edge_worker.*`` metrics through the Task SDK ``Stats`` singleton. Unlike other - components it never initialized that singleton, so Edge Worker metrics were dropped. - ``_initialize_task_sdk_stats`` (called from ``lifespan``) closes that gap. + Ensure that stats subsystem is properly initialized in API server. """ def test_initializes_task_sdk_stats_with_factory(self):