ColtenOuO commented on code in PR #64523: URL: https://github.com/apache/airflow/pull/64523#discussion_r3891129653
########## airflow-core/src/airflow/api_fastapi/common/http_metrics.py: ########## @@ -0,0 +1,124 @@ +# 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") + +_API_METRICS_PATH_PREFIXES = ("/api/v2", "/ui") + + +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_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: + return 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": 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) Review Comment: If this is indeed a bug, we should probably add corresponding tests as well. ########## airflow-core/src/airflow/api_fastapi/common/http_metrics.py: ########## @@ -0,0 +1,124 @@ +# 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") + +_API_METRICS_PATH_PREFIXES = ("/api/v2", "/ui") + + +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_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: + return 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": method, Review Comment: There might be a minor security concern here regarding metric cardinality. An attacker could potentially overload the metrics backend by sending requests with an arbitrary number of custom HTTP methods (instead of standard ones like GET, POST, etc.). Even though these requests will be rejected during authentication, the metrics middleware still records the failed attempts. Example: If an attacker sends: ``` METHOD-0001 /api/v2/dags/example METHOD-0002 /api/v2/dags/example METHOD-0003 /api/v2/dags/example ``` The metrics backend will create three distinct time series because each request has a different method tag. This can cause high-cardinality issues, waste backend resources, and pollute the metrics with junk data. We should probably restrict the method tag to an allowlist of standard HTTP methods and bucket all others as "OTHER". ########## airflow-core/src/airflow/api_fastapi/common/http_metrics.py: ########## @@ -0,0 +1,124 @@ +# 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") + +_API_METRICS_PATH_PREFIXES = ("/api/v2", "/ui") + + +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_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: + return 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": 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) Review Comment: I noticed that although the HTTP middleware passes tags to `Stats.incr()`, the regular StatsD backend ultimately calls self.statsd.incr(stat, count, rate) without forwarding them: [shared/observability/src/airflow_shared/observability/metrics/statsd_logger.py:L84-L97](https://github.com/apache/airflow/blob/f3747c331a235829aa81121eb3d042aaf669099d/shared/observability/src/airflow_shared/observability/metrics/statsd_logger.py#L84-L97) The `prepare_stat_with_tags` decorator preserves them by encoding them into the metric name only when `statsd_influxdb_enabled=True`. Since that setting defaults to False, a deployment using plain `statsd_on=True` appears to receive only aggregate metrics across all methods, routes, and status families. In particular, it would not be possible to calculate per-endpoint QPS or filter http_requests_total by status_family="5xx" as described. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
