Adding support of relay metrics

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

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 "

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,
@@ -581,6 +583,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"):
@@ -616,6 +623,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:
@@ -688,13 +696,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(