diff --git a/experimental/sgl-router/monitoring/README.md b/experimental/sgl-router/monitoring/README.md new file mode 100644 index 000000000..048dceed8 --- /dev/null +++ b/experimental/sgl-router/monitoring/README.md @@ -0,0 +1,69 @@ + + +# 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. diff --git a/experimental/sgl-router/monitoring/grafana-dashboard.json b/experimental/sgl-router/monitoring/grafana-dashboard.json new file mode 100644 index 000000000..91ae2b084 --- /dev/null +++ b/experimental/sgl-router/monitoring/grafana-dashboard.json @@ -0,0 +1,2093 @@ +{ + "annotations": { + "list": [ + { + "builtIn": 1, + "datasource": { + "type": "grafana", + "uid": "-- Grafana --" + }, + "enable": true, + "hide": true, + "iconColor": "rgba(0, 211, 255, 1)", + "name": "Annotations & Alerts", + "type": "dashboard" + } + ] + }, + "editable": true, + "fiscalYearStartMonth": 0, + "graphTooltip": 1, + "links": [], + "panels": [ + { + "type": "row", + "title": "Overview", + "collapsed": false, + "gridPos": { + "x": 0, + "y": 0, + "w": 24, + "h": 1 + }, + "id": 1, + "panels": [] + }, + { + "type": "stat", + "title": "Request rate", + "description": "Chat-completions dispatched to workers per second.", + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "x": 0, + "y": 1, + "w": 6, + "h": 4 + }, + "id": 2, + "fieldConfig": { + "defaults": { + "unit": "reqps", + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "options": { + "colorMode": "value", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "textMode": "auto" + }, + "pluginVersion": "11.6.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum(rate(sgl_router_requests_total{model_id=~\"$model_id\",worker_url=~\"$worker_url\"}[$__rate_interval]))", + "range": true, + "refId": "A", + "legendFormat": "__auto", + "format": "time_series" + } + ] + }, + { + "type": "stat", + "title": "Error ratio", + "description": "Share of dispatches whose outcome=error.", + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "x": 6, + "y": 1, + "w": 6, + "h": 4 + }, + "id": 3, + "fieldConfig": { + "defaults": { + "unit": "percent", + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "yellow", + "value": 1 + }, + { + "color": "red", + "value": 5 + } + ] + } + }, + "overrides": [] + }, + "options": { + "colorMode": "value", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "textMode": "auto" + }, + "pluginVersion": "11.6.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "100 * sum(rate(sgl_router_requests_total{outcome=\"error\",model_id=~\"$model_id\",worker_url=~\"$worker_url\"}[$__rate_interval])) / clamp_min(sum(rate(sgl_router_requests_total{model_id=~\"$model_id\",worker_url=~\"$worker_url\"}[$__rate_interval])), 1e-9)", + "range": true, + "refId": "A", + "legendFormat": "__auto", + "format": "time_series" + } + ] + }, + { + "type": "stat", + "title": "P99 latency", + "description": "99th percentile end-to-end request latency.", + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "x": 12, + "y": 1, + "w": 6, + "h": 4 + }, + "id": 4, + "fieldConfig": { + "defaults": { + "unit": "s", + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "yellow", + "value": 1 + }, + { + "color": "red", + "value": 5 + } + ] + } + }, + "overrides": [] + }, + "options": { + "colorMode": "value", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "textMode": "auto" + }, + "pluginVersion": "11.6.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "histogram_quantile(0.99, sum by (le) (rate(sgl_router_request_duration_seconds_bucket{model_id=~\"$model_id\"}[$__rate_interval])))", + "range": true, + "refId": "A", + "legendFormat": "__auto", + "format": "time_series" + } + ] + }, + { + "type": "stat", + "title": "Healthy workers", + "description": "Workers whose circuit breaker currently admits requests.", + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "x": 18, + "y": 1, + "w": 6, + "h": 4 + }, + "id": 5, + "fieldConfig": { + "defaults": { + "unit": "short", + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "red", + "value": null + }, + { + "color": "green", + "value": 1 + } + ] + } + }, + "overrides": [] + }, + "options": { + "colorMode": "value", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "textMode": "auto" + }, + "pluginVersion": "11.6.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum(sgl_router_worker_health)", + "range": true, + "refId": "A", + "legendFormat": "__auto", + "format": "time_series" + } + ] + }, + { + "type": "row", + "title": "Traffic & Errors", + "collapsed": false, + "gridPos": { + "x": 0, + "y": 5, + "w": 24, + "h": 1 + }, + "id": 6, + "panels": [] + }, + { + "type": "timeseries", + "title": "Request rate by outcome", + "description": "success / error / cancelled.", + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "x": 0, + "y": 6, + "w": 12, + "h": 8 + }, + "id": 7, + "fieldConfig": { + "defaults": { + "unit": "reqps", + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "normal" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "11.6.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by (outcome) (rate(sgl_router_requests_total{model_id=~\"$model_id\",worker_url=~\"$worker_url\"}[$__rate_interval]))", + "range": true, + "refId": "A", + "legendFormat": "{{outcome}}", + "format": "time_series" + } + ] + }, + { + "type": "timeseries", + "title": "Request rate by mode", + "description": "prefill / decode / plain dispatch split.", + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "x": 12, + "y": 6, + "w": 12, + "h": 8 + }, + "id": 8, + "fieldConfig": { + "defaults": { + "unit": "reqps", + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "normal" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "11.6.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by (mode) (rate(sgl_router_requests_total{model_id=~\"$model_id\",worker_url=~\"$worker_url\"}[$__rate_interval]))", + "range": true, + "refId": "A", + "legendFormat": "{{mode}}", + "format": "time_series" + } + ] + }, + { + "type": "timeseries", + "title": "Responses by HTTP status", + "description": "Client-visible HTTP status codes (2xx/4xx/5xx, 504=stale-cancel).", + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "x": 0, + "y": 14, + "w": 12, + "h": 8 + }, + "id": 9, + "fieldConfig": { + "defaults": { + "unit": "reqps", + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "normal" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "11.6.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by (status_code) (rate(sgl_router_responses_total[$__rate_interval]))", + "range": true, + "refId": "A", + "legendFormat": "{{status_code}}", + "format": "time_series" + } + ] + }, + { + "type": "timeseries", + "title": "Request rate by worker", + "description": "Per-worker dispatch rate \u2014 reveals routing imbalance.", + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "x": 12, + "y": 14, + "w": 12, + "h": 8 + }, + "id": 10, + "fieldConfig": { + "defaults": { + "unit": "reqps", + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "11.6.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by (worker_url) (rate(sgl_router_requests_total{model_id=~\"$model_id\",worker_url=~\"$worker_url\"}[$__rate_interval]))", + "range": true, + "refId": "A", + "legendFormat": "{{worker_url}}", + "format": "time_series" + } + ] + }, + { + "type": "row", + "title": "Latency", + "collapsed": false, + "gridPos": { + "x": 0, + "y": 22, + "w": 24, + "h": 1 + }, + "id": 11, + "panels": [] + }, + { + "type": "timeseries", + "title": "Request latency quantiles", + "description": "End-to-end request latency percentiles.", + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "x": 0, + "y": 23, + "w": 8, + "h": 8 + }, + "id": 12, + "fieldConfig": { + "defaults": { + "unit": "s", + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 0, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "11.6.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "histogram_quantile(0.5, sum by (le) (rate(sgl_router_request_duration_seconds_bucket{model_id=~\"$model_id\"}[$__rate_interval])))", + "range": true, + "refId": "A", + "legendFormat": "p50", + "format": "time_series" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "histogram_quantile(0.9, sum by (le) (rate(sgl_router_request_duration_seconds_bucket{model_id=~\"$model_id\"}[$__rate_interval])))", + "range": true, + "refId": "B", + "legendFormat": "p90", + "format": "time_series" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "histogram_quantile(0.99, sum by (le) (rate(sgl_router_request_duration_seconds_bucket{model_id=~\"$model_id\"}[$__rate_interval])))", + "range": true, + "refId": "C", + "legendFormat": "p99", + "format": "time_series" + } + ] + }, + { + "type": "timeseries", + "title": "TTFT quantiles (streaming)", + "description": "Time to first token (first upstream chunk) for streaming requests.", + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "x": 8, + "y": 23, + "w": 8, + "h": 8 + }, + "id": 13, + "fieldConfig": { + "defaults": { + "unit": "s", + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 0, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "11.6.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "histogram_quantile(0.5, sum by (le) (rate(sgl_router_ttft_seconds_bucket{model_id=~\"$model_id\"}[$__rate_interval])))", + "range": true, + "refId": "A", + "legendFormat": "p50", + "format": "time_series" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "histogram_quantile(0.9, sum by (le) (rate(sgl_router_ttft_seconds_bucket{model_id=~\"$model_id\"}[$__rate_interval])))", + "range": true, + "refId": "B", + "legendFormat": "p90", + "format": "time_series" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "histogram_quantile(0.99, sum by (le) (rate(sgl_router_ttft_seconds_bucket{model_id=~\"$model_id\"}[$__rate_interval])))", + "range": true, + "refId": "C", + "legendFormat": "p99", + "format": "time_series" + } + ] + }, + { + "type": "heatmap", + "title": "Request latency distribution", + "description": "Latency histogram over time.", + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "x": 16, + "y": 23, + "w": 8, + "h": 8 + }, + "id": 14, + "options": { + "calculate": false, + "cellGap": 1, + "color": { + "scheme": "Spectral", + "mode": "scheme", + "steps": 64, + "reverse": false, + "exponent": 0.5, + "fill": "dark-orange" + }, + "yAxis": { + "unit": "s", + "axisPlacement": "left" + }, + "legend": { + "show": true + }, + "tooltip": { + "mode": "single", + "yHistogram": true, + "showColorScale": true + }, + "rowsFrame": { + "layout": "auto" + }, + "cellValues": {}, + "filterValues": { + "le": 1e-09 + } + }, + "pluginVersion": "11.6.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by (le) (rate(sgl_router_request_duration_seconds_bucket{model_id=~\"$model_id\"}[$__rate_interval]))", + "format": "heatmap", + "legendFormat": "{{le}}", + "range": true, + "refId": "A" + } + ] + }, + { + "type": "row", + "title": "Workers & Health", + "collapsed": false, + "gridPos": { + "x": 0, + "y": 31, + "w": 24, + "h": 1 + }, + "id": 15, + "panels": [] + }, + { + "type": "timeseries", + "title": "Workers by mode", + "description": "Registered worker count per mode.", + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "x": 0, + "y": 32, + "w": 6, + "h": 8 + }, + "id": 16, + "fieldConfig": { + "defaults": { + "unit": "short", + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "11.6.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sgl_router_workers", + "range": true, + "refId": "A", + "legendFormat": "{{mode}}", + "format": "time_series" + } + ] + }, + { + "type": "timeseries", + "title": "Worker health", + "description": "1 = breaker admits requests, 0 = open.", + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "x": 6, + "y": 32, + "w": 6, + "h": 8 + }, + "id": 17, + "fieldConfig": { + "defaults": { + "unit": "short", + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 0, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "11.6.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sgl_router_worker_health{worker_url=~\"$worker_url\"}", + "range": true, + "refId": "A", + "legendFormat": "{{worker_url}}", + "format": "time_series" + } + ] + }, + { + "type": "timeseries", + "title": "Circuit breaker state", + "description": "0=closed, 1=open, 2=half_open.", + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "x": 12, + "y": 32, + "w": 6, + "h": 8 + }, + "id": 18, + "fieldConfig": { + "defaults": { + "unit": "short", + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 0, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [ + { + "type": "value", + "options": { + "0": { + "text": "closed", + "color": "green" + }, + "1": { + "text": "open", + "color": "red" + }, + "2": { + "text": "half_open", + "color": "yellow" + } + } + } + ], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "11.6.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sgl_router_worker_cb_state{worker_url=~\"$worker_url\"}", + "range": true, + "refId": "A", + "legendFormat": "{{worker_url}}", + "format": "time_series" + } + ] + }, + { + "type": "timeseries", + "title": "In-flight requests per worker", + "description": "Requests currently dispatched to each worker.", + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "x": 18, + "y": 32, + "w": 6, + "h": 8 + }, + "id": 19, + "fieldConfig": { + "defaults": { + "unit": "short", + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "11.6.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sgl_router_worker_inflight_requests{worker_url=~\"$worker_url\"}", + "range": true, + "refId": "A", + "legendFormat": "{{worker_url}}", + "format": "time_series" + } + ] + }, + { + "type": "row", + "title": "Cache & Load", + "collapsed": false, + "gridPos": { + "x": 0, + "y": 40, + "w": 24, + "h": 1 + }, + "id": 20, + "panels": [] + }, + { + "type": "timeseries", + "title": "Active load \u2014 prefill tokens", + "description": "Per-worker prefill-token load tracked by the active-load policy.", + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "x": 0, + "y": 41, + "w": 8, + "h": 8 + }, + "id": 21, + "fieldConfig": { + "defaults": { + "unit": "short", + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "11.6.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sgl_router_active_load{kind=\"prefill_tokens\",worker_url=~\"$worker_url\"}", + "range": true, + "refId": "A", + "legendFormat": "{{worker_url}}", + "format": "time_series" + } + ] + }, + { + "type": "timeseries", + "title": "Active load \u2014 decode blocks", + "description": "Per-worker decode-block load tracked by the active-load policy.", + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "x": 8, + "y": 41, + "w": 8, + "h": 8 + }, + "id": 22, + "fieldConfig": { + "defaults": { + "unit": "short", + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "11.6.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sgl_router_active_load{kind=\"decode_blocks\",worker_url=~\"$worker_url\"}", + "range": true, + "refId": "A", + "legendFormat": "{{worker_url}}", + "format": "time_series" + } + ] + }, + { + "type": "timeseries", + "title": "Overlap blocks quantiles", + "description": "Cache-aware-zmq overlap-block count at policy selection.", + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "x": 16, + "y": 41, + "w": 8, + "h": 8 + }, + "id": 23, + "fieldConfig": { + "defaults": { + "unit": "short", + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 0, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "11.6.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "histogram_quantile(0.5, sum by (le) (rate(sgl_router_overlap_blocks_bucket{model_id=~\"$model_id\"}[$__rate_interval])))", + "range": true, + "refId": "A", + "legendFormat": "p50", + "format": "time_series" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "histogram_quantile(0.99, sum by (le) (rate(sgl_router_overlap_blocks_bucket{model_id=~\"$model_id\"}[$__rate_interval])))", + "range": true, + "refId": "B", + "legendFormat": "p99", + "format": "time_series" + } + ] + }, + { + "type": "row", + "title": "Routing Policy", + "collapsed": false, + "gridPos": { + "x": 0, + "y": 49, + "w": 24, + "h": 1 + }, + "id": 24, + "panels": [] + }, + { + "type": "timeseries", + "title": "Sticky-session outcomes", + "description": "hit / assigned / remap / no_routing_key.", + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "x": 0, + "y": 50, + "w": 8, + "h": 8 + }, + "id": 25, + "fieldConfig": { + "defaults": { + "unit": "reqps", + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "normal" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "11.6.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by (outcome) (rate(sgl_router_sticky_total[$__rate_interval]))", + "range": true, + "refId": "A", + "legendFormat": "{{outcome}}", + "format": "time_series" + } + ] + }, + { + "type": "timeseries", + "title": "Decode-affinity outcomes", + "description": "same_host_picked / fallback_breaker / fallback_load_imbalance.", + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "x": 8, + "y": 50, + "w": 8, + "h": 8 + }, + "id": 26, + "fieldConfig": { + "defaults": { + "unit": "reqps", + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "normal" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "11.6.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by (outcome) (rate(sgl_router_decode_affinity_total[$__rate_interval]))", + "range": true, + "refId": "A", + "legendFormat": "{{outcome}}", + "format": "time_series" + } + ] + }, + { + "type": "timeseries", + "title": "Stale-request cancellations", + "description": "Stale-request cancellations fired by the janitor.", + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "x": 16, + "y": 50, + "w": 8, + "h": 8 + }, + "id": 27, + "fieldConfig": { + "defaults": { + "unit": "reqps", + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "11.6.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum(rate(sgl_router_stale_requests_total[$__rate_interval]))", + "range": true, + "refId": "A", + "legendFormat": "expired", + "format": "time_series" + } + ] + } + ], + "refresh": "30s", + "schemaVersion": 39, + "tags": [ + "sglang", + "sgl-router", + "experimental" + ], + "templating": { + "list": [ + { + "current": {}, + "hide": 0, + "includeAll": false, + "label": "Datasource", + "multi": false, + "name": "datasource", + "options": [], + "query": "prometheus", + "refresh": 1, + "regex": "", + "skipUrlSync": false, + "type": "datasource" + }, + { + "current": { + "text": "All", + "value": "$__all" + }, + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "definition": "label_values(sgl_router_requests_total, model_id)", + "hide": 0, + "includeAll": true, + "allValue": ".*", + "label": "model_id", + "multi": true, + "name": "model_id", + "options": [], + "query": { + "query": "label_values(sgl_router_requests_total, model_id)", + "refId": "StandardVariableQuery" + }, + "refresh": 2, + "regex": "", + "sort": 1, + "type": "query" + }, + { + "current": { + "text": "All", + "value": "$__all" + }, + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "definition": "label_values(sgl_router_requests_total, worker_url)", + "hide": 0, + "includeAll": true, + "allValue": ".*", + "label": "worker_url", + "multi": true, + "name": "worker_url", + "options": [], + "query": { + "query": "label_values(sgl_router_requests_total, worker_url)", + "refId": "StandardVariableQuery" + }, + "refresh": 2, + "regex": "", + "sort": 1, + "type": "query" + } + ] + }, + "time": { + "from": "now-1h", + "to": "now" + }, + "timepicker": {}, + "timezone": "", + "title": "SGLang Router (experimental)", + "uid": "sgl-router-experimental", + "version": 1, + "weekStart": "" +} diff --git a/experimental/sgl-router/src/health/circuit_breaker.rs b/experimental/sgl-router/src/health/circuit_breaker.rs index 35779f70e..d216d3ebb 100644 --- a/experimental/sgl-router/src/health/circuit_breaker.rs +++ b/experimental/sgl-router/src/health/circuit_breaker.rs @@ -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"); + } +} diff --git a/experimental/sgl-router/src/proxy/mod.rs b/experimental/sgl-router/src/proxy/mod.rs index a04877683..b01423e9d 100644 --- a/experimental/sgl-router/src/proxy/mod.rs +++ b/experimental/sgl-router/src/proxy/mod.rs @@ -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>, + on_first_byte: Option>, ) -> Result, 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( diff --git a/experimental/sgl-router/src/proxy/sse.rs b/experimental/sgl-router/src/proxy/sse.rs index 63f082718..01a83f01f 100644 --- a/experimental/sgl-router/src/proxy/sse.rs +++ b/experimental/sgl-router/src/proxy/sse.rs @@ -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( stream: S, stream_guards: Option>, on_complete: Option>, + on_first_byte: Option>, ) -> Body where S: futures::Stream> + 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 = 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::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> = 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> = 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(); diff --git a/experimental/sgl-router/src/server/metrics.rs b/experimental/sgl-router/src/server/metrics.rs index 2ecb22672..c8361b712 100644 --- a/experimental/sgl-router/src/server/metrics.rs +++ b/experimental/sgl-router/src/server/metrics.rs @@ -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>>, + // 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>, + ttft_seconds: Mutex>, + responses_total: Mutex>>, overlap_blocks: Mutex>, active_load: Mutex>>, stale_requests_total: Mutex>>, @@ -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, 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 (`_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] diff --git a/experimental/sgl-router/src/server/routes/chat.rs b/experimental/sgl-router/src/server/routes/chat.rs index b3f7ab1ec..d2ce1c1d2 100644 --- a/experimental/sgl-router/src/server/routes/chat.rs +++ b/experimental/sgl-router/src/server/routes/chat.rs @@ -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, } +/// 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, + 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 { + 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 = Box::new(decode_guard); + let stream_guards: Box = + 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 = Box::new((guard, active_guard)); + let stream_guards: Box = + 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", ); diff --git a/experimental/sgl-router/src/server/routes/metrics.rs b/experimental/sgl-router/src/server/routes/metrics.rs index 3d45897bf..79c87e561 100644 --- a/experimental/sgl-router/src/server/routes/metrics.rs +++ b/experimental/sgl-router/src/server/routes/metrics.rs @@ -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>) -> 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 = 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"#) + ); + } } diff --git a/experimental/sgl-router/src/workers/registry.rs b/experimental/sgl-router/src/workers/registry.rs index aa51c89bc..831297ab9 100644 --- a/experimental/sgl-router/src/workers/registry.rs +++ b/experimental/sgl-router/src/workers/registry.rs @@ -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> { 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 = 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 diff --git a/experimental/sgl-router/tests/proxy/chat_routing.rs b/experimental/sgl-router/tests/proxy/chat_routing.rs index 7425052b4..72f4e80d1 100644 --- a/experimental/sgl-router/tests/proxy/chat_routing.rs +++ b/experimental/sgl-router/tests/proxy/chat_routing.rs @@ -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;