diff --git a/python/sglang/multimodal_gen/runtime/entrypoints/http_server.py b/python/sglang/multimodal_gen/runtime/entrypoints/http_server.py index 6e559e0e6321..0a95f581ad23 100644 --- a/python/sglang/multimodal_gen/runtime/entrypoints/http_server.py +++ b/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=["*"], diff --git a/python/sglang/srt/entrypoints/http_server.py b/python/sglang/srt/entrypoints/http_server.py index 5bb8ca428935..430d86b9b67b 100644 --- a/python/sglang/srt/entrypoints/http_server.py +++ b/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: diff --git a/python/sglang/srt/observability/http_request_metrics.py b/python/sglang/srt/observability/http_request_metrics.py new file mode 100644 index 000000000000..a8298a926726 --- /dev/null +++ b/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 diff --git a/python/sglang/srt/utils/common.py b/python/sglang/srt/utils/common.py index ca7fc1963bd4..c84312cceb71 100644 --- a/python/sglang/srt/utils/common.py +++ b/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 diff --git a/test/registered/unit/observability/test_http_request_metrics.py b/test/registered/unit/observability/test_http_request_metrics.py new file mode 100644 index 000000000000..96834d31e53d --- /dev/null +++ b/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()