diff --git a/experimental/sgl-router/Cargo.lock b/experimental/sgl-router/Cargo.lock index c76d0933d..7e222fc32 100644 --- a/experimental/sgl-router/Cargo.lock +++ b/experimental/sgl-router/Cargo.lock @@ -3828,6 +3828,7 @@ dependencies = [ "futures-util", "http", "http-body", + "http-body-util", "mime", "pin-project-lite", "tokio", diff --git a/experimental/sgl-router/Cargo.toml b/experimental/sgl-router/Cargo.toml index df0ce17ef..6379a2332 100644 --- a/experimental/sgl-router/Cargo.toml +++ b/experimental/sgl-router/Cargo.toml @@ -49,7 +49,7 @@ axum = { version = "0.8", features = ["macros", "tracing", "http2"] } # `Frame`/`SizeHint` are not re-exported through `axum::body`. http-body = "1" tower = { version = "0.5", features = ["full"] } -tower-http = { version = "0.6", features = ["trace", "compression-gzip", "cors", "timeout", "request-id"] } +tower-http = { version = "0.6", features = ["trace", "compression-gzip", "cors", "timeout", "request-id", "catch-panic"] } reqwest = { version = "0.12", features = ["stream", "json", "rustls-tls", "http2"], default-features = false } sgl-kv-indexer = { path = "sgl-kv-indexer" } @@ -104,6 +104,9 @@ http-body-util = "0.1" # rather than as an unresolved import in a test. hyper = { version = "1", features = ["server", "http1", "http2"] } hyper-util = { version = "0.1", features = ["tokio"] } +# Capture the router's `tracing` access log in tests. Also a runtime dep, but +# integration tests are a separate crate and need it declared here too. +tracing-subscriber = { version = "0.3", features = ["fmt"] } serde = { version = "1", features = ["derive"] } serde_json = "1" tempfile = "3" diff --git a/experimental/sgl-router/monitoring/README.md b/experimental/sgl-router/monitoring/README.md index 073266540..1f4f9e65a 100644 --- a/experimental/sgl-router/monitoring/README.md +++ b/experimental/sgl-router/monitoring/README.md @@ -23,7 +23,7 @@ Families the router emits. The dashboard graphs all of them except the |---|---|---| | `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_worker_requests_total` | Counter | Per-worker **dispatches** by `worker_url`, `model_id`, `mode`, `outcome` (recorded after dispatch; blind to pre-dispatch drops). See [Dispatch outcomes](#dispatch-outcomes) | | `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_stream_outcome_total` | Counter | Streaming outcomes by `worker_url`, `model_id`, and `outcome` (`ok`, `stream_error_event`, `upstream_error`, or `client_disconnect`). Counts committed 2xx streams only — non-2xx responses are counted by status in `responses_total` | @@ -103,3 +103,43 @@ default to *All*) to scope the panels. The JSON is generated programmatically to keep the ~20 panels consistent. If the metric surface changes, update the generator and overwrite the JSON rather than hand-editing — hand-edits drift from the panel conventions. + +## Dispatch outcomes + +`sgl_router_worker_requests_total{outcome}` is derived from the status the +client saw, not from whether the router's internal dispatch returned `Ok` — a +worker error the router forwards is a successful *proxy* operation and a failed +*request*. + +| `outcome` | Source | Counts as a worker fault? | +|---|---|---| +| `success` | 2xx | no | +| `client_error` | 4xx except 429 | no — the caller sent something invalid | +| `backpressure` | 429, 503 | no — responsive but at capacity | +| `error` | 5xx except 503, plus transport failures, timeouts and incomplete bodies | **yes** | +| `cancelled` | the router's own stale-request deadline | no | + +`error` is the only bucket that means *this worker failed*, which is why the +Error-ratio panel uses it alone. The split matters during an incident: a +saturated fleet answering with its own queue-full 503s registers as +`backpressure`, and the circuit breaker likewise declines to open on those +statuses — so the two agree, and the error ratio keeps pointing at genuine +faults instead of pegging at 100% exactly when it is being read. + +A hung worker surfaces as `error` (the router's upstream timeout), *not* as +`cancelled`. Only the stale-request deadline produces `cancelled`; +`sgl_router_stale_requests_total{outcome="expired"}` counts the same events. + +## Access log + +The router emits one `http_request` event per request from a single middleware, +so requests that never reach a handler (a body-limit 413, an unrouted 404, a +panic-500) are logged too. Fields: `pod_id`, `request_id`, `method`, `path`, +`status`, `outcome`, `worker`, `model`, `stream`, `latency_ms`. + +`worker` and `model` are empty when the request was rejected before dispatch or +hit a route that does not dispatch — that is normal, not a gap. Successful infra +polls (`/healthz`, `/readyz`, `/metrics`) log at DEBUG so they do not bury real +traffic; a *failing* probe keeps the INFO line. For a stream the line is written +when the response head is ready, so `status=200` there does not mean the stream +finished — `sgl_router_stream_outcome_total` carries that. diff --git a/experimental/sgl-router/monitoring/grafana-dashboard.json b/experimental/sgl-router/monitoring/grafana-dashboard.json index cbef4ac50..7ff23b7ba 100644 --- a/experimental/sgl-router/monitoring/grafana-dashboard.json +++ b/experimental/sgl-router/monitoring/grafana-dashboard.json @@ -100,7 +100,7 @@ { "type": "stat", "title": "Error ratio", - "description": "Share of dispatches whose outcome=error.", + "description": "Share of dispatches that failed on the WORKER's side (outcome=error). Excludes client_error (4xx), backpressure (429/503) and cancelled, so a saturated fleet or a bad client does not read as a worker fault.", "datasource": { "type": "prometheus", "uid": "${datasource}" diff --git a/experimental/sgl-router/src/server/app.rs b/experimental/sgl-router/src/server/app.rs index a35041f4b..a5074bc02 100644 --- a/experimental/sgl-router/src/server/app.rs +++ b/experimental/sgl-router/src/server/app.rs @@ -2,45 +2,169 @@ // SPDX-License-Identifier: Apache-2.0 use crate::server::app_context::AppContext; +use crate::server::error::ApiError; +use crate::server::metrics::{outcome_from_status, RequestLogContext}; use crate::server::routes::chat::MAX_CHAT_BODY_BYTES; use axum::extract::{DefaultBodyLimit, MatchedPath, Request, State}; use axum::http::StatusCode; use axum::middleware::{self, Next}; +use axum::response::IntoResponse; use axum::response::Response; use axum::routing::{get, post}; use axum::Router; -use std::sync::Arc; +use std::sync::{Arc, OnceLock}; +use tower_http::catch_panic::CatchPanicLayer; +/// Infra endpoints whose *successful* polls are logged at DEBUG rather than +/// INFO. They are polled constantly — Prometheus scrapes `/metrics`, the kubelet +/// hits `/healthz` + `/readyz` every few seconds — so logging every hit at INFO +/// would bury real API traffic. A FAILING probe is the opposite: a pod dropping +/// out of readiness is an incident signal, so it keeps the normal INFO line (see +/// `access_log_and_record`). They are counted in both edge counters either way; +/// filter them by `route` in PromQL. +fn is_infra_path(path: &str) -> bool { + matches!(path, "/healthz" | "/readyz" | "/metrics") +} + +/// Collapse an HTTP method to a bounded allow-list for use as a metric label. +/// `http::Method` accepts any RFC-7230 extension token, and this middleware runs +/// before axum's method-router rejects an unknown verb with 405 — so the raw +/// method on `requests_total` / `responses_total` would be caller-controlled +/// unbounded-cardinality input. Unknown verbs collapse to `other`. The access +/// log below keeps the real method. +fn normalize_method(method: &axum::http::Method) -> &'static str { + use axum::http::Method; + match *method { + Method::GET => "GET", + Method::POST => "POST", + Method::PUT => "PUT", + Method::DELETE => "DELETE", + Method::PATCH => "PATCH", + Method::HEAD => "HEAD", + Method::OPTIONS => "OPTIONS", + Method::TRACE => "TRACE", + Method::CONNECT => "CONNECT", + _ => "other", + } +} + +/// Router pod identity stamped on every access-log line, so a multi-replica +/// router fleet's aggregated logs show which pod handled each request. Resolved +/// once, lazily, from the environment: `POD_NAME` (a downward-API env var an +/// operator opts into) wins, else `HOSTNAME` (Kubernetes defaults a pod's +/// hostname to its `metadata.name`, which the runtime exposes here), else +/// `"unknown"` (running outside a container with neither set). +static POD_ID: OnceLock = OnceLock::new(); + +fn pod_id() -> &'static str { + POD_ID.get_or_init(|| { + std::env::var("POD_NAME") + .or_else(|_| std::env::var("HOSTNAME")) + .unwrap_or_else(|_| "unknown".to_string()) + }) +} + +/// Outermost middleware: the single access-log and edge-counter site. +/// /// 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. +/// `route` is the matched template (not the raw URI) and `method` a known-verb +/// allow-list, so neither label's cardinality is caller-controlled. +/// /// /// Also the only place that sees every HTTP exchange on every route, so it is /// where `inflight_http` is taken — the count the termination drain reports on. /// The guard rides the response body rather than being dropped here: a /// streaming completion has barely started when this function returns. -async fn count_requests(State(ctx): State>, req: Request, next: Next) -> Response { - let method = req.method().as_str().to_owned(); +/// The access log runs at the same site, which is what lets it cover responses +/// produced before any handler runs (a 413 from the body-limit layer, a 400 from +/// the body extractor when a client drops the connection mid-upload, a +/// `CatchPanicLayer` 500) and handler short-circuits that return via `?` (a +/// body-validation 400, a model-not-found 404) — none of which can reach a +/// handler's own logging. Dispatched requests carry a [`RequestLogContext`] +/// naming the worker, model and outcome; everything else logs those empty. +/// +/// Two things it does NOT cover, both by construction: +/// * A client that disconnects before the response head exists. `next.run` +/// never resolves, so nothing after it runs; the request is already counted +/// in `requests_total`, and `requests_total - responses_total` is the only +/// evidence it happened. +/// * The final fate of a stream. The line is emitted when the response HEAD is +/// ready, which for SSE is before a single body byte is pumped — so a stream +/// that dies mid-body is logged `status=200`. The `stream` field marks those +/// lines; `stream_outcome_total` carries their real ending. +async fn access_log_and_record( + State(ctx): State>, + req: Request, + next: Next, +) -> Response { + let method = req.method().clone(); + let method_label = normalize_method(&method); + let path = req.uri().path().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 request_id = req + .headers() + .get("x-request-id") + .and_then(|v| v.to_str().ok()) + .unwrap_or("-") + .to_owned(); + let start = std::time::Instant::now(); + + ctx.metrics.record_ingress(&route, method_label); let inflight = ctx.inflight_http.enter(); let resp = next.run(req).await; + let status = resp.status(); + let latency_ms = start.elapsed().as_millis() as u64; ctx.metrics - .record_response(&route, &method, resp.status().as_u16()); + .record_response(&route, method_label, status.as_u16()); + + // A healthy probe is noise; a failing one is an incident signal. + if is_infra_path(&path) && status.is_success() { + tracing::debug!( + method = %method, + path = %path, + status = status.as_u16(), + latency_ms, + "http_request", + ); + } else { + // Per-worker fields are present only when a handler dispatched the + // request and attached them; anything rejected before dispatch logs + // them empty and falls back to the status for its outcome, which is all + // the status can say. + let log_ctx = resp.extensions().get::(); + tracing::info!( + pod_id = %pod_id(), + request_id = %request_id, + method = %method, + path = %path, + status = status.as_u16(), + outcome = log_ctx + .map(|c| c.outcome) + .unwrap_or_else(|| outcome_from_status(status.as_u16())) + .as_str(), + worker = log_ctx.map(|c| c.worker_url.as_str()).unwrap_or(""), + model = log_ctx.map(|c| c.model_id.as_str()).unwrap_or(""), + stream = log_ctx.is_some_and(|c| c.streaming), + latency_ms, + "http_request", + ); + } resp.map(|body| crate::server::inflight::track_body(body, inflight)) } -/// 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` -/// layer BEFORE the handler runs, so without this we would have no record -/// of which request was rejected. +/// Middleware: log 413 PAYLOAD_TOO_LARGE responses with the request method and +/// full URI, at WARN, so an operator investigating "client X gets 413s" has a +/// server-side breadcrumb. `access_log_and_record` already logs the 413 at INFO, +/// but only with the route template — this adds the query string and raises the +/// level, because a body-limit rejection is a client-configuration problem +/// rather than routine traffic. async fn log_413(req: Request, next: Next) -> Response { let method = req.method().clone(); let uri = req.uri().clone(); @@ -56,7 +180,7 @@ async fn log_413(req: Request, next: Next) -> Response { } pub fn build_router(ctx: Arc) -> Router { - Router::new() + let router = Router::new() .route("/healthz", get(crate::server::routes::health::healthz)) .route("/readyz", get(crate::server::routes::health::readyz)) .route("/metrics", get(crate::server::routes::metrics::metrics)) @@ -81,8 +205,382 @@ pub fn build_router(ctx: Arc) -> Router { .route( "/flush_cache", post(crate::server::routes::cache::flush_cache), - ) + ); + // A route that panics on purpose, so the panic-handling layers below are + // exercised as `build_router` actually composes them. Without it the layers + // could be deleted from this function and every test would still pass. + #[cfg(test)] + let router = router.route( + "/__test_panic", + get(|| async { + panic!("handler exploded"); + #[allow(unreachable_code)] + StatusCode::OK + }), + ); + router + // Convert a handler panic into a 500 response. hyper otherwise catches + // the panic and drops the connection WITHOUT a Response, so the failure + // never reaches the `access_log_and_record` middleware below and is + // invisible to both the edge counters and the access log. Positioned + // INNER relative to that middleware (added before it, so it sits closer + // to the handlers) so the synthesized 500 is observed and counted. + // + // The response is built from `ApiError::Internal` rather than + // tower-http's default plain-text body, so a panic answers with the same + // JSON envelope and `x-router-error-code` as every other + // router-originated error instead of punching a hole in that contract. + .layer(CatchPanicLayer::custom( + |_: Box| { + ApiError::Internal(anyhow::anyhow!("handler panicked")).into_response() + }, + )) // After routing, so MatchedPath is set for every route. - .layer(middleware::from_fn_with_state(ctx.clone(), count_requests)) + .layer(middleware::from_fn_with_state( + ctx.clone(), + access_log_and_record, + )) .with_state(ctx) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::server::metrics::RequestOutcome; + use axum::body::Body; + use axum::http::Request; + use std::sync::Mutex; + use tower::ServiceExt; + use tracing_subscriber::fmt::MakeWriter; + + /// Capture the `tracing` output of one test into a buffer. + #[derive(Clone)] + struct VecWriter(Arc>>); + + impl std::io::Write for VecWriter { + fn write(&mut self, b: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(b); + Ok(b.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + impl<'a> MakeWriter<'a> for VecWriter { + type Writer = VecWriter; + fn make_writer(&'a self) -> Self::Writer { + self.clone() + } + } + + /// Install a permissive global subscriber once per test binary. + /// + /// `tracing` caches each callsite's "interest" the first time it is hit. Under + /// the parallel test harness a thread with no subscriber of its own evaluates + /// the `http_request` callsite against `NoSubscriber`, which caches it as + /// *never* interested — after which a per-test `set_default` capture on another + /// thread records nothing, and an access-log assertion fails depending only on + /// which test ran first. A global subscriber that is interested in everything + /// keeps the callsite live; it discards what it receives, so per-test + /// `set_default` buffers stay isolated to their own thread. + fn prime_tracing_callsites() { + static PRIMED: OnceLock<()> = OnceLock::new(); + PRIMED.get_or_init(|| { + let _ = tracing::subscriber::set_global_default( + tracing_subscriber::fmt() + .with_max_level(tracing::Level::TRACE) + .with_writer(std::io::sink) + .finish(), + ); + }); + } + + /// Install a buffer-backed subscriber for the current thread. The returned + /// guard must stay alive for the duration of the capture. + fn capture_logs() -> (Arc>>, tracing::subscriber::DefaultGuard) { + prime_tracing_callsites(); + let buf = Arc::new(Mutex::new(Vec::::new())); + let subscriber = tracing_subscriber::fmt() + .with_ansi(false) + .with_writer(VecWriter(buf.clone())) + .finish(); + let guard = tracing::subscriber::set_default(subscriber); + (buf, guard) + } + + fn captured(buf: &Arc>>) -> String { + String::from_utf8(buf.lock().unwrap().clone()).unwrap() + } + + /// Metric labels must stay bounded: standard verbs pass through, but an + /// arbitrary RFC-7230 extension token (which reaches this middleware before + /// axum's 405) collapses to `other` so it can't explode label cardinality. + #[test] + fn normalize_method_collapses_unknown_verbs() { + use axum::http::Method; + assert_eq!(normalize_method(&Method::GET), "GET"); + assert_eq!(normalize_method(&Method::POST), "POST"); + let exotic = Method::from_bytes(b"BREW").unwrap(); + assert_eq!( + normalize_method(&exotic), + "other", + "an unknown verb must collapse to `other`, not mint a new label series", + ); + } + + /// An unrouted path must collapse to `route="unmatched"` and never put the + /// raw URI in a metric label — otherwise any caller could mint unbounded + /// label series by walking made-up paths. + #[tokio::test] + async fn unmatched_route_does_not_leak_the_raw_uri_into_metrics() { + let ctx = Arc::new(AppContext::stub()); + let req = Request::builder() + .method("GET") + .uri("/not/a/real/route-9f3c") + .body(Body::empty()) + .unwrap(); + let res = build_router(Arc::clone(&ctx)).oneshot(req).await.unwrap(); + assert_eq!(res.status(), StatusCode::NOT_FOUND); + + let m = ctx.metrics.render(); + assert!( + m.contains(r#"sgl_router_requests_total{route="unmatched",method="GET"} 1"#), + "an unrouted request must be counted under route=\"unmatched\": {m}", + ); + assert!( + !m.contains("route-9f3c"), + "the raw URI must never reach a metric label: {m}", + ); + } + + /// A handler panic must become a 500 that the `access_log_and_record` + /// middleware still observes. hyper catches a handler panic and drops the + /// connection WITHOUT producing a Response, so without the catch-panic layer + /// the failure is invisible to the edge counters and the access log; and the + /// middleware must be OUTER (applied after) so it counts the synthesized + /// 500. Driven through the real `build_router` — via the `#[cfg(test)]` + /// panic route it registers — so deleting either layer from production code + /// fails this test rather than only the ordering being pinned. + /// + /// The 500 must also carry the ordinary router error envelope: a panic is a + /// router-originated error, and commit-1's contract says every one of those + /// is machine-readable through `x-router-error-code`. + #[tokio::test] + async fn handler_panic_becomes_a_counted_500_with_the_router_error_envelope() { + let ctx = Arc::new(AppContext::stub()); + let req = Request::builder() + .method("GET") + .uri("/__test_panic") + .body(Body::empty()) + .unwrap(); + let res = build_router(Arc::clone(&ctx)).oneshot(req).await.unwrap(); + assert_eq!( + res.status(), + StatusCode::INTERNAL_SERVER_ERROR, + "a handler panic must surface as 500, not a dropped connection", + ); + assert_eq!( + res.headers() + .get("x-router-error-code") + .and_then(|v| v.to_str().ok()), + Some("internal_error"), + "a panic-500 must carry the same error envelope as every other \ + router-originated error", + ); + let m = ctx.metrics.render(); + assert!( + m.contains( + r#"sgl_router_responses_total{route="/__test_panic",method="GET",status_code="500"} 1"# + ), + "the middleware must observe and count the panic-500; got:\n{m}", + ); + } + + /// A request rejected BEFORE any handler runs — here an unrouted path, which + /// axum 404s with no handler involved at all — must still produce an access + /// log line. This is the gap a per-handler log cannot close: the response + /// exists, but no handler ever saw the request. The same site covers the + /// body-limit 413, the extractor 400 from a client that drops mid-upload, + /// and the `?` short-circuits inside a handler. + #[tokio::test] + async fn request_that_never_reaches_a_handler_is_still_logged() { + let (buf, _guard) = capture_logs(); + let ctx = Arc::new(AppContext::stub()); + let req = Request::builder() + .method("GET") + .uri("/nope") + .header("x-request-id", "rid-unrouted") + .body(Body::empty()) + .unwrap(); + let res = build_router(ctx).oneshot(req).await.unwrap(); + assert_eq!(res.status(), StatusCode::NOT_FOUND); + + let logs = captured(&buf); + assert!( + logs.contains("http_request") && logs.contains("rid-unrouted"), + "every request must be logged by the middleware; captured:\n{logs}", + ); + assert!( + logs.contains("status=404") && logs.contains("outcome=\"client_error\""), + "the access log must carry the final status and its outcome; captured:\n{logs}", + ); + } + + /// A routed request carries the worker and model it was dispatched to, via + /// the [`RequestLogContext`] its handler attaches to the response — so the + /// one central log line can still answer "which engine served this?". The + /// middleware cannot see that itself; without the extension the fields would + /// be blank on every line. + #[tokio::test] + async fn routed_response_context_names_the_worker_in_the_access_log() { + let (buf, _guard) = capture_logs(); + let ctx = Arc::new(AppContext::stub()); + let app = Router::new() + .route( + "/routed", + get(|| async { + let mut resp = StatusCode::OK.into_response(); + resp.extensions_mut().insert(RequestLogContext { + worker_url: "http://worker-a:30000".into(), + model_id: "tiny".into(), + streaming: false, + outcome: RequestOutcome::Cancelled, + }); + resp + }), + ) + .layer(middleware::from_fn_with_state( + Arc::clone(&ctx), + access_log_and_record, + )); + + let req = Request::builder() + .method("GET") + .uri("/routed") + .body(Body::empty()) + .unwrap(); + let res = app.oneshot(req).await.unwrap(); + assert_eq!(res.status(), StatusCode::OK); + + let logs = captured(&buf); + assert!( + logs.contains("worker=\"http://worker-a:30000\"") && logs.contains("model=\"tiny\""), + "a routed request must be logged with its worker and model; captured:\n{logs}", + ); + // The handler's outcome must win over the status-derived fallback — + // otherwise the log and `worker_requests_total` can disagree about a + // request the handler classified itself (here, a cancellation served + // with a 200 head, which no status could reveal). + assert!( + logs.contains("outcome=\"cancelled\"") && !logs.contains("outcome=\"success\""), + "the handler's own outcome must win over the status fallback; captured:\n{logs}", + ); + } + + /// A SUCCEEDING infra probe (`/healthz`, `/readyz`, `/metrics`) is polled + /// every few seconds by the kubelet and Prometheus, so it logs at DEBUG and + /// a default INFO subscriber sees nothing — otherwise probe traffic buries + /// real API traffic. It is still counted in both edge counters. + /// + /// The non-infra request in the same test is a positive control: without it + /// the negative assertion would also pass if log capture silently broke and + /// the buffer were simply empty. + #[tokio::test] + async fn successful_infra_probe_is_counted_but_not_logged_at_info() { + let (buf, _guard) = capture_logs(); + let ctx = Arc::new(AppContext::stub()); + let req = Request::builder() + .method("GET") + .uri("/healthz") + .body(Body::empty()) + .unwrap(); + let res = build_router(Arc::clone(&ctx)).oneshot(req).await.unwrap(); + assert_eq!(res.status(), StatusCode::OK); + assert!( + !captured(&buf).contains("/healthz"), + "a healthy probe must not reach the INFO access log; captured:\n{}", + captured(&buf), + ); + + // Positive control: a non-infra request through the same capture must + // produce a line, proving the absence above is the filter and not a + // broken subscriber. + let control = Request::builder() + .method("GET") + .uri("/nope") + .body(Body::empty()) + .unwrap(); + let _ = build_router(Arc::clone(&ctx)) + .oneshot(control) + .await + .unwrap(); + assert!( + captured(&buf).contains("http_request"), + "log capture is broken — the negative assertion above proves nothing; captured:\n{}", + captured(&buf), + ); + + let m = ctx.metrics.render(); + assert!( + m.contains(r#"sgl_router_requests_total{route="/healthz",method="GET"} 1"#) + && m.contains( + r#"sgl_router_responses_total{route="/healthz",method="GET",status_code="200"} 1"# + ), + "infra probes must still be counted at the edge: {m}", + ); + } + + /// A FAILING infra probe is the opposite of noise: a pod dropping out of + /// readiness is an incident signal, and demoting it to DEBUG alongside the + /// healthy polls would hide the transition an operator most needs to see. + #[tokio::test] + async fn failing_readiness_probe_is_logged_at_info() { + let (buf, _guard) = capture_logs(); + // A stub context has never been marked ready, so /readyz answers 503. + let ctx = Arc::new(AppContext::stub()); + assert!(!ctx.is_ready(), "stub context must start unready"); + + let req = Request::builder() + .method("GET") + .uri("/readyz") + .body(Body::empty()) + .unwrap(); + let res = build_router(Arc::clone(&ctx)).oneshot(req).await.unwrap(); + assert_eq!(res.status(), StatusCode::SERVICE_UNAVAILABLE); + + let logs = captured(&buf); + assert!( + logs.contains("http_request") + && logs.contains("/readyz") + && logs.contains("status=503"), + "a failing readiness probe must reach the INFO access log; captured:\n{logs}", + ); + } + + /// The `method` label must be bounded where it is USED, not merely where it + /// is computed: `normalize_method` existing is worthless if a call site + /// still passes the raw verb. An RFC-7230 extension token reaches this + /// middleware before axum's method-router can answer 405, so drive a made-up + /// verb through the real router and read the metric back. + #[tokio::test] + async fn unknown_verb_is_collapsed_in_the_metric_labels() { + let ctx = Arc::new(AppContext::stub()); + let req = Request::builder() + .method(axum::http::Method::from_bytes(b"BREW").unwrap()) + .uri("/healthz") + .body(Body::empty()) + .unwrap(); + let _ = build_router(Arc::clone(&ctx)).oneshot(req).await.unwrap(); + + let m = ctx.metrics.render(); + assert!( + m.contains(r#"method="other""#), + "an unknown verb must be counted under method=\"other\": {m}", + ); + assert!( + !m.contains("BREW"), + "the raw verb must never reach a metric label: {m}", + ); + } +} diff --git a/experimental/sgl-router/src/server/app_context.rs b/experimental/sgl-router/src/server/app_context.rs index 599d271ed..029e70b01 100644 --- a/experimental/sgl-router/src/server/app_context.rs +++ b/experimental/sgl-router/src/server/app_context.rs @@ -38,7 +38,8 @@ pub struct AppContext { /// timeout janitor, and metrics. pub active_load: Arc, /// Lightweight Prometheus-format metrics registry served via - /// `/metrics`. Shared with the chat handler (requests_total), + /// `/metrics`. Shared with the edge middleware (requests_total / + /// responses_total), the chat handler (worker_requests_total), the /// active-load registry, policy-specific counters, and PD dispatch. pub metrics: Arc, /// Shared Engine LoadStat table; ingress captures one immutable snapshot per request. diff --git a/experimental/sgl-router/src/server/metrics.rs b/experimental/sgl-router/src/server/metrics.rs index 4c86acaf7..79d383726 100644 --- a/experimental/sgl-router/src/server/metrics.rs +++ b/experimental/sgl-router/src/server/metrics.rs @@ -139,23 +139,92 @@ const OVERLAP_BLOCK_BUCKETS: &[f64] = &[ /// Recordable outcome for a request — narrowed to a handful of variants so /// the label cardinality stays bounded. +/// +/// The split exists so `outcome="error"` means *this worker failed*, matching +/// what [`crate::proxy`]'s `breaker_outcome` counts as a fault. A request can +/// fail for reasons that say nothing about the worker's health — the caller sent +/// something invalid, or the worker was merely at capacity — and folding those +/// into `error` makes the per-worker error ratio fire on client mistakes and on +/// exactly the backpressure the circuit breaker deliberately tolerates. #[derive(Debug, Clone, Copy)] pub enum RequestOutcome { Success, + /// The worker answered and rejected the request as invalid (a 4xx other than + /// 429). The caller's fault, not the worker's. + ClientError, + /// The worker was responsive but at capacity (429 / 503). Not a fault — the + /// same judgement `breaker_outcome` makes when it declines to open the + /// breaker on these statuses. + Backpressure, + /// The worker failed to serve the request: a 5xx fault, a transport failure, + /// a timeout, or a body that never completed. Error, + /// The router cancelled the request itself — today only the stale-request + /// deadline. Never derived from a status; see [`outcome_from_status`]. Cancelled, } impl RequestOutcome { - fn as_str(self) -> &'static str { + pub(crate) fn as_str(self) -> &'static str { match self { Self::Success => "success", + Self::ClientError => "client_error", + Self::Backpressure => "backpressure", Self::Error => "error", Self::Cancelled => "cancelled", } } } +/// Derive the bounded [`RequestOutcome`] label from the client-visible HTTP +/// status. +/// +/// Deriving from the status rather than from `Result::Ok`/`Err` is what keeps a +/// forwarded worker error honest: a worker 4xx/5xx the router proxies is an +/// `Ok(Response)` at the handler, so keying off `Ok` credits it as a success. +/// +/// This never returns [`RequestOutcome::Cancelled`]. A status cannot identify a +/// router-side cancellation: a 504 is produced by the stale-request deadline, by +/// the router's own upstream timeout, and by a worker 504 forwarded unchanged, +/// and only the caller holding the `ApiError` can tell them apart. Callers that +/// know they cancelled the request say so explicitly instead. +pub fn outcome_from_status(status: u16) -> RequestOutcome { + match status { + 200..=299 => RequestOutcome::Success, + // Responsive but at capacity. Listed before the 4xx arm so 429 lands + // here rather than in `ClientError`. + 429 | 503 => RequestOutcome::Backpressure, + 400..=499 => RequestOutcome::ClientError, + _ => RequestOutcome::Error, + } +} + +/// Routing context a handler attaches to its `Response` (via response +/// extensions) so the outermost access-log middleware can describe a dispatch it +/// cannot see itself. +/// +/// Attached today only by `chat_completions`. There is no compile-time +/// obligation to attach one — any handler that dispatches to a worker must do so +/// or its access-log line names no worker and falls back to a status-derived +/// outcome. A line with empty `worker`/`model` is therefore normal, not a bug: +/// it means the request was rejected before dispatch, or reached a route that +/// does not dispatch at all. +#[derive(Debug, Clone)] +pub struct RequestLogContext { + /// The worker the client-visible response actually came from. In PD mode + /// that is the decode worker, not the policy-selected prefill worker. + pub worker_url: String, + pub model_id: String, + /// Whether the client asked for an SSE stream. Only the handler knows this + /// (it is a body field, not a header or a route), and it separates + /// time-to-last-byte from time-to-headers when reading `latency_ms`. + pub streaming: bool, + /// The outcome the handler recorded for this request. Carried so the log + /// line and `worker_requests_total` cannot disagree — the middleware can + /// only see the status, which cannot express a router-side cancellation. + pub outcome: RequestOutcome, +} + /// Final outcome of a 2xx SSE stream. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum StreamOutcome { @@ -329,9 +398,8 @@ pub struct MetricsRegistry { // 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. + // Per-worker dispatch outcomes. 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 @@ -570,8 +638,8 @@ impl MetricsRegistry { } /// 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. + /// at the middleware, so it captures every response — including early-exit + /// 400/413/503s that never reach a handler. pub fn record_response(&self, route: &str, method: &str, status_code: u16) { let key = EdgeResponseKey { route: route.to_owned(), @@ -802,7 +870,7 @@ impl MetricsRegistry { } drop(guard); - // worker_requests_total — per-worker dispatch outcomes (formerly requests_total) + // worker_requests_total — per-worker dispatch outcomes out.push_str( "# HELP sgl_router_worker_requests_total Chat-completions requests dispatched to a worker, by dispatch outcome.\n", ); @@ -1805,4 +1873,38 @@ mod tests { "got:\n{out}" ); } + + /// The status → outcome mapping is the single definition shared by the + /// access log and `worker_requests_total`, so a silent change here corrupts + /// both surfaces at once. Pin every class, including the boundaries. + #[test] + fn outcome_from_status_maps_every_class() { + let cases = [ + (200, "success"), + (204, "success"), + (299, "success"), + // Backpressure is listed before the 4xx arm, so 429 must not fall + // through to client_error. + (429, "backpressure"), + (503, "backpressure"), + (400, "client_error"), + (404, "client_error"), + (499, "client_error"), + (500, "error"), + (502, "error"), + // A 504 is NOT a cancellation: the router's own upstream timeout and + // a worker's forwarded 504 both land here, and only the caller + // holding the `ApiError` can tell a real stale-cancel apart. + (504, "error"), + (199, "error"), + (300, "error"), + ]; + for (status, want) in cases { + assert_eq!( + outcome_from_status(status).as_str(), + want, + "status {status} must map to `{want}`", + ); + } + } } diff --git a/experimental/sgl-router/src/server/routes/chat.rs b/experimental/sgl-router/src/server/routes/chat.rs index 2f8711689..251fe3c47 100644 --- a/experimental/sgl-router/src/server/routes/chat.rs +++ b/experimental/sgl-router/src/server/routes/chat.rs @@ -18,13 +18,14 @@ use crate::proxy::sse::StreamEnd; use crate::server::app_context::AppContext; use crate::server::error::ApiError; use crate::server::metrics::{ - classify_stream_end, MetricsRegistry, PolicySelectionFailureReason, RequestOutcome, - StaleRequestOutcome, WorkerModeLabel, + classify_stream_end, outcome_from_status, MetricsRegistry, PolicySelectionFailureReason, + RequestLogContext, RequestOutcome, StaleRequestOutcome, WorkerModeLabel, }; use crate::workers::{LoadGuard, Worker}; use axum::body::Body; use axum::extract::State; use axum::http::{HeaderMap, HeaderName, HeaderValue, Response}; +use axum::response::IntoResponse; use bytes::Bytes; use serde::de::IgnoredAny; use serde::Deserialize; @@ -35,9 +36,7 @@ use std::sync::Arc; /// bootstrap-injected request body to BOTH the prefill and the decode /// worker concurrently; this header lets the prefill log the chosen /// peer, and is mirrored onto the response so sidecars / tests can -/// observe affinity without sniffing the proxy hop. The `x-sgl-` -/// prefix matches `x-sgl-router-error-code` so router-emitted metadata -/// stays grouped. +/// observe affinity without sniffing the proxy hop. const X_SGL_DECODE_URL: HeaderName = HeaderName::from_static("x-sgl-decode-url"); /// Optional caller requirement consumed only when a static P Bucket config is enabled. const X_SGL_TTFT_SLO_MS: HeaderName = HeaderName::from_static("x-sgl-ttft-slo-ms"); @@ -770,11 +769,22 @@ pub async fn chat_completions( // Snapshot the labels we need for metrics BEFORE moving the worker // / model_str values into the per-branch fetch futures. - let metrics_worker_url = worker.url.clone(); - let metrics_mode = match worker.mode() { - WorkerMode::Prefill => WorkerModeLabel::Prefill, - WorkerMode::Decode => WorkerModeLabel::Decode, - WorkerMode::Plain => WorkerModeLabel::Plain, + // The peer that answers is the one the outcome belongs to: in PD mode the + // client-visible response comes from the decode peer, not the prefill peer + // the policy selected, so charging prefill for a decode fault blames the + // wrong engine. Keeps this metric and the `RequestLogContext` below naming + // the same worker. + let metrics_worker_url = decode_hint_url + .clone() + .unwrap_or_else(|| worker.url.clone()); + let metrics_mode = if decode_hint_url.is_some() { + WorkerModeLabel::Decode + } else { + match worker.mode() { + WorkerMode::Prefill => WorkerModeLabel::Prefill, + WorkerMode::Decode => WorkerModeLabel::Decode, + WorkerMode::Plain => WorkerModeLabel::Plain, + } }; let metrics_model = model_str.clone(); @@ -1038,74 +1048,60 @@ pub async fn chat_completions( } }; - // Record the dispatch outcome AFTER we know whether the upstream - // accepted the request. A 504 from the stale-request branch counts as - // `cancelled` — semantically distinct from upstream errors that bubble - // through as `error`. The metric is per-worker so convergence tests - // can scrape `/metrics` and assert that ≥N requests landed on a - // single prefill worker. - let outcome = match &result { - Ok(_) => RequestOutcome::Success, - Err(ApiError::StaleRequestExpired { .. }) => { - // The janitor fired the stale-cancel and we observed it - // user-side; record both the per-request `cancelled` outcome - // AND the global `expired` count. The two views are useful for - // different alerts: per-worker request_total{cancelled} flags a - // worker that's hanging, while stale_requests_total{expired} - // tracks the global health of the janitor. - ctx.metrics - .record_stale_request(StaleRequestOutcome::Expired); - RequestOutcome::Cancelled - } - Err(_) => RequestOutcome::Error, - }; - ctx.metrics - .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 - // X-Request-Id (echoed end-to-end); `worker` is the engine the policy - // selected. The cache-aware routing rationale is logged separately at - // DEBUG by the policy. - let request_id = headers - .get("x-request-id") - .and_then(|v| v.to_str().ok()) - .unwrap_or("-"); + // Record the dispatch outcome AFTER we know the client-visible status. + // + // The status decides success-vs-failure, NOT `Result::Ok`/`Err`: a response + // the router forwards from a worker with a 4xx/5xx status is an + // `Ok(Response)` here, so keying off `Ok` would credit a forwarded engine + // error as a success. + // + // The `Err` VARIANT decides `Cancelled`, because a status cannot: the stale + // deadline, the router's own upstream timeout, and a worker's forwarded 504 + // all surface as 504, and only `StaleRequestExpired` is a router-side + // cancellation. Reading `Cancelled` off the status instead would quietly + // move every hung worker out of `outcome="error"` and blind the per-worker + // error ratio to the most common hard worker failure there is. + // + // The metric is per-worker so convergence tests can scrape `/metrics` and + // assert that ≥N requests landed on a single prefill worker. let http_status = match &result { Ok(resp) => resp.status().as_u16(), Err(e) => e.status_code().as_u16(), }; + let outcome = match &result { + Err(ApiError::StaleRequestExpired { .. }) => { + // Also record the global `expired` count. The two views drive + // different alerts: per-worker worker_requests_total{cancelled} + // flags a worker that is hanging, while stale_requests_total + // {expired} tracks how often the deadline fires at all. + ctx.metrics + .record_stale_request(StaleRequestOutcome::Expired); + RequestOutcome::Cancelled + } + // A 503 the ROUTER produced -- the breaker was open, or the worker URL + // would not parse -- not backpressure the worker reported; it was never + // asked. `outcome_from_status` cannot tell those from an engine's own + // 503, the same reason `Cancelled` is matched on the variant above. + Err(ApiError::BreakerOpen { .. } | ApiError::WorkerMisconfigured { .. }) => { + RequestOutcome::Error + } + _ => outcome_from_status(http_status), + }; + ctx.metrics + .record_worker_request(&metrics_worker_url, &metrics_model, metrics_mode, outcome); // 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(); + // The access-log line and the edge counters are emitted once, centrally, by + // the `access_log_and_record` middleware (app.rs) — a per-handler site + // cannot see the early exits that never reach a handler. if !streaming { ctx.metrics - .observe_request_duration(&metrics_model, elapsed.as_secs_f64()); + .observe_request_duration(&metrics_model, start.elapsed().as_secs_f64()); } - let outcome_str = match outcome { - RequestOutcome::Success => "success", - RequestOutcome::Error => "error", - RequestOutcome::Cancelled => "cancelled", - }; - tracing::info!( - request_id = %request_id, - method = "POST", - path = "/v1/chat/completions", - model = %metrics_model, - worker = %metrics_worker_url, - outcome = outcome_str, - http_status, - stream = streaming, - latency_ms = elapsed.as_millis() as u64, - "chat_completions", - ); // Mirror the upstream `x-sgl-decode-url` hint onto the response so // external tests / sidecars can observe the final PD Decode selection without @@ -1115,7 +1111,7 @@ pub async fn chat_completions( // resolved). A malformed URL was already rejected at the // request-side parse — we only reach this branch when the URL was // header-valid, so the second parse is safe. - match (result, decode_hint_url) { + let mut response = match (result, decode_hint_url) { (Ok(mut response), Some(url)) => { match HeaderValue::from_str(&url) { Ok(v) => { @@ -1130,10 +1126,32 @@ pub async fn chat_completions( ); } } - Ok(response) + response } - (other, _) => other, - } + (Ok(response), None) => response, + // Post-dispatch error (a worker WAS selected). Materialize it here + // instead of returning `Err` so it can carry the routing context below: + // an `Err` reaches the middleware as a bare response with no worker to + // name, and "which engine failed" is exactly what an operator wants from + // this line. + // + // The remaining `?` short-circuits return `Err` and are logged with no + // worker. All but one run before a worker is selected; the exception is + // `build_outgoing_body`, a router-side serialization failure that is not + // attributable to the worker it would have been sent to. + (Err(e), _) => e.into_response(), + }; + // Tag the routed response so the access-log middleware can describe the + // dispatch. `worker` names where the client-visible response came from — in + // PD mode the decode worker, which is the peer whose failure the client + // actually saw, not the prefill worker the policy selected. + response.extensions_mut().insert(RequestLogContext { + worker_url: metrics_worker_url, + model_id: metrics_model, + streaming, + outcome, + }); + Ok(response) } fn resolve_prefix_query( diff --git a/experimental/sgl-router/tests/proxy/chat_routing.rs b/experimental/sgl-router/tests/proxy/chat_routing.rs index 5b88f625b..000704faa 100644 --- a/experimental/sgl-router/tests/proxy/chat_routing.rs +++ b/experimental/sgl-router/tests/proxy/chat_routing.rs @@ -527,6 +527,158 @@ async fn non_streaming_upstream_500_preserved() { assert_eq!(got, upstream_body); } +/// A response the router FORWARDS from a worker with a non-2xx status is an +/// `Ok(Response)` at the router layer — only transport failures become `Err`. +/// The per-worker `worker_requests_total` outcome must therefore be derived from +/// the client-visible HTTP status, not from `Result::Ok`/`Err`: a forwarded 5xx +/// is counted `outcome="error"`, NOT credited as a success. +#[tokio::test] +async fn forwarded_5xx_records_worker_outcome_error_not_success() { + let worker = crate::common::mock_worker::MockWorker::start_returning_error( + StatusCode::INTERNAL_SERVER_ERROR, + serde_json::json!({"error": {"type": "server_error", "message": "boom"}}), + ) + .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::INTERNAL_SERVER_ERROR); + let _ = res.into_body().collect().await.unwrap().to_bytes(); + + let m = ctx.metrics.render(); + let error_line = format!( + r#"sgl_router_worker_requests_total{{worker_url="{}",model_id="tiny",mode="plain",outcome="error"}} 1"#, + worker.url, + ); + assert!( + m.contains(&error_line), + "a forwarded 5xx must be counted as outcome=\"error\"; got:\n{m}", + ); + let success_line = format!( + r#"sgl_router_worker_requests_total{{worker_url="{}",model_id="tiny",mode="plain",outcome="success"}}"#, + worker.url, + ); + assert!( + !m.contains(&success_line), + "a forwarded 5xx must NOT be credited as a success; got:\n{m}", + ); +} + +/// Buffer-backed `tracing` writer so a test can assert on the access log. +#[derive(Clone)] +struct VecWriter(Arc>>); + +impl std::io::Write for VecWriter { + fn write(&mut self, b: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(b); + Ok(b.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for VecWriter { + type Writer = VecWriter; + fn make_writer(&'a self) -> Self::Writer { + self.clone() + } +} + +/// Install a permissive global subscriber once per test binary. +/// +/// `tracing` caches each callsite's "interest" the first time it is hit. Under +/// the parallel test harness a thread with no subscriber of its own evaluates +/// the `http_request` callsite against `NoSubscriber`, which caches it as +/// *never* interested — after which a per-test `set_default` capture on another +/// thread records nothing, and an access-log assertion fails depending only on +/// which test ran first. A global subscriber that is interested in everything +/// keeps the callsite live; it discards what it receives, so per-test +/// `set_default` buffers stay isolated to their own thread. +fn prime_tracing_callsites() { + static PRIMED: std::sync::OnceLock<()> = std::sync::OnceLock::new(); + PRIMED.get_or_init(|| { + let _ = tracing::subscriber::set_global_default( + tracing_subscriber::fmt() + .with_max_level(tracing::Level::TRACE) + .with_writer(std::io::sink) + .finish(), + ); + }); +} + +/// The access-log line for a DISPATCHED request must name the worker it was +/// dispatched to. Only the chat handler knows that, so it attaches a +/// `RequestLogContext` to the response — including on the post-dispatch error +/// path, which is where "which engine failed" matters most. Without the attach +/// the line is still emitted, just anonymous, so only a log assertion catches a +/// regression here. +#[tokio::test] +async fn dispatched_error_names_its_worker_in_the_access_log() { + prime_tracing_callsites(); + let buf = Arc::new(std::sync::Mutex::new(Vec::::new())); + let subscriber = tracing_subscriber::fmt() + .with_ansi(false) + .with_writer(VecWriter(buf.clone())) + .finish(); + let _guard = tracing::subscriber::set_default(subscriber); + + // A worker that accepts the connection then drops it mid-body: dispatch + // succeeds, the request fails afterwards. + let worker = crate::common::mock_worker::MockWorker::start_returning_partial_body( + StatusCode::OK, + b"{\"partial\": ", + ) + .await; + let ctx = build_ctx_with_worker(&worker.url); + + let req = Request::builder() + .method("POST") + .uri("/v1/chat/completions") + .header("content-type", "application/json") + .header("x-request-id", "rid-dispatched-error") + .body(Body::from( + serde_json::to_vec(&serde_json::json!({ + "model": "tiny", + "messages": [{"role": "user", "content": "hi"}], + "stream": false, + })) + .unwrap(), + )) + .unwrap(); + let res = build_router(ctx.clone()).oneshot(req).await.unwrap(); + assert_eq!(res.status(), StatusCode::BAD_GATEWAY); + let _ = res.into_body().collect().await.unwrap().to_bytes(); + + let logs = String::from_utf8(buf.lock().unwrap().clone()).unwrap(); + let line = logs + .lines() + .find(|l| l.contains("rid-dispatched-error")) + .unwrap_or_else(|| panic!("no access-log line for the request; captured:\n{logs}")); + assert!( + line.contains(&format!(r#"worker="{}""#, worker.url)), + "a dispatched failure must name its worker: {line}", + ); + assert!( + line.contains(r#"model="tiny""#) && line.contains(r#"outcome="error""#), + "a dispatched failure must carry model and outcome=error: {line}", + ); +} + #[tokio::test] async fn non_streaming_upstream_4xx_body_passthrough() { // Regression: the worker's response bytes must reach the client diff --git a/experimental/sgl-router/tests/proxy/timeout.rs b/experimental/sgl-router/tests/proxy/timeout.rs index 5ea4cdc5c..7d550641e 100644 --- a/experimental/sgl-router/tests/proxy/timeout.rs +++ b/experimental/sgl-router/tests/proxy/timeout.rs @@ -77,7 +77,7 @@ async fn non_streaming_request_times_out_when_worker_hangs() { let policies = Arc::new(build_policy_registry(&cfg).unwrap()); let proxy = Arc::new(Proxy::new(Duration::from_millis(200)).unwrap()); let ctx = Arc::new(AppContext::new(cfg, tokenizers, proxy, registry, policies)); - let app = build_router(ctx); + let app = build_router(ctx.clone()); let req = Request::builder() .method("POST") @@ -111,6 +111,19 @@ async fn non_streaming_request_times_out_when_worker_hangs() { ); let bytes = res.into_body().collect().await.unwrap().to_bytes(); let body_str = String::from_utf8_lossy(&bytes); + // A hung worker is the most common hard worker failure there is, so it must + // land in `outcome="error"` — the series a per-worker error-ratio alert + // watches. Deriving the outcome from the 504 status instead would silently + // reclassify it as `cancelled` and blind that alert. + assert!( + ctx.metrics + .render() + .lines() + .any(|l| l.starts_with("sgl_router_worker_requests_total{") + && l.contains(r#"outcome="error""#)), + "an upstream timeout must be counted outcome=error, not cancelled:\n{}", + ctx.metrics.render(), + ); assert!( body_str.contains("\"code\":\"upstream_timeout\""), "body: {body_str}"