[model-gateway] Implement Layer 1 HTTP metrics instrumentation (#15121)

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<B> OnResponse<B> 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<S> Layer<S> for HttpMetricsLayer {
+    type Service = HttpMetricsMiddleware<S>;
+
+    fn layer(&self, inner: S) -> Self::Service {
+        HttpMetricsMiddleware { inner }
+    }
+}
+
+/// Tower Service for HTTP metrics collection
+#[derive(Clone)]
+pub struct HttpMetricsMiddleware<S> {
+    inner: S,
+}
+
+impl<S> Service<Request> for HttpMetricsMiddleware<S>
+where
+    S: Service<Request, Response = Response> + Send + Clone + 'static,
+    S::Future: Send + 'static,
+{
+    type Response = S::Response;
+    type Error = S::Error;
+    type Future =
+        Pin<Box<dyn std::future::Future<Output = Result<Self::Response, Self::Error>> + Send>>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        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<String> = 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<Arc<AppState>>,
     request: Request<Body>,
@@ -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"));
+    }
+}

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)