diff --git a/python/sglang/srt/entrypoints/grpc_server.py b/python/sglang/srt/entrypoints/grpc_server.py index 848fc520b..4c62d61fb 100644 --- a/python/sglang/srt/entrypoints/grpc_server.py +++ b/python/sglang/srt/entrypoints/grpc_server.py @@ -1,23 +1,43 @@ """ Thin gRPC server wrapper — delegates to smg-grpc-servicer package. -When --enable-metrics is set, a lightweight HTTP server is started on ---metrics-http-port (default: --port + 1) to expose Prometheus /metrics. +A lightweight HTTP sidecar is started alongside the gRPC server to expose: +- /metrics (Prometheus, when --enable-metrics is set) +- /start_profile, /stop_profile (profiling control) + +The sidecar is started on --grpc-http-sidecar-port (default: --port + 1) +once the gRPC request manager is ready, regardless of whether --enable-metrics +is set. """ +import json import logging +import time + +from aiohttp import web + +from sglang.srt.managers.io_struct import ProfileReq, ProfileReqType +from sglang.srt.utils.common import get_bool_env_var logger = logging.getLogger(__name__) -async def _start_metrics_server(host: str, port: int): - """Start an HTTP server exposing Prometheus /metrics. +async def _start_sidecar_server(host: str, port: int, app): + """Start the aiohttp sidecar and return the runner for cleanup.""" + runner = web.AppRunner(app) + await runner.setup() + try: + site = web.TCPSite(runner, host, port) + await site.start() + except BaseException: + await runner.cleanup() + raise + logger.info("HTTP sidecar server started on http://%s:%d", host, port) + return runner - The caller is responsible for calling ``runner.cleanup()`` on the returned - AppRunner when shutting down. The server begins accepting requests before - this function returns. - """ - from aiohttp import web + +def _add_metrics_routes(app): + """Add Prometheus /metrics endpoint to the aiohttp app.""" from prometheus_client import ( CollectorRegistry, multiprocess, @@ -29,14 +49,6 @@ async def _start_metrics_server(host: str, port: int): async def metrics_handler(request): try: - # Create a fresh registry and attach a MultiProcessCollector - # on each request. This is the recommended pattern from the - # prometheus_client multiprocess docs to ensure up-to-date - # data from PROMETHEUS_MULTIPROC_DIR. - # - # Use OpenMetrics format to match what the HTTP-mode endpoint - # returns when Prometheus scrapes it with an OpenMetrics Accept - # header (make_asgi_app performs content negotiation). registry = CollectorRegistry() multiprocess.MultiProcessCollector(registry) data = generate_latest(registry) @@ -48,19 +60,96 @@ async def _start_metrics_server(host: str, port: int): logger.exception("Failed to generate Prometheus metrics") return web.Response(status=500, text="Failed to generate metrics") - app = web.Application() app.router.add_get("/metrics", metrics_handler) - runner = web.AppRunner(app) - await runner.setup() - try: - site = web.TCPSite(runner, host, port) - await site.start() - except BaseException: - await runner.cleanup() - raise - logger.info("Prometheus metrics server started on http://%s:%d/metrics", host, port) - return runner + +def _check_communicator_results(results, action): + """Return a web.Response error if results indicate failure, else None.""" + if not results: + return web.Response(status=500, text="No response from scheduler\n") + failures = [r for r in results if not r.success] + if failures: + msgs = " | ".join(r.message for r in failures) + return web.Response(status=500, text=f"{action} failed: {msgs}\n") + return None + + +def _add_admin_routes(app, request_manager): + """Add admin endpoints to the aiohttp app. + + Endpoints: /start_profile, /stop_profile. + Business logic (request construction, env var handling, response interpretation) + lives here; request_manager only provides the transport to the scheduler. + """ + + async def start_profile_handler(request): + try: + if request.content_length and request.content_length > 0: + try: + body = await request.json() + except json.JSONDecodeError as e: + return web.Response( + status=400, + text=f"Invalid JSON in request body: {e}", + ) + else: + body = {} + + # Build ProfileReq with env var overrides (same as tokenizer_communicator_mixin) + with_stack = body.get("with_stack") + env_with_stack = get_bool_env_var("SGLANG_PROFILE_WITH_STACK", "true") + with_stack = (with_stack is not False) and env_with_stack + record_shapes = body.get("record_shapes") + env_record_shapes = get_bool_env_var("SGLANG_PROFILE_RECORD_SHAPES", "true") + record_shapes = (record_shapes is not False) and env_record_shapes + + req = ProfileReq( + type=ProfileReqType.START_PROFILE, + output_dir=body.get("output_dir"), + start_step=body.get("start_step"), + num_steps=body.get("num_steps"), + activities=body.get("activities"), + with_stack=with_stack, + record_shapes=record_shapes, + profile_by_stage=body.get("profile_by_stage", False), + profile_id=str(time.time()), + merge_profiles=body.get("merge_profiles", False), + profile_prefix=body.get("profile_prefix"), + profile_stages=body.get("profile_stages"), + ) + results = await request_manager.send_communicator_req( + req, "profile_communicator", timeout=600.0 + ) + err = _check_communicator_results(results, "Start Profile") + if err: + return err + return web.Response(text="Start profiling.\n") + except Exception as e: + logger.exception("Failed to start profile") + return web.Response( + status=500, + text=f"Internal error: {type(e).__name__}. Check server logs.\n", + ) + + async def stop_profile_handler(request): + try: + req = ProfileReq(type=ProfileReqType.STOP_PROFILE) + results = await request_manager.send_communicator_req( + req, "profile_communicator", timeout=600.0 + ) + err = _check_communicator_results(results, "Stop profile") + if err: + return err + return web.Response(text="Stop profiling. This will take some time.\n") + except Exception as e: + logger.exception("Failed to stop profile") + return web.Response( + status=500, + text=f"Internal error: {type(e).__name__}. Check server logs.\n", + ) + + app.router.add_post("/start_profile", start_profile_handler) + app.router.add_post("/stop_profile", stop_profile_handler) async def serve_grpc(server_args, model_info=None): @@ -75,46 +164,74 @@ async def serve_grpc(server_args, model_info=None): "version mismatch — see the chained exception above for details." ) from e - metrics_runner = None + sidecar_app = web.Application() + sidecar_runner = None + sidecar_port = ( + server_args.grpc_http_sidecar_port + if server_args.grpc_http_sidecar_port is not None + else server_args.port + 1 + ) + + # Metrics setup: must set PROMETHEUS_MULTIPROC_DIR before scheduler + # processes import prometheus_client, since the env var is inherited + # at fork time. if server_args.enable_metrics: try: from sglang.srt.observability.func_timer import enable_func_timer from sglang.srt.utils import set_prometheus_multiproc_dir - # Must set PROMETHEUS_MULTIPROC_DIR env var before any - # prometheus_client import. The env var is inherited by child - # processes (schedulers) that import prometheus_client later. set_prometheus_multiproc_dir() enable_func_timer() + _add_metrics_routes(sidecar_app) + except Exception as e: + logger.error( + "Failed to set up metrics: %s. Continuing without metrics.", + e, + exc_info=True, + ) - metrics_port = ( - server_args.metrics_http_port - if server_args.metrics_http_port is not None - else server_args.port + 1 + async def _on_request_manager_ready(request_manager, srv_args, sched_info): + nonlocal sidecar_runner + try: + _add_admin_routes(sidecar_app, request_manager) + except Exception as e: + logger.error( + "Failed to set up admin routes: %s. " + "Continuing without admin endpoints.", + e, + exc_info=True, + ) + try: + sidecar_runner = await _start_sidecar_server( + server_args.host, sidecar_port, sidecar_app ) - metrics_runner = await _start_metrics_server(server_args.host, metrics_port) except OSError as e: logger.error( - "Failed to start metrics server: %s. " "Continuing without metrics.", + "Failed to start HTTP sidecar server: %s. " + "Continuing without metrics/profile endpoints.", e, exc_info=True, ) except Exception as e: logger.error( - "Unexpected error starting metrics server: %s. " - "Continuing without metrics.", + "Unexpected error starting HTTP sidecar server: %s. " + "Continuing without metrics/profile endpoints.", e, exc_info=True, ) try: - await _serve_grpc(server_args, model_info) + await _serve_grpc( + server_args, + model_info, + on_request_manager_ready=_on_request_manager_ready, + ) finally: - if metrics_runner is not None: + if sidecar_runner is not None: try: - await metrics_runner.cleanup() + await sidecar_runner.cleanup() except Exception as e: logger.exception( - "Failed to cleanly shut down Prometheus metrics server: %s", + "Failed to cleanly shut down HTTP sidecar server: %s", e, ) diff --git a/python/sglang/srt/server_args.py b/python/sglang/srt/server_args.py index 4454d1926..c32dae658 100644 --- a/python/sglang/srt/server_args.py +++ b/python/sglang/srt/server_args.py @@ -410,7 +410,7 @@ class ServerArgs: crash_dump_folder: Optional[str] = None show_time_cost: bool = False enable_metrics: bool = False - metrics_http_port: Optional[int] = None + grpc_http_sidecar_port: Optional[int] = None enable_mfu_metrics: bool = False enable_metrics_for_all_schedulers: bool = False tokenizer_metrics_custom_labels_header: str = "x-custom-labels" @@ -4717,12 +4717,12 @@ class ServerArgs: help="Enable log prometheus metrics.", ) parser.add_argument( - "--metrics-http-port", + "--grpc-http-sidecar-port", type=int, - default=ServerArgs.metrics_http_port, - help="Port for the Prometheus metrics HTTP server. " - "Only used in gRPC mode (--grpc-mode); in HTTP mode, metrics are served on the main --port. " - "Defaults to --port + 1 when --enable-metrics is set.", + default=ServerArgs.grpc_http_sidecar_port, + help="Port for the HTTP sidecar server in gRPC mode (--grpc-mode). " + "Serves Prometheus metrics and profiling endpoints. " + "Defaults to --port + 1. Not used in HTTP mode.", ) parser.add_argument( "--enable-mfu-metrics",