From 67c6f1329262a82455f8e328ecf9a6548c631d97 Mon Sep 17 00:00:00 2001 From: nnshah1 Date: Wed, 1 Apr 2026 06:16:48 -0700 Subject: [PATCH 1/4] =?UTF-8?q?feat:=20trace=20infrastructure=20=E2=80=94?= =?UTF-8?q?=20spans,=20targets,=20request=20ID=20propagation?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Rename make_request_span → make_inference_request_span with target: "request_span" (always on via filter directive) - Add make_system_request_span with target: "system_span" (debug level) - Add "request_span=trace" directive in filters() - Simplify get_or_create_request_id() — validates UUID, returns Result - Update worker spans to target: "request_span" - Worker system_status_server uses make_system_request_span Co-Authored-By: Claude Opus 4.6 (1M context) --- lib/llm/src/http/service/anthropic.rs | 3 +- lib/llm/src/http/service/openai.rs | 117 +++++++++++++----- lib/llm/src/http/service/service_v2.rs | 4 +- lib/runtime/src/logging.rs | 46 ++++++- .../pipeline/network/ingress/push_endpoint.rs | 2 +- lib/runtime/src/system_status_server.rs | 4 +- 6 files changed, 133 insertions(+), 43 deletions(-) diff --git a/lib/llm/src/http/service/anthropic.rs b/lib/llm/src/http/service/anthropic.rs index 15e798cd8c04..5dc95c709085 100644 --- a/lib/llm/src/http/service/anthropic.rs +++ b/lib/llm/src/http/service/anthropic.rs @@ -123,7 +123,8 @@ async fn handler_anthropic_messages( } // Create request context - let request_id = get_or_create_request_id(None, &headers); + let request_id = get_or_create_request_id(&headers) + .map_err(|msg| anthropic_error(StatusCode::BAD_REQUEST, "invalid_request_error", &msg))?; let streaming = request.stream; let cancellation_labels = CancellationLabels { model: request.model.clone(), diff --git a/lib/llm/src/http/service/openai.rs b/lib/llm/src/http/service/openai.rs index 6d462e301bb1..e2c3d026b85b 100644 --- a/lib/llm/src/http/service/openai.rs +++ b/lib/llm/src/http/service/openai.rs @@ -290,37 +290,46 @@ pub async fn smart_json_error_middleware(request: Request, next: Next) -> } } -/// Get the request ID from a primary source, or next from the headers, or lastly create a new one if not present -// TODO: Similar function exists in lib/llm/src/grpc/service/openai.rs but with different signature and simpler logic -pub(super) fn get_or_create_request_id(primary: Option<&str>, headers: &HeaderMap) -> String { - // Try to get request id from trace context +/// Validate the `x-dynamo-request-id` header and return the request ID. +/// +/// Returns `Err(message)` if the header is present but invalid (not UTF-8 or not a UUID). +/// The caller is responsible for converting the error message into the appropriate HTTP +/// error format (OpenAI vs Anthropic). +/// +/// The request ID comes from the trace context — `make_inference_request_span()` guarantees it by +/// generating a UUID when the client doesn't provide one. +pub(super) fn get_or_create_request_id(headers: &HeaderMap) -> Result { + // Validate and extract x-dynamo-request-id header if present. + // Returns error for non-UTF-8 or non-UUID values. + let validated_header = if let Some(raw) = headers.get(DYNAMO_REQUEST_ID_HEADER) { + match raw.to_str() { + Err(_) => { + return Err(format!( + "{}{} header must be a valid UTF-8 string", + VALIDATION_PREFIX, DYNAMO_REQUEST_ID_HEADER + )); + } + Ok(s) if uuid::Uuid::parse_str(s).is_err() => { + return Err(format!( + "{}{} header must be a valid UUID, got: {}", + VALIDATION_PREFIX, DYNAMO_REQUEST_ID_HEADER, s + )); + } + Ok(s) => Some(s.to_string()), + } + } else { + None + }; + + // Prefer trace context (set by make_inference_request_span via DistributedTraceIdLayer) if let Some(trace_context) = get_distributed_tracing_context() && let Some(x_dynamo_request_id) = trace_context.x_dynamo_request_id { - return x_dynamo_request_id; - } - - // Try to get the request ID from the primary source - if let Some(primary) = primary - && let Ok(uuid) = uuid::Uuid::parse_str(primary) - { - return uuid.to_string(); + return Ok(x_dynamo_request_id); } - // Try to get the request ID header as a string slice - let request_id_opt = headers - .get(DYNAMO_REQUEST_ID_HEADER) - .and_then(|h| h.to_str().ok()); - - // Try to parse the request ID as a UUID, or generate a new one if missing/invalid - let uuid = match request_id_opt { - Some(request_id) => { - uuid::Uuid::parse_str(request_id).unwrap_or_else(|_| uuid::Uuid::new_v4()) - } - None => uuid::Uuid::new_v4(), - }; - - uuid.to_string() + // Fallback: use validated header value, or generate new UUID + Ok(validated_header.unwrap_or_else(|| uuid::Uuid::new_v4().to_string())) } /// OpenAI Completions Request Handler @@ -342,7 +351,12 @@ async fn handler_completions( request.nvext = apply_header_routing_overrides(request.nvext.take(), &headers); // create the context for the request - let request_id = get_or_create_request_id(request.inner.user.as_deref(), &headers); + let request_id = get_or_create_request_id(&headers).map_err(|msg| { + ErrorMessage::from_http_error(HttpError { + code: 400, + message: msg, + }) + })?; let streaming = request.inner.stream.unwrap_or(false); let cancellation_labels = CancellationLabels { model: request.inner.model.clone(), @@ -722,7 +736,12 @@ async fn embeddings( // return a 503 if the service is not ready check_ready(&state)?; - let request_id = get_or_create_request_id(request.inner.user.as_deref(), &headers); + let request_id = get_or_create_request_id(&headers).map_err(|msg| { + ErrorMessage::from_http_error(HttpError { + code: 400, + message: msg, + }) + })?; let request = Context::with_id(request, request_id); let request_id = request.id().to_string(); @@ -800,7 +819,12 @@ async fn handler_chat_completions( request.nvext = apply_header_routing_overrides(request.nvext.take(), &headers); // create the context for the request - let request_id = get_or_create_request_id(request.inner.user.as_deref(), &headers); + let request_id = get_or_create_request_id(&headers).map_err(|msg| { + ErrorMessage::from_http_error(HttpError { + code: 400, + message: msg, + }) + })?; let streaming = request.inner.stream.unwrap_or(false); let cancellation_labels = CancellationLabels { model: request.inner.model.clone(), @@ -1409,7 +1433,12 @@ async fn handler_responses( request.nvext = apply_header_routing_overrides(request.nvext.take(), &headers); // create the context for the request - let request_id = get_or_create_request_id(None, &headers); + let request_id = get_or_create_request_id(&headers).map_err(|msg| { + ErrorMessage::from_http_error(HttpError { + code: 400, + message: msg, + }) + })?; let streaming = request.inner.stream.unwrap_or(false); let cancellation_labels = CancellationLabels { model: request.inner.model.clone().unwrap_or_default(), @@ -1893,7 +1922,12 @@ async fn images( // return a 503 if the service is not ready check_ready(&state)?; - let request_id = get_or_create_request_id(request.inner.user.as_deref(), &headers); + let request_id = get_or_create_request_id(&headers).map_err(|msg| { + ErrorMessage::from_http_error(HttpError { + code: 400, + message: msg, + }) + })?; let request = Context::with_id(request, request_id); let request_id = request.id().to_string(); @@ -1986,7 +2020,12 @@ async fn videos( // return a 503 if the service is not ready check_ready(&state)?; - let request_id = get_or_create_request_id(request.user.as_deref(), &headers); + let request_id = get_or_create_request_id(&headers).map_err(|msg| { + ErrorMessage::from_http_error(HttpError { + code: 400, + message: msg, + }) + })?; let request = Context::with_id(request, request_id); let request_id = request.id().to_string(); @@ -2057,7 +2096,12 @@ async fn video_stream( ) -> Result { check_ready(&state)?; - let request_id = get_or_create_request_id(request.user.as_deref(), &headers); + let request_id = get_or_create_request_id(&headers).map_err(|msg| { + ErrorMessage::from_http_error(HttpError { + code: 400, + message: msg, + }) + })?; let request = Context::with_id(request, request_id); let model = request.model.clone(); @@ -2211,7 +2255,12 @@ async fn audio_speech( check_ready(&state)?; let response_format = request.response_format.clone(); - let request_id = get_or_create_request_id(request.user.as_deref(), &headers); + let request_id = get_or_create_request_id(&headers).map_err(|msg| { + ErrorMessage::from_http_error(HttpError { + code: 400, + message: msg, + }) + })?; let request = Context::with_id(request, request_id); let request_id = request.id().to_string(); diff --git a/lib/llm/src/http/service/service_v2.rs b/lib/llm/src/http/service/service_v2.rs index 7872a0005218..ae970a49c65e 100644 --- a/lib/llm/src/http/service/service_v2.rs +++ b/lib/llm/src/http/service/service_v2.rs @@ -28,7 +28,7 @@ use derive_builder::Builder; use dynamo_runtime::config::env_is_truthy; use dynamo_runtime::config::environment_names::llm as env_llm; use dynamo_runtime::discovery::Discovery; -use dynamo_runtime::logging::make_request_span; +use dynamo_runtime::logging::make_inference_request_span; use dynamo_runtime::metrics::{ frontend_perf::ensure_frontend_perf_metrics_registered_prometheus, request_plane::ensure_request_plane_metrics_registered_prometheus, @@ -526,7 +526,7 @@ impl HttpServiceConfigBuilder { // Add on_response callback for logging response status code router = router.layer( TraceLayer::new_for_http() - .make_span_with(make_request_span) + .make_span_with(make_inference_request_span) .on_response( |response: &Response, latency: Duration, _span: &tracing::Span| { let status = response.status(); diff --git a/lib/runtime/src/logging.rs b/lib/runtime/src/logging.rs index f2497fc78ab4..6df80bdc5264 100644 --- a/lib/runtime/src/logging.rs +++ b/lib/runtime/src/logging.rs @@ -288,8 +288,12 @@ impl TraceParent { } } -// Takes Axum request and returning a span -pub fn make_request_span(req: &Request) -> Span { +/// Create a span for inference request endpoints (completions, chat, embeddings, etc.). +/// +/// Uses `target: "request_span"` which is always allowed through the DYN_LOG filter +/// (via `request_span=trace` directive in `filters()`). This ensures request context +/// (request_id, model, trace_id) is always available on log events. +pub fn make_inference_request_span(req: &Request) -> Span { let method = req.method(); let uri = req.uri(); let version = format!("{:?}", req.version()); @@ -297,7 +301,15 @@ pub fn make_request_span(req: &Request) -> Span { let otel_context = extract_otel_context_from_http_headers(req.headers()); + // Generate a request ID if the client didn't provide one so that workers + // (which read x_dynamo_request_id from the span/trace context) always + // have a consistent ID to correlate with. + let x_dynamo_request_id = trace_parent + .x_dynamo_request_id + .unwrap_or_else(|| Uuid::new_v4().to_string()); + let span = tracing::info_span!( + target: "request_span", "http-request", method = %method, uri = %uri, @@ -305,7 +317,10 @@ pub fn make_request_span(req: &Request) -> Span { trace_id = trace_parent.trace_id, parent_id = trace_parent.parent_id, x_request_id = trace_parent.x_request_id, - x_dynamo_request_id = trace_parent.x_dynamo_request_id, + x_dynamo_request_id = %x_dynamo_request_id, + model = tracing::field::Empty, + input_tokens = tracing::field::Empty, + output_tokens = tracing::field::Empty, ); if let Some(context) = otel_context { @@ -315,6 +330,21 @@ pub fn make_request_span(req: &Request) -> Span { span } +/// Create a span for system endpoints (health, metrics, models, etc.). +/// +/// Uses `target: "system_span"` which follows normal DYN_LOG filtering — these +/// endpoints are polled frequently and don't need to be visible at INFO level. +pub fn make_system_request_span(req: &Request) -> Span { + let method = req.method(); + let uri = req.uri(); + tracing::debug_span!( + target: "system_span", + "http-request", + method = %method, + uri = %uri, + ) +} + /// Extract OpenTelemetry context from HTTP headers for distributed tracing fn extract_otel_context_from_http_headers( headers: &http::HeaderMap, @@ -364,6 +394,7 @@ pub fn make_handle_payload_span( if let (Some(trace_id), Some(parent_id)) = (trace_id.as_ref(), parent_span_id.as_ref()) { let span = tracing::info_span!( + target: "request_span", "handle_payload", trace_id = trace_id.as_str(), parent_id = parent_id.as_str(), @@ -382,6 +413,7 @@ pub fn make_handle_payload_span( span } else { tracing::info_span!( + target: "request_span", "handle_payload", x_request_id = trace_parent.x_request_id, x_dynamo_request_id = trace_parent.x_dynamo_request_id, @@ -409,6 +441,7 @@ pub fn make_handle_payload_span_from_tcp_headers( if let (Some(trace_id), Some(parent_id)) = (trace_id.as_ref(), parent_span_id.as_ref()) { let span = tracing::info_span!( + target: "request_span", "handle_payload", trace_id = trace_id.as_str(), parent_id = parent_id.as_str(), @@ -427,6 +460,7 @@ pub fn make_handle_payload_span_from_tcp_headers( span } else { tracing::info_span!( + target: "request_span", "handle_payload", x_request_id = x_request_id, x_dynamo_request_id = x_dynamo_request_id, @@ -1043,6 +1077,12 @@ fn filters(config: LoggingConfig) -> EnvFilter { filter_layer = filter_layer.add_directive("span_event=trace".parse().unwrap()); } + // Always allow infrastructure request spans regardless of DYN_LOG level. + // This ensures request context (request_id, model, trace_id) is always + // available on log events, even when DYN_LOG=error or DYN_LOG=warn. + // Can be overridden via DYN_LOG=request_span= if needed. + filter_layer = filter_layer.add_directive("request_span=trace".parse().unwrap()); + filter_layer } diff --git a/lib/runtime/src/pipeline/network/ingress/push_endpoint.rs b/lib/runtime/src/pipeline/network/ingress/push_endpoint.rs index 9a910d7caf4d..4d192f2f9c50 100644 --- a/lib/runtime/src/pipeline/network/ingress/push_endpoint.rs +++ b/lib/runtime/src/pipeline/network/ingress/push_endpoint.rs @@ -102,7 +102,7 @@ impl PushEndpoint { instance_id, ) } else { - tracing::info_span!("handle_payload") + tracing::info_span!(target: "request_span", "handle_payload") }; tokio::spawn(async move { diff --git a/lib/runtime/src/system_status_server.rs b/lib/runtime/src/system_status_server.rs index c46f59ed9b68..573a204f5398 100644 --- a/lib/runtime/src/system_status_server.rs +++ b/lib/runtime/src/system_status_server.rs @@ -8,7 +8,7 @@ use crate::config::HealthStatus; use crate::config::environment_names::logging as env_logging; use crate::config::environment_names::runtime::canary as env_canary; use crate::config::environment_names::runtime::system as env_system; -use crate::logging::make_request_span; +use crate::logging::make_system_request_span; use crate::metrics::MetricsHierarchy; use crate::traits::DistributedRuntimeProvider; use axum::{ @@ -221,7 +221,7 @@ pub async fn spawn_system_status_server( tracing::info!("[fallback handler] called"); (StatusCode::NOT_FOUND, "Route not found").into_response() }) - .layer(TraceLayer::new_for_http().make_span_with(make_request_span)); + .layer(TraceLayer::new_for_http().make_span_with(make_system_request_span)); let address = format!("{}:{}", host, port); tracing::info!("[spawn_system_status_server] binding to: {address}"); From f649905fbfae8f9cbfad4637033331d3c5b3f1c1 Mon Sep 17 00:00:00 2001 From: nnshah1 Date: Wed, 1 Apr 2026 06:16:48 -0700 Subject: [PATCH 2/4] =?UTF-8?q?feat:=20trace=20infrastructure=20=E2=80=94?= =?UTF-8?q?=20spans,=20targets,=20request=20ID=20propagation?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Rename make_request_span → make_inference_request_span with target: "request_span" (always on via filter directive) - Add make_system_request_span with target: "system_span" (debug level) - Add "request_span=trace" directive in filters() - Simplify get_or_create_request_id() — validates UUID, returns Result - Update worker spans to target: "request_span" - Worker system_status_server uses make_system_request_span Co-Authored-By: Claude Opus 4.6 (1M context) --- .../DIS-1643-consistent-error-tracing.md | 176 ++++++++++++++++++ 1 file changed, 176 insertions(+) create mode 100644 docs/specs/DIS-1643-consistent-error-tracing.md diff --git a/docs/specs/DIS-1643-consistent-error-tracing.md b/docs/specs/DIS-1643-consistent-error-tracing.md new file mode 100644 index 000000000000..f8f20bab3342 --- /dev/null +++ b/docs/specs/DIS-1643-consistent-error-tracing.md @@ -0,0 +1,176 @@ +# DIS-1643: Consistent Error Tracing — Implementation Plan + +## Overview + +4 stacked PRs implementing consistent structured logging across all frontend error paths and worker trace propagation. + +## PR1: Trace Infrastructure (#7733) + +**Problem:** Request spans use `info_span!` which gets filtered out by `DYN_LOG=warn/error`, losing request context on error logs. No UUID validation on `x-dynamo-request-id`. System and inference endpoints share the same span/log treatment. + +**Solution:** +- Custom span targets: `request_span` (inference, always on via `request_span=trace` directive) and `system_span` (health/metrics, debug level) +- `make_inference_request_span()` generates UUID when client doesn't provide one — captured by `DistributedTraceIdLayer` and propagated to workers +- `get_or_create_request_id()` returns `Result` — validates UUID header, callers format errors for their API (OpenAI via `from_http_error`, Anthropic via `anthropic_error`) +- Router split in `service_v2.rs`: system routes get `make_system_request_span`, inference routes get `make_inference_request_span` + +**Files:** `logging.rs`, `openai.rs`, `anthropic.rs`, `service_v2.rs`, `system_status_server.rs`, `push_endpoint.rs` + +## PR2: Request Lifecycle Logging (#7734) + +**Problem:** No consistent "request received" / "request completed" log. Error paths have different log formats. Worker has no request-level logs at INFO. + +**Solution:** +- `InflightGuard` is single source of truth: logs "request received" (INFO) on creation, "request completed" (INFO/ERROR) on Drop +- `create_inflight_guard()` takes `request_id`, records `model` on span — no separate setup calls +- All inference errors at ERROR level with `status=error`, `error_type`, `error_detail` +- `on_response` renamed to "http response sent" — system at DEBUG, inference at INFO/ERROR +- Worker `push_handler.rs` logs "request received" / "request completed" at INFO + +**Files:** `metrics.rs`, `openai.rs`, `anthropic.rs`, `service_v2.rs`, `push_handler.rs`, `grpc/{openai,tensor}.rs` + +## PR3: Token Counts, TTFT, ITL, Worker IDs (#7735) + +**Problem:** No token counts, latency metrics, or worker identification on request completion logs. + +**Solution:** +- `ResponseMetricCollector::Drop` records on span: `input_tokens`, `output_tokens`, `ttft_ms`, `avg_itl_ms`, `prefill_worker_id`, `decode_worker_id` +- TTFT stored from already-computed value; ITL accumulated from per-chunk computation +- WARN log at cancellation point in `disconnect.rs` +- Connection monitor upgraded from TRACE to WARN + +**Performance:** All additions are on cleanup path (Drop), not streaming hot path. Two f64/u64 accumulations per chunk (negligible alongside existing histogram publish). + +**Files:** `metrics.rs`, `disconnect.rs`, `logging.rs` (Empty fields) + +## PR4: E2E Tests (#7766) + +11 parallel-safe pytest tests (25s with `-n auto`): + +| Category | Tests | +|----------|-------| +| Aggregated success | unary, streaming, request ID propagation | +| Aggregated errors | 404, 400 invalid UUID, cancellation, worker crash | +| Disaggregated success | unary, streaming (both workers verified) | +| Disaggregated crashes | prefill crash, decode crash | + +## Request ID Propagation Flow + +``` +Client → x-dynamo-request-id header (optional, 400 if invalid) + ↓ +make_inference_request_span() → generates UUID if absent + ↓ +DistributedTraceIdLayer::on_new_span() → captures into trace context + ↓ +get_or_create_request_id() → validates + reads from trace context + ↓ +create_inflight_guard() → logs "request received", records model on span + ↓ +addressed_router.rs → inject_trace_headers_into_map() → worker gets UUID + ↓ +Worker span → x_dynamo_request_id, trace_id correlation + ↓ +InflightGuard::Drop → logs "request completed" with all fields +``` + +## Log Messages + +| Message | When | Success | Error | +|---|---|---|---| +| "request received" | Request starts | INFO | INFO | +| "http response sent" | HTTP headers sent | INFO | ERROR | +| "request completed" | Request fully done | INFO | ERROR | +| "request cancelled by client" | Client disconnect | — | WARN | + +## Error Details + +| error_type | error_detail | +|---|---| +| cancelled | client disconnected before completion | +| internal | internal server error during processing | +| validation | invalid request parameters | +| not_found | model or resource not found | +| overload | service overloaded or rate limited | +| not_implemented | requested feature not implemented | + +## "request completed" Fields (streaming) + +```json +{ + "level": "INFO", + "message": "request completed", + "status": "success", + "request_id": "32691d61-...", + "model": "qwen/qwen3-0.6b", + "endpoint": "chat_completions", + "request_type": "stream", + "elapsed_ms": "20", + "input_tokens": "9", + "output_tokens": "50", + "ttft_ms": "5.85", + "avg_itl_ms": "0.29", + "trace_id": "bca97f5e..." +} +``` + +## Before / After + +### Before — Cancellation +``` +(nothing — only a metric counter increment) +``` + +### Before — HTTP 500 +``` +request completed with server error status=500 latency_ms=1508 +``` + +### Before — HTTP 400 +``` +request completed with client request error status=400 latency_ms=23 +``` + +### After — Streaming Success +```json +{"level":"INFO","message":"request received","request_id":"284f18c7-...","model":"qwen/qwen3-0.6b","endpoint":"chat_completions","request_type":"stream"} +{"level":"INFO","message":"http response sent","status":"200","latency_ms":"4"} +{"level":"INFO","message":"request completed","status":"success","elapsed_ms":"14","input_tokens":"9","output_tokens":"50","ttft_ms":"5.85","avg_itl_ms":"0.29"} +``` + +### After — 404 Error +```json +{"level":"INFO","message":"request received","request_id":"4644979b-...","model":"nonexistent-model"} +{"level":"ERROR","message":"request completed","status":"error","error_type":"not_found","error_detail":"model or resource not found","elapsed_ms":"0"} +{"level":"ERROR","message":"http response sent","status":"404","latency_ms":"0"} +``` + +### After — Worker Crash (partial generation) +```json +{"level":"ERROR","message":"request completed","status":"error","error_type":"internal","error_detail":"internal server error during processing","elapsed_ms":"556","input_tokens":"9","output_tokens":"4"} +``` + +### After — Cancellation +```json +{"level":"ERROR","message":"request completed","status":"error","error_type":"cancelled","error_detail":"client disconnected before completion","elapsed_ms":"230"} +``` + +### After — Worker (trace_id correlation) +```json +{"level":"INFO","message":"request received","request_id":"6ec6ddd9-...","component":"backend","endpoint":"generate","instance_id":"2221573453717914121","trace_id":"3fe59d20..."} +{"level":"INFO","message":"request completed","request_id":"6ec6ddd9-...","trace_id":"3fe59d20..."} +``` + +## Span Targets + +| Target | Level | Always on? | Used for | +|---|---|---|---| +| `request_span` | info | Yes (`request_span=trace` in filters) | Inference HTTP spans, worker payload spans | +| `system_span` | debug | No (follows DYN_LOG) | Health, metrics, models endpoints | + +Both overridable via `DYN_LOG=request_span=off` or `DYN_LOG=system_span=trace`. + +## Follow-up Issues + +- **DIS-1652** — Propagate model name to worker via transport headers +- **DIS-1653** — Token counts/TTFT missing on unary requests (collector Drop ordering) From ff214caeae8a804980cec332fa2cc0b764a80e46 Mon Sep 17 00:00:00 2001 From: nnshah1 Date: Wed, 1 Apr 2026 06:21:19 -0700 Subject: [PATCH 3/4] feat: request lifecycle logging via InflightGuard - InflightGuard logs "request received" (INFO) and "request completed" (INFO success, ERROR failure) with structured fields - Split service_v2 router into system/inference with separate TraceLayer - System endpoints: debug spans, inference: info spans with "http response sent" - Worker logs "request received"/"request completed" at INFO - All inference errors log at ERROR level Co-Authored-By: Claude Opus 4.6 (1M context) --- lib/llm/src/grpc/service/openai.rs | 10 +- lib/llm/src/grpc/service/tensor.rs | 10 +- lib/llm/src/http/service/anthropic.rs | 10 +- lib/llm/src/http/service/metrics.rs | 166 ++++++++++++++---- lib/llm/src/http/service/openai.rs | 79 +++++---- lib/llm/src/http/service/service_v2.rs | 93 +++++----- .../pipeline/network/ingress/push_handler.rs | 9 + 7 files changed, 253 insertions(+), 124 deletions(-) diff --git a/lib/llm/src/grpc/service/openai.rs b/lib/llm/src/grpc/service/openai.rs index 124c7737cc46..0f53f8330d9b 100644 --- a/lib/llm/src/grpc/service/openai.rs +++ b/lib/llm/src/grpc/service/openai.rs @@ -88,10 +88,12 @@ pub async fn completion_response_stream( let http_queue_guard = state.metrics_clone().create_http_queue_guard(model); - let inflight_guard = - state - .metrics_clone() - .create_inflight_guard(model, Endpoint::Completions, streaming); + let inflight_guard = state.metrics_clone().create_inflight_guard( + model, + Endpoint::Completions, + streaming, + request_id.clone(), + ); let mut response_collector = state.metrics_clone().create_response_collector(model); diff --git a/lib/llm/src/grpc/service/tensor.rs b/lib/llm/src/grpc/service/tensor.rs index f8dc97249101..c1a13f3971bf 100644 --- a/lib/llm/src/grpc/service/tensor.rs +++ b/lib/llm/src/grpc/service/tensor.rs @@ -89,10 +89,12 @@ pub async fn tensor_response_stream( let http_queue_guard = state.metrics_clone().create_http_queue_guard(model); - let inflight_guard = - state - .metrics_clone() - .create_inflight_guard(model, Endpoint::Tensor, streaming); + let inflight_guard = state.metrics_clone().create_inflight_guard( + model, + Endpoint::Tensor, + streaming, + request_id.clone(), + ); let mut response_collector = state.metrics_clone().create_response_collector(model); diff --git a/lib/llm/src/http/service/anthropic.rs b/lib/llm/src/http/service/anthropic.rs index 5dc95c709085..602320653441 100644 --- a/lib/llm/src/http/service/anthropic.rs +++ b/lib/llm/src/http/service/anthropic.rs @@ -300,10 +300,12 @@ async fn anthropic_messages( Box> + Send>, > = Box::pin(engine_stream); - let mut inflight_guard = - state - .metrics_clone() - .create_inflight_guard(&model, Endpoint::AnthropicMessages, streaming); + let mut inflight_guard = state.metrics_clone().create_inflight_guard( + &model, + Endpoint::AnthropicMessages, + streaming, + ctx.id().to_string(), + ); if streaming { stream_handle.arm(); diff --git a/lib/llm/src/http/service/metrics.rs b/lib/llm/src/http/service/metrics.rs index 3cb3baa5ca7b..12e40cb14590 100644 --- a/lib/llm/src/http/service/metrics.rs +++ b/lib/llm/src/http/service/metrics.rs @@ -283,6 +283,7 @@ pub struct InflightGuard { status: Status, error_type: ErrorType, timer: Instant, + request_id: String, } /// Requests will be logged by the type of endpoint hit @@ -371,6 +372,7 @@ pub struct ResponseMetricCollector { // be computed. last_response_time: Option, osl: usize, + isl: usize, // we track if cached_tokens has been observed to ensure we only increment once per request cached_tokens_observed: bool, // we track if tokenize latency has been observed to ensure we only increment once per request @@ -914,6 +916,7 @@ impl Metrics { model: &str, endpoint: Endpoint, streaming: bool, + request_id: String, ) -> InflightGuard { let request_type = if streaming { RequestType::Stream @@ -926,6 +929,7 @@ impl Metrics { model.to_string().to_lowercase(), endpoint, request_type, + request_id, ) } @@ -965,14 +969,22 @@ impl InflightGuard { model: String, endpoint: Endpoint, request_type: RequestType, + request_id: String, ) -> Self { - // Start the timer let timer = Instant::now(); - - // Increment the inflight gauge when the guard is created metrics.inc_inflight_gauge(&model); - // Return the RAII Guard + // Record model on the enclosing span so all logs inherit it + tracing::Span::current().record("model", model.as_str()); + + tracing::info!( + request_id = %request_id, + model = %model, + endpoint = %endpoint, + request_type = %request_type, + "request received" + ); + InflightGuard { metrics, model, @@ -981,9 +993,29 @@ impl InflightGuard { status: Status::Error, error_type: ErrorType::Internal, timer, + request_id, } } + pub fn request_id(&self) -> &str { + &self.request_id + } + pub fn model(&self) -> &str { + &self.model + } + pub fn endpoint(&self) -> &Endpoint { + &self.endpoint + } + pub fn request_type(&self) -> &RequestType { + &self.request_type + } + pub fn error_type(&self) -> &ErrorType { + &self.error_type + } + pub fn elapsed_ms(&self) -> u128 { + self.timer.elapsed().as_millis() + } + pub(crate) fn mark_ok(&mut self) { self.status = Status::Success; self.error_type = ErrorType::None; @@ -998,13 +1030,7 @@ impl InflightGuard { impl Drop for InflightGuard { fn drop(&mut self) { let duration = self.timer.elapsed().as_secs_f64(); - - // Decrement the gauge when the guard is dropped self.metrics.dec_inflight_gauge(&self.model); - - // the frequency on incrementing the full request counter is relatively low - // if we were incrementing the counter on every forward pass, we'd use static CounterVec or - // discrete counter object without the more costly lookup required for the following calls self.metrics.inc_request_counter( &self.model, &self.endpoint, @@ -1012,12 +1038,48 @@ impl Drop for InflightGuard { &self.status, &self.error_type, ); - - // Record the duration of the request self.metrics .request_duration .with_label_values(&[&self.model]) .observe(duration); + + let elapsed_ms = self.timer.elapsed().as_millis(); + let status_str = self.status.as_str(); + match self.status { + Status::Error => { + let detail = match self.error_type { + ErrorType::Cancelled => "client disconnected before completion", + ErrorType::Internal => "internal server error during processing", + ErrorType::Validation => "invalid request parameters", + ErrorType::NotFound => "model or resource not found", + ErrorType::Overload => "service overloaded or rate limited", + ErrorType::NotImplemented => "requested feature not implemented", + ErrorType::None => "unknown error", + }; + tracing::error!( + request_id = %self.request_id, + model = %self.model, + endpoint = %self.endpoint, + request_type = %self.request_type, + status = %status_str, + error_type = %self.error_type, + error_detail = %detail, + elapsed_ms = %elapsed_ms, + "request completed" + ); + } + Status::Success => { + tracing::info!( + request_id = %self.request_id, + model = %self.model, + endpoint = %self.endpoint, + request_type = %self.request_type, + status = %status_str, + elapsed_ms = %elapsed_ms, + "request completed" + ); + } + } } } @@ -1062,6 +1124,12 @@ impl RequestType { } } +impl std::fmt::Display for RequestType { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(self.as_str()) + } +} + impl Status { pub fn as_str(&self) -> &'static str { match self { @@ -1085,6 +1153,12 @@ impl ErrorType { } } +impl std::fmt::Display for ErrorType { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(self.as_str()) + } +} + impl ResponseMetricCollector { fn new(metrics: Arc, model: String) -> Self { ResponseMetricCollector { @@ -1094,6 +1168,7 @@ impl ResponseMetricCollector { last_response_time: None, start_time: Instant::now(), osl: 0, + isl: 0, cached_tokens_observed: false, tokenize_latency_observed: false, detokenize_latency_total: Duration::ZERO, @@ -2084,10 +2159,12 @@ mod tests { let model = "test-model"; { - let mut guard = - metrics - .clone() - .create_inflight_guard(model, Endpoint::ChatCompletions, false); + let mut guard = metrics.clone().create_inflight_guard( + model, + Endpoint::ChatCompletions, + false, + String::new(), + ); guard.mark_ok(); } // guard drops here @@ -2114,10 +2191,12 @@ mod tests { let model = "test-model"; { - let mut guard = - metrics - .clone() - .create_inflight_guard(model, Endpoint::ChatCompletions, false); + let mut guard = metrics.clone().create_inflight_guard( + model, + Endpoint::ChatCompletions, + false, + String::new(), + ); guard.mark_error(ErrorType::Validation); } // guard drops here @@ -2144,10 +2223,12 @@ mod tests { let model = "test-model"; { - let _guard = - metrics - .clone() - .create_inflight_guard(model, Endpoint::ChatCompletions, false); + let _guard = metrics.clone().create_inflight_guard( + model, + Endpoint::ChatCompletions, + false, + String::new(), + ); // Don't call mark_ok() or mark_error() - simulate panic/unhandled error } // guard drops with default error_type=Internal @@ -2185,9 +2266,10 @@ mod tests { ]; for error_type in &error_types { - let mut guard = metrics - .clone() - .create_inflight_guard(model, endpoint, false); + let mut guard = + metrics + .clone() + .create_inflight_guard(model, endpoint, false, String::new()); guard.mark_error(error_type.clone()); drop(guard); } @@ -2223,28 +2305,34 @@ mod tests { // Record 2 validation errors, 3 internal errors, 1 success for _ in 0..2 { - let mut guard = - metrics - .clone() - .create_inflight_guard(model, Endpoint::ChatCompletions, false); + let mut guard = metrics.clone().create_inflight_guard( + model, + Endpoint::ChatCompletions, + false, + String::new(), + ); guard.mark_error(ErrorType::Validation); drop(guard); } for _ in 0..3 { - let mut guard = - metrics - .clone() - .create_inflight_guard(model, Endpoint::Completions, false); + let mut guard = metrics.clone().create_inflight_guard( + model, + Endpoint::Completions, + false, + String::new(), + ); guard.mark_error(ErrorType::Internal); drop(guard); } { - let mut guard = - metrics - .clone() - .create_inflight_guard(model, Endpoint::Embeddings, false); + let mut guard = metrics.clone().create_inflight_guard( + model, + Endpoint::Embeddings, + false, + String::new(), + ); guard.mark_ok(); drop(guard); } diff --git a/lib/llm/src/http/service/openai.rs b/lib/llm/src/http/service/openai.rs index e2c3d026b85b..29fde16c1100 100644 --- a/lib/llm/src/http/service/openai.rs +++ b/lib/llm/src/http/service/openai.rs @@ -438,10 +438,12 @@ async fn completions_single( let model = request.inner.model.clone(); // Create inflight_guard early to ensure all errors are counted - let mut inflight_guard = - state - .metrics_clone() - .create_inflight_guard(&model, Endpoint::Completions, streaming); + let mut inflight_guard = state.metrics_clone().create_inflight_guard( + &model, + Endpoint::Completions, + streaming, + request_id.clone(), + ); // Create http_queue_guard early - tracks time waiting to be processed let http_queue_guard = state.metrics_clone().create_http_queue_guard(&model); @@ -572,10 +574,12 @@ async fn completions_batch( let model = request.inner.model.clone(); // Create inflight_guard early to ensure all errors are counted - let mut inflight_guard = - state - .metrics_clone() - .create_inflight_guard(&model, Endpoint::Completions, streaming); + let mut inflight_guard = state.metrics_clone().create_inflight_guard( + &model, + Endpoint::Completions, + streaming, + request_id.clone(), + ); // Create http_queue_guard early - tracks time waiting to be processed let http_queue_guard = state.metrics_clone().create_http_queue_guard(&model); @@ -753,10 +757,12 @@ async fn embeddings( let model = &request.inner.model; // Create inflight_guard early to ensure all errors are counted - let mut inflight = - state - .metrics_clone() - .create_inflight_guard(model, Endpoint::Embeddings, streaming); + let mut inflight = state.metrics_clone().create_inflight_guard( + model, + Endpoint::Embeddings, + streaming, + request_id.clone(), + ); // Create http_queue_guard early - tracks time waiting to be processed let http_queue_guard = state.metrics_clone().create_http_queue_guard(model); @@ -1129,10 +1135,12 @@ async fn chat_completions( tracing::trace!("Received chat completions request: {:?}", request.content()); // Create inflight_guard early to ensure all errors (including validation) are counted - let mut inflight_guard = - state - .metrics_clone() - .create_inflight_guard(&model, Endpoint::ChatCompletions, streaming); + let mut inflight_guard = state.metrics_clone().create_inflight_guard( + &model, + Endpoint::ChatCompletions, + streaming, + request_id.clone(), + ); // Handle unsupported fields - if Some(resp) is returned by // validate_chat_completion_unsupported_fields, @@ -1509,10 +1517,12 @@ async fn responses( // Create http_queue_guard early - tracks time waiting to be processed let http_queue_guard = state.metrics_clone().create_http_queue_guard(&model); - let mut inflight_guard = - state - .metrics_clone() - .create_inflight_guard(&model, Endpoint::Responses, streaming); + let mut inflight_guard = state.metrics_clone().create_inflight_guard( + &model, + Endpoint::Responses, + streaming, + request.id().to_string(), + ); // Handle unsupported fields - if Some(resp) is returned by validate_unsupported_fields, // then a field was used that is unsupported. We will log an error message @@ -1956,10 +1966,12 @@ async fn images( .map_err(|_| ErrorMessage::model_not_found())?; // this will increment the inflight gauge for the model - let mut inflight = - state - .metrics_clone() - .create_inflight_guard(&model, Endpoint::Images, streaming); + let mut inflight = state.metrics_clone().create_inflight_guard( + &model, + Endpoint::Images, + streaming, + request_id.clone(), + ); let mut response_collector = state.metrics_clone().create_response_collector(&model); @@ -2045,10 +2057,12 @@ async fn videos( .map_err(|_| ErrorMessage::model_not_found())?; // this will increment the inflight gauge for the model - let mut inflight = - state - .metrics_clone() - .create_inflight_guard(&model, Endpoint::Videos, streaming); + let mut inflight = state.metrics_clone().create_inflight_guard( + &model, + Endpoint::Videos, + streaming, + request_id.clone(), + ); let mut response_collector = state.metrics_clone().create_response_collector(&model); @@ -2112,9 +2126,12 @@ async fn video_stream( .get_videos_engine(&model) .map_err(|_| ErrorMessage::model_not_found())?; - let mut inflight = state - .metrics_clone() - .create_inflight_guard(&model, Endpoint::Videos, true); + let mut inflight = state.metrics_clone().create_inflight_guard( + &model, + Endpoint::Videos, + true, + request.id().to_string(), + ); let mut response_collector = state.metrics_clone().create_response_collector(&model); diff --git a/lib/llm/src/http/service/service_v2.rs b/lib/llm/src/http/service/service_v2.rs index ae970a49c65e..c4d82097772b 100644 --- a/lib/llm/src/http/service/service_v2.rs +++ b/lib/llm/src/http/service/service_v2.rs @@ -28,7 +28,7 @@ use derive_builder::Builder; use dynamo_runtime::config::env_is_truthy; use dynamo_runtime::config::environment_names::llm as env_llm; use dynamo_runtime::discovery::Discovery; -use dynamo_runtime::logging::make_inference_request_span; +use dynamo_runtime::logging::{make_inference_request_span, make_system_request_span}; use dynamo_runtime::metrics::{ frontend_perf::ensure_frontend_perf_metrics_registered_prometheus, request_plane::ensure_request_plane_metrics_registered_prometheus, @@ -491,11 +491,36 @@ impl HttpServiceConfigBuilder { tracing::warn!("Failed to register transport metrics: {}", e); } - let mut router = axum::Router::new(); - let mut all_docs = Vec::new(); - let mut routes = vec![ + // on_response callback shared by both system and inference TraceLayer + let on_response_system = |response: &Response, + latency: Duration, + _span: &tracing::Span| { + let status = response.status(); + let latency_ms = latency.as_millis(); + if status.is_server_error() { + tracing::error!(status = %status.as_u16(), latency_ms = %latency_ms, "http response sent"); + } else if status.is_client_error() { + tracing::warn!(status = %status.as_u16(), latency_ms = %latency_ms, "http response sent"); + } else { + tracing::debug!(status = %status.as_u16(), latency_ms = %latency_ms, "http response sent"); + } + }; + let on_response_inference = |response: &Response, + latency: Duration, + _span: &tracing::Span| { + let status = response.status(); + let latency_ms = latency.as_millis(); + if status.is_server_error() || status.is_client_error() { + tracing::error!(status = %status.as_u16(), latency_ms = %latency_ms, "http response sent"); + } else { + tracing::info!(status = %status.as_u16(), latency_ms = %latency_ms, "http response sent"); + } + }; + + // System routes (health, metrics, models) — debug-level spans + let system_routes = vec![ metrics::router( registry, var(HTTP_SVC_METRICS_PATH_ENV).ok(), @@ -506,54 +531,38 @@ impl HttpServiceConfigBuilder { super::health::live_check_router(state.clone(), var(HTTP_SVC_LIVE_PATH_ENV).ok()), super::busy_threshold::busy_threshold_router(state.clone(), None), ]; + let mut system_router = axum::Router::new(); + for (route_docs, route) in system_routes { + system_router = system_router.merge(route); + all_docs.extend(route_docs); + } + system_router = system_router.layer( + TraceLayer::new_for_http() + .make_span_with(make_system_request_span) + .on_response(on_response_system), + ); + // Inference routes (completions, chat, embeddings, etc.) — info-level spans let endpoint_routes = HttpServiceConfigBuilder::get_endpoints_router(state.clone(), &config.request_template); - routes.extend(endpoint_routes); - for (route_docs, route) in routes { - router = router.merge(route); + let mut inference_router = axum::Router::new(); + for (route_docs, route) in endpoint_routes { + inference_router = inference_router.merge(route); all_docs.extend(route_docs); } + inference_router = inference_router.layer( + TraceLayer::new_for_http() + .make_span_with(make_inference_request_span) + .on_response(on_response_inference), + ); - // Add OpenAPI documentation routes (must be after all other routes so it can document them) - // Note: The path parameter is currently unused as SwaggerUi requires static paths + // OpenAPI documentation routes (system) let (openapi_docs, openapi_route) = super::openapi_docs::openapi_router(all_docs.clone(), None); - router = router.merge(openapi_route); + system_router = system_router.merge(openapi_route); all_docs.extend(openapi_docs); - // Add span for tracing - // Add on_response callback for logging response status code - router = router.layer( - TraceLayer::new_for_http() - .make_span_with(make_inference_request_span) - .on_response( - |response: &Response, latency: Duration, _span: &tracing::Span| { - let status = response.status(); - let latency_ms = latency.as_millis(); - - if status.is_server_error() { - tracing::error!( - status = %status.as_u16(), - latency_ms = %latency_ms, - "request completed with server error" - ); - } else if status.is_client_error() { - tracing::warn!( - status = %status.as_u16(), - latency_ms = %latency_ms, - "request completed with client request error" - ); - } else { - tracing::debug!( - status = %status.as_u16(), - latency_ms = %latency_ms, - "request completed" - ); - } - }, - ), - ); + let router = system_router.merge(inference_router); Ok(HttpService { state, diff --git a/lib/runtime/src/pipeline/network/ingress/push_handler.rs b/lib/runtime/src/pipeline/network/ingress/push_handler.rs index fdacf587035a..b1c41269c307 100644 --- a/lib/runtime/src/pipeline/network/ingress/push_handler.rs +++ b/lib/runtime/src/pipeline/network/ingress/push_handler.rs @@ -219,6 +219,10 @@ where // extend request with context tracing::trace!("received control message: {:?}", control_msg); + tracing::info!( + request_id = %control_msg.id, + "request received" + ); tracing::trace!("received request: {:?}", request); let request: context::Context = Context::with_id(request, control_msg.id); @@ -359,6 +363,11 @@ where // Ensure the metrics guard is not dropped until the end of the function. drop(_inflight_guard); + tracing::info!( + request_id = %context.id(), + "request completed" + ); + Ok(()) } } From 0943ded8962319540d720670c91ba4043f2139e7 Mon Sep 17 00:00:00 2001 From: nnshah1 Date: Wed, 1 Apr 2026 06:58:16 -0700 Subject: [PATCH 4/4] feat: token counts, TTFT, ITL, worker IDs on span + cancellation logging - ResponseMetricCollector records on span in Drop: input_tokens, output_tokens, ttft_ms, avg_itl_ms, prefill_worker_id, decode_worker_id - Stores already-computed TTFT and accumulates ITL for average - WARN log at cancellation point with request context - Connection monitor disconnects upgraded from TRACE to WARN Co-Authored-By: Claude Opus 4.6 (1M context) --- lib/llm/src/http/service/disconnect.rs | 16 ++++++++--- lib/llm/src/http/service/metrics.rs | 37 +++++++++++++++++++++++++- lib/runtime/src/logging.rs | 4 +++ 3 files changed, 53 insertions(+), 4 deletions(-) diff --git a/lib/llm/src/http/service/disconnect.rs b/lib/llm/src/http/service/disconnect.rs index 1454fdb3382a..df9280ba582c 100644 --- a/lib/llm/src/http/service/disconnect.rs +++ b/lib/llm/src/http/service/disconnect.rs @@ -135,7 +135,7 @@ async fn connection_monitor( match connection_rx.await { Err(_) | Ok(ConnectionStatus::ClosedUnexpectedly) => { // the client has disconnected, no need to gracefully cancel, just kill the context - tracing::trace!("Connection closed unexpectedly; issuing cancellation"); + tracing::warn!("Connection closed unexpectedly; issuing cancellation"); if let Some(metrics) = &metrics { metrics.inc_client_disconnect(); metrics.inc_cancellation(&cancellation_labels); @@ -150,7 +150,7 @@ async fn connection_monitor( match stream_rx.await { Err(_) | Ok(ConnectionStatus::ClosedUnexpectedly) => { - tracing::trace!("Stream closed unexpectedly; issuing cancellation"); + tracing::warn!("Stream closed unexpectedly; issuing cancellation"); if let Some(metrics) = &metrics { metrics.inc_client_disconnect(); metrics.inc_cancellation(&cancellation_labels); @@ -211,9 +211,19 @@ pub fn monitor_for_disconnects( } } _ = context.stopped() => { - tracing::trace!("Context stopped; breaking stream"); // Mark as cancelled when context is stopped (client disconnect or timeout) inflight_guard.mark_error(ErrorType::Cancelled); + // Token counts (input_tokens, output_tokens) are recorded on + // the enclosing span by ResponseMetricCollector::Drop. + tracing::warn!( + request_id = %inflight_guard.request_id(), + model = %inflight_guard.model(), + endpoint = %inflight_guard.endpoint(), + request_type = %inflight_guard.request_type(), + error_type = "cancelled", + elapsed_ms = %inflight_guard.elapsed_ms(), + "request cancelled by client" + ); break; } } diff --git a/lib/llm/src/http/service/metrics.rs b/lib/llm/src/http/service/metrics.rs index 12e40cb14590..66ff9bfe1ec3 100644 --- a/lib/llm/src/http/service/metrics.rs +++ b/lib/llm/src/http/service/metrics.rs @@ -373,6 +373,9 @@ pub struct ResponseMetricCollector { last_response_time: Option, osl: usize, isl: usize, + ttft_ms: Option, + itl_sum_secs: f64, + itl_count: u64, // we track if cached_tokens has been observed to ensure we only increment once per request cached_tokens_observed: bool, // we track if tokenize latency has been observed to ensure we only increment once per request @@ -1169,6 +1172,9 @@ impl ResponseMetricCollector { start_time: Instant::now(), osl: 0, isl: 0, + ttft_ms: None, + itl_sum_secs: 0.0, + itl_count: 0, cached_tokens_observed: false, tokenize_latency_observed: false, detokenize_latency_total: Duration::ZERO, @@ -1269,6 +1275,9 @@ impl ResponseMetricCollector { return; } + // Store ISL for span recording on drop + self.isl = isl; + // Increment the real-time output tokens counter self.metrics .output_tokens_counter @@ -1280,8 +1289,9 @@ impl ResponseMetricCollector { // we use the full response time as TTFT and ignore the ITL self.is_first_token = false; - // Publish TTFT + // Publish TTFT and store for span recording let ttft = self.start_time.elapsed().as_secs_f64(); + self.ttft_ms = Some(ttft * 1000.0); self.metrics .time_to_first_token .with_label_values(&[&self.model]) @@ -1322,6 +1332,8 @@ impl ResponseMetricCollector { if let Some(last_response_time) = self.last_response_time { let response_duration = current_duration - last_response_time; let itl = response_duration.as_secs_f64() / num_tokens as f64; + self.itl_sum_secs += itl * num_tokens as f64; + self.itl_count += num_tokens as u64; for _ in 0..num_tokens { self.metrics .inter_token_latency @@ -1367,6 +1379,29 @@ impl Drop for ResponseMetricCollector { .output_sequence_length .with_label_values(&[&self.model]) .observe(self.osl as f64); + + // Record request summary on the enclosing span. + // InflightGuard::Drop and on_response logs will inherit these. + let span = tracing::Span::current(); + if self.isl > 0 { + span.record("input_tokens", self.isl as u32); + } + if self.osl > 0 { + span.record("output_tokens", self.osl as u32); + } + if let Some(ttft_ms) = self.ttft_ms { + span.record("ttft_ms", format!("{:.2}", ttft_ms).as_str()); + } + if self.itl_count > 0 { + let avg_ms = (self.itl_sum_secs / self.itl_count as f64) * 1000.0; + span.record("avg_itl_ms", format!("{:.2}", avg_ms).as_str()); + } + if let Some(worker_id) = self.prefill_worker_id { + span.record("prefill_worker_id", worker_id); + } + if let Some(worker_id) = self.decode_worker_id { + span.record("decode_worker_id", worker_id); + } } } diff --git a/lib/runtime/src/logging.rs b/lib/runtime/src/logging.rs index 6df80bdc5264..28f0ce8215f9 100644 --- a/lib/runtime/src/logging.rs +++ b/lib/runtime/src/logging.rs @@ -321,6 +321,10 @@ pub fn make_inference_request_span(req: &Request) -> Span { model = tracing::field::Empty, input_tokens = tracing::field::Empty, output_tokens = tracing::field::Empty, + ttft_ms = tracing::field::Empty, + avg_itl_ms = tracing::field::Empty, + prefill_worker_id = tracing::field::Empty, + decode_worker_id = tracing::field::Empty, ); if let Some(context) = otel_context {