diff --git a/components/src/dynamo/common/metrics_relay.py b/components/src/dynamo/common/metrics_relay.py new file mode 100644 index 000000000000..849369dad872 --- /dev/null +++ b/components/src/dynamo/common/metrics_relay.py @@ -0,0 +1,116 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Async HTTP client for sending LLM metrics to the ss-agent metrics relay service. + +Reads METRICS_RELAY_ADDR env var (default: disabled). When set, emits TTFT, +input throughput, and output throughput with a streaming label via fire-and-forget +background tasks so the hot path is never blocked. + +Retry behaviour: up to _MAX_RETRIES additional attempts on connection errors or +5xx responses, with exponential back-off (_RETRY_DELAYS). 4xx responses are not +retried (client error — retrying would just fail again). +""" + +import asyncio +import logging +import os +from typing import Optional + +logger = logging.getLogger(__name__) + +_RELAY_ADDR_ENV = "METRICS_RELAY_ADDR" +_MAX_RETRIES = 3 +_RETRY_DELAYS = (0.1, 0.3, 0.9) # seconds before each successive retry + +_client: Optional["MetricsRelayClient"] = None + + +class MetricsRelayClient: + def __init__(self, relay_addr: str) -> None: + self._relay_addr = relay_addr.rstrip("/") + # Lazily created and reused; recreated if closed. + self._session: Optional[object] = None # aiohttp.ClientSession + + def _get_session(self) -> object: + import aiohttp + + if self._session is None or self._session.closed: # type: ignore[union-attr] + self._session = aiohttp.ClientSession() + return self._session + + async def _post_with_retry(self, path: str, payload: dict) -> None: + import aiohttp + + url = f"{self._relay_addr}{path}" + + for attempt in range(_MAX_RETRIES + 1): + if attempt > 0: + await asyncio.sleep(_RETRY_DELAYS[attempt - 1]) + + try: + session = self._get_session() + async with session.post( # type: ignore[union-attr] + url, + json=payload, + timeout=aiohttp.ClientTimeout(total=2.0), + ) as resp: + if resp.status < 500: + # 2xx/3xx = success; 4xx = client error (don't retry) + if resp.status >= 400: + logger.debug( + "metrics relay client error %d for %s", + resp.status, + path, + ) + return + # 5xx server error — fall through to retry + logger.debug( + "metrics relay server error %d (attempt %d/%d)", + resp.status, + attempt + 1, + _MAX_RETRIES + 1, + ) + except Exception as exc: + logger.debug( + "metrics relay post failed (attempt %d/%d): %s", + attempt + 1, + _MAX_RETRIES + 1, + exc, + ) + + logger.debug("metrics relay gave up after %d attempts for %s", _MAX_RETRIES + 1, path) + + def capture_generic_metric( + self, + metric_type: str, + deployment: str, + value: int, + streaming: bool, + ) -> None: + """Schedule a background task to send one metric sample; never blocks.""" + payload = { + "metric_type": metric_type, + "deployment": deployment, + "key": f"dynamo_frontend_{metric_type}_{deployment}", + "value": value, + "metadata": {"streaming": streaming}, + } + try: + asyncio.get_running_loop().create_task( + self._post_with_retry("/custom-metric", payload) + ) + except RuntimeError: + pass + + +def get_metrics_relay_client() -> Optional[MetricsRelayClient]: + """Return the singleton client, or None if METRICS_RELAY_ADDR is not set.""" + global _client + if _client is not None: + return _client + addr = os.environ.get(_RELAY_ADDR_ENV) + if not addr: + return None + _client = MetricsRelayClient(addr) + return _client diff --git a/components/src/dynamo/frontend/sglang_processor.py b/components/src/dynamo/frontend/sglang_processor.py index 985222fcc45c..886853cbb14b 100644 --- a/components/src/dynamo/frontend/sglang_processor.py +++ b/components/src/dynamo/frontend/sglang_processor.py @@ -22,6 +22,8 @@ from dynamo.llm import ModelCardInstanceId, PythonAsyncEngine, RoutedEngine, fetch_model from dynamo.llm.exceptions import InvalidArgument, Unknown +from dynamo.common.metrics_relay import get_metrics_relay_client + from .sglang_prepost import ( SglangStreamingPostProcessor, ToolCallParserType, @@ -466,6 +468,10 @@ async def _generate_and_stream( post_proc_total_ms = 0.0 created_ts = int(time.time()) stream_interval = self.stream_interval + t_start = time.monotonic() + t_first: float | None = None + total_output_tokens = 0 + last_usage: dict[str, Any] | None = None try: dynamo_stream = await self.routed_engine.generate( @@ -506,12 +512,14 @@ async def _generate_and_stream( break new_ids = engine_response["token_ids"] + total_output_tokens += len(new_ids) raw_finish = engine_response.get("finish_reason") finish_reason = _map_finish_reason(raw_finish) stop_reason = engine_response.get("stop_reason") if usage := engine_response.get("completion_usage"): pending_usage = usage + last_usage = usage pending_token_ids.extend(new_ids) @@ -549,6 +557,8 @@ async def _generate_and_stream( ): dynamo_out["nvext"] = {"stop_reason": stop_reason} + if t_first is None: + t_first = time.monotonic() yield dynamo_out pending_token_ids = [] @@ -562,6 +572,46 @@ async def _generate_and_stream( f"Error generating response for request {request_id}: {e}" ) from e finally: + metrics_client = get_metrics_relay_client() + if metrics_client is not None and t_first is not None: + t_end = time.monotonic() + streaming = bool(request.get("stream", False)) + deployment = request.get("model", "unknown") + ttft_sec = (t_first - t_start) if streaming else (t_end - t_start) + total_sec = max(t_end - t_start, 1e-9) + input_tokens = len(tokens) + output_tokens = ( + last_usage.get("completion_tokens", total_output_tokens) + if last_usage + else total_output_tokens + ) + input_denom = max(ttft_sec, 1e-9) if streaming else total_sec + output_denom = max(t_end - t_first, 1e-9) if streaming else total_sec + metrics_client.capture_generic_metric( + "ttft", deployment, int(ttft_sec * 1000), streaming + ) + if input_tokens > 0: + metrics_client.capture_generic_metric( + "input_throughput", + deployment, + int(input_tokens / input_denom), + streaming, + ) + if output_tokens > 0: + metrics_client.capture_generic_metric( + "output_throughput", + deployment, + int(output_tokens / output_denom), + streaming, + ) + total_tokens = input_tokens + output_tokens + if total_tokens > 0: + metrics_client.capture_generic_metric( + "total_throughput", + deployment, + int(total_tokens / total_sec), + streaming, + ) if self.debug_perf and token_count > 0: logger.info( "[perf] sglang stream done: request=%s tokens=%d " diff --git a/components/src/dynamo/frontend/vllm_processor.py b/components/src/dynamo/frontend/vllm_processor.py index 10c35e71d843..11198a72ea88 100644 --- a/components/src/dynamo/frontend/vllm_processor.py +++ b/components/src/dynamo/frontend/vllm_processor.py @@ -36,6 +36,8 @@ from dynamo.frontend.frontend_args import FrontendConfig from dynamo.llm import ModelCardInstanceId, PythonAsyncEngine, RoutedEngine, fetch_model +from dynamo.common.metrics_relay import get_metrics_relay_client + from .prepost import StreamingPostProcessor, preprocess_chat_request from .utils import ( extract_mm_urls, @@ -585,6 +587,11 @@ async def _generate_and_stream( output_request_ids[output_idx] = child_request_id registered_request_ids.append(child_request_id) + t_start = time.monotonic() + t_first: float | None = None + total_output_tokens = 0 + last_usage: dict[str, Any] | None = None + try: _inject_routing_metadata(dynamo_preproc, dynamo_preproc, mm_routing_info) with _nvtx.annotate("mm_frontend:routed_engine_generate", color="red"): @@ -620,6 +627,7 @@ async def _generate_and_stream( yield handle_engine_error(engine_response, request_id, logger) break + total_output_tokens += len(engine_response["token_ids"]) output_idx = engine_response.get("index", 0) or 0 output_request_id = output_request_ids.get(output_idx) if output_request_id is None: @@ -692,13 +700,56 @@ async def _generate_and_stream( } if usage := engine_response.get("completion_usage"): dynamo_out["usage"] = usage + last_usage = usage + if t_first is None: + t_first = time.monotonic() yield dynamo_out _nvtx.end_range(rng_stream) except Exception as e: logger.exception("Error generating response for request %s", request_id) yield make_internal_error(request_id, str(e)) finally: + metrics_client = get_metrics_relay_client() + if metrics_client is not None and t_first is not None: + t_end = time.monotonic() + streaming = bool(request.get("stream", False)) + deployment = request.get("model", "unknown") + ttft_sec = (t_first - t_start) if streaming else (t_end - t_start) + total_sec = max(t_end - t_start, 1e-9) + input_tokens = len(tokens) + output_tokens = ( + last_usage.get("completion_tokens", total_output_tokens) + if last_usage + else total_output_tokens + ) + input_denom = max(ttft_sec, 1e-9) if streaming else total_sec + output_denom = max(t_end - t_first, 1e-9) if streaming else total_sec + metrics_client.capture_generic_metric( + "ttft", deployment, int(ttft_sec * 1000), streaming + ) + if input_tokens > 0: + metrics_client.capture_generic_metric( + "input_throughput", + deployment, + int(input_tokens / input_denom), + streaming, + ) + if output_tokens > 0: + metrics_client.capture_generic_metric( + "output_throughput", + deployment, + int(output_tokens / output_denom), + streaming, + ) + total_tokens = input_tokens + output_tokens + if total_tokens > 0: + metrics_client.capture_generic_metric( + "total_throughput", + deployment, + int(total_tokens / total_sec), + streaming, + ) for output_request_id in registered_request_ids: if output_request_id in self.output_processor.request_states: self.output_processor.abort_requests(