[Observability] Add HTTP sidecar endpoints and FlushCache gRPC RPC for gRPC mode (#22500)
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Kangyan-Zhou
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
3c5b1f0810
commit
f1a70b4666
@@ -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,
|
||||
)
|
||||
|
||||
@@ -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",
|
||||
|
||||
Reference in New Issue
Block a user