From df9cccf4d1befca325fdcdd40820915dbccc49fb Mon Sep 17 00:00:00 2001 From: Kangyan Zhou Date: Thu, 9 Apr 2026 23:29:11 -0700 Subject: [PATCH 1/3] [Observability] Add profiling HTTP endpoints for gRPC mode Add /start_profile and /stop_profile endpoints to the HTTP sidecar that runs alongside the gRPC server. This enables torch profiler trace collection in gRPC mode, matching the existing HTTP-mode profiling API. The sidecar (previously metrics-only) is refactored into a unified HTTP server that always starts on --metrics-http-port (default: --port + 1), serving both Prometheus /metrics (when --enable-metrics) and profiling endpoints. Profiling is triggered via an async callback from smg-grpc-servicer's serve_grpc(), which provides access to the GrpcRequestManager after scheduler initialization. Co-Authored-By: Claude Opus 4.6 (1M context) --- python/sglang/srt/entrypoints/grpc_server.py | 237 +++++++++++++++---- python/sglang/srt/server_args.py | 12 +- 2 files changed, 200 insertions(+), 49 deletions(-) diff --git a/python/sglang/srt/entrypoints/grpc_server.py b/python/sglang/srt/entrypoints/grpc_server.py index 848fc520b0e3..a9625cc68c89 100644 --- a/python/sglang/srt/entrypoints/grpc_server.py +++ b/python/sglang/srt/entrypoints/grpc_server.py @@ -1,8 +1,16 @@ """ 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, for direct engine access) +- /server_info (server configuration and internal state) + +The sidecar always starts on --grpc-http-sidecar-port (default: --port + 1), +even if --enable-metrics is not set, to serve these endpoints. + +Note: /flush_cache is served natively via the FlushCache gRPC RPC (proxied +by the SMG router's HTTP /flush_cache endpoint). """ import logging @@ -10,13 +18,24 @@ 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.""" + from aiohttp import web - The caller is responsible for calling ``runner.cleanup()`` on the returned - AppRunner when shutting down. The server begins accepting requests before - this function returns. - """ + 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 + + +def _add_metrics_routes(app): + """Add Prometheus /metrics endpoint to the aiohttp app.""" from aiohttp import web from prometheus_client import ( CollectorRegistry, @@ -29,14 +48,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 +59,131 @@ async def metrics_handler(request): 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 _add_admin_routes(app, request_manager, server_args, scheduler_info): + """Add admin endpoints to the aiohttp app. + + Endpoints: /start_profile, /stop_profile, /server_info. + Business logic (request construction, env var handling, response interpretation) + lives here; request_manager only provides the ZMQ transport layer. + + Note: /flush_cache is handled natively via the FlushCache gRPC RPC. + """ + import dataclasses + import json + import time + from functools import partial + + from aiohttp import web + + from sglang.srt.managers.io_struct import ( + GetInternalStateReq, + ProfileReq, + ProfileReqType, + ) + from sglang.srt.utils.common import get_bool_env_var + from sglang.version import __version__ + + async def start_profile_handler(request): + try: + if request.content_length and request.content_length > 0: + try: + body = await request.json() + except (json.JSONDecodeError, Exception) 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 = ( + False if with_stack is False or env_with_stack is False else True + ) + 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 + ) + 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"Profile failed: {msgs}\n") + return web.Response(text="Start profiling.\n") + except Exception as e: + logger.exception("Failed to start profile") + return web.Response(status=500, text=str(e)) + + 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 + ) + 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"Stop profile failed: {msgs}\n") + 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=str(e)) + + async def server_info_handler(request): + try: + results = await request_manager.send_communicator_req( + GetInternalStateReq(), "get_internal_state_communicator" + ) + internal_states = [r.internal_state for r in results] if results else [] + # Serialize server_args safely without mutating the shared object + import copy + + args_copy = copy.copy(server_args) + for attr in ("model_config", "custom_sigquit_handler"): + if hasattr(args_copy, attr): + delattr(args_copy, attr) + server_args_dict = dataclasses.asdict(args_copy) + return web.json_response( + { + **server_args_dict, + **scheduler_info, + "internal_states": internal_states, + "version": __version__, + }, + dumps=partial(json.dumps, default=str), + ) + except Exception as e: + logger.exception("Failed to get server info") + return web.json_response({"error": str(e)}, status=500) + + app.router.add_post("/start_profile", start_profile_handler) + app.router.add_post("/stop_profile", stop_profile_handler) + app.router.add_get("/server_info", server_info_handler) async def serve_grpc(server_args, model_info=None): @@ -75,46 +198,74 @@ async def serve_grpc(server_args, model_info=None): "version mismatch — see the chained exception above for details." ) from e - metrics_runner = None + from aiohttp import web + + sidecar_app = web.Application() + sidecar_runner = None + sidecar_port = ( + getattr(server_args, "grpc_http_sidecar_port", None) or 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, srv_args, sched_info) + 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 753d18c21eb7..ed869e211e7a 100644 --- a/python/sglang/srt/server_args.py +++ b/python/sglang/srt/server_args.py @@ -405,7 +405,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" @@ -4517,12 +4517,12 @@ def add_cli_args(parser: argparse.ArgumentParser): 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, profiling, and server_info endpoints. " + "Defaults to --port + 1. Not used in HTTP mode.", ) parser.add_argument( "--enable-mfu-metrics", From f2a4198ffa93b161a45169e990403018ecc993b2 Mon Sep 17 00:00:00 2001 From: Kangyan Zhou Date: Mon, 13 Apr 2026 19:06:38 -0700 Subject: [PATCH 2/3] =?UTF-8?q?Remove=20/server=5Finfo=20from=20HTTP=20sid?= =?UTF-8?q?ecar=20=E2=80=94=20already=20available=20as=20gRPC=20RPC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reviewer pointed out that server_info is already enabled natively in smg-grpc-servicer, so the duplicate HTTP sidecar endpoint is unnecessary. Co-Authored-By: Claude Opus 4.6 (1M context) --- python/sglang/srt/entrypoints/grpc_server.py | 44 ++------------------ python/sglang/srt/server_args.py | 2 +- 2 files changed, 4 insertions(+), 42 deletions(-) diff --git a/python/sglang/srt/entrypoints/grpc_server.py b/python/sglang/srt/entrypoints/grpc_server.py index a9625cc68c89..28389eae40de 100644 --- a/python/sglang/srt/entrypoints/grpc_server.py +++ b/python/sglang/srt/entrypoints/grpc_server.py @@ -4,13 +4,9 @@ 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, for direct engine access) -- /server_info (server configuration and internal state) The sidecar always starts on --grpc-http-sidecar-port (default: --port + 1), even if --enable-metrics is not set, to serve these endpoints. - -Note: /flush_cache is served natively via the FlushCache gRPC RPC (proxied -by the SMG router's HTTP /flush_cache endpoint). """ import logging @@ -62,29 +58,23 @@ async def metrics_handler(request): app.router.add_get("/metrics", metrics_handler) -def _add_admin_routes(app, request_manager, server_args, scheduler_info): +def _add_admin_routes(app, request_manager): """Add admin endpoints to the aiohttp app. - Endpoints: /start_profile, /stop_profile, /server_info. + Endpoints: /start_profile, /stop_profile. Business logic (request construction, env var handling, response interpretation) lives here; request_manager only provides the ZMQ transport layer. - - Note: /flush_cache is handled natively via the FlushCache gRPC RPC. """ - import dataclasses import json import time - from functools import partial from aiohttp import web from sglang.srt.managers.io_struct import ( - GetInternalStateReq, ProfileReq, ProfileReqType, ) from sglang.srt.utils.common import get_bool_env_var - from sglang.version import __version__ async def start_profile_handler(request): try: @@ -154,36 +144,8 @@ async def stop_profile_handler(request): logger.exception("Failed to stop profile") return web.Response(status=500, text=str(e)) - async def server_info_handler(request): - try: - results = await request_manager.send_communicator_req( - GetInternalStateReq(), "get_internal_state_communicator" - ) - internal_states = [r.internal_state for r in results] if results else [] - # Serialize server_args safely without mutating the shared object - import copy - - args_copy = copy.copy(server_args) - for attr in ("model_config", "custom_sigquit_handler"): - if hasattr(args_copy, attr): - delattr(args_copy, attr) - server_args_dict = dataclasses.asdict(args_copy) - return web.json_response( - { - **server_args_dict, - **scheduler_info, - "internal_states": internal_states, - "version": __version__, - }, - dumps=partial(json.dumps, default=str), - ) - except Exception as e: - logger.exception("Failed to get server info") - return web.json_response({"error": str(e)}, status=500) - app.router.add_post("/start_profile", start_profile_handler) app.router.add_post("/stop_profile", stop_profile_handler) - app.router.add_get("/server_info", server_info_handler) async def serve_grpc(server_args, model_info=None): @@ -227,7 +189,7 @@ async def serve_grpc(server_args, model_info=None): async def _on_request_manager_ready(request_manager, srv_args, sched_info): nonlocal sidecar_runner try: - _add_admin_routes(sidecar_app, request_manager, srv_args, sched_info) + _add_admin_routes(sidecar_app, request_manager) except Exception as e: logger.error( "Failed to set up admin routes: %s. " diff --git a/python/sglang/srt/server_args.py b/python/sglang/srt/server_args.py index ed869e211e7a..77ebba2c41d3 100644 --- a/python/sglang/srt/server_args.py +++ b/python/sglang/srt/server_args.py @@ -4521,7 +4521,7 @@ def add_cli_args(parser: argparse.ArgumentParser): type=int, default=ServerArgs.grpc_http_sidecar_port, help="Port for the HTTP sidecar server in gRPC mode (--grpc-mode). " - "Serves Prometheus metrics, profiling, and server_info endpoints. " + "Serves Prometheus metrics and profiling endpoints. " "Defaults to --port + 1. Not used in HTTP mode.", ) parser.add_argument( From 5fe6350a761d55b22ef5cecf5a1e02c2089ac5dd Mon Sep 17 00:00:00 2001 From: Kangyan Zhou Date: Mon, 13 Apr 2026 22:44:25 -0700 Subject: [PATCH 3/3] Address code review: fix error handling, imports, and docstrings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Fix redundant except (JSONDecodeError, Exception) → JSONDecodeError only - Fix module docstring: remove misleading "always starts" and "direct engine access" - Stop leaking raw str(e) to HTTP callers; return type name only - Replace getattr+or with is-not-None check for sidecar port - Extract _check_communicator_results helper to deduplicate result checking - Unify with_stack/record_shapes boolean idioms - Move lightweight imports (json, time, aiohttp, io_struct) to top level - Fix docstring: "ZMQ transport layer" → "transport to the scheduler" Co-Authored-By: Claude Opus 4.6 (1M context) --- python/sglang/srt/entrypoints/grpc_server.py | 80 ++++++++++---------- 1 file changed, 42 insertions(+), 38 deletions(-) diff --git a/python/sglang/srt/entrypoints/grpc_server.py b/python/sglang/srt/entrypoints/grpc_server.py index 28389eae40de..4c62d61fb987 100644 --- a/python/sglang/srt/entrypoints/grpc_server.py +++ b/python/sglang/srt/entrypoints/grpc_server.py @@ -3,21 +3,27 @@ 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, for direct engine access) +- /start_profile, /stop_profile (profiling control) -The sidecar always starts on --grpc-http-sidecar-port (default: --port + 1), -even if --enable-metrics is not set, to serve these endpoints. +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_sidecar_server(host: str, port: int, app): """Start the aiohttp sidecar and return the runner for cleanup.""" - from aiohttp import web - runner = web.AppRunner(app) await runner.setup() try: @@ -32,7 +38,6 @@ async def _start_sidecar_server(host: str, port: int, app): def _add_metrics_routes(app): """Add Prometheus /metrics endpoint to the aiohttp app.""" - from aiohttp import web from prometheus_client import ( CollectorRegistry, multiprocess, @@ -58,30 +63,31 @@ async def metrics_handler(request): app.router.add_get("/metrics", metrics_handler) +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 ZMQ transport layer. + lives here; request_manager only provides the transport to the scheduler. """ - import json - 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 async def start_profile_handler(request): try: if request.content_length and request.content_length > 0: try: body = await request.json() - except (json.JSONDecodeError, Exception) as e: + except json.JSONDecodeError as e: return web.Response( status=400, text=f"Invalid JSON in request body: {e}", @@ -92,9 +98,7 @@ async def start_profile_handler(request): # 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 = ( - False if with_stack is False or env_with_stack is False else 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 @@ -116,16 +120,16 @@ async def start_profile_handler(request): results = await request_manager.send_communicator_req( req, "profile_communicator", timeout=600.0 ) - 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"Profile failed: {msgs}\n") + 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=str(e)) + return web.Response( + status=500, + text=f"Internal error: {type(e).__name__}. Check server logs.\n", + ) async def stop_profile_handler(request): try: @@ -133,16 +137,16 @@ async def stop_profile_handler(request): results = await request_manager.send_communicator_req( req, "profile_communicator", timeout=600.0 ) - 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"Stop profile failed: {msgs}\n") + 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=str(e)) + 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) @@ -160,12 +164,12 @@ async def serve_grpc(server_args, model_info=None): "version mismatch — see the chained exception above for details." ) from e - from aiohttp import web - sidecar_app = web.Application() sidecar_runner = None sidecar_port = ( - getattr(server_args, "grpc_http_sidecar_port", None) or server_args.port + 1 + 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