diff --git a/internal/canary/canary.go b/internal/canary/canary.go new file mode 100644 index 0000000..bc16bb3 --- /dev/null +++ b/internal/canary/canary.go @@ -0,0 +1,79 @@ +// Package canary compares a model's candidate with the model it may +// replace, from what each side served, and says when to give up on it. +package canary + +import ( + "fmt" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" +) + +// Side is what one side of a canary served since it began. +type Side struct { + Requests int64 `json:"requests"` + // Errors are answers of 500 and up: the engine's failures, not the + // client's mistakes. + Errors int64 `json:"errors"` + // TTFT is over streamed requests, Latency over the rest, both means in + // milliseconds with how many they are over. + TTFTMs float64 `json:"ttft_ms"` + TTFTCount int64 `json:"ttft_count"` + LatencyMs float64 `json:"latency_ms"` + LatencyCount int64 `json:"latency_count"` +} + +// ErrorRate is errors over requests, 0 with none. +func (v Side) ErrorRate() float64 { + if v.Requests == 0 { + return 0 + } + return float64(v.Errors) / float64(v.Requests) +} + +// Add counts one finished request. +func (v *Side) Add(status int, streamed bool, ttft, took time.Duration) { + v.Requests++ + if status >= 500 { + v.Errors++ + } + switch { + case streamed && ttft > 0: + v.TTFTMs += (float64(ttft.Milliseconds()) - v.TTFTMs) / float64(v.TTFTCount+1) + v.TTFTCount++ + case !streamed && status < 400: + v.LatencyMs += (float64(took.Milliseconds()) - v.LatencyMs) / float64(v.LatencyCount+1) + v.LatencyCount++ + } +} + +// Stat compares the two sides of one model's canary. +type Stat struct { + Canary string `json:"canary"` + Since time.Time `json:"since"` + Stable Side `json:"stable"` + Candidate Side `json:"candidate"` +} + +// minTimed is how many timed requests a side needs before its mean counts. +const minTimed = 10 + +// Verdict says whether the candidate has done worse than the canary allows, +// and how. Nothing is decided before it has served MinRequests. +func Verdict(s Stat, cn catalog.Canary) (rollback bool, why string) { + minReq, maxErr, maxRatio := cn.Limits() + c, st := s.Candidate, s.Stable + if c.Requests < int64(minReq) { + return false, "" + } + if d := c.ErrorRate() - st.ErrorRate(); d > maxErr { + return true, fmt.Sprintf("errors in %.1f%% of its requests against %.1f%%", 100*c.ErrorRate(), 100*st.ErrorRate()) + } + if c.TTFTCount >= minTimed && st.TTFTCount >= minTimed && st.TTFTMs > 0 && c.TTFTMs > maxRatio*st.TTFTMs { + return true, fmt.Sprintf("first token after %.0f ms on average against %.0f ms", c.TTFTMs, st.TTFTMs) + } + if c.LatencyCount >= minTimed && st.LatencyCount >= minTimed && st.LatencyMs > 0 && c.LatencyMs > maxRatio*st.LatencyMs { + return true, fmt.Sprintf("answers in %.0f ms on average against %.0f ms", c.LatencyMs, st.LatencyMs) + } + return false, "" +} diff --git a/internal/canary/canary_test.go b/internal/canary/canary_test.go new file mode 100644 index 0000000..f9008dc --- /dev/null +++ b/internal/canary/canary_test.go @@ -0,0 +1,42 @@ +package canary + +import ( + "testing" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" +) + +func side(n, errs int, ms time.Duration, streamed bool) Side { + var s Side + for i := range n { + status := 200 + if i < errs { + status = 500 + } + s.Add(status, streamed, ms, ms) + } + return s +} + +func TestTheVerdict(t *testing.T) { + cn := catalog.Canary{Model: "c", Percent: 10} + for name, c := range map[string]struct { + stable, cand Side + rollback bool + }{ + "too few to say": {side(500, 0, 100*time.Millisecond, true), side(49, 49, 100*time.Millisecond, true), false}, + "as good": {side(500, 5, 100*time.Millisecond, true), side(60, 1, 110*time.Millisecond, true), false}, + "more errors": {side(500, 5, 100*time.Millisecond, true), side(60, 6, 100*time.Millisecond, true), true}, + "slower first token": {side(500, 0, 100*time.Millisecond, true), side(60, 0, 200*time.Millisecond, true), true}, + "slower answers": {side(500, 0, time.Second, false), side(60, 0, 2*time.Second, false), true}, + } { + got, why := Verdict(Stat{Stable: c.stable, Candidate: c.cand}, cn) + if got != c.rollback { + t.Errorf("%s: rollback=%v (%s)", name, got, why) + } + if got && why == "" { + t.Errorf("%s: no reason given", name) + } + } +} diff --git a/internal/control/catalog/catalog.go b/internal/control/catalog/catalog.go index 9564fc5..9f4429c 100644 --- a/internal/control/catalog/catalog.go +++ b/internal/control/catalog/catalog.go @@ -175,6 +175,10 @@ type Model struct { // every adapter, applying one only to a request that names it. Adapters []Adapter `json:"adapters,omitempty" koanf:"adapters"` Fallbacks []string `json:"fallbacks,omitempty" koanf:"fallbacks"` + // Canary sends a share of this model's requests to another catalog + // model — the same model at new weights — to compare the two before + // moving to it (`llmgw ctl model canary`). + Canary *Canary `json:"canary,omitempty" koanf:"canary"` // RoutingProfile selects how the router scores endpoints for this model. RoutingProfile string `json:"routing_profile,omitempty" koanf:"routing_profile"` SLO SLO `json:"slo,omitempty" koanf:"slo"` @@ -347,9 +351,79 @@ func New(models []Model) (*Catalog, error) { } } } + // A canary is another model in the catalog, a candidate for this one's + // place: not itself, not a judge, not a canary of its own, and the + // candidate of one model only. + canaryOf := map[string]string{} + for _, name := range c.order { + cn := c.models[name].Canary + if cn == nil { + continue + } + target, ok := c.Resolve(cn.Model) + switch { + case !ok: + return nil, fmt.Errorf("model %q: canary %q is not in the catalog", name, cn.Model) + case target.Name == name: + return nil, fmt.Errorf("model %q: a model cannot be its own canary", name) + case target.IsGuard(): + return nil, fmt.Errorf("model %q: canary %q is a guard model", name, cn.Model) + case target.Canary != nil: + return nil, fmt.Errorf("model %q: canary %q has a canary of its own", name, cn.Model) + case canaryOf[target.Name] != "": + return nil, fmt.Errorf("model %q: %q is already the canary of %q", name, cn.Model, canaryOf[target.Name]) + case cn.Percent < 0 || cn.Percent > 100: + return nil, fmt.Errorf("model %q: canary percent %d is not 0 to 100", name, cn.Percent) + } + canaryOf[target.Name] = name + } return c, nil } +// Canary is a share of a model's requests sent to a candidate, and when to +// give up on it. +type Canary struct { + // Model is the candidate: a catalog model, usually this one at newer + // weights, which clients cannot name themselves. + Model string `json:"model" koanf:"model"` + // Percent of requests go to it, 0 to 100. + Percent int `json:"percent" koanf:"percent"` + // The candidate is taken out — rolled back — when, over at least + // MinRequests of its own, its error rate is MaxErrorIncrease above + // this model's (0.05 is five points), or its mean time to the first + // token is MaxTTFTRatio times this model's. Zero is 50, 0.05 and 1.5. + MinRequests int `json:"min_requests,omitempty" koanf:"min_requests"` + MaxErrorIncrease float64 `json:"max_error_increase,omitempty" koanf:"max_error_increase"` + MaxTTFTRatio float64 `json:"max_ttft_ratio,omitempty" koanf:"max_ttft_ratio"` +} + +// Limits are the rollback thresholds with their defaults filled in. +func (c Canary) Limits() (minRequests int, maxErrorIncrease, maxTTFTRatio float64) { + minRequests, maxErrorIncrease, maxTTFTRatio = c.MinRequests, c.MaxErrorIncrease, c.MaxTTFTRatio + if minRequests <= 0 { + minRequests = 50 + } + if maxErrorIncrease <= 0 { + maxErrorIncrease = 0.05 + } + if maxTTFTRatio <= 0 { + maxTTFTRatio = 1.5 + } + return +} + +// CanaryOf is the model whose canary name is, or "". +func (c *Catalog) CanaryOf(name string) string { + for _, n := range c.order { + if cn := c.models[n].Canary; cn != nil { + if t, ok := c.Resolve(cn.Model); ok && t.Name == name { + return n + } + } + } + return "" +} + func (m Model) normalize() Model { if m.Scaling.Min < 0 { m.Scaling.Min = 0 diff --git a/internal/router/canary.go b/internal/router/canary.go new file mode 100644 index 0000000..99f1a6f --- /dev/null +++ b/internal/router/canary.go @@ -0,0 +1,124 @@ +package router + +import ( + "hash/fnv" + "net/http" + "strings" + "sync" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/canary" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" +) + +// A canary: a share of a model's requests sent to a candidate, usually the +// same model at newer weights, and the two compared as they are served. + +// variantHeader pins a request to one side, for testing either on purpose: +// "canary" or "stable". Either is the model the key asked for, so pinning +// grants nothing the key did not have. +const variantHeader = "x-llmgw-variant" + +// chooseVariant says whether a request goes to the canary. A session — the +// x-session-id header or OpenAI's user field — always lands on the same +// side, so one conversation does not hop between weights; a request without +// one is placed by its own id. +func chooseVariant(req *parsedRequest, cn *catalog.Canary) bool { + switch strings.ToLower(req.Header.Get(variantHeader)) { + case "canary": + return true + case "stable": + return false + } + if cn.Percent <= 0 { + return false + } + if cn.Percent >= 100 { + return true + } + key := req.SessionID + if key == "" { + key = req.RequestID + } + h := fnv.New32a() + h.Write([]byte(key)) + return int(h.Sum32()%100) < cn.Percent +} + +type canaryStats struct { + mu sync.Mutex + by map[string]*canary.Stat // by the model the canary is of +} + +func (c *canaryStats) observe(parent, cand string, onCanary bool, status int, streamed bool, ttft, took time.Duration) { + c.mu.Lock() + defer c.mu.Unlock() + if c.by == nil { + c.by = map[string]*canary.Stat{} + } + s := c.by[parent] + if s == nil || s.Canary != cand { + // A new candidate starts from nothing: the last one's numbers are + // not this one's. + s = &canary.Stat{Canary: cand, Since: time.Now()} + c.by[parent] = s + } + if onCanary { + s.Candidate.Add(status, streamed, ttft, took) + } else { + s.Stable.Add(status, streamed, ttft, took) + } +} + +// CanaryStats is how each model's canary and its stable side have served +// since the canary began, kept by this router. +func (r *Router) CanaryStats() map[string]canary.Stat { + r.canaries.mu.Lock() + defer r.canaries.mu.Unlock() + out := map[string]canary.Stat{} + for k, v := range r.canaries.by { + out[k] = *v + } + return out +} + +// ResetCanary forgets what a model's canary served, for a canary that has +// been taken out or promoted. +func (r *Router) ResetCanary(model string) { + r.canaries.mu.Lock() + defer r.canaries.mu.Unlock() + delete(r.canaries.by, model) +} + +// observeCanary counts a finished request against its side of a canary. +func (r *Router) observeCanary(req *parsedRequest, res dispatchResult, status int, took time.Duration) { + if res.canaryOf == "" || status == 499 { + return + } + r.canaries.observe(res.canaryOf, res.canary, res.onCanary, status, req.Stream, res.outcome.TTFT, took) +} + +// isCanaryTarget reports whether a name is some model's canary: reached +// through that model, never by its own name. +func (r *Router) isCanaryTarget(name string) bool { + canonical, _ := r.resolve(name) + if p := r.canaryTargets.Load(); p != nil { + _, ok := (*p)[canonical] + return ok + } + return false +} + +// hiddenModel is a model clients cannot name: a judge, or a canary. +func (r *Router) hiddenModel(name string) bool { + return r.isGuardModel(name) || r.isCanaryTarget(name) +} + +// setVariantHeader tells the client which side served it. +func setVariantHeader(h http.Header, onCanary bool) { + if onCanary { + h.Set(variantHeader, "canary") + } else { + h.Set(variantHeader, "stable") + } +} diff --git a/internal/router/canary_test.go b/internal/router/canary_test.go new file mode 100644 index 0000000..a4bb51f --- /dev/null +++ b/internal/router/canary_test.go @@ -0,0 +1,82 @@ +package router + +import ( + "fmt" + "net/http" + "strings" + "testing" + + "gitea.mynodes.xyz/ilker/llmgw/internal/config" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" +) + +func canaryRouter(t *testing.T, percent int) (*Router, *fakeEngine, *fakeEngine) { + t.Helper() + stable := newFakeEngine(t, "qwen") + cand := newFakeEngine(t, "qwen-canary") + cfg := baseConfig( + config.Endpoint{ID: "s", URL: stable.srv.URL, Models: []string{"qwen"}}, + config.Endpoint{ID: "c", URL: cand.srv.URL, Models: []string{"qwen-canary"}}, + ) + m := catModel("qwen", nil) + m.Canary = &catalog.Canary{Model: "qwen-canary", Percent: percent} + cfg.Models = []catalog.Model{m, catModel("qwen-canary", nil)} + return newTestRouter(t, cfg), stable, cand +} + +func TestACanaryTakesItsShareAndIsComparedWithTheRest(t *testing.T) { + rt, stable, cand := canaryRouter(t, 30) + for i := range 400 { + rec := post(t, rt.Handler(), map[string]any{"model": "qwen", "messages": []any{}}, + map[string]string{"X-Request-Id": fmt.Sprintf("r-%d", i)}) + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + } + c, s := cand.calls.Load(), stable.calls.Load() + if c < 80 || c > 160 || c+s != 400 { + t.Errorf("canary %d, stable %d of 400 at 30%%", c, s) + } + if b, _ := cand.last.Load().([]byte); !strings.Contains(string(b), `"qwen-canary"`) { + t.Errorf("the candidate was asked for another name: %s", b) + } + st := rt.CanaryStats()["qwen"] + if st.Canary != "qwen-canary" || st.Candidate.Requests != c || st.Stable.Requests != s { + t.Errorf("stats %+v", st) + } +} + +func TestASessionStaysOnOneSideAndAHeaderPinsIt(t *testing.T) { + rt, _, cand := canaryRouter(t, 50) + for range 20 { + post(t, rt.Handler(), map[string]any{"model": "qwen", "user": "alice", "messages": []any{}}, nil) + } + if n := cand.calls.Load(); n != 0 && n != 20 { + t.Errorf("one session went to the canary %d times of 20", n) + } + before := cand.calls.Load() + rec := post(t, rt.Handler(), map[string]any{"model": "qwen", "user": "alice", "messages": []any{}}, + map[string]string{"X-Llmgw-Variant": "canary"}) + if cand.calls.Load() != before+1 || rec.Header().Get("x-llmgw-variant") != "canary" { + t.Errorf("the pin was not followed: %q", rec.Header().Get("x-llmgw-variant")) + } +} + +func TestACanaryCannotBeNamedAndFallsBackToStable(t *testing.T) { + rt, _, _ := canaryRouter(t, 100) + if rec := post(t, rt.Handler(), map[string]any{"model": "qwen-canary", "messages": []any{}}, nil); rec.Code != http.StatusNotFound { + t.Errorf("a canary named directly: %d", rec.Code) + } + + // With nothing ready on the canary side, the stable side serves. + stable := newFakeEngine(t, "qwen") + cfg := baseConfig(config.Endpoint{ID: "s", URL: stable.srv.URL, Models: []string{"qwen"}}) + m := catModel("qwen", nil) + m.Canary = &catalog.Canary{Model: "qwen-canary", Percent: 100} + cfg.Models = []catalog.Model{m, catModel("qwen-canary", nil)} + rt = newTestRouter(t, cfg) + rec := post(t, rt.Handler(), map[string]any{"model": "qwen", "messages": []any{}}, nil) + if rec.Code != http.StatusOK || stable.calls.Load() != 1 || rec.Header().Get("x-llmgw-variant") != "stable" { + t.Errorf("status %d, stable calls %d, variant %q", rec.Code, stable.calls.Load(), rec.Header().Get("x-llmgw-variant")) + } +} diff --git a/internal/router/handlers.go b/internal/router/handlers.go index f18a05c..7cc1a4c 100644 --- a/internal/router/handlers.go +++ b/internal/router/handlers.go @@ -396,7 +396,7 @@ func (r *Router) handleInference(path string) http.HandlerFunc { } } - if r.isGuardModel(parsed.Model) { + if r.hiddenModel(parsed.Model) { // Said as a model that is not there: it is not, for clients. writeErrorCode(w, http.StatusNotFound, "invalid_request_error", "model_not_found", fmt.Sprintf("the model %q does not exist", parsed.Model)) @@ -491,6 +491,7 @@ func (r *Router) handleInference(path string) http.HandlerFunc { res := r.dispatch(req.Context(), w, parsed) status := r.finish(w, parsed, res, start) + r.observeCanary(parsed, res, status, time.Since(start)) r.observeOutput(parsed, res, res.judged) span.SetAttributes(attribute.Int("http.response.status_code", status), attribute.String("llmgw.served_model", res.model), @@ -516,8 +517,9 @@ func (r *Router) handleInference(path string) http.HandlerFunc { // around the allowlist; matching only the canonical name would silently ignore // an allowlist written in the names clients actually use. func (r *Router) permits(p auth.Principal, requested string) bool { - // A judge model is the guardrails', not anybody's to ask. - if r.isGuardModel(requested) { + // A judge model is the guardrails', not anybody's to ask; a canary is + // reached through the model it is a canary of. + if r.hiddenModel(requested) { return false } // An adapter named in the allowlist allows that adapter: not its model, @@ -852,7 +854,7 @@ func noEndpointFor(model string, served []string) string { // handing it a simulated model would send someone's first request to an engine // that replies in Greek letters. func (r *Router) exampleModels() []string { - names := slices.DeleteFunc(r.servedModels(), r.isGuardModel) + names := slices.DeleteFunc(r.servedModels(), r.hiddenModel) slices.SortStableFunc(names, func(a, b string) int { as, bs := r.modelMap()[a].Simulated(), r.modelMap()[b].Simulated() switch { @@ -910,7 +912,7 @@ func (r *Router) handleModels(w http.ResponseWriter, req *http.Request) { principal, authed := auth.FromContext(req.Context()) names := make([]string, 0, len(seen)) for n := range seen { - if authed && !r.permits(principal, n) || r.isGuardModel(n) { + if authed && !r.permits(principal, n) || r.hiddenModel(n) { continue } names = append(names, n) diff --git a/internal/router/router.go b/internal/router/router.go index 52ed428..77fa7ac 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -64,6 +64,10 @@ type Router struct { models atomic.Pointer[map[string]catalog.Model] // adapters are the LoRA adapters by the name clients ask for them by. adapters atomic.Pointer[map[string]adapterRef] + // canaryTargets maps each model that is another's canary to that one; + // canaries counts how the two sides served. + canaryTargets atomic.Pointer[map[string]string] + canaries canaryStats // extraMetrics are collectors other components publish on /metrics. extraMetrics []prometheus.Collector @@ -635,6 +639,15 @@ func (r *Router) storeModels(models []catalog.Model) { } r.models.Store(&m) r.adapters.Store(&ads) + targets := map[string]string{} + for _, entry := range models { + if entry.Canary != nil { + if t, ok := m[entry.Canary.Model]; ok { + targets[t.Name] = entry.Name + } + } + } + r.canaryTargets.Store(&targets) } // adapterRef is where an adapter runs and how it is applied. @@ -732,6 +745,10 @@ type dispatchResult struct { // judged whether there was. guard guard.Verdict judged bool + // canaryOf is the model whose canary this request was split by, canary + // the candidate, and onCanary whether the candidate served it. + canaryOf, canary string + onCanary bool } type servedModel struct { @@ -749,6 +766,39 @@ type servedModel struct { func (r *Router) dispatch(ctx context.Context, w http.ResponseWriter, req *parsedRequest) dispatchResult { canonical, entry := r.resolve(req.Model) + // A canary takes its share before anything else, and falls back to the + // stable side, not to a fallback, when it has nothing ready. + // Not an adapter: the candidate's replicas do not run it under the + // stable model's name. + _, isAdapter := r.adapter(req.Model) + if cn := entry.Canary; cn != nil && !isAdapter { + onCanary := chooseVariant(req, cn) + target, _ := r.resolve(cn.Model) + if onCanary && len(r.reg.Snapshots(target)) == 0 { + onCanary = false + } + setVariantHeader(w.Header(), onCanary) + if onCanary { + // The engine is asked for the model by the name it serves. + creq := *req + creq.UpstreamBody = withModel(req.UpstreamBody, target) + res := r.dispatchOne(ctx, w, &creq, target, entry) + res.model, res.canaryOf, res.canary, res.onCanary = target, canonical, target, true + if res.err == nil || res.outcome.Wrote || errors.Is(res.err, proxy.ErrClientGone) || + !errors.Is(res.err, errNoCapacity) { + return res + } + setVariantHeader(w.Header(), false) + } + res := r.dispatchStable(ctx, w, req, canonical, entry) + res.canaryOf, res.canary = canonical, target + return res + } + return r.dispatchStable(ctx, w, req, canonical, entry) +} + +// dispatchStable serves a model as asked, then its fallbacks. +func (r *Router) dispatchStable(ctx context.Context, w http.ResponseWriter, req *parsedRequest, canonical string, entry catalog.Model) dispatchResult { // The requested model first, then its fallbacks. A fallback is a different // model, so the client is told about it via x-llmgw-served-model. chain := []string{canonical} @@ -824,7 +874,7 @@ func (r *Router) dispatchOne(ctx context.Context, w http.ResponseWriter, req *pa return dispatchResult{err: fmt.Errorf("%w: %s", errLoading, "the model is loading; retry shortly")} } - return dispatchResult{err: fmt.Errorf("%w: %s", errNoCapacity, noEndpointFor(model, slices.DeleteFunc(r.servedModels(), r.isGuardModel)))} + return dispatchResult{err: fmt.Errorf("%w: %s", errNoCapacity, noEndpointFor(model, slices.DeleteFunc(r.servedModels(), r.hiddenModel)))} } profile := r.profileFor(entry, req.ProfileOverride)