[Fix] metrics: record HTTP request metrics in tokenizer workers

python/sglang/multimodal_gen/runtime/entrypoints/http_server.py

@@ -46,11 +46,12 @@
     globally_suppress_loggers,
     init_logger,
 )
-from sglang.srt.observability.trace import fastapi_telemetry_disabled_kwargs
-from sglang.srt.utils.common import (
-    add_prometheus_middleware,
-    add_prometheus_track_response_middleware,
+from sglang.srt.observability.http_request_metrics import (
+    add_http_request_metrics_middleware,
+    init_http_request_metrics,
 )
+from sglang.srt.observability.trace import fastapi_telemetry_disabled_kwargs
+from sglang.srt.utils.common import add_prometheus_middleware
 from sglang.srt.utils.json_response import orjson_response
 from sglang.version import __version__
 
@@ -435,7 +436,8 @@ def create_app(server_args: ServerArgs):
     if server_args.enable_metrics:
         configure_metrics()
         add_prometheus_middleware(app)
-        add_prometheus_track_response_middleware(app)
+        add_http_request_metrics_middleware(app)
+        init_http_request_metrics()
     app.add_middleware(
         CORSMiddleware,
         allow_origins=["*"],

python/sglang/srt/entrypoints/http_server.py

@@ -169,6 +169,10 @@
 )
 from sglang.srt.managers.tokenizer_manager import ServerStatus, TokenizerManager
 from sglang.srt.observability.func_timer import enable_func_timer
+from sglang.srt.observability.http_request_metrics import (
+    add_http_request_metrics_middleware,
+    init_http_request_metrics,
+)
 from sglang.srt.observability.trace import (
     fastapi_telemetry_disabled_kwargs,
     process_tracing_init,
@@ -181,7 +185,6 @@
 from sglang.srt.server_args import PortArgs, ServerArgs
 from sglang.srt.utils import (
     add_prometheus_middleware,
-    add_prometheus_track_response_middleware,
     build_server_info,
     delete_directory,
     freeze_gc,
@@ -305,6 +308,7 @@ async def lifespan(fast_api_app: FastAPI):
     if get_observability().enable_metrics:
         add_prometheus_middleware(app)
         enable_func_timer()
+        init_http_request_metrics()
 
     # Init tracing
     if get_observability().enable_trace:
@@ -502,6 +506,9 @@ async def custom_handler(request: Request):
 
     app.add_middleware(RequestDecompressionMiddleware)
 
+# Registered at import so that tokenizer workers, which re-import this module, get it too.
+add_http_request_metrics_middleware(app)
+
 # Include routers
 from sglang.srt.entrypoints.v1_loads import router as v1_loads_router
 
@@ -2667,9 +2674,6 @@ def _setup_and_run_http_server(
     if tokenizer_manager is not None:
         tokenizer_manager._subprocess_watchdog = subprocess_watchdog
 
-    if get_observability().enable_metrics:
-        add_prometheus_track_response_middleware(app)
-
     # Pass additional arguments to the lifespan function.
     # They will be used for additional initialization setups.
     if get_serving().tokenizer_worker_num == 1:

python/sglang/srt/observability/http_request_metrics.py

@@ -0,0 +1,117 @@
+from typing import Dict, Optional, Tuple
+
+
+class RefCountedGauge:
+    def __init__(self, gauge):
+        self._gauge = gauge
+        self._refcount: Dict[str, int] = {}
+
+    def inc(self, key: str):
+        if key in self._refcount:
+            self._refcount[key] += 1
+        else:
+            self._refcount[key] = 1
+            self._gauge.inc()
+
+    def dec(self, key: str):
+        if key in self._refcount:
+            self._refcount[key] -= 1
+            if self._refcount[key] == 0:
+                del self._refcount[key]
+                self._gauge.dec()
+
+
+class _HttpRequestMetrics:
+    def __init__(self):
+        from prometheus_client import Counter, Gauge
+
+        self.requests_total = Counter(
+            name="sglang:http_requests_total",
+            documentation="Total number of HTTP requests by endpoint and method",
+            labelnames=["endpoint", "method"],
+        )
+
+        self.responses_total = Counter(
+            name="sglang:http_responses_total",
+            documentation="Total number of HTTP responses by endpoint and status code",
+            labelnames=["endpoint", "status_code", "method"],
+        )
+
+        self.requests_active = Gauge(
+            name="sglang:http_requests_active",
+            documentation="Number of currently active HTTP requests",
+            labelnames=["endpoint", "method"],
+            multiprocess_mode="livesum",
+        )
+
+        self.routing_keys_active = RefCountedGauge(
+            Gauge(
+                name="sglang:routing_keys_active",
+                documentation="Number of unique routing keys with active requests",
+                multiprocess_mode="livesum",
+            )
+        )
+
+
+_http_request_metrics: Optional[_HttpRequestMetrics] = None
+
+
+def init_http_request_metrics():
+    # Call after PROMETHEUS_MULTIPROC_DIR is set: prometheus_client picks its
+    # value backend on import.
+    global _http_request_metrics
+    if _http_request_metrics is None:
+        _http_request_metrics = _HttpRequestMetrics()
+
+
+def add_http_request_metrics_middleware(app):
+    """Records nothing until init_http_request_metrics() runs in this process."""
+    # Fix: replace BaseHTTPMiddleware's call_next with a pure ASGI version
+    # that passes `receive` through, so request.is_disconnected() keeps working.
+    from sglang.srt.utils.http_middleware_patch import patch_app_http_middleware
+
+    patch_app_http_middleware(app)
+
+    @app.middleware("http")
+    async def track_http_status_code(request, call_next):
+        metrics = _http_request_metrics
+        if metrics is None:
+            return await call_next(request)
+
+        # With recording all requests, we have the risk of high cardinality if requests have arbitrary unhandled paths.
+        # But given that SGLang engines with metrics enabled are usually behind routers this looks safe.
+        path, is_handled_path = _get_fastapi_request_path(request)
+        method = request.method
+        routing_key = request.headers.get("x-smg-routing-key")
+
+        metrics.requests_total.labels(endpoint=path, method=method).inc()
+        metrics.requests_active.labels(endpoint=path, method=method).inc()
+        if routing_key:
+            metrics.routing_keys_active.inc(routing_key)
+
+        try:
+            response = await call_next(request)
+
+            metrics.responses_total.labels(
+                endpoint=path,
+                method=method,
+                status_code=str(response.status_code),
+            ).inc()
+
+            return response
+        finally:
+            metrics.requests_active.labels(endpoint=path, method=method).dec()
+            if routing_key:
+                metrics.routing_keys_active.dec(routing_key)
+
+
+# https://github.com/blueswen/fastapi-observability/blob/132a3c576f8b09e5311c68bd553215013bc75685/fastapi_app/utils.py#L98
+def _get_fastapi_request_path(request) -> Tuple[str, bool]:
+    from starlette.routing import Match
+
+    for route in request.app.routes:
+        match, child_scope = route.matches(request.scope)
+        if match == Match.FULL:
+            return getattr(route, "path", request.url.path), True
+
+    return request.url.path, False

python/sglang/srt/utils/common.py

@@ -2844,111 +2844,6 @@ def add_prometheus_middleware(app):
     app.routes.append(metrics_route)
 
 
-class RefCountedGauge:
-    def __init__(self, gauge):
-        self._gauge = gauge
-        self._refcount: Dict[str, int] = {}
-
-    def inc(self, key: str):
-        if key in self._refcount:
-            self._refcount[key] += 1
-        else:
-            self._refcount[key] = 1
-            self._gauge.inc()
-
-    def dec(self, key: str):
-        if key in self._refcount:
-            self._refcount[key] -= 1
-            if self._refcount[key] == 0:
-                del self._refcount[key]
-                self._gauge.dec()
-
-
-def add_prometheus_track_response_middleware(
-    app, extra_labels: Optional[Dict[str, str]] = None
-):
-    from prometheus_client import Counter, Gauge
-
-    extra_labels = extra_labels or {}
-    extra_label_names = list(extra_labels.keys())
-
-    http_request_counter = Counter(
-        name="sglang:http_requests_total",
-        documentation="Total number of HTTP requests by endpoint and method",
-        labelnames=extra_label_names + ["endpoint", "method"],
-    )
-
-    http_response_counter = Counter(
-        name="sglang:http_responses_total",
-        documentation="Total number of HTTP responses by endpoint and status code",
-        labelnames=extra_label_names + ["endpoint", "status_code", "method"],
-    )
-
-    http_requests_active = Gauge(
-        name="sglang:http_requests_active",
-        documentation="Number of currently active HTTP requests",
-        labelnames=extra_label_names + ["endpoint", "method"],
-        multiprocess_mode="livesum",
-    )
-
-    routing_keys_active = RefCountedGauge(
-        Gauge(
-            name="sglang:routing_keys_active",
-            documentation="Number of unique routing keys with active requests",
-            multiprocess_mode="livesum",
-        )
-    )
-
-    # Fix: replace BaseHTTPMiddleware's call_next with a pure ASGI version
-    # that passes `receive` through, so request.is_disconnected() keeps working.
-    from sglang.srt.utils.http_middleware_patch import patch_app_http_middleware
-
-    patch_app_http_middleware(app)
-
-    @app.middleware("http")
-    async def track_http_status_code(request, call_next):
-        # With recording all requests, we have the risk of high cardinality if requests have arbitrary unhandled paths.
-        # But given that SGLang engines with metrics enabled are usually behind routers this looks safe.
-        path, is_handled_path = _get_fastapi_request_path(request)
-        method = request.method
-        routing_key = request.headers.get("x-smg-routing-key")
-
-        http_request_counter.labels(**extra_labels, endpoint=path, method=method).inc()
-        http_requests_active.labels(**extra_labels, endpoint=path, method=method).inc()
-        if routing_key:
-            routing_keys_active.inc(routing_key)
-
-        try:
-            response = await call_next(request)
-
-            http_response_counter.labels(
-                **extra_labels,
-                endpoint=path,
-                method=method,
-                status_code=str(response.status_code),
-            ).inc()
-
-            return response
-        finally:
-            http_requests_active.labels(
-                **extra_labels, endpoint=path, method=method
-            ).dec()
-            if routing_key:
-                routing_keys_active.dec(routing_key)
-
-
-# https://github.com/blueswen/fastapi-observability/blob/132a3c576f8b09e5311c68bd553215013bc75685/fastapi_app/utils.py#L98
-def _get_fastapi_request_path(request) -> Tuple[str, bool]:
-    from starlette.routing import Match
-
-    for route in request.app.routes:
-        match, child_scope = route.matches(request.scope)
-        if match == Match.FULL:
-            return getattr(route, "path", request.url.path), True
-
-    return request.url.path, False
-
-
 # Copy from pytorch and OpenRLHF to allow creating multiple main groups.
 # https://github.com/pytorch/pytorch/blob/main/torch/distributed/distributed_c10d.py
 # https://github.com/OpenRLHF/OpenRLHF/blob/main/openrlhf/utils/distributed_util.py

test/registered/unit/observability/test_http_request_metrics.py

@@ -0,0 +1,37 @@
+from sglang.test.ci.ci_register import register_cpu_ci
+
+register_cpu_ci(est_time=15, suite="base-a-test-cpu")
+
+import unittest
+
+from fastapi.testclient import TestClient
+from prometheus_client import REGISTRY
+
+import sglang.srt.entrypoints.http_server as http_server
+from sglang.srt.observability import http_request_metrics
+from sglang.test.test_utils import CustomTestCase
+
+UNKNOWN_PATH = "/__no_such_route__"
+
+
+class TestHttpRequestMetrics(CustomTestCase):
+    def test_module_level_app_records_requests(self):
+        # Tokenizer workers re-import http_server and serve its module-level app,
+        # so the middleware must be on it, while the metrics must wait for
+        # init_http_request_metrics().
+        self.assertIsNone(http_request_metrics._http_request_metrics)
+
+        http_request_metrics.init_http_request_metrics()
+        TestClient(http_server.app).get(UNKNOWN_PATH)
+
+        self.assertEqual(
+            1.0,
+            REGISTRY.get_sample_value(
+                "sglang:http_requests_total",
+                {"endpoint": UNKNOWN_PATH, "method": "GET"},
+            ),
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()