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