[Observability] Add HTTP sidecar endpoints and FlushCache gRPC RPC for gRPC mode

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 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 _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,
                 )

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 and profiling endpoints. "
+            "Defaults to --port + 1. Not used in HTTP mode.",
         )
         parser.add_argument(
             "--enable-mfu-metrics",