[Router] Log every request at one site; derive its outcome from the final status (3/3) (#39465)

Co-authored-by: Kangyan Zhou <kangyan.zhou@radixark.ai>
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
Co-authored-by: Shangming Cai <csmthu@gmail.com>
This commit is contained in:
Kangyan-Zhou
2026-09-19 03:11:00 +08:00
committed by GitHub
co-authored by Kangyan Zhou Claude Opus 5 Shangming Cai
parent 248c202b46
commit aed3fb1cdd
10 changed files with 923 additions and 95 deletions
+1
View File
@@ -3828,6 +3828,7 @@ dependencies = [
"futures-util",
"http",
"http-body",
"http-body-util",
"mime",
"pin-project-lite",
"tokio",
+4 -1
View File
@@ -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"
+41 -1
View File
@@ -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.
@@ -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}"
+512 -14
View File
@@ -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<String> = 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<Arc<AppContext>>, 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<Arc<AppContext>>,
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::<MatchedPath>()
.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::<RequestLogContext>();
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<AppContext>) -> 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<AppContext>) -> 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<dyn std::any::Any + Send>| {
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<Mutex<Vec<u8>>>);
impl std::io::Write for VecWriter {
fn write(&mut self, b: &[u8]) -> std::io::Result<usize> {
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<Mutex<Vec<u8>>>, tracing::subscriber::DefaultGuard) {
prime_tracing_callsites();
let buf = Arc::new(Mutex::new(Vec::<u8>::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<Mutex<Vec<u8>>>) -> 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}",
);
}
}
@@ -38,7 +38,8 @@ pub struct AppContext {
/// timeout janitor, and metrics.
pub active_load: Arc<ActiveLoadRegistry>,
/// 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<MetricsRegistry>,
/// Shared Engine LoadStat table; ingress captures one immutable snapshot per request.
+109 -7
View File
@@ -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<HashMap<EdgeKey, Arc<AtomicU64>>>,
responses_total: Mutex<HashMap<EdgeResponseKey, Arc<AtomicU64>>>,
// 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<HashMap<RequestKey, Arc<AtomicU64>>>,
// 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}`",
);
}
}
}
@@ -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(
@@ -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<std::sync::Mutex<Vec<u8>>>);
impl std::io::Write for VecWriter {
fn write(&mut self, b: &[u8]) -> std::io::Result<usize> {
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::<u8>::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
+14 -1
View File
@@ -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}"