diff --git a/experimental/sgl-router/monitoring/README.md b/experimental/sgl-router/monitoring/README.md index 048dceed8..cafa27ad8 100644 --- a/experimental/sgl-router/monitoring/README.md +++ b/experimental/sgl-router/monitoring/README.md @@ -20,10 +20,11 @@ The dashboard graphs every family the router emits: | Metric | Type | What it shows | |---|---|---| -| `sgl_router_requests_total` | Counter | Dispatches by `worker_url`, `model_id`, `mode`, `outcome` | +| `sgl_router_requests_total` | Counter | **Edge intake** — every request received at the router HTTP boundary, by `route`, `method`, counted before worker dispatch (true intake) | +| `sgl_router_responses_total` | Counter | **Edge responses** — every response returned, by `route`, `method`, `status_code` (incl. early-exit 400/413/503). `requests_total - responses_total` = received-but-not-answered | +| `sgl_router_worker_requests_total` | Counter | Per-worker **dispatches** by `worker_url`, `model_id`, `mode`, `outcome` (recorded after dispatch; blind to pre-dispatch drops) | | `sgl_router_request_duration_seconds` | Histogram | End-to-end request latency by `model_id` | | `sgl_router_ttft_seconds` | Histogram | Time to first token (streaming) by `model_id` | -| `sgl_router_responses_total` | Counter | Client-visible HTTP `status_code` | | `sgl_router_overlap_blocks` | Histogram | Cache-aware-zmq overlap blocks by `model_id` | | `sgl_router_active_load` | Gauge | Per-worker prefill-token / decode-block load | | `sgl_router_workers` | Gauge | Registered worker count by `mode` | diff --git a/experimental/sgl-router/monitoring/grafana-dashboard.json b/experimental/sgl-router/monitoring/grafana-dashboard.json index 91ae2b084..9892e56a9 100644 --- a/experimental/sgl-router/monitoring/grafana-dashboard.json +++ b/experimental/sgl-router/monitoring/grafana-dashboard.json @@ -89,7 +89,7 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "sum(rate(sgl_router_requests_total{model_id=~\"$model_id\",worker_url=~\"$worker_url\"}[$__rate_interval]))", + "expr": "sum(rate(sgl_router_worker_requests_total{model_id=~\"$model_id\",worker_url=~\"$worker_url\"}[$__rate_interval]))", "range": true, "refId": "A", "legendFormat": "__auto", @@ -161,7 +161,7 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "100 * sum(rate(sgl_router_requests_total{outcome=\"error\",model_id=~\"$model_id\",worker_url=~\"$worker_url\"}[$__rate_interval])) / clamp_min(sum(rate(sgl_router_requests_total{model_id=~\"$model_id\",worker_url=~\"$worker_url\"}[$__rate_interval])), 1e-9)", + "expr": "100 * sum(rate(sgl_router_worker_requests_total{outcome=\"error\",model_id=~\"$model_id\",worker_url=~\"$worker_url\"}[$__rate_interval])) / clamp_min(sum(rate(sgl_router_worker_requests_total{model_id=~\"$model_id\",worker_url=~\"$worker_url\"}[$__rate_interval])), 1e-9)", "range": true, "refId": "A", "legendFormat": "__auto", @@ -407,7 +407,7 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "sum by (outcome) (rate(sgl_router_requests_total{model_id=~\"$model_id\",worker_url=~\"$worker_url\"}[$__rate_interval]))", + "expr": "sum by (outcome) (rate(sgl_router_worker_requests_total{model_id=~\"$model_id\",worker_url=~\"$worker_url\"}[$__rate_interval]))", "range": true, "refId": "A", "legendFormat": "{{outcome}}", @@ -500,7 +500,7 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "sum by (mode) (rate(sgl_router_requests_total{model_id=~\"$model_id\",worker_url=~\"$worker_url\"}[$__rate_interval]))", + "expr": "sum by (mode) (rate(sgl_router_worker_requests_total{model_id=~\"$model_id\",worker_url=~\"$worker_url\"}[$__rate_interval]))", "range": true, "refId": "A", "legendFormat": "{{mode}}", @@ -686,7 +686,7 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "sum by (worker_url) (rate(sgl_router_requests_total{model_id=~\"$model_id\",worker_url=~\"$worker_url\"}[$__rate_interval]))", + "expr": "sum by (worker_url) (rate(sgl_router_worker_requests_total{model_id=~\"$model_id\",worker_url=~\"$worker_url\"}[$__rate_interval]))", "range": true, "refId": "A", "legendFormat": "{{worker_url}}", @@ -2035,7 +2035,7 @@ "type": "prometheus", "uid": "${datasource}" }, - "definition": "label_values(sgl_router_requests_total, model_id)", + "definition": "label_values(sgl_router_worker_requests_total, model_id)", "hide": 0, "includeAll": true, "allValue": ".*", @@ -2044,7 +2044,7 @@ "name": "model_id", "options": [], "query": { - "query": "label_values(sgl_router_requests_total, model_id)", + "query": "label_values(sgl_router_worker_requests_total, model_id)", "refId": "StandardVariableQuery" }, "refresh": 2, @@ -2061,7 +2061,7 @@ "type": "prometheus", "uid": "${datasource}" }, - "definition": "label_values(sgl_router_requests_total, worker_url)", + "definition": "label_values(sgl_router_worker_requests_total, worker_url)", "hide": 0, "includeAll": true, "allValue": ".*", @@ -2070,7 +2070,7 @@ "name": "worker_url", "options": [], "query": { - "query": "label_values(sgl_router_requests_total, worker_url)", + "query": "label_values(sgl_router_worker_requests_total, worker_url)", "refId": "StandardVariableQuery" }, "refresh": 2, diff --git a/experimental/sgl-router/src/server/app.rs b/experimental/sgl-router/src/server/app.rs index 4412809db..a769bc150 100644 --- a/experimental/sgl-router/src/server/app.rs +++ b/experimental/sgl-router/src/server/app.rs @@ -3,7 +3,7 @@ use crate::server::app_context::AppContext; use crate::server::routes::chat::MAX_CHAT_BODY_BYTES; -use axum::extract::{DefaultBodyLimit, Request}; +use axum::extract::{DefaultBodyLimit, MatchedPath, Request, State}; use axum::http::StatusCode; use axum::middleware::{self, Next}; use axum::response::Response; @@ -11,6 +11,25 @@ use axum::routing::{get, post}; use axum::Router; use std::sync::Arc; +/// Edge counters: `requests_total{route,method}` at entry (true intake, incl. +/// requests parked/shed/cancelled before dispatch), `responses_total{..., +/// status_code}` on exit (incl. early-exit 400/413/503). Their difference = +/// received-but-not-answered, invisible to post-dispatch `worker_requests_total`. +/// `route` is the matched template (not raw URI) to bound label cardinality. +async fn count_requests(State(ctx): State>, req: Request, next: Next) -> Response { + let method = req.method().as_str().to_owned(); + let route = req + .extensions() + .get::() + .map(|m| m.as_str().to_owned()) + .unwrap_or_else(|| "unmatched".to_owned()); + ctx.metrics.record_ingress(&route, &method); + let resp = next.run(req).await; + ctx.metrics + .record_response(&route, &method, resp.status().as_u16()); + resp +} + /// Middleware: log 413 PAYLOAD_TOO_LARGE responses with the request method /// and URI so an operator investigating "client X gets 413s" has a /// server-side breadcrumb. The 413 is produced by axum's `DefaultBodyLimit` @@ -57,5 +76,7 @@ pub fn build_router(ctx: Arc) -> Router { "/flush_cache", post(crate::server::routes::cache::flush_cache), ) + // After routing, so MatchedPath is set for every route. + .layer(middleware::from_fn_with_state(ctx.clone(), count_requests)) .with_state(ctx) } diff --git a/experimental/sgl-router/src/server/metrics.rs b/experimental/sgl-router/src/server/metrics.rs index f2dae7f16..cb1c8c5c6 100644 --- a/experimental/sgl-router/src/server/metrics.rs +++ b/experimental/sgl-router/src/server/metrics.rs @@ -18,10 +18,11 @@ //! //! | Metric | Type | Labels | //! |---|---|---| -//! | `sgl_router_requests_total` | Counter | `worker_url`, `model_id`, `mode`, `outcome` | +//! | `sgl_router_requests_total` | Counter | `route`, `method` | +//! | `sgl_router_responses_total` | Counter | `route`, `method`, `status_code` | +//! | `sgl_router_worker_requests_total` | Counter | `worker_url`, `model_id`, `mode`, `outcome` | //! | `sgl_router_request_duration_seconds` | Histogram | `model_id` | //! | `sgl_router_ttft_seconds` | Histogram | `model_id` | -//! | `sgl_router_responses_total` | Counter | `status_code` | //! | `sgl_router_overlap_blocks` | Histogram | `model_id` | //! | `sgl_router_active_load` | Gauge | `worker_url`, `kind` | //! | `sgl_router_workers` | Gauge | `mode` | @@ -203,14 +204,21 @@ impl ActiveLoadKind { /// internal state is `Arc`/`Atomic`/`Mutex`-protected. #[derive(Debug, Default)] pub struct MetricsRegistry { - requests_total: Mutex>>, + // Edge counters (recorded at the app.rs middleware): intake at entry, + // responses at exit. `requests_total - responses_total` = received but + // never answered, which `worker_requests_total` (post-dispatch) can't see. + requests_total: Mutex>>, + responses_total: Mutex>>, + // Per-worker dispatch outcomes (formerly `requests_total`). Recorded after + // dispatch, so blind to pre-dispatch drops; kept per-worker for the + // routing-convergence tests. + worker_requests_total: Mutex>>, // Keyed by `model_id` only: a model's pool is either all-plain or all-PD // (the registry rejects mixed pools), so the worker `mode` would be a pure // function of `model_id` here — a redundant label. Per-worker `mode` lives - // on `requests_total` / the worker gauges instead. + // on `worker_requests_total` / the worker gauges instead. request_duration: Mutex>, ttft_seconds: Mutex>, - responses_total: Mutex>>, overlap_blocks: Mutex>, active_load: Mutex>>, stale_requests_total: Mutex>>, @@ -227,6 +235,22 @@ struct RequestKey { outcome: &'static str, } +/// Labels for the edge `requests_total` (intake) counter. `route` is the matched +/// template (small fixed set), so cardinality is bounded. +#[derive(Debug, Hash, Eq, PartialEq, Clone)] +struct EdgeKey { + route: String, + method: String, +} + +/// Labels for the edge `responses_total` counter: `EdgeKey` + final HTTP status. +#[derive(Debug, Hash, Eq, PartialEq, Clone)] +struct EdgeResponseKey { + route: String, + method: String, + status_code: u16, +} + /// Per-worker state sampled from the [`crate::workers::WorkerRegistry`] at /// scrape time and rendered as the `sgl_router_workers` / /// `sgl_router_worker_*` gauge families. Built by the `/metrics` route from @@ -302,8 +326,25 @@ impl MetricsRegistry { Arc::new(Self::default()) } - /// Bump `sgl_router_requests_total` for the given worker / model / mode / outcome. - pub fn record_request( + /// Bump the edge intake counter `requests_total{route,method}`. Called at the + /// middleware before worker pick, so it sees pre-dispatch drops. + pub fn record_ingress(&self, route: &str, method: &str) { + let key = EdgeKey { + route: route.to_owned(), + method: method.to_owned(), + }; + let mut guard = self.requests_total.lock(); + let counter = guard + .entry(key) + .or_insert_with(|| Arc::new(AtomicU64::new(0))) + .clone(); + drop(guard); + counter.fetch_add(1, Ordering::Relaxed); + } + + /// Bump `worker_requests_total`. Recorded after dispatch — see `record_ingress` + /// for true intake. + pub fn record_worker_request( &self, worker_url: &str, model_id: &str, @@ -316,7 +357,7 @@ impl MetricsRegistry { mode: mode.as_str(), outcome: outcome.as_str(), }; - let mut guard = self.requests_total.lock(); + let mut guard = self.worker_requests_total.lock(); let counter = guard .entry(key) .or_insert_with(|| Arc::new(AtomicU64::new(0))) @@ -373,14 +414,18 @@ impl MetricsRegistry { hist.observe(seconds); } - /// Bump `sgl_router_responses_total{status_code}` for the HTTP status the - /// client ultimately saw. Cardinality is bounded by the small set of - /// status codes the router returns (2xx success, 4xx client, 5xx - /// upstream/proxy, 504 stale-cancel). - pub fn record_response(&self, status_code: u16) { + /// Bump the edge counter `responses_total{route,method,status_code}`. Called + /// at the middleware, so it captures every outcome — incl. early-exit + /// 400/413/503 that the old per-handler site skipped. + pub fn record_response(&self, route: &str, method: &str, status_code: u16) { + let key = EdgeResponseKey { + route: route.to_owned(), + method: method.to_owned(), + status_code, + }; let mut guard = self.responses_total.lock(); let counter = guard - .entry(status_code) + .entry(key) .or_insert_with(|| Arc::new(AtomicU64::new(0))) .clone(); drop(guard); @@ -472,12 +517,33 @@ impl MetricsRegistry { pub fn render_with_workers(&self, workers: &[WorkerSnapshot]) -> String { let mut out = String::new(); - // requests_total + // requests_total — edge intake (every request, counted before dispatch) out.push_str( - "# HELP sgl_router_requests_total Total chat-completions requests dispatched to a worker.\n", + "# HELP sgl_router_requests_total Total requests received at the router HTTP edge, counted before worker dispatch (true intake).\n", ); out.push_str("# TYPE sgl_router_requests_total counter\n"); let guard = self.requests_total.lock(); + let mut entries: Vec<(&EdgeKey, u64)> = guard + .iter() + .map(|(k, v)| (k, v.load(Ordering::Relaxed))) + .collect(); + entries.sort_by(|a, b| (&a.0.route, &a.0.method).cmp(&(&b.0.route, &b.0.method))); + for (key, value) in entries { + out.push_str(&format!( + "sgl_router_requests_total{{route=\"{}\",method=\"{}\"}} {}\n", + escape_label(&key.route), + escape_label(&key.method), + value, + )); + } + drop(guard); + + // worker_requests_total — per-worker dispatch outcomes (formerly requests_total) + out.push_str( + "# HELP sgl_router_worker_requests_total Chat-completions requests dispatched to a worker, by dispatch outcome.\n", + ); + out.push_str("# TYPE sgl_router_worker_requests_total counter\n"); + let guard = self.worker_requests_total.lock(); // Sort for stable output — easier for tests. let mut entries: Vec<(&RequestKey, u64)> = guard .iter() @@ -493,7 +559,7 @@ impl MetricsRegistry { }); for (key, value) in entries { out.push_str(&format!( - "sgl_router_requests_total{{worker_url=\"{}\",model_id=\"{}\",mode=\"{}\",outcome=\"{}\"}} {}\n", + "sgl_router_worker_requests_total{{worker_url=\"{}\",model_id=\"{}\",mode=\"{}\",outcome=\"{}\"}} {}\n", escape_label(&key.worker_url), escape_label(&key.model_id), key.mode, @@ -538,21 +604,30 @@ impl MetricsRegistry { } drop(guard); - // responses_total + // responses_total — edge, by route/method/status (incl. early-exit 400/413/503) out.push_str( - "# HELP sgl_router_responses_total Chat-completions responses returned to clients, by HTTP status code (recorded after worker dispatch).\n", + "# HELP sgl_router_responses_total Responses returned at the router HTTP edge, by route, method and HTTP status code.\n", ); out.push_str("# TYPE sgl_router_responses_total counter\n"); let guard = self.responses_total.lock(); - let mut entries: Vec<(u16, u64)> = guard + let mut entries: Vec<(&EdgeResponseKey, u64)> = guard .iter() - .map(|(k, v)| (*k, v.load(Ordering::Relaxed))) + .map(|(k, v)| (k, v.load(Ordering::Relaxed))) .collect(); - entries.sort_by_key(|e| e.0); - for (status_code, value) in entries { + entries.sort_by(|a, b| { + (&a.0.route, &a.0.method, a.0.status_code).cmp(&( + &b.0.route, + &b.0.method, + b.0.status_code, + )) + }); + for (key, value) in entries { out.push_str(&format!( - "sgl_router_responses_total{{status_code=\"{}\"}} {}\n", - status_code, value, + "sgl_router_responses_total{{route=\"{}\",method=\"{}\",status_code=\"{}\"}} {}\n", + escape_label(&key.route), + escape_label(&key.method), + key.status_code, + value, )); } drop(guard); @@ -918,16 +993,35 @@ mod tests { } #[test] - fn record_response_counts_by_status_code() { + fn record_response_counts_by_route_method_status_code() { let reg = MetricsRegistry::new(); - reg.record_response(200); - reg.record_response(200); - reg.record_response(502); - reg.record_response(504); + reg.record_response("/v1/chat/completions", "POST", 200); + reg.record_response("/v1/chat/completions", "POST", 200); + reg.record_response("/v1/chat/completions", "POST", 502); + reg.record_response("/v1/chat/completions", "POST", 504); let out = reg.render(); - assert!(out.contains(r#"sgl_router_responses_total{status_code="200"} 2"#)); - assert!(out.contains(r#"sgl_router_responses_total{status_code="502"} 1"#)); - assert!(out.contains(r#"sgl_router_responses_total{status_code="504"} 1"#)); + assert!(out.contains( + r#"sgl_router_responses_total{route="/v1/chat/completions",method="POST",status_code="200"} 2"# + )); + assert!(out.contains( + r#"sgl_router_responses_total{route="/v1/chat/completions",method="POST",status_code="502"} 1"# + )); + assert!(out.contains( + r#"sgl_router_responses_total{route="/v1/chat/completions",method="POST",status_code="504"} 1"# + )); + } + + #[test] + fn record_ingress_counts_by_route_method() { + let reg = MetricsRegistry::new(); + reg.record_ingress("/v1/chat/completions", "POST"); + reg.record_ingress("/v1/chat/completions", "POST"); + reg.record_ingress("/v1/models", "GET"); + let out = reg.render(); + assert!(out.contains( + r#"sgl_router_requests_total{route="/v1/chat/completions",method="POST"} 2"# + )); + assert!(out.contains(r#"sgl_router_requests_total{route="/v1/models",method="GET"} 1"#)); } #[test] @@ -981,15 +1075,15 @@ mod tests { } #[test] - fn record_request_emits_labelled_counter_line() { + fn record_worker_request_emits_labelled_counter_line() { let reg = MetricsRegistry::new(); - reg.record_request( + reg.record_worker_request( "http://worker-a:30000", "tiny", WorkerModeLabel::Prefill, RequestOutcome::Success, ); - reg.record_request( + reg.record_worker_request( "http://worker-a:30000", "tiny", WorkerModeLabel::Prefill, @@ -997,7 +1091,7 @@ mod tests { ); let out = reg.render(); assert!( - out.contains(r#"sgl_router_requests_total{worker_url="http://worker-a:30000",model_id="tiny",mode="prefill",outcome="success"} 2"#), + out.contains(r#"sgl_router_worker_requests_total{worker_url="http://worker-a:30000",model_id="tiny",mode="prefill",outcome="success"} 2"#), "render did not include the expected counter line; got:\n{out}", ); } @@ -1111,7 +1205,7 @@ mod tests { #[test] fn label_values_escape_quotes_and_backslashes() { let reg = MetricsRegistry::new(); - reg.record_request( + reg.record_worker_request( r#"http://"weird":30000"#, r"back\slash", WorkerModeLabel::Plain, diff --git a/experimental/sgl-router/src/server/routes/chat.rs b/experimental/sgl-router/src/server/routes/chat.rs index 85afbf33d..bb27d2f11 100644 --- a/experimental/sgl-router/src/server/routes/chat.rs +++ b/experimental/sgl-router/src/server/routes/chat.rs @@ -544,7 +544,7 @@ pub async fn chat_completions( Err(_) => RequestOutcome::Error, }; ctx.metrics - .record_request(&metrics_worker_url, &metrics_model, metrics_mode, outcome); + .record_worker_request(&metrics_worker_url, &metrics_model, metrics_mode, outcome); // Per-request access log — always on at INFO so incoming traffic and its // status are visible without DEBUG. `request_id` is the client/gateway @@ -560,19 +560,19 @@ pub async fn chat_completions( Err(e) => e.status_code().as_u16(), }; - // Record the client-visible HTTP status now that the outcome is known. - // For non-streaming requests the body is already buffered here, so - // `start.elapsed()` is true end-to-end latency — record it directly. For - // streaming, the body is still pending; the `RecordDurationOnDrop` guard - // packed into `stream_guards` records it at stream completion instead (so - // we don't capture only time-to-headers). `elapsed` still feeds the - // access-log `latency_ms` below for both. + // Record end-to-end latency now that the outcome is known. Non-streaming: + // body is buffered here, so `start.elapsed()` is true e2e — record directly. + // Streaming: body still pending, so the `RecordDurationOnDrop` guard in + // `stream_guards` records it at stream completion (not just time-to-headers). + // `elapsed` still feeds the access-log `latency_ms` for both. + // + // HTTP status is counted into `responses_total` by the edge middleware + // (app.rs), not here — the old per-handler site skipped early exits. let elapsed = start.elapsed(); if !streaming { ctx.metrics .observe_request_duration(&metrics_model, elapsed.as_secs_f64()); } - ctx.metrics.record_response(http_status); let outcome_str = match outcome { RequestOutcome::Success => "success", RequestOutcome::Error => "error", diff --git a/experimental/sgl-router/src/server/routes/metrics.rs b/experimental/sgl-router/src/server/routes/metrics.rs index 79c87e561..7953aab8a 100644 --- a/experimental/sgl-router/src/server/routes/metrics.rs +++ b/experimental/sgl-router/src/server/routes/metrics.rs @@ -102,7 +102,7 @@ mod tests { #[tokio::test] async fn metrics_endpoint_reflects_recorded_counters() { let ctx = Arc::new(AppContext::stub()); - ctx.metrics.record_request( + ctx.metrics.record_worker_request( "http://w-test:30000", "tiny", WorkerModeLabel::Prefill, diff --git a/experimental/sgl-router/tests/e2e/chat_completions/test_load_based_policy.py b/experimental/sgl-router/tests/e2e/chat_completions/test_load_based_policy.py index 4683e572d..c835f5135 100644 --- a/experimental/sgl-router/tests/e2e/chat_completions/test_load_based_policy.py +++ b/experimental/sgl-router/tests/e2e/chat_completions/test_load_based_policy.py @@ -19,7 +19,7 @@ _ACTIVE_RE = re.compile( r'^sgl_router_active_load\{worker_url="([^"]+)",kind="prefill_tokens"\}\s+(-?\d+)' ) _REQ_TOTAL_RE = re.compile( - r"^sgl_router_requests_total\{([^}]*)\}\s+(\d+(?:\.\d+)?)\s*$" + r"^sgl_router_worker_requests_total\{([^}]*)\}\s+(\d+(?:\.\d+)?)\s*$" ) _LABEL_RE = re.compile(r'(\w+)="([^"]*)"') diff --git a/experimental/sgl-router/tests/e2e/chat_completions/test_two_router_convergence.py b/experimental/sgl-router/tests/e2e/chat_completions/test_two_router_convergence.py index efcd35f61..a46f44c79 100644 --- a/experimental/sgl-router/tests/e2e/chat_completions/test_two_router_convergence.py +++ b/experimental/sgl-router/tests/e2e/chat_completions/test_two_router_convergence.py @@ -74,7 +74,7 @@ PREFIX_Y = (_PREFIX_Y_BODY * 8).strip() _REQ_TOTAL_RE = re.compile( - r"^sgl_router_requests_total\{([^}]*)\}\s+(\d+(?:\.\d+)?)\s*$" + r"^sgl_router_worker_requests_total\{([^}]*)\}\s+(\d+(?:\.\d+)?)\s*$" ) _LABEL_RE = re.compile(r'(\w+)="([^"]*)"') diff --git a/experimental/sgl-router/tests/proxy/chat_routing.rs b/experimental/sgl-router/tests/proxy/chat_routing.rs index b91404eb1..6d4399147 100644 --- a/experimental/sgl-router/tests/proxy/chat_routing.rs +++ b/experimental/sgl-router/tests/proxy/chat_routing.rs @@ -90,6 +90,48 @@ async fn non_streaming_returns_200() { assert_eq!(v["choices"][0]["message"]["content"], "ok"); } +/// Edge counters fire through the real middleware: `requests_total` at entry + +/// `responses_total` on exit, with matched-route/method labels. The unit tests +/// call record_* directly, so this is the only check that the middleware is +/// actually wired (MatchedPath -> record_ingress / record_response). +#[tokio::test] +async fn edge_counters_recorded_through_middleware() { + let worker = crate::common::mock_worker::MockWorker::start(vec![]).await; + let ctx = build_ctx_with_worker(&worker.url); + let app = build_router(ctx.clone()); + + let req = Request::builder() + .method("POST") + .uri("/v1/chat/completions") + .header("content-type", "application/json") + .body(Body::from( + serde_json::to_vec(&serde_json::json!({ + "model": "tiny", + "messages": [{"role": "user", "content": "hi"}], + "stream": false + })) + .unwrap(), + )) + .unwrap(); + let res = app.oneshot(req).await.unwrap(); + assert_eq!(res.status(), StatusCode::OK); + let _ = res.into_body().collect().await; + + let m = ctx.metrics.render(); + // intake — counted at entry by the middleware (the path unit tests miss) + assert!( + m.contains(r#"sgl_router_requests_total{route="/v1/chat/completions",method="POST"} 1"#), + "edge intake counter missing; got:\n{m}", + ); + // response — counted on the way out by the middleware + assert!( + m.contains( + r#"sgl_router_responses_total{route="/v1/chat/completions",method="POST",status_code="200"} 1"# + ), + "edge response counter missing; got:\n{m}", + ); +} + #[tokio::test] async fn non_streaming_upstream_unreachable_returns_502_unreachable() { // Bind a port, drop it — guarantees a closed/refused TCP destination. @@ -288,7 +330,9 @@ async fn streaming_5xx_request_records_duration_and_status_but_not_ttft() { "TTFT must NOT be recorded for a non-2xx streaming response; got:\n{m}", ); assert!( - m.contains(r#"sgl_router_responses_total{status_code="500"} 1"#), + m.contains( + r#"sgl_router_responses_total{route="/v1/chat/completions",method="POST",status_code="500"} 1"# + ), "the 500 status must be counted; got:\n{m}", ); assert!( diff --git a/experimental/sgl-router/tests/proxy/sticky_routing.rs b/experimental/sgl-router/tests/proxy/sticky_routing.rs index 72a8aac35..8d3fb2bff 100644 --- a/experimental/sgl-router/tests/proxy/sticky_routing.rs +++ b/experimental/sgl-router/tests/proxy/sticky_routing.rs @@ -93,12 +93,12 @@ fn chat_request(header: Option<(&str, &str)>) -> Request { .unwrap() } -/// Parse `sgl_router_requests_total{...,outcome="success"} N` lines into a +/// Parse `sgl_router_worker_requests_total{...,outcome="success"} N` lines into a /// map of worker_url -> success count. fn success_counts(metrics: &str) -> std::collections::HashMap { let mut counts = std::collections::HashMap::new(); for line in metrics.lines() { - let Some(rest) = line.strip_prefix("sgl_router_requests_total{") else { + let Some(rest) = line.strip_prefix("sgl_router_worker_requests_total{") else { continue; }; if !rest.contains(r#"outcome="success""#) {