[router] Add request/TTFT/worker metrics + Grafana dashboard to experimental sgl-router (#27591)
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Kangyan-Zhou
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
1368717248
commit
badab6b136
@@ -0,0 +1,69 @@
|
||||
<!--
|
||||
SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors
|
||||
SPDX-License-Identifier: Apache-2.0
|
||||
-->
|
||||
|
||||
# sgl-router (experimental) monitoring
|
||||
|
||||
Grafana dashboard for the experimental router's Prometheus metrics, exposed
|
||||
on `/metrics` (text/plain, version 0.0.4) on the router's serving port
|
||||
(default `30000`).
|
||||
|
||||
## Files
|
||||
|
||||
- `grafana-dashboard.json` — importable Grafana dashboard, **SGLang Router
|
||||
(experimental)** (uid `sgl-router-experimental`).
|
||||
|
||||
## Metrics covered
|
||||
|
||||
The dashboard graphs every family the router emits:
|
||||
|
||||
| Metric | Type | What it shows |
|
||||
|---|---|---|
|
||||
| `sgl_router_requests_total` | Counter | Dispatches by `worker_url`, `model_id`, `mode`, `outcome` |
|
||||
| `sgl_router_request_duration_seconds` | Histogram | End-to-end request latency by `model_id` |
|
||||
| `sgl_router_ttft_seconds` | Histogram | Time to first token (streaming) by `model_id` |
|
||||
| `sgl_router_responses_total` | Counter | Client-visible HTTP `status_code` |
|
||||
| `sgl_router_overlap_blocks` | Histogram | Cache-aware-zmq overlap blocks by `model_id` |
|
||||
| `sgl_router_active_load` | Gauge | Per-worker prefill-token / decode-block load |
|
||||
| `sgl_router_workers` | Gauge | Registered worker count by `mode` |
|
||||
| `sgl_router_worker_health` | Gauge | Per-worker health (1=breaker admits, 0=open) |
|
||||
| `sgl_router_worker_cb_state` | Gauge | Per-worker circuit breaker state (0=closed, 1=open, 2=half_open) |
|
||||
| `sgl_router_worker_inflight_requests` | Gauge | In-flight requests per worker |
|
||||
| `sgl_router_stale_requests_total` | Counter | Stale-request cancellations |
|
||||
| `sgl_router_decode_affinity_total` | Counter | PD decode-affinity outcomes |
|
||||
| `sgl_router_sticky_total` | Counter | Sticky-session selection outcomes |
|
||||
|
||||
The `sgl_router_workers` / `sgl_router_worker_*` gauges are sampled from the
|
||||
live worker registry on every scrape, so a removed worker stops emitting
|
||||
series immediately rather than leaving a stale value.
|
||||
|
||||
## Prometheus scrape config
|
||||
|
||||
Point Prometheus at the router's `/metrics` endpoint:
|
||||
|
||||
```yaml
|
||||
scrape_configs:
|
||||
- job_name: sgl-router
|
||||
metrics_path: /metrics
|
||||
static_configs:
|
||||
- targets:
|
||||
- '127.0.0.1:30000' # router host:port
|
||||
```
|
||||
|
||||
## Import into Grafana
|
||||
|
||||
1. **Dashboards → New → Import**.
|
||||
2. Upload `grafana-dashboard.json` (or paste its contents).
|
||||
3. When prompted, select your Prometheus data source for the `Datasource`
|
||||
variable. The dashboard uses a templated data source, so it imports into
|
||||
any Grafana without editing the JSON.
|
||||
|
||||
The top bar exposes `model_id` and `worker_url` template variables (both
|
||||
default to *All*) to scope the panels.
|
||||
|
||||
## Regenerating
|
||||
|
||||
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.
|
||||
File diff suppressed because it is too large
Load Diff
@@ -13,6 +13,16 @@ use std::sync::Mutex;
|
||||
use std::time::Duration;
|
||||
use tokio::time::Instant;
|
||||
|
||||
/// Consistent `(admit, state_code)` pair read under a single breaker lock.
|
||||
/// See [`CircuitBreaker::snapshot`].
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
pub struct CircuitSnapshot {
|
||||
/// Would the breaker admit a request right now (`would_allow` semantics).
|
||||
pub admit: bool,
|
||||
/// State code: 0=closed, 1=open, 2=half_open.
|
||||
pub state_code: u8,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct CircuitBreakerConfig {
|
||||
pub threshold: NonZeroU32,
|
||||
@@ -88,6 +98,27 @@ impl CircuitBreaker {
|
||||
}
|
||||
}
|
||||
|
||||
/// Single-lock snapshot of `(admit, state_code)` for the `/metrics`
|
||||
/// scrape path, feeding `sgl_router_worker_health` and
|
||||
/// `sgl_router_worker_cb_state` (0=closed, 1=open, 2=half_open). Reading
|
||||
/// admit and state separately would take the lock twice and could observe
|
||||
/// a transition between the two reads, emitting a self-contradictory pair
|
||||
/// for one scrape. This reads both under one lock so they always agree.
|
||||
///
|
||||
/// Note `admit` and `state_code` can still legitimately disagree within
|
||||
/// a *consistent* read: an `Open` breaker past its cooldown returns
|
||||
/// `admit=true` (a probe slot is available) while `state_code=1`. That
|
||||
/// is the breaker's real state, not a race.
|
||||
pub fn snapshot(&self) -> CircuitSnapshot {
|
||||
let g = self.inner.lock().unwrap();
|
||||
let (admit, state_code) = match g.state {
|
||||
State::Closed => (true, 0),
|
||||
State::Open { opened_at } => (opened_at.elapsed() >= self.config.cool_down, 1),
|
||||
State::HalfOpen { probe_in_flight } => (!probe_in_flight, 2),
|
||||
};
|
||||
CircuitSnapshot { admit, state_code }
|
||||
}
|
||||
|
||||
/// True if a request may proceed. Mutates state when transitioning
|
||||
/// from Open → HalfOpen.
|
||||
pub fn allow(&self) -> bool {
|
||||
@@ -148,3 +179,70 @@ impl Default for CircuitBreaker {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn cb(threshold: u32, cool_down_secs: u64) -> CircuitBreaker {
|
||||
CircuitBreaker::with_config(CircuitBreakerConfig {
|
||||
threshold: NonZeroU32::new(threshold).unwrap(),
|
||||
cool_down: Duration::from_secs(cool_down_secs),
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn state_code_is_closed_by_default() {
|
||||
assert_eq!(CircuitBreaker::new().snapshot().state_code, 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn state_code_reports_open_only_after_threshold() {
|
||||
let b = cb(2, 30);
|
||||
b.record_failure();
|
||||
assert_eq!(
|
||||
b.snapshot().state_code,
|
||||
0,
|
||||
"1 failure < threshold 2 stays closed",
|
||||
);
|
||||
b.record_failure();
|
||||
assert_eq!(
|
||||
b.snapshot().state_code,
|
||||
1,
|
||||
"reaching threshold opens the breaker",
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn state_code_reports_half_open_after_cooldown_probe() {
|
||||
let b = cb(1, 10);
|
||||
b.record_failure();
|
||||
assert_eq!(
|
||||
b.snapshot().state_code,
|
||||
1,
|
||||
"threshold=1 opens on first failure"
|
||||
);
|
||||
tokio::time::advance(Duration::from_secs(11)).await;
|
||||
// `allow()` claims the probe slot, transitioning Open -> HalfOpen.
|
||||
assert!(b.allow());
|
||||
assert_eq!(b.snapshot().state_code, 2);
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn snapshot_reports_open_but_admittable_after_cooldown() {
|
||||
// The contract the scrape path depends on: a single read can show an
|
||||
// Open breaker (state_code=1) that nonetheless admits (admit=true)
|
||||
// once cooldown has elapsed — and the two halves never disagree due
|
||||
// to a torn read because they come from one lock acquisition.
|
||||
let b = cb(1, 10);
|
||||
b.record_failure();
|
||||
let s = b.snapshot();
|
||||
assert!(!s.admit, "open within cooldown must not admit");
|
||||
assert_eq!(s.state_code, 1);
|
||||
|
||||
tokio::time::advance(Duration::from_secs(11)).await;
|
||||
let s = b.snapshot();
|
||||
assert!(s.admit, "open past cooldown admits a probe");
|
||||
assert_eq!(s.state_code, 1, "...but is still reported as open");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -163,6 +163,11 @@ impl Proxy {
|
||||
/// `active_requests` counter and the per-request active-load entry alive
|
||||
/// for the full streaming lifetime — without which a long-running SSE
|
||||
/// response would under-report load.
|
||||
// Each parameter is a distinct, required input to a single upstream
|
||||
// forward (target, breaker, path, headers, body, plus the two
|
||||
// streaming-lifetime callbacks). Bundling them into a struct purely to
|
||||
// satisfy the arg-count heuristic would add indirection without clarity.
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub async fn forward_streaming_to(
|
||||
&self,
|
||||
worker_url: &str,
|
||||
@@ -171,6 +176,7 @@ impl Proxy {
|
||||
headers: &HeaderMap,
|
||||
body: Bytes,
|
||||
stream_guards: Option<Box<dyn Send + 'static>>,
|
||||
on_first_byte: Option<Box<dyn FnOnce() + Send + 'static>>,
|
||||
) -> Result<Response<Body>, ApiError> {
|
||||
if !breaker.allow() {
|
||||
return Err(ApiError::BreakerOpen {
|
||||
@@ -225,7 +231,20 @@ impl Proxy {
|
||||
}
|
||||
}))
|
||||
};
|
||||
let body = sse::bytes_stream_to_body(resp.bytes_stream(), stream_guards, on_complete);
|
||||
// Only record TTFT for successful streams — a 4xx/5xx error body
|
||||
// streaming back is not a generated token, so drop the hook for
|
||||
// non-2xx responses.
|
||||
let first_byte_hook = if status.is_success() {
|
||||
on_first_byte
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let body = sse::bytes_stream_to_body(
|
||||
resp.bytes_stream(),
|
||||
stream_guards,
|
||||
on_complete,
|
||||
first_byte_hook,
|
||||
);
|
||||
let mut out = Response::new(body);
|
||||
*out.status_mut() = status;
|
||||
out.headers_mut().insert(
|
||||
|
||||
@@ -55,10 +55,18 @@ use tokio_stream::wrappers::ReceiverStream;
|
||||
/// circuit-breaker outcome — without this hook, a worker that returns
|
||||
/// 2xx headers and then drops the stream mid-flight would stay credited
|
||||
/// as healthy.
|
||||
///
|
||||
/// # First-byte hook
|
||||
/// When `on_first_byte` is `Some`, the closure runs exactly once, the moment
|
||||
/// the first `Ok` chunk is read from the upstream stream — i.e. time to first
|
||||
/// token. It does NOT fire if the stream ends or errors before any `Ok` chunk
|
||||
/// arrives. `forward_streaming_to` passes a closure that records
|
||||
/// `sgl_router_ttft_seconds` for successful streaming responses.
|
||||
pub fn bytes_stream_to_body<S, E>(
|
||||
stream: S,
|
||||
stream_guards: Option<Box<dyn Send + 'static>>,
|
||||
on_complete: Option<Box<dyn FnOnce(bool) + Send + 'static>>,
|
||||
on_first_byte: Option<Box<dyn FnOnce() + Send + 'static>>,
|
||||
) -> Body
|
||||
where
|
||||
S: futures::Stream<Item = Result<Bytes, E>> + Send + Unpin + 'static,
|
||||
@@ -78,6 +86,7 @@ where
|
||||
// underscore suppresses the "unused variable" lint while
|
||||
// keeping intent explicit.
|
||||
let _hold = stream_guards;
|
||||
let mut on_first_byte = on_first_byte;
|
||||
let mut s = stream;
|
||||
while let Some(chunk) = s.next().await {
|
||||
let item: Result<Bytes, std::io::Error> = chunk.map_err(|e| {
|
||||
@@ -86,6 +95,15 @@ where
|
||||
std::io::Error::other(msg)
|
||||
});
|
||||
let is_err_chunk = item.is_err();
|
||||
// Fire the time-to-first-token hook on the first successful
|
||||
// chunk from upstream. `take()` makes it fire at most once;
|
||||
// an error-first stream never produced a token, so it's left
|
||||
// unfired (and dropped on task end).
|
||||
if !is_err_chunk {
|
||||
if let Some(hook) = on_first_byte.take() {
|
||||
hook();
|
||||
}
|
||||
}
|
||||
if is_err_chunk {
|
||||
*outcome_setter.lock() = false;
|
||||
}
|
||||
@@ -142,11 +160,66 @@ mod tests {
|
||||
Ok(Bytes::from_static(b"world")),
|
||||
];
|
||||
let s = stream::iter(chunks);
|
||||
let body = bytes_stream_to_body(s, None, None);
|
||||
let body = bytes_stream_to_body(s, None, None, None);
|
||||
let bytes = body.collect().await.unwrap().to_bytes();
|
||||
assert_eq!(&bytes[..], b"hello world");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn on_first_byte_fires_once_on_first_ok_chunk() {
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::Arc;
|
||||
|
||||
let fired = Arc::new(AtomicUsize::new(0));
|
||||
let fired_c = Arc::clone(&fired);
|
||||
let chunks = vec![
|
||||
Ok::<Bytes, std::io::Error>(Bytes::from_static(b"a")),
|
||||
Ok(Bytes::from_static(b"b")),
|
||||
];
|
||||
let s = stream::iter(chunks);
|
||||
let body = bytes_stream_to_body(
|
||||
s,
|
||||
None,
|
||||
None,
|
||||
Some(Box::new(move || {
|
||||
fired_c.fetch_add(1, Ordering::SeqCst);
|
||||
})),
|
||||
);
|
||||
let _ = body.collect().await.unwrap();
|
||||
assert_eq!(
|
||||
fired.load(Ordering::SeqCst),
|
||||
1,
|
||||
"first-byte hook must fire exactly once across the whole stream",
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn on_first_byte_not_fired_when_stream_errors_first() {
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::Arc;
|
||||
|
||||
let fired = Arc::new(AtomicUsize::new(0));
|
||||
let fired_c = Arc::clone(&fired);
|
||||
let chunks: Vec<Result<Bytes, std::io::Error>> = vec![Err(std::io::Error::other(
|
||||
"upstream failed before any token",
|
||||
))];
|
||||
let s = stream::iter(chunks);
|
||||
let body = bytes_stream_to_body(
|
||||
s,
|
||||
None,
|
||||
None,
|
||||
Some(Box::new(move || {
|
||||
fired_c.fetch_add(1, Ordering::SeqCst);
|
||||
})),
|
||||
);
|
||||
let _ = body.collect().await;
|
||||
assert_eq!(
|
||||
fired.load(Ordering::SeqCst),
|
||||
0,
|
||||
"first-byte hook must not fire when no Ok chunk is ever produced",
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn upstream_error_surfaces_to_consumer() {
|
||||
let chunks: Vec<Result<Bytes, std::io::Error>> = vec![
|
||||
@@ -154,7 +227,7 @@ mod tests {
|
||||
Err(std::io::Error::other("upstream blew up mid-stream")),
|
||||
];
|
||||
let s = stream::iter(chunks);
|
||||
let body = bytes_stream_to_body(s, None, None);
|
||||
let body = bytes_stream_to_body(s, None, None, None);
|
||||
// Collecting a body that terminates with an error must return Err.
|
||||
let result = body.collect().await;
|
||||
assert!(
|
||||
@@ -216,7 +289,7 @@ mod tests {
|
||||
// that arm, the closure unwrap-or-elses would panic itself or
|
||||
// produce an empty message, which this test catches.
|
||||
let s = PanicAnyOnSecondPoll { polls: 0 };
|
||||
let body = bytes_stream_to_body(s, None, None);
|
||||
let body = bytes_stream_to_body(s, None, None, None);
|
||||
let result = body.collect().await;
|
||||
assert!(
|
||||
result.is_err(),
|
||||
@@ -239,7 +312,7 @@ mod tests {
|
||||
// The pump task panics mid-stream. The client must see a loud Err,
|
||||
// NOT a silently-truncated success.
|
||||
let s = PanicOnSecondPoll { polls: 0 };
|
||||
let body = bytes_stream_to_body(s, None, None);
|
||||
let body = bytes_stream_to_body(s, None, None, None);
|
||||
let result = body.collect().await;
|
||||
assert!(
|
||||
result.is_err(),
|
||||
@@ -298,7 +371,7 @@ mod tests {
|
||||
yielded: 0,
|
||||
max: 1000, // way more than we'll let it consume
|
||||
};
|
||||
let body = bytes_stream_to_body(stream, None, None);
|
||||
let body = bytes_stream_to_body(stream, None, None, None);
|
||||
|
||||
// Read exactly one frame, then drop the body to simulate client disconnect.
|
||||
let mut data_stream = body.into_data_stream();
|
||||
|
||||
@@ -19,12 +19,25 @@
|
||||
//! | Metric | Type | Labels |
|
||||
//! |---|---|---|
|
||||
//! | `sgl_router_requests_total` | Counter | `worker_url`, `model_id`, `mode`, `outcome` |
|
||||
//! | `sgl_router_request_duration_seconds` | Histogram | `model_id` |
|
||||
//! | `sgl_router_ttft_seconds` | Histogram | `model_id` |
|
||||
//! | `sgl_router_responses_total` | Counter | `status_code` |
|
||||
//! | `sgl_router_overlap_blocks` | Histogram | `model_id` |
|
||||
//! | `sgl_router_active_load` | Gauge | `worker_url`, `kind` |
|
||||
//! | `sgl_router_workers` | Gauge | `mode` |
|
||||
//! | `sgl_router_worker_health` | Gauge | `worker_url` |
|
||||
//! | `sgl_router_worker_cb_state` | Gauge | `worker_url` |
|
||||
//! | `sgl_router_worker_inflight_requests` | Gauge | `worker_url` |
|
||||
//! | `sgl_router_stale_requests_total` | Counter | `outcome` |
|
||||
//! | `sgl_router_decode_affinity_total` | Counter | `outcome` |
|
||||
//! | `sgl_router_sticky_total` | Counter | `outcome` |
|
||||
//!
|
||||
//! The four `sgl_router_worker*` gauges and `sgl_router_workers` are sampled
|
||||
//! at scrape time from the live [`crate::workers::WorkerRegistry`] (passed to
|
||||
//! [`MetricsRegistry::render_with_workers`]) rather than pushed — there is no
|
||||
//! health-check loop to push from, and pull-on-scrape means a removed worker
|
||||
//! stops emitting series immediately instead of leaving a stale gauge.
|
||||
//!
|
||||
//! The exposition is text/plain; version=0.0.4 per the Prometheus spec.
|
||||
|
||||
use parking_lot::Mutex;
|
||||
@@ -41,6 +54,14 @@ const OVERLAP_BLOCKS_BUCKETS: &[f64] = &[
|
||||
0.0, 1.0, 2.0, 4.0, 8.0, 16.0, 32.0, 64.0, 128.0, 256.0, 512.0, 1000.0,
|
||||
];
|
||||
|
||||
/// Histogram bucket upper bounds (seconds) for
|
||||
/// `sgl_router_request_duration_seconds` and `sgl_router_ttft_seconds`.
|
||||
/// Standard latency ladder spanning 5 ms → 30 s; the `+Inf` bucket catches
|
||||
/// anything slower (a request that outlives the upstream's own timeouts).
|
||||
const REQUEST_DURATION_BUCKETS: &[f64] = &[
|
||||
0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0,
|
||||
];
|
||||
|
||||
/// Recordable outcome for a request — narrowed to a handful of variants so
|
||||
/// the label cardinality stays bounded.
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
@@ -158,6 +179,13 @@ impl ActiveLoadKind {
|
||||
#[derive(Debug, Default)]
|
||||
pub struct MetricsRegistry {
|
||||
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
|
||||
// function of `model_id` here — a redundant label. Per-worker `mode` lives
|
||||
// on `requests_total` / the worker gauges instead.
|
||||
request_duration: Mutex<HashMap<String, Histogram>>,
|
||||
ttft_seconds: Mutex<HashMap<String, Histogram>>,
|
||||
responses_total: Mutex<HashMap<u16, Arc<AtomicU64>>>,
|
||||
overlap_blocks: Mutex<HashMap<String, Histogram>>,
|
||||
active_load: Mutex<HashMap<ActiveLoadKey, Arc<AtomicI64>>>,
|
||||
stale_requests_total: Mutex<HashMap<&'static str, Arc<AtomicU64>>>,
|
||||
@@ -173,6 +201,23 @@ struct RequestKey {
|
||||
outcome: &'static str,
|
||||
}
|
||||
|
||||
/// Per-worker state sampled from the [`crate::workers::WorkerRegistry`] at
|
||||
/// scrape time and rendered as the `sgl_router_workers` /
|
||||
/// `sgl_router_worker_*` gauge families. Built by the `/metrics` route from
|
||||
/// the live registry on every scrape — see [`MetricsRegistry::render_with_workers`].
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct WorkerSnapshot {
|
||||
pub worker_url: String,
|
||||
/// `"plain"`, `"prefill"`, or `"decode"`.
|
||||
pub mode: &'static str,
|
||||
/// Circuit breaker would currently admit a request (`would_allow`).
|
||||
pub healthy: bool,
|
||||
/// Circuit breaker state code: 0=closed, 1=open, 2=half_open.
|
||||
pub cb_state: u8,
|
||||
/// In-flight request count for this worker (`Worker::active_load`).
|
||||
pub inflight: i64,
|
||||
}
|
||||
|
||||
#[derive(Debug, Hash, Eq, PartialEq, Clone)]
|
||||
struct ActiveLoadKey {
|
||||
worker_url: String,
|
||||
@@ -181,18 +226,27 @@ struct ActiveLoadKey {
|
||||
|
||||
#[derive(Debug)]
|
||||
struct Histogram {
|
||||
/// One counter per bucket boundary in [`OVERLAP_BLOCKS_BUCKETS`], plus
|
||||
/// one for `+Inf`. Buckets are cumulative on render but stored as
|
||||
/// non-cumulative counts here.
|
||||
/// Bucket upper bounds this histogram observes against (e.g.
|
||||
/// [`OVERLAP_BLOCKS_BUCKETS`] or [`REQUEST_DURATION_BUCKETS`]). Held
|
||||
/// per-instance so a single `Histogram` type backs metrics with
|
||||
/// different bucket ladders.
|
||||
bounds: &'static [f64],
|
||||
/// One counter per boundary in `bounds`, plus one for `+Inf`. Buckets
|
||||
/// are cumulative on render but stored as non-cumulative counts here.
|
||||
buckets: Vec<u64>,
|
||||
sum: f64,
|
||||
count: u64,
|
||||
}
|
||||
|
||||
impl Histogram {
|
||||
fn new() -> Self {
|
||||
fn new(bounds: &'static [f64]) -> Self {
|
||||
debug_assert!(
|
||||
bounds.windows(2).all(|w| w[0] <= w[1]),
|
||||
"histogram bounds must be ascending; `observe` relies on first-match placement",
|
||||
);
|
||||
Self {
|
||||
buckets: vec![0; OVERLAP_BLOCKS_BUCKETS.len() + 1],
|
||||
bounds,
|
||||
buckets: vec![0; bounds.len() + 1],
|
||||
sum: 0.0,
|
||||
count: 0,
|
||||
}
|
||||
@@ -200,7 +254,7 @@ impl Histogram {
|
||||
|
||||
fn observe(&mut self, value: f64) {
|
||||
let mut placed = false;
|
||||
for (i, &bound) in OVERLAP_BLOCKS_BUCKETS.iter().enumerate() {
|
||||
for (i, &bound) in self.bounds.iter().enumerate() {
|
||||
if value <= bound {
|
||||
self.buckets[i] += 1;
|
||||
placed = true;
|
||||
@@ -250,10 +304,62 @@ impl MetricsRegistry {
|
||||
let mut guard = self.overlap_blocks.lock();
|
||||
let hist = guard
|
||||
.entry(model_id.to_owned())
|
||||
.or_insert_with(Histogram::new);
|
||||
.or_insert_with(|| Histogram::new(OVERLAP_BLOCKS_BUCKETS));
|
||||
hist.observe(blocks as f64);
|
||||
}
|
||||
|
||||
/// Observe end-to-end request latency (seconds) for
|
||||
/// `sgl_router_request_duration_seconds`. Recorded once the upstream
|
||||
/// outcome is known, regardless of success or error — a slow error is
|
||||
/// still latency the operator cares about.
|
||||
pub fn observe_request_duration(&self, model_id: &str, seconds: f64) {
|
||||
// Drop non-finite observations before touching the map: a NaN would
|
||||
// poison the series `sum` permanently (NaN propagates through every
|
||||
// later add). Guarding here (not in `Histogram::observe`) also avoids
|
||||
// materializing an empty series for a dropped observation. Current
|
||||
// callers feed `Instant::elapsed`, so this is defense-in-depth.
|
||||
if !seconds.is_finite() {
|
||||
return;
|
||||
}
|
||||
let mut guard = self.request_duration.lock();
|
||||
let hist = guard
|
||||
.entry(model_id.to_owned())
|
||||
.or_insert_with(|| Histogram::new(REQUEST_DURATION_BUCKETS));
|
||||
hist.observe(seconds);
|
||||
}
|
||||
|
||||
/// Observe time-to-first-token (seconds) for `sgl_router_ttft_seconds` —
|
||||
/// the interval from request receipt to the first response chunk arriving
|
||||
/// from the upstream worker. Recorded only for successful *streaming*
|
||||
/// responses; non-streaming "first token" equals total latency, which
|
||||
/// `sgl_router_request_duration_seconds` already captures. Shares the
|
||||
/// latency bucket ladder ([`REQUEST_DURATION_BUCKETS`]).
|
||||
pub fn observe_ttft(&self, model_id: &str, seconds: f64) {
|
||||
// See `observe_request_duration` — drop non-finite before the map.
|
||||
if !seconds.is_finite() {
|
||||
return;
|
||||
}
|
||||
let mut guard = self.ttft_seconds.lock();
|
||||
let hist = guard
|
||||
.entry(model_id.to_owned())
|
||||
.or_insert_with(|| Histogram::new(REQUEST_DURATION_BUCKETS));
|
||||
hist.observe(seconds);
|
||||
}
|
||||
|
||||
/// Bump `sgl_router_responses_total{status_code}` for the HTTP status the
|
||||
/// client ultimately saw. Cardinality is bounded by the small set of
|
||||
/// status codes the router returns (2xx success, 4xx client, 5xx
|
||||
/// upstream/proxy, 504 stale-cancel).
|
||||
pub fn record_response(&self, status_code: u16) {
|
||||
let mut guard = self.responses_total.lock();
|
||||
let counter = guard
|
||||
.entry(status_code)
|
||||
.or_insert_with(|| Arc::new(AtomicU64::new(0)))
|
||||
.clone();
|
||||
drop(guard);
|
||||
counter.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
|
||||
/// Set `sgl_router_active_load` for the given worker + kind. Replaces the
|
||||
/// previous value (gauge semantics).
|
||||
pub fn set_active_load(&self, worker_url: &str, kind: ActiveLoadKind, value: i64) {
|
||||
@@ -303,8 +409,19 @@ impl MetricsRegistry {
|
||||
counter.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
|
||||
/// Render the registry as a Prometheus 0.0.4 exposition-format string.
|
||||
/// Render the registry as a Prometheus 0.0.4 exposition-format string
|
||||
/// with no live worker snapshot. The per-worker gauges emit only their
|
||||
/// HELP/TYPE headers and a zeroed pool-size series. Production scrapes
|
||||
/// go through [`Self::render_with_workers`]; this exists for callers
|
||||
/// (and tests) that have no [`crate::workers::WorkerRegistry`] handy.
|
||||
pub fn render(&self) -> String {
|
||||
self.render_with_workers(&[])
|
||||
}
|
||||
|
||||
/// Render the full exposition, sampling the supplied per-worker
|
||||
/// [`WorkerSnapshot`]s into the `sgl_router_workers` /
|
||||
/// `sgl_router_worker_*` gauge families.
|
||||
pub fn render_with_workers(&self, workers: &[WorkerSnapshot]) -> String {
|
||||
let mut out = String::new();
|
||||
|
||||
// requests_total
|
||||
@@ -338,6 +455,60 @@ impl MetricsRegistry {
|
||||
}
|
||||
drop(guard);
|
||||
|
||||
// request_duration histogram
|
||||
out.push_str(
|
||||
"# HELP sgl_router_request_duration_seconds End-to-end latency of chat-completions requests dispatched to a worker, in seconds (streaming requests are measured to stream completion).\n",
|
||||
);
|
||||
out.push_str("# TYPE sgl_router_request_duration_seconds histogram\n");
|
||||
let guard = self.request_duration.lock();
|
||||
let mut models: Vec<&String> = guard.keys().collect();
|
||||
models.sort();
|
||||
for model_id in models {
|
||||
let hist = guard.get(model_id).unwrap();
|
||||
let label_body = format!("model_id=\"{}\"", escape_label(model_id));
|
||||
render_histogram(
|
||||
&mut out,
|
||||
"sgl_router_request_duration_seconds",
|
||||
&label_body,
|
||||
hist,
|
||||
);
|
||||
}
|
||||
drop(guard);
|
||||
|
||||
// ttft histogram
|
||||
out.push_str(
|
||||
"# HELP sgl_router_ttft_seconds Time to first token (first upstream response chunk) for streaming requests, in seconds.\n",
|
||||
);
|
||||
out.push_str("# TYPE sgl_router_ttft_seconds histogram\n");
|
||||
let guard = self.ttft_seconds.lock();
|
||||
let mut models: Vec<&String> = guard.keys().collect();
|
||||
models.sort();
|
||||
for model_id in models {
|
||||
let hist = guard.get(model_id).unwrap();
|
||||
let label_body = format!("model_id=\"{}\"", escape_label(model_id));
|
||||
render_histogram(&mut out, "sgl_router_ttft_seconds", &label_body, hist);
|
||||
}
|
||||
drop(guard);
|
||||
|
||||
// responses_total
|
||||
out.push_str(
|
||||
"# HELP sgl_router_responses_total Chat-completions responses returned to clients, by HTTP status code (recorded after worker dispatch).\n",
|
||||
);
|
||||
out.push_str("# TYPE sgl_router_responses_total counter\n");
|
||||
let guard = self.responses_total.lock();
|
||||
let mut entries: Vec<(u16, u64)> = guard
|
||||
.iter()
|
||||
.map(|(k, v)| (*k, v.load(Ordering::Relaxed)))
|
||||
.collect();
|
||||
entries.sort_by_key(|e| e.0);
|
||||
for (status_code, value) in entries {
|
||||
out.push_str(&format!(
|
||||
"sgl_router_responses_total{{status_code=\"{}\"}} {}\n",
|
||||
status_code, value,
|
||||
));
|
||||
}
|
||||
drop(guard);
|
||||
|
||||
// overlap_blocks histogram
|
||||
out.push_str(
|
||||
"# HELP sgl_router_overlap_blocks Overlap-block count observed at cache-aware-zmq policy selection.\n",
|
||||
@@ -348,32 +519,8 @@ impl MetricsRegistry {
|
||||
models.sort();
|
||||
for model_id in models {
|
||||
let hist = guard.get(model_id).unwrap();
|
||||
let mut cumulative: u64 = 0;
|
||||
for (i, &bound) in OVERLAP_BLOCKS_BUCKETS.iter().enumerate() {
|
||||
cumulative += hist.buckets[i];
|
||||
out.push_str(&format!(
|
||||
"sgl_router_overlap_blocks_bucket{{model_id=\"{}\",le=\"{}\"}} {}\n",
|
||||
escape_label(model_id),
|
||||
bound,
|
||||
cumulative,
|
||||
));
|
||||
}
|
||||
cumulative += hist.buckets[OVERLAP_BLOCKS_BUCKETS.len()];
|
||||
out.push_str(&format!(
|
||||
"sgl_router_overlap_blocks_bucket{{model_id=\"{}\",le=\"+Inf\"}} {}\n",
|
||||
escape_label(model_id),
|
||||
cumulative,
|
||||
));
|
||||
out.push_str(&format!(
|
||||
"sgl_router_overlap_blocks_sum{{model_id=\"{}\"}} {}\n",
|
||||
escape_label(model_id),
|
||||
hist.sum,
|
||||
));
|
||||
out.push_str(&format!(
|
||||
"sgl_router_overlap_blocks_count{{model_id=\"{}\"}} {}\n",
|
||||
escape_label(model_id),
|
||||
hist.count,
|
||||
));
|
||||
let label_body = format!("model_id=\"{}\"", escape_label(model_id));
|
||||
render_histogram(&mut out, "sgl_router_overlap_blocks", &label_body, hist);
|
||||
}
|
||||
drop(guard);
|
||||
|
||||
@@ -398,6 +545,66 @@ impl MetricsRegistry {
|
||||
}
|
||||
drop(guard);
|
||||
|
||||
// Worker gauges — sampled from the live registry snapshot passed in,
|
||||
// not stored. Rendering from the snapshot (rather than a pushed map)
|
||||
// means a removed worker stops emitting series on the very next
|
||||
// scrape instead of leaving a stale gauge pinned at its last value.
|
||||
|
||||
// workers (pool size by mode). Emit all three modes so the series
|
||||
// exist (at 0) even before any worker of that mode is discovered.
|
||||
out.push_str("# HELP sgl_router_workers Registered workers by mode.\n");
|
||||
out.push_str("# TYPE sgl_router_workers gauge\n");
|
||||
for mode in ["plain", "prefill", "decode"] {
|
||||
let count = workers.iter().filter(|w| w.mode == mode).count();
|
||||
out.push_str(&format!(
|
||||
"sgl_router_workers{{mode=\"{}\"}} {}\n",
|
||||
mode, count,
|
||||
));
|
||||
}
|
||||
|
||||
// Sort the per-worker series by URL for stable output (tests + diffs).
|
||||
let mut sorted: Vec<&WorkerSnapshot> = workers.iter().collect();
|
||||
sorted.sort_by(|a, b| a.worker_url.cmp(&b.worker_url));
|
||||
|
||||
// worker_health (1=breaker would admit a request, 0=breaker open)
|
||||
out.push_str(
|
||||
"# HELP sgl_router_worker_health Worker health: 1 = circuit breaker admits requests, 0 = rejecting (open within cooldown, or half-open with a probe in flight). May read 1 while sgl_router_worker_cb_state=1 (open but cooldown elapsed).\n",
|
||||
);
|
||||
out.push_str("# TYPE sgl_router_worker_health gauge\n");
|
||||
for w in &sorted {
|
||||
out.push_str(&format!(
|
||||
"sgl_router_worker_health{{worker_url=\"{}\"}} {}\n",
|
||||
escape_label(&w.worker_url),
|
||||
u8::from(w.healthy),
|
||||
));
|
||||
}
|
||||
|
||||
// worker_cb_state (0=closed, 1=open, 2=half_open)
|
||||
out.push_str(
|
||||
"# HELP sgl_router_worker_cb_state Circuit breaker state per worker (0=closed, 1=open, 2=half_open).\n",
|
||||
);
|
||||
out.push_str("# TYPE sgl_router_worker_cb_state gauge\n");
|
||||
for w in &sorted {
|
||||
out.push_str(&format!(
|
||||
"sgl_router_worker_cb_state{{worker_url=\"{}\"}} {}\n",
|
||||
escape_label(&w.worker_url),
|
||||
w.cb_state,
|
||||
));
|
||||
}
|
||||
|
||||
// worker_inflight_requests (in-flight request count per worker)
|
||||
out.push_str(
|
||||
"# HELP sgl_router_worker_inflight_requests In-flight requests currently dispatched to each worker.\n",
|
||||
);
|
||||
out.push_str("# TYPE sgl_router_worker_inflight_requests gauge\n");
|
||||
for w in &sorted {
|
||||
out.push_str(&format!(
|
||||
"sgl_router_worker_inflight_requests{{worker_url=\"{}\"}} {}\n",
|
||||
escape_label(&w.worker_url),
|
||||
w.inflight,
|
||||
));
|
||||
}
|
||||
|
||||
// stale_requests_total
|
||||
out.push_str(
|
||||
"# HELP sgl_router_stale_requests_total Total stale-request cancellations fired by the janitor.\n",
|
||||
@@ -459,6 +666,28 @@ impl MetricsRegistry {
|
||||
}
|
||||
}
|
||||
|
||||
/// Render one labelled histogram family (`<name>_bucket` / `_sum` /
|
||||
/// `_count`) into `out`. `label_body` is the inside-of-braces label set
|
||||
/// WITHOUT the trailing `le` (e.g. `model_id="tiny"`) and is
|
||||
/// emitted verbatim — callers escape their own label values. Buckets are
|
||||
/// rendered cumulatively per the Prometheus histogram contract, with a
|
||||
/// final `+Inf` bucket.
|
||||
fn render_histogram(out: &mut String, name: &str, label_body: &str, hist: &Histogram) {
|
||||
let mut cumulative: u64 = 0;
|
||||
for (i, &bound) in hist.bounds.iter().enumerate() {
|
||||
cumulative += hist.buckets[i];
|
||||
out.push_str(&format!(
|
||||
"{name}_bucket{{{label_body},le=\"{bound}\"}} {cumulative}\n"
|
||||
));
|
||||
}
|
||||
cumulative += hist.buckets[hist.bounds.len()];
|
||||
out.push_str(&format!(
|
||||
"{name}_bucket{{{label_body},le=\"+Inf\"}} {cumulative}\n"
|
||||
));
|
||||
out.push_str(&format!("{name}_sum{{{label_body}}} {}\n", hist.sum));
|
||||
out.push_str(&format!("{name}_count{{{label_body}}} {}\n", hist.count));
|
||||
}
|
||||
|
||||
/// Prometheus label-value escape rule per
|
||||
/// https://prometheus.io/docs/instrumenting/exposition_formats/.
|
||||
/// We only escape `\`, `"`, and newline — the three characters the
|
||||
@@ -486,11 +715,177 @@ mod tests {
|
||||
let out = reg.render();
|
||||
// Should at least carry HELP / TYPE for every metric family.
|
||||
assert!(out.contains("# TYPE sgl_router_requests_total counter"));
|
||||
assert!(out.contains("# TYPE sgl_router_request_duration_seconds histogram"));
|
||||
assert!(out.contains("# TYPE sgl_router_ttft_seconds histogram"));
|
||||
assert!(out.contains("# TYPE sgl_router_responses_total counter"));
|
||||
assert!(out.contains("# TYPE sgl_router_overlap_blocks histogram"));
|
||||
assert!(out.contains("# TYPE sgl_router_active_load gauge"));
|
||||
assert!(out.contains("# TYPE sgl_router_workers gauge"));
|
||||
assert!(out.contains("# TYPE sgl_router_worker_health gauge"));
|
||||
assert!(out.contains("# TYPE sgl_router_worker_cb_state gauge"));
|
||||
assert!(out.contains("# TYPE sgl_router_worker_inflight_requests gauge"));
|
||||
assert!(out.contains("# TYPE sgl_router_stale_requests_total counter"));
|
||||
assert!(out.contains("# TYPE sgl_router_decode_affinity_total counter"));
|
||||
assert!(out.contains("# TYPE sgl_router_sticky_total counter"));
|
||||
// Pool-size series exist (at 0) for all three modes even with no
|
||||
// workers, so dashboards have a stable series to graph.
|
||||
assert!(out.contains(r#"sgl_router_workers{mode="plain"} 0"#));
|
||||
assert!(out.contains(r#"sgl_router_workers{mode="prefill"} 0"#));
|
||||
assert!(out.contains(r#"sgl_router_workers{mode="decode"} 0"#));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn observe_request_duration_writes_buckets_sum_and_count() {
|
||||
let reg = MetricsRegistry::new();
|
||||
// 25 ms, 120 ms, 600 ms for model "tiny".
|
||||
reg.observe_request_duration("tiny", 0.025);
|
||||
reg.observe_request_duration("tiny", 0.12);
|
||||
reg.observe_request_duration("tiny", 0.6);
|
||||
let out = reg.render();
|
||||
assert!(
|
||||
out.contains(r#"sgl_router_request_duration_seconds_count{model_id="tiny"} 3"#),
|
||||
"expected count=3; got:\n{out}",
|
||||
);
|
||||
// 0.025 <= 0.025, so the le=0.025 bucket is 1 (cumulative).
|
||||
assert!(
|
||||
out.contains(
|
||||
r#"sgl_router_request_duration_seconds_bucket{model_id="tiny",le="0.025"} 1"#
|
||||
),
|
||||
"expected le=0.025 bucket = 1; got:\n{out}",
|
||||
);
|
||||
// le=1 is cumulative over all three observations.
|
||||
assert!(
|
||||
out.contains(r#"sgl_router_request_duration_seconds_bucket{model_id="tiny",le="1"} 3"#),
|
||||
"expected le=1 bucket = 3; got:\n{out}",
|
||||
);
|
||||
assert!(out.contains(
|
||||
r#"sgl_router_request_duration_seconds_bucket{model_id="tiny",le="+Inf"} 3"#
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn request_duration_separates_by_model() {
|
||||
let reg = MetricsRegistry::new();
|
||||
reg.observe_request_duration("a", 0.01);
|
||||
reg.observe_request_duration("b", 0.01);
|
||||
let out = reg.render();
|
||||
assert!(out.contains(r#"sgl_router_request_duration_seconds_count{model_id="a"} 1"#));
|
||||
assert!(out.contains(r#"sgl_router_request_duration_seconds_count{model_id="b"} 1"#));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn request_duration_overflow_lands_in_plus_inf_bucket_only() {
|
||||
let reg = MetricsRegistry::new();
|
||||
// 45s is beyond the top finite bound (30s) — the operationally
|
||||
// critical "outlived the upstream timeout" case the +Inf bucket exists
|
||||
// for. It must NOT appear in le="30" but must be in le="+Inf"/_count,
|
||||
// and _sum must reflect the full value.
|
||||
reg.observe_request_duration("m", 45.0);
|
||||
let out = reg.render();
|
||||
assert!(
|
||||
out.contains(r#"sgl_router_request_duration_seconds_bucket{model_id="m",le="30"} 0"#),
|
||||
"45s must not fall in the le=30 bucket; got:\n{out}",
|
||||
);
|
||||
assert!(
|
||||
out.contains(r#"sgl_router_request_duration_seconds_bucket{model_id="m",le="+Inf"} 1"#)
|
||||
);
|
||||
assert!(out.contains(r#"sgl_router_request_duration_seconds_count{model_id="m"} 1"#));
|
||||
assert!(out.contains(r#"sgl_router_request_duration_seconds_sum{model_id="m"} 45"#));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn observe_request_duration_ignores_non_finite() {
|
||||
let reg = MetricsRegistry::new();
|
||||
reg.observe_request_duration("m", f64::NAN);
|
||||
reg.observe_request_duration("m", f64::INFINITY);
|
||||
let out = reg.render();
|
||||
// Nothing recorded — no count series for the model (sum stays uncorrupted).
|
||||
assert!(
|
||||
!out.contains(r#"sgl_router_request_duration_seconds_count{model_id="m"}"#),
|
||||
"non-finite observations must be dropped, not bucketed; got:\n{out}",
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn observe_ttft_writes_buckets_sum_and_count() {
|
||||
let reg = MetricsRegistry::new();
|
||||
reg.observe_ttft("tiny", 0.04);
|
||||
reg.observe_ttft("tiny", 0.2);
|
||||
let out = reg.render();
|
||||
assert!(
|
||||
out.contains(r#"sgl_router_ttft_seconds_count{model_id="tiny"} 2"#),
|
||||
"expected ttft count=2; got:\n{out}",
|
||||
);
|
||||
// 0.04 <= 0.05, so the le=0.05 bucket is 1 (cumulative).
|
||||
assert!(
|
||||
out.contains(r#"sgl_router_ttft_seconds_bucket{model_id="tiny",le="0.05"} 1"#),
|
||||
"expected le=0.05 bucket = 1; got:\n{out}",
|
||||
);
|
||||
// le=0.25 is cumulative over both observations.
|
||||
assert!(out.contains(r#"sgl_router_ttft_seconds_bucket{model_id="tiny",le="0.25"} 2"#));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn record_response_counts_by_status_code() {
|
||||
let reg = MetricsRegistry::new();
|
||||
reg.record_response(200);
|
||||
reg.record_response(200);
|
||||
reg.record_response(502);
|
||||
reg.record_response(504);
|
||||
let out = reg.render();
|
||||
assert!(out.contains(r#"sgl_router_responses_total{status_code="200"} 2"#));
|
||||
assert!(out.contains(r#"sgl_router_responses_total{status_code="502"} 1"#));
|
||||
assert!(out.contains(r#"sgl_router_responses_total{status_code="504"} 1"#));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn render_with_workers_emits_per_worker_gauges_and_pool_size() {
|
||||
let reg = MetricsRegistry::new();
|
||||
let workers = vec![
|
||||
WorkerSnapshot {
|
||||
worker_url: "http://p0:30000".into(),
|
||||
mode: "prefill",
|
||||
healthy: true,
|
||||
cb_state: 0,
|
||||
inflight: 5,
|
||||
},
|
||||
WorkerSnapshot {
|
||||
worker_url: "http://d0:30000".into(),
|
||||
mode: "decode",
|
||||
healthy: false,
|
||||
cb_state: 1,
|
||||
inflight: 0,
|
||||
},
|
||||
];
|
||||
let out = reg.render_with_workers(&workers);
|
||||
// Pool size by mode.
|
||||
assert!(out.contains(r#"sgl_router_workers{mode="prefill"} 1"#));
|
||||
assert!(out.contains(r#"sgl_router_workers{mode="decode"} 1"#));
|
||||
assert!(out.contains(r#"sgl_router_workers{mode="plain"} 0"#));
|
||||
// Health: healthy prefill = 1, unhealthy decode = 0.
|
||||
assert!(out.contains(r#"sgl_router_worker_health{worker_url="http://p0:30000"} 1"#));
|
||||
assert!(out.contains(r#"sgl_router_worker_health{worker_url="http://d0:30000"} 0"#));
|
||||
// Circuit breaker state codes.
|
||||
assert!(out.contains(r#"sgl_router_worker_cb_state{worker_url="http://p0:30000"} 0"#));
|
||||
assert!(out.contains(r#"sgl_router_worker_cb_state{worker_url="http://d0:30000"} 1"#));
|
||||
// In-flight request counts.
|
||||
assert!(
|
||||
out.contains(r#"sgl_router_worker_inflight_requests{worker_url="http://p0:30000"} 5"#)
|
||||
);
|
||||
assert!(
|
||||
out.contains(r#"sgl_router_worker_inflight_requests{worker_url="http://d0:30000"} 0"#)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn render_without_workers_emits_no_per_worker_series() {
|
||||
let reg = MetricsRegistry::new();
|
||||
let out = reg.render();
|
||||
// Headers present, but no per-worker series lines.
|
||||
assert!(out.contains("# TYPE sgl_router_worker_health gauge"));
|
||||
assert!(!out.contains("sgl_router_worker_health{"));
|
||||
assert!(!out.contains("sgl_router_worker_cb_state{"));
|
||||
assert!(!out.contains("sgl_router_worker_inflight_requests{"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -6,7 +6,9 @@ use crate::policies::registry::{PdPoolResolver, PdResolveError};
|
||||
use crate::policies::SelectionContext;
|
||||
use crate::server::app_context::AppContext;
|
||||
use crate::server::error::ApiError;
|
||||
use crate::server::metrics::{RequestOutcome, StaleRequestOutcome, WorkerModeLabel};
|
||||
use crate::server::metrics::{
|
||||
MetricsRegistry, RequestOutcome, StaleRequestOutcome, WorkerModeLabel,
|
||||
};
|
||||
use crate::workers::{LoadGuard, Worker};
|
||||
use axum::body::Body;
|
||||
use axum::extract::State;
|
||||
@@ -66,6 +68,27 @@ struct RequestProbe {
|
||||
model: Option<String>,
|
||||
}
|
||||
|
||||
/// RAII guard that records `sgl_router_request_duration_seconds` when
|
||||
/// dropped. For streaming requests the handler returns at response-headers
|
||||
/// time, so recording end-to-end latency at the dispatch site would capture
|
||||
/// only time-to-headers (≈ TTFT). Instead this guard is packed into the SSE
|
||||
/// pump's `stream_guards`, so it drops — and records — when the stream
|
||||
/// completes (or the client disconnects), yielding true end-to-end latency.
|
||||
/// Non-streaming requests record at the dispatch site directly (the body is
|
||||
/// already buffered there) and do not use this guard.
|
||||
struct RecordDurationOnDrop {
|
||||
metrics: Arc<MetricsRegistry>,
|
||||
model: String,
|
||||
start: std::time::Instant,
|
||||
}
|
||||
|
||||
impl Drop for RecordDurationOnDrop {
|
||||
fn drop(&mut self) {
|
||||
self.metrics
|
||||
.observe_request_duration(&self.model, self.start.elapsed().as_secs_f64());
|
||||
}
|
||||
}
|
||||
|
||||
/// POST /v1/chat/completions — parse model from body, select a healthy
|
||||
/// worker via the per-model policy, then proxy the request. If the
|
||||
/// request opts into streaming (`stream: true`), we pipe SSE bytes back;
|
||||
@@ -221,6 +244,30 @@ pub async fn chat_completions(
|
||||
};
|
||||
let metrics_model = model_str.clone();
|
||||
|
||||
// Builds the time-to-first-token hook the SSE pump fires when the first
|
||||
// upstream chunk lands. Installed only on the streaming arms below —
|
||||
// non-streaming "first token" equals total latency, already captured by
|
||||
// `sgl_router_request_duration_seconds`. The proxy drops the hook for
|
||||
// non-2xx responses so error bodies don't pollute TTFT.
|
||||
let make_ttft_hook = || -> Box<dyn FnOnce() + Send + 'static> {
|
||||
let metrics = Arc::clone(&ctx.metrics);
|
||||
let model = metrics_model.clone();
|
||||
let started = start;
|
||||
Box::new(move || {
|
||||
metrics.observe_ttft(&model, started.elapsed().as_secs_f64());
|
||||
})
|
||||
};
|
||||
|
||||
// Builds the end-to-end-latency guard for streaming requests. Packed into
|
||||
// `stream_guards` so it records when the SSE pump finishes (stream end or
|
||||
// client disconnect), not at response-headers time. Non-streaming records
|
||||
// at the dispatch site instead (see below).
|
||||
let make_duration_guard = || RecordDurationOnDrop {
|
||||
metrics: Arc::clone(&ctx.metrics),
|
||||
model: metrics_model.clone(),
|
||||
start,
|
||||
};
|
||||
|
||||
let result = if let Some(decode_worker) = decode_peer {
|
||||
// PD-disagg dispatch (Pattern B — spawn prefill, await decode).
|
||||
//
|
||||
@@ -310,7 +357,8 @@ pub async fn chat_completions(
|
||||
// cache-aware-zmq decisions on the decode side.
|
||||
let decode_guard = decode_worker.load_guard();
|
||||
if streaming {
|
||||
let stream_guards: Box<dyn Send + 'static> = Box::new(decode_guard);
|
||||
let stream_guards: Box<dyn Send + 'static> =
|
||||
Box::new((decode_guard, make_duration_guard()));
|
||||
let fetch = ctx.proxy.forward_streaming_to(
|
||||
&decode_worker.url,
|
||||
&decode_worker.breaker,
|
||||
@@ -318,6 +366,7 @@ pub async fn chat_completions(
|
||||
&headers,
|
||||
injected_body,
|
||||
Some(stream_guards),
|
||||
Some(make_ttft_hook()),
|
||||
);
|
||||
tokio::select! {
|
||||
biased;
|
||||
@@ -343,7 +392,8 @@ pub async fn chat_completions(
|
||||
// Plain mode, streaming. Both guards ride the SSE pump until
|
||||
// the body completes — see the matching comment in the
|
||||
// non-streaming arm.
|
||||
let stream_guards: Box<dyn Send + 'static> = Box::new((guard, active_guard));
|
||||
let stream_guards: Box<dyn Send + 'static> =
|
||||
Box::new((guard, active_guard, make_duration_guard()));
|
||||
let fetch = ctx.proxy.forward_streaming_to(
|
||||
&worker.url,
|
||||
&worker.breaker,
|
||||
@@ -351,6 +401,7 @@ pub async fn chat_completions(
|
||||
&headers,
|
||||
body,
|
||||
Some(stream_guards),
|
||||
Some(make_ttft_hook()),
|
||||
);
|
||||
// Bias `fetch` over the cancellation branch: a successful
|
||||
// response that completes in the same poll as the token firing
|
||||
@@ -423,6 +474,20 @@ pub async fn chat_completions(
|
||||
Ok(resp) => resp.status().as_u16(),
|
||||
Err(e) => e.status_code().as_u16(),
|
||||
};
|
||||
|
||||
// Record the client-visible HTTP status now that the outcome is known.
|
||||
// For non-streaming requests the body is already buffered here, so
|
||||
// `start.elapsed()` is true end-to-end latency — record it directly. For
|
||||
// streaming, the body is still pending; the `RecordDurationOnDrop` guard
|
||||
// packed into `stream_guards` records it at stream completion instead (so
|
||||
// we don't capture only time-to-headers). `elapsed` still feeds the
|
||||
// access-log `latency_ms` below for both.
|
||||
let elapsed = start.elapsed();
|
||||
if !streaming {
|
||||
ctx.metrics
|
||||
.observe_request_duration(&metrics_model, elapsed.as_secs_f64());
|
||||
}
|
||||
ctx.metrics.record_response(http_status);
|
||||
let outcome_str = match outcome {
|
||||
RequestOutcome::Success => "success",
|
||||
RequestOutcome::Error => "error",
|
||||
@@ -437,7 +502,7 @@ pub async fn chat_completions(
|
||||
outcome = outcome_str,
|
||||
http_status,
|
||||
stream = streaming,
|
||||
latency_ms = start.elapsed().as_millis() as u64,
|
||||
latency_ms = elapsed.as_millis() as u64,
|
||||
"chat_completions",
|
||||
);
|
||||
|
||||
|
||||
@@ -9,7 +9,9 @@
|
||||
//! while the router is warming up so the "router started but no workers
|
||||
//! discovered" failure mode is observable.
|
||||
|
||||
use crate::discovery::WorkerMode;
|
||||
use crate::server::app_context::AppContext;
|
||||
use crate::server::metrics::WorkerSnapshot;
|
||||
use axum::extract::State;
|
||||
use axum::http::header::CONTENT_TYPE;
|
||||
use axum::http::StatusCode;
|
||||
@@ -20,7 +22,34 @@ use std::sync::Arc;
|
||||
const PROMETHEUS_CONTENT_TYPE: &str = "text/plain; version=0.0.4; charset=utf-8";
|
||||
|
||||
pub async fn metrics(State(ctx): State<Arc<AppContext>>) -> impl IntoResponse {
|
||||
let body = ctx.metrics.render();
|
||||
// Sample the live registry into a snapshot for the worker gauges. These
|
||||
// are pull-on-scrape (not pushed) so removed workers stop emitting series
|
||||
// immediately; see `MetricsRegistry::render_with_workers`.
|
||||
let workers: Vec<WorkerSnapshot> = ctx
|
||||
.registry
|
||||
.all()
|
||||
.into_iter()
|
||||
.map(|w| {
|
||||
// One lock acquisition for both health + state so the two gauges
|
||||
// can't report a torn (self-contradictory) pair for one scrape.
|
||||
let cb = w.breaker.snapshot();
|
||||
WorkerSnapshot {
|
||||
worker_url: w.url.clone(),
|
||||
mode: match w.mode() {
|
||||
WorkerMode::Plain => "plain",
|
||||
WorkerMode::Prefill => "prefill",
|
||||
WorkerMode::Decode => "decode",
|
||||
},
|
||||
healthy: cb.admit,
|
||||
cb_state: cb.state_code,
|
||||
// Saturating rather than `as i64`: a guard-accounting
|
||||
// underflow would wrap usize and render as a nonsensical
|
||||
// negative gauge; clamp to a large positive ceiling instead.
|
||||
inflight: i64::try_from(w.active_load()).unwrap_or(i64::MAX),
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
let body = ctx.metrics.render_with_workers(&workers);
|
||||
(
|
||||
StatusCode::OK,
|
||||
[(CONTENT_TYPE, PROMETHEUS_CONTENT_TYPE)],
|
||||
@@ -96,4 +125,43 @@ mod tests {
|
||||
"metrics did not include the recorded worker_url; got:\n{body}",
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn metrics_endpoint_samples_worker_gauges_from_registry() {
|
||||
use crate::discovery::{ModelId, WorkerId, WorkerMode, WorkerSpec};
|
||||
|
||||
let ctx = Arc::new(AppContext::stub());
|
||||
ctx.registry
|
||||
.add(WorkerSpec {
|
||||
id: WorkerId("p0".into()),
|
||||
url: "http://p0:30000".into(),
|
||||
mode: WorkerMode::Prefill,
|
||||
model_ids: vec![ModelId("m".into())],
|
||||
bootstrap_port: None,
|
||||
})
|
||||
.unwrap();
|
||||
let app = crate::server::app::build_router(ctx.clone());
|
||||
let res = app
|
||||
.oneshot(
|
||||
Request::builder()
|
||||
.uri("/metrics")
|
||||
.body(Body::empty())
|
||||
.unwrap(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let body = res.into_body().collect().await.unwrap().to_bytes();
|
||||
let body = std::str::from_utf8(&body).unwrap();
|
||||
// Pool size reflects the registered prefill worker, and the per-worker
|
||||
// gauges are sampled (fresh breaker => healthy, closed, 0 inflight).
|
||||
assert!(
|
||||
body.contains(r#"sgl_router_workers{mode="prefill"} 1"#),
|
||||
"got:\n{body}"
|
||||
);
|
||||
assert!(body.contains(r#"sgl_router_worker_health{worker_url="http://p0:30000"} 1"#));
|
||||
assert!(body.contains(r#"sgl_router_worker_cb_state{worker_url="http://p0:30000"} 0"#));
|
||||
assert!(
|
||||
body.contains(r#"sgl_router_worker_inflight_requests{worker_url="http://p0:30000"} 0"#)
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -197,11 +197,17 @@ impl WorkerRegistry {
|
||||
self.by_id.get(id).map(|w| Arc::clone(&w))
|
||||
}
|
||||
|
||||
/// Snapshot of every registered worker, across all models and modes.
|
||||
/// Snapshot of every registered worker, across all models and modes,
|
||||
/// regardless of breaker state. Order is unspecified (iterates the
|
||||
/// underlying `DashMap`).
|
||||
///
|
||||
/// Used by fleet-wide admin fan-out (e.g. `/flush_cache`) that targets
|
||||
/// every worker the router knows about rather than one model's pool.
|
||||
/// Order is unspecified (iterates the underlying `DashMap`).
|
||||
/// every worker the router knows about rather than one model's pool, and
|
||||
/// by the `/metrics` scrape path to render per-worker gauges
|
||||
/// (`sgl_router_worker_health`, `_cb_state`, `_inflight_requests`) plus
|
||||
/// the pool-size gauge. The metrics path samples this fresh on each
|
||||
/// scrape rather than pushing, so a removed worker stops appearing
|
||||
/// immediately.
|
||||
pub fn all(&self) -> Vec<Arc<Worker>> {
|
||||
self.by_id.iter().map(|e| Arc::clone(e.value())).collect()
|
||||
}
|
||||
@@ -278,6 +284,18 @@ mod tests {
|
||||
assert!(r.workers_for(&ModelId("m2".into())).is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn all_lists_multi_model_worker_once() {
|
||||
let r = WorkerRegistry::default();
|
||||
// "a" serves two models; `all` must still list it once,
|
||||
// unlike a per-model enumeration which would double-count.
|
||||
let _ = r.add(spec("a", WorkerMode::Plain, &["m1", "m2"]));
|
||||
let _ = r.add(spec("b", WorkerMode::Plain, &["m1"]));
|
||||
let mut urls: Vec<String> = r.all().iter().map(|w| w.url.clone()).collect();
|
||||
urls.sort();
|
||||
assert_eq!(urls, vec!["http://a:30000", "http://b:30000"]);
|
||||
}
|
||||
|
||||
/// `healthy_workers_for` must drop workers whose breaker is Open.
|
||||
/// An earlier version of this test asserted `healthy.len() == 2`
|
||||
/// against two workers with untouched breakers — i.e., it pinned
|
||||
|
||||
@@ -201,6 +201,101 @@ async fn streaming_first_chunk_before_completion() {
|
||||
assert!(bytes.windows(5).any(|w| w == b"first"));
|
||||
}
|
||||
|
||||
/// A successful (2xx) streaming request records both TTFT (fired by the SSE
|
||||
/// pump on the first chunk) and end-to-end request_duration (recorded by the
|
||||
/// drop-guard when the stream completes). End-to-end coverage of the chat
|
||||
/// handler installing the hooks — the sse-level unit tests only cover the
|
||||
/// pump primitive in isolation.
|
||||
#[tokio::test]
|
||||
async fn streaming_2xx_request_records_ttft_and_duration() {
|
||||
let chunks: Vec<&'static str> = vec![
|
||||
"data: {\"choices\":[{\"delta\":{\"content\":\"hi\"}}]}\n\n",
|
||||
"data: [DONE]\n\n",
|
||||
];
|
||||
let worker = crate::common::mock_worker::MockWorker::start(chunks).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": true
|
||||
}))
|
||||
.unwrap(),
|
||||
))
|
||||
.unwrap();
|
||||
let res = app.oneshot(req).await.unwrap();
|
||||
assert_eq!(res.status(), StatusCode::OK);
|
||||
// Draining drives the pump to completion: fires the TTFT hook on the
|
||||
// first chunk and drops the duration guard at stream end.
|
||||
let _ = res.into_body().collect().await.unwrap().to_bytes();
|
||||
// The duration guard records from the pump task; give it a beat to drop,
|
||||
// matching the active-load streaming tests' synchronization.
|
||||
tokio::time::sleep(Duration::from_millis(20)).await;
|
||||
|
||||
let m = ctx.metrics.render();
|
||||
assert!(
|
||||
m.contains(r#"sgl_router_ttft_seconds_count{model_id="tiny"} 1"#),
|
||||
"TTFT must be recorded once for a 2xx streaming request; got:\n{m}",
|
||||
);
|
||||
assert!(
|
||||
m.contains(r#"sgl_router_request_duration_seconds_count{model_id="tiny"} 1"#),
|
||||
"request_duration must be recorded at stream completion; got:\n{m}",
|
||||
);
|
||||
}
|
||||
|
||||
/// A non-2xx streaming response must NOT record TTFT (the error body is not a
|
||||
/// generated token — the gate lives in `Proxy::forward_streaming_to`), but it
|
||||
/// MUST still record request_duration (latency of a failed request matters)
|
||||
/// and the response status. Guards the 2xx-gating decision end-to-end.
|
||||
#[tokio::test]
|
||||
async fn streaming_5xx_request_records_duration_and_status_but_not_ttft() {
|
||||
let worker = crate::common::mock_worker::MockWorker::start_returning_error(
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
serde_json::json!({"error": "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": true
|
||||
}))
|
||||
.unwrap(),
|
||||
))
|
||||
.unwrap();
|
||||
let res = app.oneshot(req).await.unwrap();
|
||||
assert_eq!(res.status(), StatusCode::INTERNAL_SERVER_ERROR);
|
||||
let _ = res.into_body().collect().await;
|
||||
tokio::time::sleep(Duration::from_millis(20)).await;
|
||||
|
||||
let m = ctx.metrics.render();
|
||||
assert!(
|
||||
!m.contains("sgl_router_ttft_seconds_count{"),
|
||||
"TTFT must NOT be recorded for a non-2xx streaming response; got:\n{m}",
|
||||
);
|
||||
assert!(
|
||||
m.contains(r#"sgl_router_responses_total{status_code="500"} 1"#),
|
||||
"the 500 status must be counted; got:\n{m}",
|
||||
);
|
||||
assert!(
|
||||
m.contains(r#"sgl_router_request_duration_seconds_count{model_id="tiny"} 1"#),
|
||||
"request_duration must be recorded even for a failed streaming request; got:\n{m}",
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn concurrent_streams_are_isolated() {
|
||||
let chunks_a: Vec<&'static str> = vec![
|
||||
@@ -803,6 +898,7 @@ async fn forward_streaming_to_records_failure_on_mid_stream_drop() {
|
||||
&headers,
|
||||
body,
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await;
|
||||
|
||||
|
||||
Reference in New Issue
Block a user