diff --git a/BACKLOG.md b/BACKLOG.md index 3b46c91..c8878c3 100644 --- a/BACKLOG.md +++ b/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, diff --git a/CHANGELOG.md b/CHANGELOG.md index 0b537b8..d4305d6 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/deploy/monitoring/llmgw-rules.yml b/deploy/monitoring/llmgw-rules.yml index 1055462..341df17 100644 --- a/deploy/monitoring/llmgw-rules.yml +++ b/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. diff --git a/docs/install.md b/docs/install.md index aaf8fdc..a58060f 100644 --- a/docs/install.md +++ b/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 diff --git a/internal/router/handlers.go b/internal/router/handlers.go index 21fe18f..b313eaa 100644 --- a/internal/router/handlers.go +++ b/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 diff --git a/internal/router/metrics.go b/internal/router/metrics.go index cdea858..0d583d2 100644 --- a/internal/router/metrics.go +++ b/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. diff --git a/internal/router/proxy/proxy.go b/internal/router/proxy/proxy.go index 7a8e864..1415937 100644 --- a/internal/router/proxy/proxy.go +++ b/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())) } diff --git a/internal/router/router.go b/internal/router/router.go index 6776e41..50912ab 100644 --- a/internal/router/router.go +++ b/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) diff --git a/internal/router/tokenmetrics.go b/internal/router/tokenmetrics.go new file mode 100644 index 0000000..b149d5e --- /dev/null +++ b/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()) + } + } +} diff --git a/internal/router/tokenmetrics_test.go b/internal/router/tokenmetrics_test.go new file mode 100644 index 0000000..afccf71 --- /dev/null +++ b/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") + } +}