Token-level timing and admission by SLO: TPOT, inter-token and tokens a second by model and tenant; a model past its SLO is saturated for the lower bands

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

BACKLOG.md

@@ -474,7 +474,12 @@ portal is not done — the page shows; the console and ctl change.
     **`server restore` (S).** Restore is manual shell today
     (`docs/install.md:1187`); add the command with version check, dry run,
     off-host copy target; a documented DR drill.
-13. **SLO-based admission and token-level metrics (S–M).** TPOT and
+13. **Done (branch `features-c2`):** TPOT, inter-token and tokens/s
+    histograms by model, pool and tenant; `slo` pressure in admission's
+    saturation; `llmgw_model_slo_seconds` and `llmgw-slo` alerts. Left: the
+    p99 is per control plane, from its own requests; an SLO per band or per
+    tenant; placement does not yet add replicas for an SLO (item 14).
+    **SLO-based admission and token-level metrics (S–M).** TPOT and
     inter-token histograms, output tokens/s per model and tenant in
     `router/metrics.go`; use `catalog.SLO` in `admission/saturation.go`.
 14. **Scheduled and warm-pool autoscaling (S–M).** Time-of-day min/max,

CHANGELOG.md

@@ -71,6 +71,16 @@ APIs change between minors without a deprecation period.
   - **Other ways in:** `GET /api/v1/usage/export` (CSV, or `format=json`)
     and **E** on the console's Usage view.
 
+### Token-level timing, and admission by SLO
+
+- **Per-token metrics:** `llmgw_tpot_seconds`, `llmgw_inter_token_seconds`
+  and `llmgw_output_tokens_per_second` by model and pool, and TPOT and
+  tokens a second by tenant.
+- **A model's `slo` counts in admission:** past its `ttft_p99` or
+  `tpot_p99` (the recent p99), it is treated as saturated, so the lower bands
+  wait. `llmgw_model_slo_seconds` and the `llmgw-slo` alerts make the
+  promise visible in Prometheus.
+
 ### Restoring a backup, and copies off the machine
 
 - **`llmgw server restore ARCHIVE`** puts a backup back: checks it first

deploy/monitoring/llmgw-rules.yml

@@ -555,3 +555,32 @@ groups:
             It was stopped or rewritten if the policy enforces; either way a
             model has seen it, in its prompt or its training. The usage rows
             with guard_category secret say which request.
+
+  - name: llmgw-slo
+    rules:
+      - alert: ModelTTFTOverSLO
+        expr: >-
+          histogram_quantile(0.99, sum by (le, model) (rate(llmgw_ttft_seconds_bucket[10m])))
+            > on (model) group_left llmgw_model_slo_seconds{kind="ttft_p99"}
+        for: 15m
+        labels: { severity: warning }
+        annotations:
+          summary: "{{ $labels.model }} answers its first token past its SLO at p99"
+          description: >-
+            The p99 time to the first token over ten minutes is past the
+            model's slo.ttft_p99. Admission already holds the lower bands back
+            while it is; this says it has lasted. More replicas, or less
+            traffic in the bands above.
+
+      - alert: ModelTPOTOverSLO
+        expr: >-
+          histogram_quantile(0.99, sum by (le, model) (rate(llmgw_tpot_seconds_bucket[10m])))
+            > on (model) group_left llmgw_model_slo_seconds{kind="tpot_p99"}
+        for: 15m
+        labels: { severity: warning }
+        annotations:
+          summary: "{{ $labels.model }} produces tokens slower than its SLO at p99"
+          description: >-
+            The p99 time per output token is past slo.tpot_p99: replicas are
+            batching more than they can decode quickly, or a slower card is
+            serving. llmgw_output_tokens_per_second by pool says which.

docs/install.md

@@ -1011,6 +1011,28 @@ fleet means the measurement is missing, not that the fleet is idle. And
 `llmgw_gpu_nvlink_peers` at zero everywhere means the topology is unknown,
 which is what a binary built without the nvml tag reports.
 
+### Token by token, and SLOs
+
+Beside time to the first token, every streamed answer is timed per token:
+`llmgw_tpot_seconds` (each output token after the first),
+`llmgw_inter_token_seconds` (the gaps between streamed events, which show the
+stalls a mean hides) and `llmgw_output_tokens_per_second`, by model and pool,
+and `llmgw_tenant_tpot_seconds` and `llmgw_tenant_output_tokens_per_second` by
+tenant.
+
+A model can say what it promises:
+
+```yaml
+slo: {ttft_p99: 2s, tpot_p99: 80ms}
+```
+
+Then its recent p99 — the last five minutes, once there are twenty streamed
+requests — counts in admission: a model past its SLO is treated as saturated
+by as much as it is past, so the lower bands wait while the queue and the
+cache still look healthy. `llmgw_model_slo_seconds` publishes the SLOs, and
+the `llmgw-slo` alerts in deploy/monitoring fire when a p99 stays past one for
+fifteen minutes.
+
 ### Being told
 
 Prometheus alerts travel once `alertmanager.yml` names a receiver; it ships

internal/router/handlers.go

@@ -605,6 +605,9 @@ func (r *Router) finish(w http.ResponseWriter, req *parsedRequest, res dispatchR
 	}
 
 	r.m.observeRequest(r.metricModel(req.Model), res.pool, status, time.Since(start), res.outcome.Usage, res.outcome.TTFT)
+	if status < 400 {
+		r.observeTokens(req, res, time.Since(start))
+	}
 	r.account(req, res, status, time.Since(start))
 	r.accessLog(req, res, status, time.Since(start))
 	return status

internal/router/metrics.go

@@ -1,6 +1,7 @@
 package router
 
 import (
+	"slices"
 	"strconv"
 	"time"
 
@@ -32,6 +33,7 @@ type metrics struct {
 
 	tenant *tenantMetrics
 	guard  *guardMetrics
+	token  *tokenMetrics
 }
 
 func newMetrics() *metrics {
@@ -91,6 +93,7 @@ func newMetrics() *metrics {
 		}, []string{"band", "reason"}),
 		tenant: newTenantMetrics(),
 		guard:  newGuardMetrics(),
+		token:  newTokenMetrics(),
 	}
 }
 
@@ -100,7 +103,7 @@ func (m *metrics) collectors() []prometheus.Collector {
 		m.requests, m.duration, m.ttft, m.tokens,
 		m.scorerValue, m.filtered, m.considered,
 		m.admissionWait, m.admissionRejected, m.prefixDecisions,
-	}, append(m.tenant.collectors(), m.guard.collectors()...)...)
+	}, slices.Concat(m.tenant.collectors(), m.guard.collectors(), m.token.collectors())...)
 }
 
 // observePrefix records which branch a prefix decision took.

internal/router/proxy/proxy.go

@@ -168,6 +168,9 @@ type Request struct {
 	// ContentType is the body's, when it is not JSON: an audio upload's
 	// multipart form.
 	ContentType string
+	// OnEvent, when set, is told the time since the previous streamed event
+	// for each event after the first: about one token's time each.
+	OnEvent func(gap time.Duration)
 
 	// Filter, when set, sees a whole non-streamed answer and the headers
 	// about to be sent with it, and returns the body to send: a guard's
@@ -207,6 +210,9 @@ type Outcome struct {
 	Retryable bool
 	// TTFT is the time until the first streamed event, zero if none arrived.
 	TTFT time.Duration
+	// Decode is the time from the first streamed event to the last: what
+	// the tokens after the first took.
+	Decode time.Duration
 	// Usage is the engine-reported token usage, if it was present.
 	Usage *Usage
 	// Bytes is how many bytes were relayed to the client.
@@ -351,6 +357,7 @@ func (p *Proxy) relayStream(ctx context.Context, abort context.CancelFunc, w htt
 
 	br := bufio.NewReaderSize(resp.Body, 16<<10)
 	var ev eventBuffer
+	var last time.Time
 
 	for {
 		line, err := br.ReadString('\n')
@@ -359,10 +366,17 @@ func (p *Proxy) relayStream(ctx context.Context, abort context.CancelFunc, w htt
 
 			ev.add(line)
 			if ev.complete() {
+				now := time.Now()
 				if out.TTFT == 0 {
-					out.TTFT = time.Since(start)
+					out.TTFT = now.Sub(start)
 					wd.setInterval(p.cfg.IdleTimeout)
+				} else {
+					if r.OnEvent != nil {
+						r.OnEvent(now.Sub(last))
+					}
+					out.Decode = now.Sub(start) - out.TTFT
 				}
+				last = now
 				if r.CaptureReply > 0 && r.StreamFilter == nil {
 					reply.add(replyDelta(ev.data.String()))
 				}

internal/router/router.go

@@ -68,6 +68,8 @@ type Router struct {
 	// canaries counts how the two sides served.
 	canaryTargets atomic.Pointer[map[string]string]
 	canaries      canaryStats
+	// slo keeps each model's recent token timings, for admission.
+	slo sloTracker
 	// batchWake has the batch worker look for work at once.
 	batchWake chan struct{}
 	// reqLogs is each project's keeping of requests, as last read.
@@ -400,10 +402,14 @@ func (r *Router) saturation(model string, outstanding int) (float64, bool) {
 	for _, e := range eps {
 		dispatched += e.Inflight
 	}
-	return admission.Saturation(eps, max(0, outstanding-dispatched), admission.Thresholds{
+	sat := admission.Saturation(eps, max(0, outstanding-dispatched), admission.Thresholds{
 		TargetWaitingPerReplica: r.cfg.Scoring.TargetWaiting,
 		KVCeiling:               r.cfg.Scoring.KVCeiling,
-	}), true
+	})
+	// A model with an SLO is as saturated as it is past it: the queue and
+	// the cache can look healthy while answers come too slowly, and that
+	// is when the lower bands should wait.
+	return max(sat, r.sloPressure(model)), true
 }
 
 func toInfos(eps []config.Endpoint) []endpoint.Info {
@@ -655,6 +661,9 @@ func (r *Router) storeModels(models []catalog.Model) {
 		}
 	}
 	r.canaryTargets.Store(&targets)
+	if r.m != nil {
+		r.m.token.setSLOs(models)
+	}
 }
 
 // adapterRef is where an adapter runs and how it is applied.
@@ -956,6 +965,10 @@ func (r *Router) dispatchOne(ctx context.Context, w http.ResponseWriter, req *pa
 			CaptureReply:       req.captureBytes(),
 			ContentType:        req.ContentType,
 		}
+		if req.Stream {
+			gaps := r.m.token.gap.WithLabelValues(r.metricModel(model))
+			preq.OnEvent = func(d time.Duration) { gaps.Observe(d.Seconds()) }
+		}
 		// A filter per attempt, for the model this attempt is served by.
 		filter := r.outputFilter(uctx, req, model, &preq)
 		out, perr := r.px.Do(uctx, w, preq)

internal/router/tokenmetrics.go

@@ -0,0 +1,227 @@
+package router
+
+import (
+	"slices"
+	"sync"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/router/proxy"
+)
+
+// Token-level timing: how long each output token took, the gaps between
+// streamed tokens, and output tokens a second — per model and per tenant.
+// What a model's SLO is written in, and what admission reads back.
+
+type tokenMetrics struct {
+	tpot       *prometheus.HistogramVec
+	gap        *prometheus.HistogramVec
+	rate       *prometheus.HistogramVec
+	tenantTPOT *prometheus.HistogramVec
+	tenantRate *prometheus.HistogramVec
+	slo        *prometheus.GaugeVec
+}
+
+var tokenBuckets = []float64{0.005, 0.01, 0.02, 0.035, 0.05, 0.075, 0.1, 0.15, 0.25, 0.5, 1, 2}
+
+func newTokenMetrics() *tokenMetrics {
+	rateBuckets := []float64{1, 2, 5, 10, 15, 20, 30, 50, 75, 100, 150, 250}
+	return &tokenMetrics{
+		tpot: prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Name:    "llmgw_tpot_seconds",
+			Help:    "Time per output token after the first, per streamed request: what a model's slo.tpot_p99 is written in.",
+			Buckets: tokenBuckets,
+		}, []string{"model", "pool"}),
+		gap: prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Name:    "llmgw_inter_token_seconds",
+			Help:    "Time between streamed events, about one token each: the stalls a mean per-token time hides.",
+			Buckets: tokenBuckets,
+		}, []string{"model"}),
+		rate: prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Name:    "llmgw_output_tokens_per_second",
+			Help:    "Output tokens a second per request: after the first token when streamed, over the whole request otherwise.",
+			Buckets: rateBuckets,
+		}, []string{"model", "pool"}),
+		tenantTPOT: prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Name:    "llmgw_tenant_tpot_seconds",
+			Help:    "Time per output token by tenant (the level router.tenant_label names), for streamed requests.",
+			Buckets: tokenBuckets,
+		}, []string{"tenant"}),
+		slo: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Name: "llmgw_model_slo_seconds",
+			Help: "A model's SLO from the catalog, by kind: ttft_p99 or tpot_p99. For alerting against the histograms.",
+		}, []string{"model", "kind"}),
+		tenantRate: prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Name:    "llmgw_tenant_output_tokens_per_second",
+			Help:    "Output tokens a second per request, by tenant.",
+			Buckets: rateBuckets,
+		}, []string{"tenant"}),
+	}
+}
+
+func (t *tokenMetrics) collectors() []prometheus.Collector {
+	return []prometheus.Collector{t.tpot, t.gap, t.rate, t.tenantTPOT, t.tenantRate, t.slo}
+}
+
+// tokenTiming is one finished request's per-token figures; ok is false when
+// there are none to give.
+func tokenTiming(streamed bool, o proxy.Outcome, took time.Duration) (tpot time.Duration, rate float64, ok bool) {
+	if o.Usage == nil || o.Usage.CompletionTokens <= 0 {
+		return 0, 0, false
+	}
+	n := o.Usage.CompletionTokens
+	if streamed {
+		if n < 2 || o.Decode <= 0 {
+			return 0, 0, false
+		}
+		tpot = o.Decode / time.Duration(n-1)
+		return tpot, float64(n-1) / o.Decode.Seconds(), true
+	}
+	if took <= 0 {
+		return 0, 0, false
+	}
+	return 0, float64(n) / took.Seconds(), true
+}
+
+func (r *Router) observeTokens(req *parsedRequest, res dispatchResult, took time.Duration) {
+	tpot, rate, ok := tokenTiming(req.Stream, res.outcome, took)
+	if !ok {
+		return
+	}
+	model, pool := r.metricModel(req.Model), res.pool
+	if pool == "" {
+		pool = "none"
+	}
+	tenant := r.m.tenant.tenantOf(req)
+	if tpot > 0 {
+		r.m.token.tpot.WithLabelValues(model, pool).Observe(tpot.Seconds())
+		if tenant != "" {
+			r.m.token.tenantTPOT.WithLabelValues(tenant).Observe(tpot.Seconds())
+		}
+	}
+	r.m.token.rate.WithLabelValues(model, pool).Observe(rate)
+	if tenant != "" {
+		r.m.token.tenantRate.WithLabelValues(tenant).Observe(rate)
+	}
+	if req.Stream {
+		r.slo.add(r.canonicalModel(req.Model), res.outcome.TTFT, tpot)
+	}
+}
+
+func (r *Router) canonicalModel(name string) string {
+	c, _ := r.resolve(name)
+	return c
+}
+
+// sloTracker keeps each model's recent first-token and per-token times, for
+// admission to compare against its SLO.
+type sloTracker struct {
+	mu sync.Mutex
+	by map[string]*sloSamples
+}
+
+const (
+	sloSamplesKept = 256
+	sloWindow      = 5 * time.Minute
+	// sloMinSamples is how many recent requests a p99 needs before it is
+	// believed.
+	sloMinSamples = 20
+)
+
+type sloSample struct {
+	at         time.Time
+	ttft, tpot time.Duration
+}
+
+type sloSamples struct {
+	ring []sloSample
+	next int
+}
+
+func (t *sloTracker) add(model string, ttft, tpot time.Duration) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	if t.by == nil {
+		t.by = map[string]*sloSamples{}
+	}
+	s := t.by[model]
+	if s == nil {
+		s = &sloSamples{}
+		t.by[model] = s
+	}
+	x := sloSample{at: time.Now(), ttft: ttft, tpot: tpot}
+	if len(s.ring) < sloSamplesKept {
+		s.ring = append(s.ring, x)
+		return
+	}
+	s.ring[s.next] = x
+	s.next = (s.next + 1) % sloSamplesKept
+}
+
+// p99 is a model's recent 99th percentiles, and false with too few.
+func (t *sloTracker) p99(model string, now time.Time) (ttft, tpot time.Duration, ok bool) {
+	t.mu.Lock()
+	s := t.by[model]
+	var a, b []time.Duration
+	if s != nil {
+		for _, x := range s.ring {
+			if now.Sub(x.at) > sloWindow {
+				continue
+			}
+			if x.ttft > 0 {
+				a = append(a, x.ttft)
+			}
+			if x.tpot > 0 {
+				b = append(b, x.tpot)
+			}
+		}
+	}
+	t.mu.Unlock()
+	pick := func(v []time.Duration) time.Duration {
+		if len(v) < sloMinSamples {
+			return 0
+		}
+		slices.Sort(v)
+		return v[(len(v)*99)/100]
+	}
+	ttft, tpot = pick(a), pick(b)
+	return ttft, tpot, ttft > 0 || tpot > 0
+}
+
+// sloPressure is how far a model is over its SLO: 1 at it, above 1 past
+// it, 0 when it has none or too little is known.
+func (r *Router) sloPressure(model string) float64 {
+	_, entry := r.resolve(model)
+	slo := entry.SLO
+	if slo.TTFTp99 <= 0 && slo.TPOTp99 <= 0 {
+		return 0
+	}
+	ttft, tpot, ok := r.slo.p99(entry.Name, time.Now())
+	if !ok {
+		return 0
+	}
+	p := 0.0
+	if slo.TTFTp99 > 0 && ttft > 0 {
+		p = max(p, float64(ttft)/float64(slo.TTFTp99))
+	}
+	if slo.TPOTp99 > 0 && tpot > 0 {
+		p = max(p, float64(tpot)/float64(slo.TPOTp99))
+	}
+	return p
+}
+
+// setSLOs publishes the catalog's SLOs, all at once, so a model removed or
+// given none stops being reported.
+func (t *tokenMetrics) setSLOs(models []catalog.Model) {
+	t.slo.Reset()
+	for _, m := range models {
+		if m.SLO.TTFTp99 > 0 {
+			t.slo.WithLabelValues(m.Name, "ttft_p99").Set(m.SLO.TTFTp99.Seconds())
+		}
+		if m.SLO.TPOTp99 > 0 {
+			t.slo.WithLabelValues(m.Name, "tpot_p99").Set(m.SLO.TPOTp99.Seconds())
+		}
+	}
+}

internal/router/tokenmetrics_test.go

@@ -0,0 +1,88 @@
+package router
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus/testutil"
+
+	"gitea.mynodes.xyz/ilker/llmgw/internal/config"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/router/proxy"
+)
+
+func TestTokenTiming(t *testing.T) {
+	o := proxy.Outcome{Usage: &proxy.Usage{CompletionTokens: 11}, TTFT: 200 * time.Millisecond, Decode: time.Second}
+	tpot, rate, ok := tokenTiming(true, o, 2*time.Second)
+	if !ok || tpot != 100*time.Millisecond || rate != 10 {
+		t.Errorf("streamed: %s %.1f %v", tpot, rate, ok)
+	}
+	if _, rate, ok := tokenTiming(false, o, 2*time.Second); !ok || rate != 5.5 {
+		t.Errorf("whole: %.1f %v", rate, ok)
+	}
+	if _, _, ok := tokenTiming(true, proxy.Outcome{Usage: &proxy.Usage{CompletionTokens: 1}}, time.Second); ok {
+		t.Error("one token has no time per token")
+	}
+}
+
+func TestAModelPastItsSLOIsSaturatedForAdmission(t *testing.T) {
+	e := newFakeEngine(t, "m")
+	cfg := baseConfig(config.Endpoint{ID: "a", URL: e.srv.URL, Models: []string{"m"}})
+	m := catModel("m", nil)
+	m.SLO = catalog.SLO{TTFTp99: 100 * time.Millisecond, TPOTp99: 50 * time.Millisecond}
+	cfg.Models = []catalog.Model{m}
+	rt := newTestRouter(t, cfg)
+	waitForEndpoint(t, rt)
+
+	sat, _ := rt.saturation("m", 0)
+	if sat >= 1 {
+		t.Fatalf("idle and unmeasured, saturation %.2f", sat)
+	}
+	for range sloMinSamples - 1 {
+		rt.slo.add("m", 300*time.Millisecond, 20*time.Millisecond)
+	}
+	if p := rt.sloPressure("m"); p != 0 {
+		t.Errorf("believed with too few: %.2f", p)
+	}
+	rt.slo.add("m", 300*time.Millisecond, 20*time.Millisecond)
+	if sat, _ := rt.saturation("m", 0); sat < 2.9 {
+		t.Errorf("first tokens at three times the SLO read as saturation %.2f", sat)
+	}
+}
+
+func TestAStreamedAnswerIsTimedPerToken(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/v1/chat/completions" {
+			return
+		}
+		w.Header().Set("Content-Type", "text/event-stream")
+		f := w.(http.Flusher)
+		for i := range 5 {
+			fmt.Fprintf(w, "data: {\"choices\":[{\"delta\":{\"content\":\"t%d\"}}]}\n\n", i)
+			f.Flush()
+			time.Sleep(20 * time.Millisecond)
+		}
+		fmt.Fprint(w, "data: {\"choices\":[],\"usage\":{\"prompt_tokens\":3,\"completion_tokens\":5}}\n\ndata: [DONE]\n\n")
+	}))
+	t.Cleanup(srv.Close)
+	rt := newTestRouter(t, baseConfig(config.Endpoint{ID: "a", URL: srv.URL, Models: []string{"m"}}))
+	b, _ := json.Marshal(map[string]any{"model": "m", "stream": true, "messages": []any{}})
+	req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(b))
+	rec := httptest.NewRecorder()
+	rt.Handler().ServeHTTP(rec, req)
+	if rec.Code != 200 {
+		t.Fatalf("%d", rec.Code)
+	}
+	if testutil.CollectAndCount(rt.m.token.tpot) != 1 || testutil.CollectAndCount(rt.m.token.gap) != 1 ||
+		testutil.CollectAndCount(rt.m.token.rate) != 1 {
+		t.Error("not timed")
+	}
+	if _, tpot, _ := rt.slo.p99("m", time.Now()); tpot != 0 {
+		t.Error("one request was believed as a p99")
+	}
+}