Canary routing: a model's canary takes its share of requests, a session stays on one side, x-llmgw-variant pins one, and the two are compared

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

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, ""
+}

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)
+		}
+	}
+}

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

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")
+	}
+}

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"))
+	}
+}

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)

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)