This is an automated email from the ASF dual-hosted git repository.
pierrejeambrun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 64ff9e40a40 Add REST API metrics (#64523)
64ff9e40a40 is described below
commit 64ff9e40a400eee895372759048549456effaacd
Author: Henry Chen <[email protected]>
AuthorDate: Tue Oct 13 17:21:51 2026 +0800
Add REST API metrics (#64523)
* add metrics
* allow user define the mapping themself
* Move API metrics config validation into configuration parser
* Simplify REST API metrics tagging
Remove the configurable path-to-surface mapping and emit metrics only
for core API and UI routes. Use normalized route templates with method
and status family tags to keep metric cardinality bounded.
* Report REST API request duration under a milliseconds metric name
Stats.timing records milliseconds on every supported backend, so publishing
the
timer as http_request_duration_seconds would have left every consumer of the
metric wrong by a factor of 1000.
Route tagging also kept a fallback that reverse-looked-up the endpoint in
scope["router"].routes whenever scope["route"] was absent. Only FastAPI's
APIRoute populates scope["route"], and all 151 routes under /api/v2 and /ui
are
APIRoute, so that branch was unreachable. For a mounted sub-app -- the one
case
it was meant to cover -- scope["router"] stays the root router, so the
lookup
could not have succeeded there either. The tests built a plain Starlette
app,
which never sets scope["route"], and so covered only that dead branch
instead of
the path the API server actually takes.
* Bound the REST API metrics method tag to methods the API serves
The router rejects a method mismatch with 405 before any authentication
dependency runs, and uvicorn's h11 parser accepts any token as a method.
That
let an anonymous client create a new metric series per request just by
inventing method names, defeating the bounded-cardinality goal of these
metrics. Only the five methods registered on API routes now get their own
tag
value; everything else is recorded as OTHER, so the series count is fixed by
the route table rather than by whatever clients send.
* Cover the Task SDK execution API in the REST API metrics
`/execution` carries every task heartbeat, state transition and XCom,
making it
the busiest surface the API server has, and it emitted no metrics at all.
Two things kept it out. The middleware was registered from
`init_middlewares`,
which `create_app` only calls for the core app, so `airflow api-server
--apps
execution` -- the shape used to run the execution API as its own deployment
--
installed nothing, and no configuration could recover the metrics. The path
prefix allowlist also covered only `/api/v2` and `/ui`.
Install the middleware for every `--apps` selection from a new
`init_api_metrics`, the shape #72026 settled on after the identical gap left
execution-only deployments with no access logs, and add `/execution` to the
allowlist. Registering it on the outer app rather than inside the execution
sub-app keeps a single instance, so a request cannot be counted twice, and
keeps it outside `GZipMiddleware` so execution API durations include
compression time the way core API durations already do.
A mounted sub-app declares its routes relative to the mount point, so the
route
tag now carries `root_path`. Without it,
`/execution/task-instances/{id}/run`
would be tagged `/task-instances/{id}/run` and collide with any core route
sharing that template.
* Build the API metrics test app once per module
Review feedback: every test rebuilt the whole FastAPI app just to issue one
request, so the app construction cost was paid once per test case for no
benefit. A module-scoped fixture is enough -- the middleware under test
holds
no state, and each test still gets its own client.
* Document that REST API metric tags need a tagging backend
Review feedback: the metric tables promised a breakdown by method, route and
status_family without saying that classic StatsD drops tags, so an operator
reading the docs would configure statsd_on and then wonder why they only
ever
see one aggregate series.
---
airflow-core/src/airflow/api_fastapi/app.py | 10 +
.../src/airflow/api_fastapi/common/http_metrics.py | 137 ++++++++++
.../src/airflow/api_fastapi/core_api/app.py | 7 +-
.../airflow/observability/metrics/stats_utils.py | 10 +
.../unit/api_fastapi/common/test_http_metrics.py | 297 +++++++++++++++++++++
airflow-core/tests/unit/api_fastapi/test_app.py | 23 ++
.../tests/unit/observability/metrics/test_stats.py | 13 +
.../observability/metrics/metrics_template.yaml | 18 ++
8 files changed, 511 insertions(+), 4 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/app.py
b/airflow-core/src/airflow/api_fastapi/app.py
index 974fb2cee75..49255373024 100644
--- a/airflow-core/src/airflow/api_fastapi/app.py
+++ b/airflow-core/src/airflow/api_fastapi/app.py
@@ -31,6 +31,7 @@ from airflow import settings
from airflow.api_fastapi.common.dagbag import create_dag_bag
from airflow.api_fastapi.common.exceptions import init_error_handlers
from airflow.api_fastapi.common.http_access_log import HttpAccessLogMiddleware
+from airflow.api_fastapi.common.http_metrics import HttpMetricsMiddleware
from airflow.api_fastapi.core_api.app import (
init_config,
init_flask_plugins,
@@ -40,6 +41,7 @@ from airflow.api_fastapi.core_api.app import (
from airflow.api_fastapi.execution_api.app import create_task_execution_api_app
from airflow.configuration import conf
from airflow.exceptions import AirflowConfigException
+from airflow.observability.metrics import stats_utils
from airflow.utils.providers_configuration_loader import
providers_configuration_loaded
if TYPE_CHECKING:
@@ -154,6 +156,7 @@ def create_app(apps: str = "all") -> FastAPI:
init_error_handlers(app)
init_middlewares(app)
+ init_api_metrics(app)
init_access_logging(app)
init_config(app)
@@ -228,6 +231,13 @@ def init_access_logging(app: FastAPI) -> None:
app.add_middleware(HttpAccessLogMiddleware)
+def init_api_metrics(app: FastAPI) -> None:
+ """Install the API metrics middleware, the only producer of API request
metrics."""
+ if not stats_utils.is_metrics_enabled():
+ return
+ app.add_middleware(HttpMetricsMiddleware)
+
+
def init_plugins(app: FastAPI) -> None:
"""Integrate FastAPI app, middlewares and UI plugins."""
from airflow import plugins_manager
diff --git a/airflow-core/src/airflow/api_fastapi/common/http_metrics.py
b/airflow-core/src/airflow/api_fastapi/common/http_metrics.py
new file mode 100644
index 00000000000..feabd66d4d4
--- /dev/null
+++ b/airflow-core/src/airflow/api_fastapi/common/http_metrics.py
@@ -0,0 +1,137 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""HTTP API metrics middleware."""
+
+from __future__ import annotations
+
+import time
+from typing import TYPE_CHECKING
+
+import structlog
+
+from airflow._shared.observability.metrics.stats import Stats
+
+if TYPE_CHECKING:
+ from starlette.types import ASGIApp, Message, Receive, Scope, Send
+
+logger = structlog.get_logger(logger_name="http.metrics")
+
+# `/execution` is the Task SDK execution API, mounted as a sub-app. The
middleware runs on the
+# outer app, so a request to it passes through here like any other.
+_API_METRICS_PATH_PREFIXES = ("/api/v2", "/ui", "/execution")
+# Only methods registered on API routes get their own tag value. A method
mismatch is rejected
+# before authentication, so any other value would let anonymous clients mint
metric series.
+_SERVED_HTTP_METHODS = frozenset({"GET", "POST", "PUT", "PATCH", "DELETE"})
+_OTHER_HTTP_METHOD = "OTHER"
+
+
+def _is_api_metrics_path(path: str) -> bool:
+ return any(path == prefix or path.startswith(f"{prefix}/") for prefix in
_API_METRICS_PATH_PREFIXES)
+
+
+def _get_status_family(status_code: int) -> str:
+ return f"{status_code // 100}xx"
+
+
+def _get_method_tag(method: str) -> str:
+ return method if method in _SERVED_HTTP_METHODS else _OTHER_HTTP_METHOD
+
+
+def _get_route_template(scope: Scope) -> str:
+ # FastAPI's APIRoute.matches() stores the matched route in the scope
before the endpoint runs.
+ # Requests matching no route (404s, redirects) share one bucket so tag
cardinality stays bounded.
+ route_path = getattr(scope.get("route"), "path", None)
+ if isinstance(route_path, str) and route_path:
+ # A mounted sub-app declares its routes relative to the mount point,
which Starlette keeps
+ # in root_path. Without it, `/execution/task-instances/{id}` would be
tagged
+ # `/task-instances/{id}` and collide with core routes sharing that
template.
+ return f"{scope.get('root_path', '')}{route_path}"
+ return "unmatched"
+
+
+def _emit_api_metrics(
+ *,
+ scope: Scope,
+ path: str,
+ method: str,
+ status_code: int,
+ duration_us: int,
+) -> None:
+ if not _is_api_metrics_path(path):
+ return
+
+ # Keep tags bounded so API metrics remain usable across supported backends.
+ tags = {
+ "method": _get_method_tag(method),
+ "route": _get_route_template(scope),
+ "status_family": _get_status_family(status_code),
+ }
+ duration_ms = duration_us / 1000.0
+
+ Stats.incr("http_requests_total", tags=tags)
+ Stats.timing("http_request_duration_milliseconds", duration_ms, tags=tags)
+
+
+class HttpMetricsMiddleware:
+ """Emit REST API metrics for completed HTTP requests."""
+
+ def __init__(self, app: ASGIApp) -> None:
+ self.app = app
+
+ async def __call__(self, scope: Scope, receive: Receive, send: Send) ->
None:
+ if scope["type"] != "http":
+ await self.app(scope, receive, send)
+ return
+
+ start = time.monotonic_ns()
+ response: Message | None = None
+
+ async def capture_send(message: Message) -> None:
+ nonlocal response
+ if message["type"] == "http.response.start":
+ response = message
+ await send(message)
+
+ try:
+ await self.app(scope, receive, capture_send)
+ except Exception:
+ if response is None:
+ response = {"status": 500}
+ raise
+ finally:
+ path = scope["path"]
+ duration_us = (time.monotonic_ns() - start) // 1000
+ status = response["status"] if response is not None else 0
+ method = scope["method"]
+
+ # Include health checks because their latency can reveal database
contention before probes fail.
+ # Observability failures must never affect serving the request.
+ try:
+ _emit_api_metrics(
+ scope=scope,
+ path=path,
+ method=method,
+ status_code=status,
+ duration_us=duration_us,
+ )
+ except Exception:
+ logger.exception(
+ "failed to emit API metrics",
+ method=method,
+ path=path,
+ status_code=status,
+ )
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/app.py
b/airflow-core/src/airflow/api_fastapi/core_api/app.py
index 841c46c2291..d3a66a5b68d 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/app.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/app.py
@@ -287,9 +287,8 @@ def init_middlewares(app: FastAPI) -> None:
for middleware_cls, middleware_kwargs in
get_auth_manager().get_fastapi_middlewares():
app.add_middleware(middleware_cls, **middleware_kwargs)
- # GZipMiddleware must be inside HttpAccessLogMiddleware so that access
logs capture
- # the full end-to-end duration including compression time.
HttpAccessLogMiddleware is
- # installed by ``init_access_logging`` in ``create_app``, which runs after
this
- # function — do not reorder those calls.
+ # GZipMiddleware must be inside HttpMetricsMiddleware and
HttpAccessLogMiddleware so that both
+ # capture the full end-to-end duration including compression time. Both
are installed by
+ # ``create_app`` after this function returns — do not reorder those calls.
# See https://github.com/apache/airflow/issues/60165
app.add_middleware(GZipMiddleware, minimum_size=1024, compresslevel=5)
diff --git a/airflow-core/src/airflow/observability/metrics/stats_utils.py
b/airflow-core/src/airflow/observability/metrics/stats_utils.py
index 9e0f1263bff..15099850b7d 100644
--- a/airflow-core/src/airflow/observability/metrics/stats_utils.py
+++ b/airflow-core/src/airflow/observability/metrics/stats_utils.py
@@ -23,6 +23,16 @@ from airflow._shared.observability.metrics.base_stats_logger
import NoStatsLogge
from airflow.configuration import conf
+def is_metrics_enabled() -> bool:
+ return any(
+ (
+ conf.getboolean("metrics", "statsd_datadog_enabled",
fallback=False),
+ conf.getboolean("metrics", "statsd_on", fallback=False),
+ conf.getboolean("metrics", "otel_on", fallback=False),
+ )
+ )
+
+
def get_stats_factory() -> Callable:
if conf.getboolean("metrics", "statsd_datadog_enabled"):
from airflow.observability.metrics import datadog_logger
diff --git a/airflow-core/tests/unit/api_fastapi/common/test_http_metrics.py
b/airflow-core/tests/unit/api_fastapi/common/test_http_metrics.py
new file mode 100644
index 00000000000..af508cfb8b1
--- /dev/null
+++ b/airflow-core/tests/unit/api_fastapi/common/test_http_metrics.py
@@ -0,0 +1,297 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import asyncio
+from unittest import mock
+
+import pytest
+import structlog.testing
+from fastapi import FastAPI
+from starlette.responses import PlainTextResponse
+from starlette.testclient import TestClient
+
+from airflow.api_fastapi.common.http_metrics import (
+ HttpMetricsMiddleware,
+ _emit_api_metrics,
+ _get_status_family,
+)
+
+
+def _make_execution_app() -> FastAPI:
+ """Stand in for the execution API, which the real app mounts as a
sub-app."""
+ app = FastAPI()
+
+ @app.get("/task-instances/{task_instance_id}/run")
+ async def run_task_instance(task_instance_id: str):
+ return PlainTextResponse("ok")
+
+ return app
+
+
[email protected](scope="module")
+def test_app() -> FastAPI:
+ """Build a FastAPI app so route resolution behaves as it does in the real
API server."""
+ app = FastAPI()
+
+ @app.get("/")
+ async def homepage():
+ return PlainTextResponse("ok")
+
+ @app.get("/api/v2/items/{item_id}")
+ async def api_item(item_id: str):
+ return PlainTextResponse("ok")
+
+
@app.get("/api/v2/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}")
+ async def task_instance(dag_id: str, dag_run_id: str, task_id: str):
+ return PlainTextResponse("ok")
+
+ @app.get("/ui/items/{item_id}")
+ async def ui_item(item_id: str):
+ return PlainTextResponse("ok")
+
+ @app.get("/plugin/items/{item_id}")
+ async def plugin_item(item_id: str):
+ return PlainTextResponse("ok")
+
+ @app.get("/api/v2/fail")
+ async def api_fail():
+ raise RuntimeError("boom")
+
+ @app.get("/api/v2/monitor/health")
+ async def health():
+ return PlainTextResponse("healthy")
+
+ app.mount("/execution", _make_execution_app())
+ app.add_middleware(HttpMetricsMiddleware)
+ return app
+
+
[email protected]
+def test_client(test_app) -> TestClient:
+ return TestClient(test_app, raise_server_exceptions=False)
+
+
+def test_metric_emission_failure_is_logged(test_client):
+ with (
+ mock.patch(
+ "airflow.api_fastapi.common.http_metrics._emit_api_metrics",
+ side_effect=RuntimeError("metrics boom"),
+ ),
+ structlog.testing.capture_logs() as logs,
+ ):
+ response = test_client.get("/api/v2/items/42")
+
+ assert response.status_code == 200
+ assert [record["event"] for record in logs] == ["failed to emit API
metrics"]
+ assert logs[0]["method"] == "GET"
+ assert logs[0]["path"] == "/api/v2/items/42"
+ assert logs[0]["status_code"] == 200
+
+
+def test_non_http_scope_does_not_emit_metrics():
+ """Non-HTTP scopes (e.g. lifespan) are passed through without metrics
emission."""
+
+ async def lifespan_app(scope, receive, send):
+ pass
+
+ middleware = HttpMetricsMiddleware(lifespan_app)
+
+ with (
+ mock.patch("airflow.api_fastapi.common.http_metrics.Stats.incr") as
mock_incr,
+ mock.patch("airflow.api_fastapi.common.http_metrics.Stats.timing") as
mock_timing,
+ ):
+ asyncio.run(middleware({"type": "lifespan"}, None, None))
+
+ mock_incr.assert_not_called()
+ mock_timing.assert_not_called()
+
+
[email protected](
+ ("status_code", "expected"),
+ [
+ pytest.param(0, "0xx", id="no-response"),
+ pytest.param(200, "2xx", id="success"),
+ pytest.param(404, "4xx", id="client-error"),
+ pytest.param(500, "5xx", id="server-error"),
+ ],
+)
+def test_get_status_family(status_code, expected):
+ assert _get_status_family(status_code) == expected
+
+
+def test_request_duration_is_emitted_in_milliseconds():
+ with (
+ mock.patch("airflow.api_fastapi.common.http_metrics.Stats.incr"),
+ mock.patch("airflow.api_fastapi.common.http_metrics.Stats.timing") as
mock_timing,
+ ):
+ _emit_api_metrics(
+ scope={},
+ path="/api/v2/items/42",
+ method="GET",
+ status_code=200,
+ duration_us=1_500_000,
+ )
+
+ mock_timing.assert_called_once_with("http_request_duration_milliseconds",
1500.0, tags=mock.ANY)
+
+
[email protected](
+ ("request_path", "route_tag"),
+ [
+ pytest.param("/api/v2/items/42", "/api/v2/items/{item_id}",
id="public"),
+ pytest.param(
+
"/api/v2/dags/example/dagRuns/manual__2026-07-13/taskInstances/extract",
+
"/api/v2/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}",
+ id="route-template-with-parameters",
+ ),
+ pytest.param("/ui/items/42", "/ui/items/{item_id}", id="ui"),
+ pytest.param(
+ "/execution/task-instances/abc/run",
+ "/execution/task-instances/{task_instance_id}/run",
+ id="execution-route-tag-keeps-the-mount-prefix",
+ ),
+ ],
+)
+def test_api_requests_emit_metrics(request_path, route_tag, test_client):
+ with (
+ mock.patch("airflow.api_fastapi.common.http_metrics.Stats.incr") as
mock_incr,
+ mock.patch("airflow.api_fastapi.common.http_metrics.Stats.timing") as
mock_timing,
+ ):
+ response = test_client.get(request_path)
+
+ assert response.status_code == 200
+ expected_tags = {
+ "method": "GET",
+ "route": route_tag,
+ "status_family": "2xx",
+ }
+ mock_incr.assert_called_once_with("http_requests_total",
tags=expected_tags)
+ mock_timing.assert_called_once_with("http_request_duration_milliseconds",
mock.ANY, tags=expected_tags)
+
+
[email protected](
+ ("request_path", "expected_status_code"),
+ [
+ pytest.param("/plugin/items/42", 200, id="plugin"),
+ pytest.param("/api/v20/items/42", 404, id="similar-prefix"),
+ ],
+)
+def test_paths_outside_api_metrics_surfaces_do_not_emit_metrics(
+ request_path, expected_status_code, test_client
+):
+ with (
+ mock.patch("airflow.api_fastapi.common.http_metrics.Stats.incr") as
mock_incr,
+ mock.patch("airflow.api_fastapi.common.http_metrics.Stats.timing") as
mock_timing,
+ ):
+ response = test_client.get(request_path)
+
+ assert response.status_code == expected_status_code
+ mock_incr.assert_not_called()
+ mock_timing.assert_not_called()
+
+
+def test_health_path_emits_metrics(test_client):
+ with (
+ mock.patch("airflow.api_fastapi.common.http_metrics.Stats.incr") as
mock_incr,
+ mock.patch("airflow.api_fastapi.common.http_metrics.Stats.timing") as
mock_timing,
+ ):
+ response = test_client.get("/api/v2/monitor/health")
+
+ assert response.status_code == 200
+ expected_tags = {
+ "method": "GET",
+ "route": "/api/v2/monitor/health",
+ "status_family": "2xx",
+ }
+ mock_incr.assert_called_once_with("http_requests_total",
tags=expected_tags)
+ mock_timing.assert_called_once_with("http_request_duration_milliseconds",
mock.ANY, tags=expected_tags)
+
+
+def
test_failed_api_requests_emit_metrics_with_server_error_status(test_client):
+ with (
+ mock.patch("airflow.api_fastapi.common.http_metrics.Stats.incr") as
mock_incr,
+ mock.patch("airflow.api_fastapi.common.http_metrics.Stats.timing") as
mock_timing,
+ ):
+ response = test_client.get("/api/v2/fail")
+
+ assert response.status_code == 500
+ expected_tags = {
+ "method": "GET",
+ "route": "/api/v2/fail",
+ "status_family": "5xx",
+ }
+ mock_incr.assert_called_once_with("http_requests_total",
tags=expected_tags)
+ mock_timing.assert_called_once_with("http_request_duration_milliseconds",
mock.ANY, tags=expected_tags)
+
+
[email protected](
+ ("method", "expected_status_code", "expected_tags"),
+ [
+ pytest.param(
+ "GET",
+ 200,
+ {"method": "GET", "route": "/api/v2/items/{item_id}",
"status_family": "2xx"},
+ id="served-method",
+ ),
+ pytest.param(
+ "POST",
+ 405,
+ {"method": "POST", "route": "/api/v2/items/{item_id}",
"status_family": "4xx"},
+ id="served-method-not-allowed-on-route",
+ ),
+ pytest.param(
+ "HEAD",
+ 405,
+ {"method": "OTHER", "route": "/api/v2/items/{item_id}",
"status_family": "4xx"},
+ id="standard-method-never-served",
+ ),
+ pytest.param(
+ "METHOD-0001",
+ 405,
+ {"method": "OTHER", "route": "/api/v2/items/{item_id}",
"status_family": "4xx"},
+ id="made-up-method",
+ ),
+ ],
+)
+def test_method_tag_is_bounded_to_served_methods(method, expected_status_code,
expected_tags, test_client):
+ with (
+ mock.patch("airflow.api_fastapi.common.http_metrics.Stats.incr") as
mock_incr,
+ mock.patch("airflow.api_fastapi.common.http_metrics.Stats.timing") as
mock_timing,
+ ):
+ response = test_client.request(method, "/api/v2/items/42")
+
+ assert response.status_code == expected_status_code
+ mock_incr.assert_called_once_with("http_requests_total",
tags=expected_tags)
+ mock_timing.assert_called_once_with("http_request_duration_milliseconds",
mock.ANY, tags=expected_tags)
+
+
+def test_unmatched_api_requests_use_unmatched_route_tag(test_client):
+ with (
+ mock.patch("airflow.api_fastapi.common.http_metrics.Stats.incr") as
mock_incr,
+ mock.patch("airflow.api_fastapi.common.http_metrics.Stats.timing") as
mock_timing,
+ ):
+ response = test_client.get("/api/v2/missing")
+
+ assert response.status_code == 404
+ expected_tags = {
+ "method": "GET",
+ "route": "unmatched",
+ "status_family": "4xx",
+ }
+ mock_incr.assert_called_once_with("http_requests_total",
tags=expected_tags)
+ mock_timing.assert_called_once_with("http_request_duration_milliseconds",
mock.ANY, tags=expected_tags)
diff --git a/airflow-core/tests/unit/api_fastapi/test_app.py
b/airflow-core/tests/unit/api_fastapi/test_app.py
index dcbbc0a593a..c994f90a0d5 100644
--- a/airflow-core/tests/unit/api_fastapi/test_app.py
+++ b/airflow-core/tests/unit/api_fastapi/test_app.py
@@ -26,11 +26,13 @@ from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import event, text
from sqlalchemy.engine import Engine
+from starlette.middleware.gzip import GZipMiddleware
import airflow.api_fastapi.app as app_module
import airflow.plugins_manager as plugins_manager
from airflow import settings
from airflow.api_fastapi.common.http_access_log import HttpAccessLogMiddleware
+from airflow.api_fastapi.common.http_metrics import HttpMetricsMiddleware
from airflow.utils.session import create_session_async
from tests_common.test_utils.config import conf_vars
@@ -189,6 +191,27 @@ def
test_access_log_middleware_installed_outermost_for_every_apps_selection(apps
assert installed[0] is HttpAccessLogMiddleware
[email protected]("apps", ["all", "core", "execution"])
+@conf_vars({("metrics", "statsd_on"): "True"})
+def test_api_metrics_middleware_installed_for_every_apps_selection(apps):
+ """The execution API is mounted as a sub-app and `--apps execution` never
reaches
+ ``init_middlewares``, so installing this middleware there would leave the
surface every
+ running task depends on unmetered. It must also wrap GZipMiddleware so the
recorded
+ duration includes compression time (see #60165)."""
+ installed = [m.cls for m in
app_module.create_app(apps=apps).user_middleware]
+
+ assert installed.count(HttpMetricsMiddleware) == 1
+ if GZipMiddleware in installed:
+ assert installed.index(HttpMetricsMiddleware) <
installed.index(GZipMiddleware)
+
+
[email protected]("apps", ["all", "core", "execution"])
+def test_api_metrics_middleware_skipped_without_a_metrics_backend(apps):
+ installed = [m.cls for m in
app_module.create_app(apps=apps).user_middleware]
+
+ assert HttpMetricsMiddleware not in installed
+
+
def test_catch_all_route_last():
"""
Ensure the catch all route that returns the initial html is the last route
in the fastapi app.
diff --git a/airflow-core/tests/unit/observability/metrics/test_stats.py
b/airflow-core/tests/unit/observability/metrics/test_stats.py
index e417d42e5a9..404920e4a36 100644
--- a/airflow-core/tests/unit/observability/metrics/test_stats.py
+++ b/airflow-core/tests/unit/observability/metrics/test_stats.py
@@ -39,6 +39,19 @@ class InvalidCustomStatsd:
class TestStats:
+ @pytest.mark.parametrize(
+ ("metrics_config", "expected"),
+ [
+ pytest.param({}, False, id="disabled"),
+ pytest.param({("metrics", "statsd_on"): "True"}, True,
id="statsd"),
+ pytest.param({("metrics", "statsd_datadog_enabled"): "True"},
True, id="datadog"),
+ pytest.param({("metrics", "otel_on"): "True"}, True, id="otel"),
+ ],
+ )
+ def test_is_metrics_enabled(self, metrics_config, expected):
+ with conf_vars(metrics_config):
+ assert stats_utils.is_metrics_enabled() is expected
+
@pytest.mark.parametrize(
("extra_conf", "expected_message"),
[
diff --git
a/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml
b/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml
index 93d09231136..9aa2b7a4a9d 100644
---
a/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml
+++
b/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml
@@ -703,9 +703,27 @@ metrics:
legacy_name: "-"
name_variables: ["status"]
+ - name: "http_requests_total"
+ description: "Number of completed REST API requests, tagged by method,
route, and status_family.
+ Breaking the total down by those tags needs a backend that carries them
(``otel_on``,
+ ``statsd_influxdb_enabled`` or ``statsd_datadog_enabled``); classic StatsD
drops the tags and
+ reports only the overall request count."
+ type: "counter"
+ legacy_name: "-"
+ name_variables: []
+
# ==========
# Timers
# ==========
+ - name: "http_request_duration_milliseconds"
+ description: "Milliseconds taken to complete a REST API request, tagged by
method, route,
+ and status_family. Breaking the timings down by those tags needs a backend
that carries them
+ (``otel_on``, ``statsd_influxdb_enabled`` or ``statsd_datadog_enabled``);
classic StatsD drops
+ the tags and reports only the overall request duration."
+ type: "timer"
+ legacy_name: "-"
+ name_variables: []
+
- name: "dagrun.dependency-check"
description: "Milliseconds taken to check Dag dependencies"
type: "timer"
