From 81e22317f4c288e4cc6686562086fd882aa728b1 Mon Sep 17 00:00:00 2001 From: nnshah1 Date: Thu, 2 Apr 2026 08:40:21 -0700 Subject: [PATCH 1/7] feat: request lifecycle logging via InflightGuard Add "request received" (INFO) and "request completed" (INFO/ERROR) logs to InflightGuard with structured fields (request_id, model, endpoint, request_type, status, elapsed_ms). Add cancellation WARN in disconnect monitor. Add worker lifecycle logs in push_handler. All call sites now pass request_id to create_inflight_guard. 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/disconnect.rs | 16 +- lib/llm/src/http/service/metrics.rs | 169 ++++++++++++++---- lib/llm/src/http/service/openai.rs | 89 +++++---- lib/llm/tests/http_metrics.rs | 3 + .../pipeline/network/ingress/push_handler.rs | 9 + 8 files changed, 227 insertions(+), 89 deletions(-) diff --git a/lib/llm/src/grpc/service/openai.rs b/lib/llm/src/grpc/service/openai.rs index 9012538ae00a..b08b4b33ffa7 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 d6f56a5a75f4..9db582432622 100644 --- a/lib/llm/src/http/service/anthropic.rs +++ b/lib/llm/src/http/service/anthropic.rs @@ -305,10 +305,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/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 3cb3baa5ca7b..eb910c94df3a 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, @@ -1194,6 +1269,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 @@ -2084,10 +2162,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 +2194,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 +2226,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 +2269,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 +2308,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 b13def23dc4d..848627c52cf9 100644 --- a/lib/llm/src/http/service/openai.rs +++ b/lib/llm/src/http/service/openai.rs @@ -443,10 +443,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); @@ -577,10 +579,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); @@ -1124,10 +1130,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, @@ -1499,10 +1507,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 @@ -1949,10 +1959,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); @@ -2033,10 +2045,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); @@ -2095,9 +2109,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); @@ -2261,10 +2278,12 @@ async fn audio_speech( .get_audios_engine(&model) .map_err(|_| ErrorMessage::model_not_found())?; - let mut inflight = - state - .metrics_clone() - .create_inflight_guard(&model, Endpoint::Audios, streaming); + let mut inflight = state.metrics_clone().create_inflight_guard( + &model, + Endpoint::Audios, + streaming, + request_id.clone(), + ); let mut response_collector = state.metrics_clone().create_response_collector(&model); diff --git a/lib/llm/tests/http_metrics.rs b/lib/llm/tests/http_metrics.rs index d16417d98900..1ba059ff75f6 100644 --- a/lib/llm/tests/http_metrics.rs +++ b/lib/llm/tests/http_metrics.rs @@ -78,6 +78,7 @@ async fn test_metrics_prefix_default() { "test-model", Endpoint::ChatCompletions, false, + String::new(), ); } @@ -117,6 +118,7 @@ async fn test_metrics_prefix_custom() { "test-model", Endpoint::ChatCompletions, true, + String::new(), ); } @@ -151,6 +153,7 @@ async fn test_metrics_prefix_sanitized() { "test-model", Endpoint::ChatCompletions, true, + String::new(), ); } 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 187ada6705e2b56167e1088aec1e258b0b408ceb Mon Sep 17 00:00:00 2001 From: nnshah1 Date: Thu, 2 Apr 2026 08:41:42 -0700 Subject: [PATCH 2/7] feat: token counts, TTFT, ITL, worker IDs on span + cancellation logging Record input_tokens, output_tokens, ttft_ms, avg_itl_ms, prefill_worker_id, decode_worker_id on the enclosing tracing span via ResponseMetricCollector::Drop so they appear in JSONL logs alongside request lifecycle events. Co-Authored-By: Claude Opus 4.6 (1M context) --- lib/llm/src/http/service/metrics.rs | 34 ++++++++++++++++++++++++++++- 1 file changed, 33 insertions(+), 1 deletion(-) diff --git a/lib/llm/src/http/service/metrics.rs b/lib/llm/src/http/service/metrics.rs index eb910c94df3a..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, @@ -1283,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]) @@ -1325,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 @@ -1370,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); + } } } From 9121f9c5431288f14a89bce8a2d09cd092ca21aa Mon Sep 17 00:00:00 2001 From: nnshah1 Date: Thu, 2 Apr 2026 16:30:23 -0700 Subject: [PATCH 3/7] fix: always record token counts, use neutral cancellation wording Always record input_tokens/output_tokens on span (zero is meaningful, distinguishes "no tokens" from "field never set"). Change cancellation messages from "client disconnected" to "cancelled" since context.stopped() can also be triggered by server-side timeouts. Co-Authored-By: Claude Opus 4.6 (1M context) --- lib/llm/src/http/service/disconnect.rs | 2 +- lib/llm/src/http/service/metrics.rs | 10 +++------- 2 files changed, 4 insertions(+), 8 deletions(-) diff --git a/lib/llm/src/http/service/disconnect.rs b/lib/llm/src/http/service/disconnect.rs index df9280ba582c..a5c8dab2f725 100644 --- a/lib/llm/src/http/service/disconnect.rs +++ b/lib/llm/src/http/service/disconnect.rs @@ -222,7 +222,7 @@ pub fn monitor_for_disconnects( request_type = %inflight_guard.request_type(), error_type = "cancelled", elapsed_ms = %inflight_guard.elapsed_ms(), - "request cancelled by client" + "request cancelled" ); break; } diff --git a/lib/llm/src/http/service/metrics.rs b/lib/llm/src/http/service/metrics.rs index 66ff9bfe1ec3..b3b8214449ea 100644 --- a/lib/llm/src/http/service/metrics.rs +++ b/lib/llm/src/http/service/metrics.rs @@ -1051,7 +1051,7 @@ impl Drop for InflightGuard { match self.status { Status::Error => { let detail = match self.error_type { - ErrorType::Cancelled => "client disconnected before completion", + ErrorType::Cancelled => "cancelled before completion", ErrorType::Internal => "internal server error during processing", ErrorType::Validation => "invalid request parameters", ErrorType::NotFound => "model or resource not found", @@ -1383,12 +1383,8 @@ impl Drop for ResponseMetricCollector { // 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); - } + span.record("input_tokens", self.isl as u32); + 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()); } From a2cf91c143c2a939eca656963adfbe1b70df7d81 Mon Sep 17 00:00:00 2001 From: nnshah1 Date: Thu, 2 Apr 2026 16:50:31 -0700 Subject: [PATCH 4/7] fix: move InflightGuard before engine.generate() in anthropic handler Match the OpenAI handler pattern ("create inflight_guard early to ensure all errors are counted"). Previously the guard was created after engine.generate(), so backend failures were not tracked. Closes #7843 Co-Authored-By: Claude Opus 4.6 (1M context) --- lib/llm/src/http/service/anthropic.rs | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/lib/llm/src/http/service/anthropic.rs b/lib/llm/src/http/service/anthropic.rs index 9db582432622..83c01f1b706b 100644 --- a/lib/llm/src/http/service/anthropic.rs +++ b/lib/llm/src/http/service/anthropic.rs @@ -273,6 +273,14 @@ async fn anthropic_messages( let mut response_collector = state.metrics_clone().create_response_collector(&model); + // Create inflight_guard early to ensure all errors are counted + let mut inflight_guard = state.metrics_clone().create_inflight_guard( + &model, + Endpoint::AnthropicMessages, + streaming, + request.id().to_string(), + ); + tracing::trace!("Issuing generate call for Anthropic messages"); let engine_stream = engine.generate(request).await.map_err(|e| { @@ -305,13 +313,6 @@ async fn anthropic_messages( Box> + Send>, > = Box::pin(engine_stream); - let mut inflight_guard = state.metrics_clone().create_inflight_guard( - &model, - Endpoint::AnthropicMessages, - streaming, - ctx.id().to_string(), - ); - if streaming { stream_handle.arm(); From c368ef9b01d69f1b748b92b4c669bd87f74e5d61 Mon Sep 17 00:00:00 2001 From: nnshah1 Date: Thu, 2 Apr 2026 16:55:58 -0700 Subject: [PATCH 5/7] fix: move worker lifecycle logs into RequestMetricsGuard (RAII) Move "request received"/"request completed" logs from manual calls into RequestMetricsGuard via set_request_id() and Drop. This ensures "request completed" fires on all exit paths (errors, panics), matching the frontend's InflightGuard pattern. Closes #7844 Co-Authored-By: Claude Opus 4.6 (1M context) --- lib/runtime/src/pipeline/network.rs | 6 +++- .../pipeline/network/ingress/http_endpoint.rs | 2 +- .../pipeline/network/ingress/push_endpoint.rs | 15 +++++++++- .../pipeline/network/ingress/push_handler.rs | 28 +++++++++++-------- .../network/ingress/shared_tcp_endpoint.rs | 22 ++++++++++++--- 5 files changed, 55 insertions(+), 18 deletions(-) diff --git a/lib/runtime/src/pipeline/network.rs b/lib/runtime/src/pipeline/network.rs index 52657330629d..b93472abb9c6 100644 --- a/lib/runtime/src/pipeline/network.rs +++ b/lib/runtime/src/pipeline/network.rs @@ -358,7 +358,11 @@ impl Ingress { #[async_trait] pub trait PushWorkHandler: Send + Sync { - async fn handle_payload(&self, payload: Bytes) -> Result<(), PipelineError>; + async fn handle_payload( + &self, + payload: Bytes, + request_id: Option, + ) -> Result<(), PipelineError>; /// Add metrics to the handler fn add_metrics( diff --git a/lib/runtime/src/pipeline/network/ingress/http_endpoint.rs b/lib/runtime/src/pipeline/network/ingress/http_endpoint.rs index c299bb537778..5a4e849eb1ef 100644 --- a/lib/runtime/src/pipeline/network/ingress/http_endpoint.rs +++ b/lib/runtime/src/pipeline/network/ingress/http_endpoint.rs @@ -253,7 +253,7 @@ async fn handle_shared_request( tokio::spawn(async move { tracing::trace!(instance_id, "handling new HTTP request"); let result = service_handler - .handle_payload(body) + .handle_payload(body, traceparent.request_id.clone()) .instrument(tracing::info_span!( "handle_payload", component = component_name.as_ref(), diff --git a/lib/runtime/src/pipeline/network/ingress/push_endpoint.rs b/lib/runtime/src/pipeline/network/ingress/push_endpoint.rs index 4d192f2f9c50..e1cd7e1818a4 100644 --- a/lib/runtime/src/pipeline/network/ingress/push_endpoint.rs +++ b/lib/runtime/src/pipeline/network/ingress/push_endpoint.rs @@ -105,10 +105,23 @@ impl PushEndpoint { tracing::info_span!(target: "request_span", "handle_payload") }; + // Extract request_id from headers before passing payload + let request_id = req + .message + .headers + .as_ref() + .and_then(|h| h.get("request-id").map(|v| v.to_string())) + .or_else(|| { + req.message + .headers + .as_ref() + .and_then(|h| h.get("x-dynamo-request-id").map(|v| v.to_string())) + }); + tokio::spawn(async move { tracing::trace!(instance_id, "handling new request"); let result = ingress - .handle_payload(req.message.payload) + .handle_payload(req.message.payload, request_id) .instrument(span) .await; match result { diff --git a/lib/runtime/src/pipeline/network/ingress/push_handler.rs b/lib/runtime/src/pipeline/network/ingress/push_handler.rs index b1c41269c307..cb5f45898eaa 100644 --- a/lib/runtime/src/pipeline/network/ingress/push_handler.rs +++ b/lib/runtime/src/pipeline/network/ingress/push_handler.rs @@ -111,17 +111,23 @@ impl WorkHandlerMetrics { } } -// RAII guard to ensure inflight gauge is decremented and request duration is observed on all code paths. +// RAII guard to ensure inflight gauge is decremented, request duration is observed, +// and lifecycle logs are emitted on all code paths. struct RequestMetricsGuard { inflight_requests: prometheus::IntGauge, request_duration: prometheus::Histogram, start_time: Instant, + request_id: Option, } + impl Drop for RequestMetricsGuard { fn drop(&mut self) { self.inflight_requests.dec(); self.request_duration .observe(self.start_time.elapsed().as_secs_f64()); + if let Some(request_id) = &self.request_id { + tracing::info!(request_id = %request_id, "request completed"); + } } } @@ -149,7 +155,11 @@ where Ok(()) } - async fn handle_payload(&self, payload: Bytes) -> Result<(), PipelineError> { + async fn handle_payload( + &self, + payload: Bytes, + request_id: Option, + ) -> Result<(), PipelineError> { let t2_wallclock_ns = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() @@ -161,10 +171,14 @@ where m.request_counter.inc(); m.inflight_requests.inc(); m.request_bytes.inc_by(payload.len() as u64); + if let Some(rid) = &request_id { + tracing::info!(request_id = %rid, "request received"); + } RequestMetricsGuard { inflight_requests: m.inflight_requests.clone(), request_duration: m.request_duration.clone(), start_time, + request_id: request_id.clone(), } }); @@ -219,10 +233,6 @@ 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); @@ -361,13 +371,9 @@ where } // Ensure the metrics guard is not dropped until the end of the function. + // Drop fires "request completed" log via RAII. drop(_inflight_guard); - tracing::info!( - request_id = %context.id(), - "request completed" - ); - Ok(()) } } diff --git a/lib/runtime/src/pipeline/network/ingress/shared_tcp_endpoint.rs b/lib/runtime/src/pipeline/network/ingress/shared_tcp_endpoint.rs index ffc20ee15729..e17ba9c15ff9 100644 --- a/lib/runtime/src/pipeline/network/ingress/shared_tcp_endpoint.rs +++ b/lib/runtime/src/pipeline/network/ingress/shared_tcp_endpoint.rs @@ -209,9 +209,15 @@ impl SharedTcpServer { work_item.instance_id, ); + let request_id = work_item + .headers + .get("request-id") + .or_else(|| work_item.headers.get("x-dynamo-request-id")) + .cloned(); + let result = work_item .service_handler - .handle_payload(work_item.payload) + .handle_payload(work_item.payload, request_id) .instrument(span) .await; @@ -657,7 +663,11 @@ mod tests { #[async_trait] impl PushWorkHandler for SlowMockHandler { - async fn handle_payload(&self, _payload: Bytes) -> Result<(), PipelineError> { + async fn handle_payload( + &self, + _payload: Bytes, + _request_id: Option, + ) -> Result<(), PipelineError> { self.request_in_flight.store(true, Ordering::SeqCst); self.request_started.notify_one(); @@ -738,7 +748,7 @@ mod tests { let handler = handler.clone(); async move { let payload = Bytes::from("test payload"); - handler.handle_payload(payload).await + handler.handle_payload(payload, None).await } }); @@ -861,7 +871,11 @@ mod tests { #[async_trait] impl PushWorkHandler for ConcurrencyTrackingHandler { - async fn handle_payload(&self, _payload: Bytes) -> Result<(), PipelineError> { + async fn handle_payload( + &self, + _payload: Bytes, + _request_id: Option, + ) -> Result<(), PipelineError> { // Increment concurrent count let current = self.concurrent_count.fetch_add(1, Ordering::SeqCst) + 1; From fce0ac67ee7d7c328503c0011e140bc8a9beed06 Mon Sep 17 00:00:00 2001 From: nnshah1 Date: Thu, 2 Apr 2026 17:19:35 -0700 Subject: [PATCH 6/7] fix: use &str for request_id in create_inflight_guard, remove stale comment Change create_inflight_guard request_id param from String to &str to avoid unnecessary clones at call sites. Remove misleading comment about span field inheritance. Co-Authored-By: Claude Opus 4.6 (1M context) --- lib/llm/src/grpc/service/openai.rs | 2 +- lib/llm/src/grpc/service/tensor.rs | 2 +- lib/llm/src/http/service/anthropic.rs | 2 +- lib/llm/src/http/service/metrics.rs | 65 +++++++++++---------------- lib/llm/src/http/service/openai.rs | 26 +++++------ lib/llm/tests/http_metrics.rs | 6 +-- 6 files changed, 44 insertions(+), 59 deletions(-) diff --git a/lib/llm/src/grpc/service/openai.rs b/lib/llm/src/grpc/service/openai.rs index b08b4b33ffa7..6f0d7a05f3e8 100644 --- a/lib/llm/src/grpc/service/openai.rs +++ b/lib/llm/src/grpc/service/openai.rs @@ -92,7 +92,7 @@ pub async fn completion_response_stream( model, Endpoint::Completions, streaming, - request_id.clone(), + &request_id, ); 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 c1a13f3971bf..c791a62b0e4d 100644 --- a/lib/llm/src/grpc/service/tensor.rs +++ b/lib/llm/src/grpc/service/tensor.rs @@ -93,7 +93,7 @@ pub async fn tensor_response_stream( model, Endpoint::Tensor, streaming, - request_id.clone(), + &request_id, ); 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 83c01f1b706b..960fb0df366e 100644 --- a/lib/llm/src/http/service/anthropic.rs +++ b/lib/llm/src/http/service/anthropic.rs @@ -278,7 +278,7 @@ async fn anthropic_messages( &model, Endpoint::AnthropicMessages, streaming, - request.id().to_string(), + request.id(), ); tracing::trace!("Issuing generate call for Anthropic messages"); diff --git a/lib/llm/src/http/service/metrics.rs b/lib/llm/src/http/service/metrics.rs index b3b8214449ea..98c6f9876aa2 100644 --- a/lib/llm/src/http/service/metrics.rs +++ b/lib/llm/src/http/service/metrics.rs @@ -919,7 +919,7 @@ impl Metrics { model: &str, endpoint: Endpoint, streaming: bool, - request_id: String, + request_id: &str, ) -> InflightGuard { let request_type = if streaming { RequestType::Stream @@ -932,7 +932,7 @@ impl Metrics { model.to_string().to_lowercase(), endpoint, request_type, - request_id, + request_id.to_string(), ) } @@ -977,7 +977,6 @@ impl InflightGuard { let timer = Instant::now(); metrics.inc_inflight_gauge(&model); - // Record model on the enclosing span so all logs inherit it tracing::Span::current().record("model", model.as_str()); tracing::info!( @@ -2190,12 +2189,10 @@ mod tests { let model = "test-model"; { - let mut guard = metrics.clone().create_inflight_guard( - model, - Endpoint::ChatCompletions, - false, - String::new(), - ); + let mut guard = + metrics + .clone() + .create_inflight_guard(model, Endpoint::ChatCompletions, false, ""); guard.mark_ok(); } // guard drops here @@ -2222,12 +2219,10 @@ mod tests { let model = "test-model"; { - let mut guard = metrics.clone().create_inflight_guard( - model, - Endpoint::ChatCompletions, - false, - String::new(), - ); + let mut guard = + metrics + .clone() + .create_inflight_guard(model, Endpoint::ChatCompletions, false, ""); guard.mark_error(ErrorType::Validation); } // guard drops here @@ -2254,12 +2249,10 @@ mod tests { let model = "test-model"; { - let _guard = metrics.clone().create_inflight_guard( - model, - Endpoint::ChatCompletions, - false, - String::new(), - ); + let _guard = + metrics + .clone() + .create_inflight_guard(model, Endpoint::ChatCompletions, false, ""); // Don't call mark_ok() or mark_error() - simulate panic/unhandled error } // guard drops with default error_type=Internal @@ -2336,34 +2329,28 @@ 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, - String::new(), - ); + let mut guard = + metrics + .clone() + .create_inflight_guard(model, Endpoint::ChatCompletions, false, ""); guard.mark_error(ErrorType::Validation); drop(guard); } for _ in 0..3 { - let mut guard = metrics.clone().create_inflight_guard( - model, - Endpoint::Completions, - false, - String::new(), - ); + let mut guard = + metrics + .clone() + .create_inflight_guard(model, Endpoint::Completions, false, ""); guard.mark_error(ErrorType::Internal); drop(guard); } { - let mut guard = metrics.clone().create_inflight_guard( - model, - Endpoint::Embeddings, - false, - String::new(), - ); + let mut guard = + metrics + .clone() + .create_inflight_guard(model, Endpoint::Embeddings, false, ""); guard.mark_ok(); drop(guard); } diff --git a/lib/llm/src/http/service/openai.rs b/lib/llm/src/http/service/openai.rs index 848627c52cf9..e83e489d4db0 100644 --- a/lib/llm/src/http/service/openai.rs +++ b/lib/llm/src/http/service/openai.rs @@ -447,7 +447,7 @@ async fn completions_single( &model, Endpoint::Completions, streaming, - request_id.clone(), + &request_id, ); // Create http_queue_guard early - tracks time waiting to be processed @@ -583,7 +583,7 @@ async fn completions_batch( &model, Endpoint::Completions, streaming, - request_id.clone(), + &request_id, ); // Create http_queue_guard early - tracks time waiting to be processed @@ -761,7 +761,7 @@ async fn embeddings( model, Endpoint::Embeddings, streaming, - request_id.clone(), + &request_id, ); // Create http_queue_guard early - tracks time waiting to be processed @@ -1134,7 +1134,7 @@ async fn chat_completions( &model, Endpoint::ChatCompletions, streaming, - request_id.clone(), + &request_id, ); // Handle unsupported fields - if Some(resp) is returned by @@ -1511,7 +1511,7 @@ async fn responses( &model, Endpoint::Responses, streaming, - request.id().to_string(), + request.id(), ); // Handle unsupported fields - if Some(resp) is returned by validate_unsupported_fields, @@ -1963,7 +1963,7 @@ async fn images( &model, Endpoint::Images, streaming, - request_id.clone(), + &request_id, ); let mut response_collector = state.metrics_clone().create_response_collector(&model); @@ -2049,7 +2049,7 @@ async fn videos( &model, Endpoint::Videos, streaming, - request_id.clone(), + &request_id, ); let mut response_collector = state.metrics_clone().create_response_collector(&model); @@ -2109,12 +2109,10 @@ 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, - request.id().to_string(), - ); + let mut inflight = + state + .metrics_clone() + .create_inflight_guard(&model, Endpoint::Videos, true, request.id()); let mut response_collector = state.metrics_clone().create_response_collector(&model); @@ -2282,7 +2280,7 @@ async fn audio_speech( &model, Endpoint::Audios, streaming, - request_id.clone(), + &request_id, ); let mut response_collector = state.metrics_clone().create_response_collector(&model); diff --git a/lib/llm/tests/http_metrics.rs b/lib/llm/tests/http_metrics.rs index 1ba059ff75f6..733f04667966 100644 --- a/lib/llm/tests/http_metrics.rs +++ b/lib/llm/tests/http_metrics.rs @@ -78,7 +78,7 @@ async fn test_metrics_prefix_default() { "test-model", Endpoint::ChatCompletions, false, - String::new(), + "", ); } @@ -118,7 +118,7 @@ async fn test_metrics_prefix_custom() { "test-model", Endpoint::ChatCompletions, true, - String::new(), + "", ); } @@ -153,7 +153,7 @@ async fn test_metrics_prefix_sanitized() { "test-model", Endpoint::ChatCompletions, true, - String::new(), + "", ); } From c458b001d586a8e4fac91a5251711e5a62f285d8 Mon Sep 17 00:00:00 2001 From: nnshah1 Date: Thu, 2 Apr 2026 17:36:52 -0700 Subject: [PATCH 7/7] fix: inject request-id header directly from request context Inject request-id into transport headers from the request context directly, independent of DistributedTraceIdLayer. This ensures worker lifecycle logs have request_id in both JSONL and READABLE modes. Co-Authored-By: Claude Opus 4.6 (1M context) --- lib/llm/src/http/service/metrics.rs | 7 +++---- .../src/pipeline/network/egress/addressed_router.rs | 1 + 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/lib/llm/src/http/service/metrics.rs b/lib/llm/src/http/service/metrics.rs index 98c6f9876aa2..a58b0448c5f6 100644 --- a/lib/llm/src/http/service/metrics.rs +++ b/lib/llm/src/http/service/metrics.rs @@ -2290,10 +2290,9 @@ mod tests { ]; for error_type in &error_types { - let mut guard = - metrics - .clone() - .create_inflight_guard(model, endpoint, false, String::new()); + let mut guard = metrics + .clone() + .create_inflight_guard(model, endpoint, false, ""); guard.mark_error(error_type.clone()); drop(guard); } diff --git a/lib/runtime/src/pipeline/network/egress/addressed_router.rs b/lib/runtime/src/pipeline/network/egress/addressed_router.rs index b7b31551361b..26fc3a675638 100644 --- a/lib/runtime/src/pipeline/network/egress/addressed_router.rs +++ b/lib/runtime/src/pipeline/network/egress/addressed_router.rs @@ -241,6 +241,7 @@ where // Prepare trace headers using shared helper let mut headers = std::collections::HashMap::new(); inject_trace_headers_into_map(&mut headers); + headers.insert("request-id".to_string(), request_id.clone()); // Stamp send time right before the transport write so the network // transit metric excludes serialization/encoding overhead.