From f9bceea064f439668e2de0b7026594222afe6a82 Mon Sep 17 00:00:00 2001 From: Simo Lin Date: Sun, 14 Dec 2025 09:39:15 -0800 Subject: [PATCH] [model-gateway] Implement Layer 1 HTTP metrics instrumentation (#15121) --- sgl-model-gateway/src/middleware.rs | 227 +++++++++++++++++++++++++++- sgl-model-gateway/src/server.rs | 1 + 2 files changed, 227 insertions(+), 1 deletion(-) diff --git a/sgl-model-gateway/src/middleware.rs b/sgl-model-gateway/src/middleware.rs index 8d71f65b3..89bc08b29 100644 --- a/sgl-model-gateway/src/middleware.rs +++ b/sgl-model-gateway/src/middleware.rs @@ -26,7 +26,7 @@ use tracing::{debug, error, field::Empty, info, info_span, warn, Span}; pub use crate::core::token_bucket::TokenBucket; use crate::{ - observability::metrics::RouterMetrics, + observability::metrics::{smg_labels, RouterMetrics, SmgMetrics}, routers::error::extract_error_code_from_response, server::AppState, wasm::{ @@ -343,6 +343,9 @@ impl OnResponse for ResponseLogger { RouterMetrics::record_http_status_code(status_code, error_code); RouterMetrics::record_request_duration(latency); + // New SMG metrics (Layer 1: HTTP) + SmgMetrics::record_http_response(status_code, error_code); + // Record these in the span for structured logging/observability tools span.record("status_code", status_code); // Use microseconds as integer to avoid format! string allocation @@ -517,6 +520,7 @@ pub async fn concurrency_limit_middleware( // Try to acquire token immediately if token_bucket.try_acquire(1.0).await.is_ok() { debug!("Acquired token immediately"); + SmgMetrics::record_http_rate_limit(smg_labels::RATE_LIMIT_ALLOWED); let response = next.run(request).await; // Wrap the response body with TokenGuardBody to return token when stream ends @@ -551,6 +555,7 @@ pub async fn concurrency_limit_middleware( match permit_rx.await { Ok(Ok(())) => { debug!("Acquired token from queue"); + SmgMetrics::record_http_rate_limit(smg_labels::RATE_LIMIT_ALLOWED); // Dequeue for embeddings if is_embeddings { let new_val = @@ -567,6 +572,7 @@ pub async fn concurrency_limit_middleware( } Ok(Err(status)) => { warn!("Queue returned error status: {}", status); + SmgMetrics::record_http_rate_limit(smg_labels::RATE_LIMIT_REJECTED); // Dequeue for embeddings on error if is_embeddings { let new_val = @@ -577,6 +583,7 @@ pub async fn concurrency_limit_middleware( } Err(_) => { error!("Queue response channel closed"); + SmgMetrics::record_http_rate_limit(smg_labels::RATE_LIMIT_REJECTED); // Dequeue for embeddings on channel error if is_embeddings { let new_val = @@ -589,16 +596,165 @@ pub async fn concurrency_limit_middleware( } Err(_) => { warn!("Request queue is full, returning 429"); + SmgMetrics::record_http_rate_limit(smg_labels::RATE_LIMIT_REJECTED); StatusCode::TOO_MANY_REQUESTS.into_response() } } } else { warn!("No tokens available and queuing is disabled, returning 429"); + SmgMetrics::record_http_rate_limit(smg_labels::RATE_LIMIT_REJECTED); StatusCode::TOO_MANY_REQUESTS.into_response() } } } +// ============================================================================ +// HTTP Metrics Layer (Layer 1: SMG metrics) +// ============================================================================ + +/// Global counter for active HTTP connections +static ACTIVE_HTTP_CONNECTIONS: AtomicU64 = AtomicU64::new(0); + +/// Tower Layer for HTTP metrics collection (SMG Layer 1 metrics) +#[derive(Clone, Copy, Default)] +pub struct HttpMetricsLayer; + +impl HttpMetricsLayer { + pub fn new() -> Self { + Self + } +} + +impl Layer for HttpMetricsLayer { + type Service = HttpMetricsMiddleware; + + fn layer(&self, inner: S) -> Self::Service { + HttpMetricsMiddleware { inner } + } +} + +/// Tower Service for HTTP metrics collection +#[derive(Clone)] +pub struct HttpMetricsMiddleware { + inner: S, +} + +impl Service for HttpMetricsMiddleware +where + S: Service + Send + Clone + 'static, + S::Future: Send + 'static, +{ + type Response = S::Response; + type Error = S::Error; + type Future = + Pin> + Send>>; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + self.inner.poll_ready(cx) + } + + fn call(&mut self, req: Request) -> Self::Future { + let method = req.method().as_str().to_owned(); + let path = normalize_path_for_metrics(req.uri().path()); + let start = Instant::now(); + + let mut inner = self.inner.clone(); + + Box::pin(async move { + // Increment inside async block - ensures no leak if future is dropped before polling + let active = ACTIVE_HTTP_CONNECTIONS.fetch_add(1, Ordering::Relaxed) + 1; + SmgMetrics::set_http_connections_active(active as usize); + + // Capture result before decrementing to ensure decrement happens on error too + let result = inner.call(req).await; + + // Always decrement, regardless of success or failure + let active = ACTIVE_HTTP_CONNECTIONS.fetch_sub(1, Ordering::Relaxed) - 1; + SmgMetrics::set_http_connections_active(active as usize); + + let response = result?; + + let duration = start.elapsed(); + let status_class = status_to_class(response.status().as_u16()); + + SmgMetrics::record_http_request(&method, &path, status_class); + SmgMetrics::record_http_duration(&method, &path, duration); + + Ok(response) + }) + } +} + +#[inline] +fn status_to_class(status: u16) -> &'static str { + match status { + 100..=199 => "1xx", + 200..=299 => "2xx", + 300..=399 => "3xx", + 400..=499 => "4xx", + 500..=599 => "5xx", + _ => "unknown", + } +} + +/// Normalize path for metrics to avoid high cardinality. +/// Replaces dynamic segments (IDs, UUIDs) with `{id}` placeholder. +/// Only allocates when normalization is needed; uses single-pass with byte offsets. +fn normalize_path_for_metrics(path: &str) -> String { + let bytes = path.as_bytes(); + let mut segment_start = 0; + let mut segment_idx = 0; + let mut result: Option = None; + + for (pos, &b) in bytes.iter().enumerate() { + if b == b'/' || pos == bytes.len() - 1 { + // Determine segment end (include last char if not a slash) + let segment_end = if b == b'/' { pos } else { pos + 1 }; + let segment = &path[segment_start..segment_end]; + + // Check segments after index 2 for dynamic IDs + if segment_idx > 2 && !segment.is_empty() && is_dynamic_id(segment) { + // Initialize result with everything before this segment + let result = result.get_or_insert_with(|| { + let mut s = String::with_capacity(path.len()); + s.push_str(&path[..segment_start]); + s + }); + result.push_str("{id}"); + } else if let Some(ref mut r) = result { + // Already normalizing, append this segment as-is + r.push_str(segment); + } + + // Add slash after segment (except at end) + if b == b'/' { + if let Some(ref mut r) = result { + r.push('/'); + } + segment_start = pos + 1; + segment_idx += 1; + } + } + } + + result.unwrap_or_else(|| path.to_owned()) +} + +/// Check if segment looks like a dynamic ID (prefixed ID, UUID, or numeric). +#[inline] +fn is_dynamic_id(s: &str) -> bool { + // Prefixed IDs: resp_xxx, chatcmpl_xxx (len > 10 with underscore) + if s.len() > 10 && s.contains('_') { + return true; + } + // UUIDs: 32+ hex chars with dashes + if s.len() >= 32 && s.bytes().all(|b| b.is_ascii_hexdigit() || b == b'-') { + return true; + } + // Numeric IDs + !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) +} + pub async fn wasm_middleware( State(app_state): State>, request: Request, @@ -820,3 +976,72 @@ pub async fn wasm_middleware( *final_response.headers_mut() = headers; Ok(final_response) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_normalize_path_no_ids() { + // Common API paths should pass through unchanged + assert_eq!( + normalize_path_for_metrics("/v1/chat/completions"), + "/v1/chat/completions" + ); + assert_eq!( + normalize_path_for_metrics("/v1/completions"), + "/v1/completions" + ); + assert_eq!(normalize_path_for_metrics("/v1/models"), "/v1/models"); + assert_eq!(normalize_path_for_metrics("/health"), "/health"); + } + + #[test] + fn test_normalize_path_with_prefixed_id() { + // Prefixed IDs (resp_xxx, chatcmpl_xxx) should be normalized + assert_eq!( + normalize_path_for_metrics("/v1/responses/resp_abc123def456"), + "/v1/responses/{id}" + ); + assert_eq!( + normalize_path_for_metrics("/v1/chat/completions/chatcmpl_abc123xyz"), + "/v1/chat/completions/{id}" + ); + } + + #[test] + fn test_normalize_path_with_uuid() { + assert_eq!( + normalize_path_for_metrics("/v1/responses/550e8400-e29b-41d4-a716-446655440000"), + "/v1/responses/{id}" + ); + } + + #[test] + fn test_normalize_path_with_numeric_id() { + assert_eq!( + normalize_path_for_metrics("/v1/workers/12345"), + "/v1/workers/{id}" + ); + } + + #[test] + fn test_is_dynamic_id() { + // Prefixed IDs + assert!(is_dynamic_id("resp_abc123def")); + assert!(is_dynamic_id("chatcmpl_xyz789")); + assert!(!is_dynamic_id("short_id")); // Too short + + // UUIDs + assert!(is_dynamic_id("550e8400-e29b-41d4-a716-446655440000")); + assert!(is_dynamic_id("550e8400e29b41d4a716446655440000")); // No dashes + + // Numeric + assert!(is_dynamic_id("12345")); + assert!(!is_dynamic_id("")); // Empty + + // Regular words + assert!(!is_dynamic_id("completions")); + assert!(!is_dynamic_id("chat")); + } +} diff --git a/sgl-model-gateway/src/server.rs b/sgl-model-gateway/src/server.rs index de1874bb3..3133a3f8f 100644 --- a/sgl-model-gateway/src/server.rs +++ b/sgl-model-gateway/src/server.rs @@ -734,6 +734,7 @@ pub fn build_app( max_payload_size, )) .layer(middleware::create_logging_layer()) + .layer(middleware::HttpMetricsLayer::new()) .layer(middleware::RequestIdLayer::new(request_id_headers)) .layer(create_cors_layer(cors_allowed_origins)) .fallback(sink_handler)