From a1dfd4df1dcc4eee1253f2f66ff3b6be3af5f97f Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Sun, 4 Oct 2026 20:31:26 +0200 Subject: [PATCH] Canary of a model's newer weights: ctl model canary, promote and canaries, rolled back by itself when it serves worse Co-Authored-By: Claude Opus 5.5 (1M context) --- internal/cli/ctl.go | 9 +- internal/cli/model.go | 3 +- internal/cli/model_canary.go | 148 ++++++++++ internal/cli/server.go | 3 + internal/control/adminapi/api.go | 6 + internal/control/adminapi/canary.go | 84 ++++++ internal/control/modelserve/canary.go | 304 ++++++++++++++++++++ internal/control/modelserve/canary_test.go | 101 +++++++ internal/control/modelserve/modelserve.go | 10 +- internal/control/modelserve/updates.go | 53 +++- internal/control/modelserve/updates_test.go | 25 ++ internal/control/modelsvc/modelsvc.go | 17 ++ internal/notify/notify.go | 5 +- 13 files changed, 750 insertions(+), 18 deletions(-) create mode 100644 internal/cli/model_canary.go create mode 100644 internal/control/adminapi/canary.go create mode 100644 internal/control/modelserve/canary.go create mode 100644 internal/control/modelserve/canary_test.go diff --git a/internal/cli/ctl.go b/internal/cli/ctl.go index def0aed2..4cf0b822 100644 --- a/internal/cli/ctl.go +++ b/internal/cli/ctl.go @@ -351,8 +351,15 @@ func newModelsCmd(client func() *ctlClient, asJSON *bool) *cobra.Command { if m.Ready > 0 { serving = "yes" } + name := m.Name + if m.Canary != nil { + name += fmt.Sprintf(" (%d%% to canary)", m.Canary.Percent) + } + if m.CanaryOf != "" { + name += " (canary of " + m.CanaryOf + ")" + } fmt.Fprintf(w, "%s\t%s\t%d/%d\t%s\t%s\t%s\t%s\t%s\t%s\n", - m.Name, serving, m.Ready, m.Min, modelStatus(m), m.Class, reserves(m), + name, serving, m.Ready, m.Min, modelStatus(m), m.Class, reserves(m), join(m.Aliases), join(m.Profiles), join(m.Fallbacks)) for _, x := range m.Warnings { warnings = append(warnings, m.Name+": "+x) diff --git a/internal/cli/model.go b/internal/cli/model.go index 66c6d283..2a19bffb 100644 --- a/internal/cli/model.go +++ b/internal/cli/model.go @@ -36,7 +36,8 @@ func newModelCmd(client func() *ctlClient, asJSON *bool) *cobra.Command { newResourcesCmd(client), newClassCmd(client), newOwnerCmd(client), newTargetCmd(client, asJSON), newAdaptersCmd(client), newModelServeCmd(client), newModelUpdatesCmd(client, asJSON), newModelUpdateCmd(client), - newModelRollbackCmd(client), newModelOriginCmd(client)) + newModelRollbackCmd(client), newModelOriginCmd(client), + newModelCanaryCmd(client), newModelPromoteCmd(client), newModelCanariesCmd(client, asJSON)) return c } diff --git a/internal/cli/model_canary.go b/internal/cli/model_canary.go new file mode 100644 index 00000000..ad353e98 --- /dev/null +++ b/internal/cli/model_canary.go @@ -0,0 +1,148 @@ +package cli + +import ( + "encoding/json" + "fmt" + "slices" + "text/tabwriter" + + "github.com/spf13/cobra" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/modelserve" +) + +func newModelCanaryCmd(client func() *ctlClient) *cobra.Command { + var ( + cn catalog.Canary + abort, detach bool + ) + c := &cobra.Command{ + Use: "canary MODEL", + Short: "Try a model's newer weights on a share of its requests before moving to them", + Long: "The newer files of the model's repository are fetched and run beside it\n" + + "as MODEL-canary, one replica, and given --percent of MODEL's requests. A\n" + + "session (x-session-id, or the user field) stays on one side; the\n" + + "x-llmgw-variant header (canary or stable) pins a request to one. Clients\n" + + "cannot name the canary itself.\n\n" + + "The two sides are compared as they serve. Once the canary has served\n" + + "--min-requests, it is taken out by itself if its error rate is\n" + + "--max-error-increase above the model's, or its time to the first token\n" + + "(or, unstreamed, to the answer) --max-ttft-ratio times the model's, and\n" + + "canary_rolled_back goes to the notification webhooks.\n\n" + + "Run again with --percent to change its share, --abort to take it out;\n" + + "`llmgw ctl model promote MODEL` moves the model to it.", + Example: " llmgw ctl model canary qwen-7b --percent 10\n" + + " llmgw ctl model canary qwen-7b --percent 50\n" + + " llmgw ctl model canary qwen-7b --abort", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + name := args[0] + var out map[string]string + if abort { + if err := client().delete(cmd.Context(), "/api/v1/models/"+name+"/canary", &out); err != nil { + return err + } + fmt.Fprintln(cmd.OutOrStdout(), out["result"]) + return nil + } + var have []modelserve.CanaryView + if err := client().get(cmd.Context(), "/api/v1/canaries", &have); err != nil { + return err + } + if slices.ContainsFunc(have, func(v modelserve.CanaryView) bool { return v.Model == name }) { + if !cmd.Flags().Changed("percent") { + return fmt.Errorf("%s already has a canary: give --percent to change its share, or --abort", name) + } + body, _ := json.Marshal(map[string]int{"percent": cn.Percent}) + if err := client().put(cmd.Context(), "/api/v1/models/"+name+"/canary", body, &out); err != nil { + return err + } + fmt.Fprintln(cmd.OutOrStdout(), out["result"]) + return nil + } + body, _ := json.Marshal(cn) + var job modelserve.Job + if err := client().post(cmd.Context(), "/api/v1/models/"+name+"/canary", body, &job); err != nil { + return err + } + if detach { + return detached(cmd, job) + } + return followJob(cmd, client, job, "has a canary") + }, + } + f := c.Flags() + f.IntVar(&cn.Percent, "percent", 10, "share of the model's requests the canary is given, 0 to 100") + f.IntVar(&cn.MinRequests, "min-requests", 50, "requests the canary serves before it is judged") + f.Float64Var(&cn.MaxErrorIncrease, "max-error-increase", 0.05, "error rate above the model's that rolls it back (0.05 is five points)") + f.Float64Var(&cn.MaxTTFTRatio, "max-ttft-ratio", 1.5, "times the model's time to the first token that rolls it back") + f.BoolVar(&abort, "abort", false, "take the canary out; the model serves all its requests again") + f.BoolVar(&detach, "detach", false, "start it and return: it runs on the control plane; `llmgw ctl jobs` follows it") + return c +} + +func newModelPromoteCmd(client func() *ctlClient) *cobra.Command { + var detach bool + c := &cobra.Command{ + Use: "promote MODEL", + Short: "Move a model to its canary's weights, and take the canary out", + Long: "The update the canary was a trial of: the catalog names the canary's\n" + + "files, the model's replicas are replaced one at a time, and the canary\n" + + "is removed. `llmgw ctl model rollback MODEL` goes back.", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + var job modelserve.Job + if err := client().post(cmd.Context(), "/api/v1/models/"+args[0]+"/promote", []byte("{}"), &job); err != nil { + return err + } + if detach { + return detached(cmd, job) + } + return followJob(cmd, client, job, "runs the canary's weights") + }, + } + c.Flags().BoolVar(&detach, "detach", false, "start it and return: it runs on the control plane; `llmgw ctl jobs` follows it") + return c +} + +func newModelCanariesCmd(client func() *ctlClient, asJSON *bool) *cobra.Command { + return &cobra.Command{ + Use: "canaries", + Short: "Models trying a canary, and how each side has served since it began", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + var vs []modelserve.CanaryView + if err := client().get(cmd.Context(), "/api/v1/canaries", &vs); err != nil { + return err + } + if *asJSON { + return json.NewEncoder(cmd.OutOrStdout()).Encode(vs) + } + if len(vs) == 0 { + fmt.Fprintln(cmd.OutOrStdout(), "No model has a canary.") + return nil + } + w := tabwriter.NewWriter(cmd.OutOrStdout(), 0, 0, 2, ' ', 0) + fmt.Fprintln(w, "MODEL\tCANARY\tSHARE\tREQUESTS\tERRORS\tFIRST TOKEN\tANSWER") + for _, v := range vs { + side := func(label string, req int64, errRate, ttft, lat float64) { + fmt.Fprintf(w, "%s\t%s\t\t%d\t%.1f%%\t%s\t%s\n", "", label, req, 100*errRate, meanMs(ttft), meanMs(lat)) + } + fmt.Fprintf(w, "%s\t%s\t%d%%\t\t\t\t\n", v.Model, v.Canary.Model, v.Canary.Percent) + if s := v.Serving; s != nil { + side(" canary", s.Candidate.Requests, s.Candidate.ErrorRate(), s.Candidate.TTFTMs, s.Candidate.LatencyMs) + side(" stable", s.Stable.Requests, s.Stable.ErrorRate(), s.Stable.TTFTMs, s.Stable.LatencyMs) + } + } + return w.Flush() + }, + } +} + +func meanMs(v float64) string { + if v <= 0 { + return "-" + } + return fmt.Sprintf("%.0f ms", v) +} diff --git a/internal/cli/server.go b/internal/cli/server.go index aa783ce0..e9aadfea 100644 --- a/internal/cli/server.go +++ b/internal/cli/server.go @@ -405,7 +405,10 @@ func newServerCmd() *cobra.Command { Current: func() []catalog.Model { return admin.Catalog().Models() }, Models: catalogSvc, Restart: admin.Reconciler().Restart, Pool: adminAPI.PoolNodes, Notify: notifier, + // Canaries: how the router saw each side serve. + CanaryStats: rt.CanaryStats, ResetCanary: rt.ResetCanary, }) + go serve.WatchCanaries(ctx, 30*time.Second) adminAPI.SetServe(serve) updateMetrics.Set(serve) go serve.WatchUpdates(ctx, cfg.Mirror.UpdateCheck) diff --git a/internal/control/adminapi/api.go b/internal/control/adminapi/api.go index 19aecd00..c880ad2d 100644 --- a/internal/control/adminapi/api.go +++ b/internal/control/adminapi/api.go @@ -154,6 +154,7 @@ func (a *API) Register(mux *http.ServeMux) { mux.HandleFunc("POST /api/v1/updates/check", a.checkUpdates) mux.HandleFunc("POST /api/v1/models/{name}/update", a.updateModel) mux.HandleFunc("POST /api/v1/models/{name}/rollback", a.rollbackModel) + a.registerCanary(mux) mux.HandleFunc("POST /api/v1/models/{name}/origin", a.setOrigin) mux.HandleFunc("GET /api/v1/plan", a.planModel) mux.HandleFunc("POST /api/v1/metrics/range", a.metricsBatch) @@ -330,6 +331,10 @@ func loadView(l *inventory.Load, now time.Time) *LoadView { // ModelView is a model as `ctl models` shows it. type ModelView struct { Name string `json:"name"` + // Canary is a share of its requests sent to a candidate; CanaryOf is + // set on the candidate, naming the model it is a canary of. + Canary *catalog.Canary `json:"canary,omitempty"` + CanaryOf string `json:"canary_of,omitempty"` // Description is what the model is for, when somebody said. Description string `json:"description,omitempty"` Aliases []string `json:"aliases,omitempty"` @@ -651,6 +656,7 @@ func (a *API) Models() []ModelView { Parked: h.Parked, Reason: h.Reason, Unplaceable: h.Unplaceable, Class: m.EffectiveClass(), Owner: m.Owner, Preempted: h.Preempted, HeldFor: h.HeldFor, External: externalLabel(m), Adapters: m.Adapters, + Canary: m.Canary, CanaryOf: a.ctl.Catalog().CanaryOf(m.Name), } for _, p := range m.Profiles { v.Profiles = append(v.Profiles, p.Name) diff --git a/internal/control/adminapi/canary.go b/internal/control/adminapi/canary.go new file mode 100644 index 00000000..bba08211 --- /dev/null +++ b/internal/control/adminapi/canary.go @@ -0,0 +1,84 @@ +package adminapi + +import ( + "encoding/json" + "io" + "net/http" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" +) + +func (a *API) registerCanary(mux *http.ServeMux) { + need := func(w http.ResponseWriter) bool { + if a.serve == nil { + writeError(w, http.StatusNotImplemented, "a canary needs a mirror and a store, and this control plane lacks one") + return false + } + return true + } + mux.HandleFunc("GET /api/v1/canaries", func(w http.ResponseWriter, r *http.Request) { + if !need(w) { + return + } + writeJSON(w, http.StatusOK, a.serve.Canaries()) + }) + // Start one: the newer weights fetched and run beside, then given + // their share. + mux.HandleFunc("POST /api/v1/models/{name}/canary", func(w http.ResponseWriter, r *http.Request) { + if !need(w) { + return + } + var cn catalog.Canary + if err := json.NewDecoder(io.LimitReader(r.Body, 1<<12)).Decode(&cn); err != nil { + writeError(w, http.StatusBadRequest, "the request is not valid JSON: "+err.Error()) + return + } + job, err := a.serve.StartCanary(r.PathValue("name"), cn, actorOf(r)) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + writeJSON(w, http.StatusAccepted, job) + }) + // Change its share. + mux.HandleFunc("PUT /api/v1/models/{name}/canary", func(w http.ResponseWriter, r *http.Request) { + if !need(w) { + return + } + var req struct { + Percent int `json:"percent"` + } + if err := json.NewDecoder(io.LimitReader(r.Body, 1<<12)).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "the request is not valid JSON: "+err.Error()) + return + } + msg, err := a.serve.SetCanaryPercent(r.Context(), r.PathValue("name"), req.Percent, actorOf(r)) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + writeJSON(w, http.StatusOK, map[string]string{"result": msg}) + }) + mux.HandleFunc("DELETE /api/v1/models/{name}/canary", func(w http.ResponseWriter, r *http.Request) { + if !need(w) { + return + } + msg, err := a.serve.AbortCanary(r.Context(), r.PathValue("name"), actorOf(r)) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + writeJSON(w, http.StatusOK, map[string]string{"result": msg}) + }) + mux.HandleFunc("POST /api/v1/models/{name}/promote", func(w http.ResponseWriter, r *http.Request) { + if !need(w) { + return + } + job, err := a.serve.StartPromote(r.PathValue("name"), actorOf(r)) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + writeJSON(w, http.StatusAccepted, job) + }) +} diff --git a/internal/control/modelserve/canary.go b/internal/control/modelserve/canary.go new file mode 100644 index 00000000..4eea4bc1 --- /dev/null +++ b/internal/control/modelserve/canary.go @@ -0,0 +1,304 @@ +package modelserve + +import ( + "context" + "errors" + "fmt" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/canary" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" + "gitea.mynodes.xyz/ilker/llmgw/internal/notify" +) + +// A canary: a model's repository's newer weights run beside it as another +// model, given a share of its requests, compared, and then promoted to its +// place or taken out — by hand, or by itself when it serves worse. + +// canaryName is the candidate's name in the catalog. +func canaryName(model string) string { return model + "-canary" } + +// StartCanary fetches a model's newer weights and runs them as its canary, +// with the share and limits given. +func (s *Server) StartCanary(name string, cn catalog.Canary, by string) (Job, error) { + if s.deps.Models == nil || s.deps.Restart == nil { + return Job{}, errors.New("a canary needs the store, which this control plane lacks") + } + m, err := s.current(name) + if err != nil { + return Job{}, err + } + if s.deps.Models.Declared(name) { + return Job{}, fmt.Errorf("%s is declared in server.yaml; a canary is for a model added at runtime", name) + } + if m.Canary != nil { + return Job{}, fmt.Errorf("%s already has a canary, %s at %d%%: change its share with --percent, "+ + "or promote or abort it first", name, m.Canary.Model, m.Canary.Percent) + } + if cn.Percent < 0 || cn.Percent > 100 { + return Job{}, fmt.Errorf("percent %d is not 0 to 100", cn.Percent) + } + s.mu.Lock() + for _, j := range s.jobs { + if (j.Kind == "update" || j.Kind == "canary" || j.Kind == "promote") && j.Model == name && j.State == Running { + s.mu.Unlock() + return Job{}, fmt.Errorf("%s has a job running: %s", name, j.ID) + } + } + s.seq++ + j := &Job{ID: fmt.Sprintf("canary-%d-%d", time.Now().Unix(), s.seq), Kind: "canary", Model: name, + Request: Request{Name: name, By: by}, State: Running, Started: time.Now()} + s.keepJob(j) + s.mu.Unlock() + go s.runCanary(j, cn) + return s.snapshot(j), nil +} + +func (s *Server) runCanary(j *Job, cn catalog.Canary) { + ctx := context.Background() + name := j.Model + p, ok := s.prepare(ctx, j, name) + if !ok { + return + } + + // 4. The candidate, beside the model: the same in everything but its + // weights, and one replica. + cand := canaryName(name) + s.step(j, "catalog", Running, "adding "+cand) + c := p.m + c.Name, c.Aliases, c.Fallbacks, c.Canary, c.Adapters = cand, nil, nil, nil, nil + c.Source = p.source + c.Origin = &catalog.Origin{Repo: p.repo.ID, Revision: p.repo.Revision, Files: originFiles(p.set)} + c.Scaling = catalog.Scaling{Min: 1, Max: 1} + c.Description = fmt.Sprintf("Canary of %s at %s.", name, p.repo.ShortRevision()) + if err := s.deps.Models.Add(ctx, c, j.Request.By); err != nil { + s.fail(j, "catalog", err) + return + } + s.step(j, "catalog", Done, cand+" runs "+p.repo.ID+" at "+p.repo.ShortRevision()) + + // 5. Serving, before it is given anyone's requests. + r, err := s.waitReady(ctx, j, cand, false) + if err == nil { + s.step(j, "smoke test", Running, "asking "+r.ID+" to say hello") + var reply string + var tps float64 + if reply, tps, err = s.smoke(ctx, r, cand); err == nil { + s.step(j, "smoke test", Done, fmt.Sprintf("answered at %.1f tokens/s: %q", tps, reply)) + } else { + s.step(j, "smoke test", Failed, err.Error()) + } + } + if err != nil { + _ = s.deps.Models.Remove(ctx, cand) + s.fail(j, "placement", fmt.Errorf("%w; %s was taken out again", err, cand)) + return + } + + // 6. Its share. + cn.Model = cand + if _, err := s.deps.Models.SetCanary(ctx, name, &cn, j.Request.By); err != nil { + _ = s.deps.Models.Remove(ctx, cand) + s.fail(j, "split", err) + return + } + if s.deps.ResetCanary != nil { + s.deps.ResetCanary(name) + } + s.step(j, "split", Done, fmt.Sprintf("%d%% of %s's requests go to %s", cn.Percent, name, cand)) + s.finish(j) +} + +// SetCanaryPercent changes the share of a model's requests its canary is +// given; 0 keeps it running and gives it none. +func (s *Server) SetCanaryPercent(ctx context.Context, name string, percent int, by string) (string, error) { + m, err := s.current(name) + if err != nil { + return "", err + } + if m.Canary == nil { + return "", fmt.Errorf("%s has no canary; start one with `llmgw ctl model canary %s --percent N`", name, name) + } + cn := *m.Canary + cn.Percent = percent + if _, err := s.deps.Models.SetCanary(ctx, name, &cn, by); err != nil { + return "", err + } + return fmt.Sprintf("%d%% of %s's requests go to %s", percent, name, cn.Model), nil +} + +// AbortCanary takes a model's canary out: its share first, then the +// candidate. +func (s *Server) AbortCanary(ctx context.Context, name, by string) (string, error) { + m, err := s.current(name) + if err != nil { + return "", err + } + if m.Canary == nil { + return "", fmt.Errorf("%s has no canary", name) + } + cand := m.Canary.Model + if _, err := s.deps.Models.SetCanary(ctx, name, nil, by); err != nil { + return "", err + } + if err := s.deps.Models.Remove(ctx, cand); err != nil { + return "", fmt.Errorf("%s gets no more requests, but could not be removed: %w", cand, err) + } + if s.deps.ResetCanary != nil { + s.deps.ResetCanary(name) + } + return fmt.Sprintf("%s is out; %s serves all of %s's requests", cand, name, name), nil +} + +// StartPromote moves a model to its canary's weights — the update a canary +// was the trial of — and takes the canary out. +func (s *Server) StartPromote(name, by string) (Job, error) { + m, err := s.current(name) + if err != nil { + return Job{}, err + } + if m.Canary == nil { + return Job{}, fmt.Errorf("%s has no canary to promote", name) + } + s.mu.Lock() + s.seq++ + j := &Job{ID: fmt.Sprintf("promote-%d-%d", time.Now().Unix(), s.seq), Kind: "promote", Model: name, + Request: Request{Name: name, By: by}, State: Running, Started: time.Now()} + s.keepJob(j) + s.mu.Unlock() + go s.runPromote(j) + return s.snapshot(j), nil +} + +func (s *Server) runPromote(j *Job) { + ctx := context.Background() + name := j.Model + m, err := s.current(name) + if err != nil || m.Canary == nil { + s.fail(j, "catalog", fmt.Errorf("%s no longer has a canary", name)) + return + } + c, err := s.current(m.Canary.Model) + if err != nil { + s.fail(j, "catalog", err) + return + } + s.step(j, "catalog", Running, fmt.Sprintf("moving %s to %s's weights", name, c.Name)) + next := &catalog.Origin{} + if c.Origin != nil { + *next = *c.Origin + } + next.Previous = &catalog.PreviousOrigin{Source: m.Source} + if m.Origin != nil { + next.Previous.Revision, next.Previous.Files = m.Origin.Revision, m.Origin.Files + } + if _, err := s.deps.Models.SetSource(ctx, name, c.Source, next, j.Request.By); err != nil { + s.fail(j, "catalog", err) + return + } + // Its share back to the model, which is about to run the same weights. + if _, err := s.deps.Models.SetCanary(ctx, name, nil, j.Request.By); err != nil { + s.fail(j, "catalog", err) + return + } + s.step(j, "catalog", Done, "the catalog names "+c.Source) + + if err := s.replace(ctx, j, name); err != nil { + s.fail(j, "replicas", fmt.Errorf("%w. `llmgw ctl model rollback %s` goes back", err, name)) + return + } + if err := s.deps.Models.Remove(ctx, c.Name); err != nil { + s.step(j, "canary", Failed, fmt.Sprintf("%s could not be removed: %v", c.Name, err)) + } else { + s.step(j, "canary", Done, c.Name+" removed") + } + if s.deps.ResetCanary != nil { + s.deps.ResetCanary(name) + } + if c.Origin != nil { + s.mu.Lock() + s.updates[name] = UpdateInfo{Model: name, Repo: c.Origin.Repo, Current: c.Origin.Revision, + Latest: c.Origin.Revision, Previous: next.Previous.Revision, CheckedAt: time.Now()} + s.mu.Unlock() + } + s.finish(j) +} + +// WatchCanaries takes out, every so often, each canary serving worse than +// its model by the limits it was given. Until ctx ends. +func (s *Server) WatchCanaries(ctx context.Context, every time.Duration) { + if s.deps.CanaryStats == nil || s.deps.Current == nil { + return + } + if every <= 0 { + every = 30 * time.Second + } + t := time.NewTicker(every) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + s.checkCanaries(ctx) + } + } +} + +func (s *Server) checkCanaries(ctx context.Context) { + stats := s.deps.CanaryStats() + for _, m := range s.deps.Current() { + if m.Canary == nil || m.Canary.Percent == 0 { + continue + } + st, ok := stats[m.Name] + if !ok || st.Canary != m.Canary.Model { + continue + } + bad, why := canary.Verdict(st, *m.Canary) + if !bad { + continue + } + cand := m.Canary.Model + if _, err := s.AbortCanary(ctx, m.Name, "llmgw"); err != nil { + s.deps.Log.Error("a canary serving worse could not be taken out", "model", m.Name, "canary", cand, "err", err) + continue + } + s.deps.Log.Warn("canary rolled back: it served worse than the model", "model", m.Name, "canary", cand, "why", why) + s.deps.Notify.Notify(notify.CanaryRolledBack, m.Name+"/"+cand, + fmt.Sprintf("the canary of %s was rolled back: %s, over %d requests", m.Name, why, st.Candidate.Requests), + map[string]any{"model": m.Name, "canary": cand, "reason": why, + "canary_requests": st.Candidate.Requests, "stable_requests": st.Stable.Requests}) + } +} + +// CanaryView is one model's canary as it stands. +type CanaryView struct { + Model string `json:"model"` + Canary catalog.Canary `json:"canary"` + Serving *canary.Stat `json:"serving,omitempty"` +} + +// Canaries are the models with a canary, and how each side has served. +func (s *Server) Canaries() []CanaryView { + if s.deps.Current == nil { + return nil + } + var stats map[string]canary.Stat + if s.deps.CanaryStats != nil { + stats = s.deps.CanaryStats() + } + var out []CanaryView + for _, m := range s.deps.Current() { + if m.Canary == nil { + continue + } + v := CanaryView{Model: m.Name, Canary: *m.Canary} + if st, ok := stats[m.Name]; ok && st.Canary == m.Canary.Model { + v.Serving = &st + } + out = append(out, v) + } + return out +} diff --git a/internal/control/modelserve/canary_test.go b/internal/control/modelserve/canary_test.go new file mode 100644 index 00000000..6eedadfc --- /dev/null +++ b/internal/control/modelserve/canary_test.go @@ -0,0 +1,101 @@ +package modelserve + +import ( + "context" + "log/slog" + "strings" + "testing" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/canary" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler" +) + +func canarySetup(t *testing.T) (*Server, *updModels, *restartFleet) { + t.Helper() + s, _, models, _, fleet := updSetup(t) + models.onAdd = func(m catalog.Model) { + fleet.mu.Lock() + fleet.reps = append(fleet.reps, reconciler.Replica{ID: m.Name + "-1", Model: m.Name, + State: reconciler.Ready, Endpoint: fleet.engine}) + fleet.mu.Unlock() + } + s.deps.Log = slog.New(slog.DiscardHandler) + return s, models, fleet +} + +func TestACanaryRunsTheNewWeightsBesideAndIsPromoted(t *testing.T) { + s, models, fleet := canarySetup(t) + oldSource := models.models["qwen"].Source + job, err := s.StartCanary("qwen", catalog.Canary{Percent: 10}, "tester") + if err != nil { + t.Fatal(err) + } + if j := wait(t, s, job.ID); j.State != Done { + t.Fatalf("%s: %+v", j.Error, j.Steps) + } + m, c := models.models["qwen"], models.models["qwen-canary"] + if m.Source != oldSource || m.Canary == nil || m.Canary.Model != "qwen-canary" || m.Canary.Percent != 10 { + t.Fatalf("the model was changed, or given no canary: %s %+v", m.Source, m.Canary) + } + if !strings.HasSuffix(c.Source, "/qwen/def98765/q4.gguf") || c.Origin.Revision != newRev || c.Scaling.Max != 1 { + t.Fatalf("candidate: %s %+v %+v", c.Source, c.Origin, c.Scaling) + } + if _, err := s.StartCanary("qwen", catalog.Canary{Percent: 10}, "tester"); err == nil { + t.Error("a second canary was started") + } + if _, err := s.SetCanaryPercent(context.Background(), "qwen", 50, "tester"); err != nil || models.models["qwen"].Canary.Percent != 50 { + t.Errorf("share: %v", err) + } + + job, err = s.StartPromote("qwen", "tester") + if err != nil { + t.Fatal(err) + } + if j := wait(t, s, job.ID); j.State != Done { + t.Fatalf("%s: %+v", j.Error, j.Steps) + } + m = models.models["qwen"] + if m.Source != c.Source || m.Canary != nil || m.Origin.Previous == nil || m.Origin.Previous.Source != oldSource { + t.Fatalf("promoted: %s %+v %+v", m.Source, m.Canary, m.Origin) + } + if _, ok := models.models["qwen-canary"]; ok { + t.Error("the canary was left in the catalog") + } + for _, r := range fleet.Replicas() { + if r.Model == "qwen" && !strings.Contains(r.ID, "new") { + t.Errorf("%s was not replaced", r.ID) + } + } +} + +func TestACanaryServingWorseIsRolledBackByItself(t *testing.T) { + s, models, _ := canarySetup(t) + job, _ := s.StartCanary("qwen", catalog.Canary{Percent: 10, MinRequests: 20}, "tester") + if j := wait(t, s, job.ID); j.State != Done { + t.Fatalf("%s: %+v", j.Error, j.Steps) + } + var stat canary.Stat + stat.Canary = "qwen-canary" + for i := range 200 { + stat.Stable.Add(200, true, 100*time.Millisecond, time.Second) + if i < 30 { + status := 200 + if i%3 == 0 { + status = 502 + } + stat.Candidate.Add(status, true, 100*time.Millisecond, time.Second) + } + } + reset := "" + s.deps.CanaryStats = func() map[string]canary.Stat { return map[string]canary.Stat{"qwen": stat} } + s.deps.ResetCanary = func(m string) { reset = m } + s.checkCanaries(context.Background()) + if models.models["qwen"].Canary != nil || reset != "qwen" { + t.Fatalf("a canary failing a third of its requests was kept: %+v", models.models["qwen"].Canary) + } + if _, ok := models.models["qwen-canary"]; ok { + t.Error("the candidate was left in the catalog") + } +} diff --git a/internal/control/modelserve/modelserve.go b/internal/control/modelserve/modelserve.go index 294d14ec..0676d4ac 100644 --- a/internal/control/modelserve/modelserve.go +++ b/internal/control/modelserve/modelserve.go @@ -25,6 +25,7 @@ import ( "sync" "time" + "gitea.mynodes.xyz/ilker/llmgw/internal/canary" "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" "gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler" "gitea.mynodes.xyz/ilker/llmgw/internal/gguf" @@ -86,9 +87,14 @@ type Deps struct { // UnplacedGrace is how long a model may go unplaced before the job // stops waiting for it: a minute unless set. UnplacedGrace time.Duration - // Notify is told when a model's repository publishes new weights; nil - // tells no one. + // Notify is told when a model's repository publishes new weights or a + // canary is rolled back; nil tells no one. Notify *notify.Notifier + // CanaryStats is how each model's canary and stable side served, by + // model; ResetCanary forgets one. From the router; without them a + // canary is never rolled back by itself. + CanaryStats func() map[string]canary.Stat + ResetCanary func(model string) } // errBusy is a model the cluster has the hardware for, all of it in use now. diff --git a/internal/control/modelserve/updates.go b/internal/control/modelserve/updates.go index d3a6a28c..18b1f4db 100644 --- a/internal/control/modelserve/updates.go +++ b/internal/control/modelserve/updates.go @@ -33,6 +33,11 @@ type Models interface { Declared(name string) bool SetSource(ctx context.Context, name, source string, origin *catalog.Origin, by string) (catalog.Model, error) SetOrigin(ctx context.Context, name string, origin *catalog.Origin) error + // For a canary: the candidate added beside the model and taken out + // again, and the share of requests it is given. + Add(ctx context.Context, m catalog.Model, by string) error + Remove(ctx context.Context, name string) error + SetCanary(ctx context.Context, name string, c *catalog.Canary, by string) (catalog.Model, error) } // UpdateInfo is what the registry has for a model against what it runs. @@ -451,32 +456,43 @@ func (s *Server) StartUpdate(name, by string) (Job, error) { return s.snapshot(j), nil } -func (s *Server) runUpdate(j *Job) { - ctx := context.Background() - name := j.Model +// prepared is a model's new weights, fetched and checked: what an update and +// a canary both start from. +type prepared struct { + m catalog.Model + o *catalog.Origin + repo hf.Repo + set hf.FileSet + source string +} + +// prepare reads where a model came from, fetches its repository's newer +// files into the mirror beside the ones in use, and checks the nodes can +// load them. It fails the job and says false when any of that cannot be. +func (s *Server) prepare(ctx context.Context, j *Job, name string) (prepared, bool) { // 1. The registry: which files, at which commit. s.step(j, "registry", Running, "reading where "+name+" came from") m, err := s.current(name) if err != nil { s.fail(j, "registry", err) - return + return prepared{}, false } list, err := s.deps.Mirror.List(ctx) if err != nil { s.fail(j, "registry", fmt.Errorf("the mirror: %w", err)) - return + return prepared{}, false } o := s.originOf(ctx, m, list) p, onMirror := placeOf(m.Source, list) if o == nil || !onMirror { s.fail(j, "registry", fmt.Errorf("%s did not come from the registry through the mirror", name)) - return + return prepared{}, false } repo, err := s.deps.Registry.Repo(ctx, o.Repo) if err != nil { s.fail(j, "registry", err) - return + return prepared{}, false } s.mu.Lock() j.Request.Repo = repo.ID @@ -485,18 +501,18 @@ func (s *Server) runUpdate(j *Job) { switch { case sameRevision(repo.Revision, o.Revision): s.fail(j, "registry", fmt.Errorf("%s already runs %s's latest commit, %s", name, repo.ID, catalog.ShortRevision(repo.Revision))) - return + return prepared{}, false case len(u.Missing) > 0: s.fail(j, "registry", fmt.Errorf("%s at %s no longer has %s; choose one of its files and serve it under a new name", repo.ID, repo.ShortRevision(), strings.Join(u.Missing, ", "))) - return + return prepared{}, false case len(u.Changed) == 0: s.fail(j, "registry", fmt.Errorf("%s has a newer commit, %s, but the files %s runs are the same in it", repo.ID, repo.ShortRevision(), name)) - return + return prepared{}, false case repo.Gated.Is(): s.fail(j, "registry", fmt.Errorf("%s is gated: it needs an account's approval the mirror does not have", repo.ID)) - return + return prepared{}, false } set := hf.FileSet{} for _, f := range o.Files { @@ -512,7 +528,7 @@ func (s *Server) runUpdate(j *Job) { source, err := s.mirror(ctx, j, repo, set, p.model) if err != nil { s.fail(j, "mirror", err) - return + return prepared{}, false } // 3. Whether the nodes load it, and whether it still fits what is @@ -523,7 +539,7 @@ func (s *Server) runUpdate(j *Job) { v := gguf.Check(ctx, u, s.deps.LlamaVersion) if !v.Servable { s.fail(j, "check", fmt.Errorf("llama.cpp %s on the nodes cannot run the new weights: %s", s.deps.LlamaVersion, v.Reason)) - return + return prepared{}, false } detail = orUnknown(v.Header.Architecture) if len(m.Profiles) > 0 { @@ -540,6 +556,17 @@ func (s *Server) runUpdate(j *Job) { } } s.step(j, "check", Done, detail) + return prepared{m: m, o: o, repo: repo, set: set, source: source}, true +} + +func (s *Server) runUpdate(j *Job) { + ctx := context.Background() + name := j.Model + p, ok := s.prepare(ctx, j, name) + if !ok { + return + } + m, o, repo, set, source := p.m, p.o, p.repo, p.set, p.source // 4. The catalog: the new source, and the old one kept to go back to. s.step(j, "catalog", Running, "moving "+name+" to "+repo.ShortRevision()) diff --git a/internal/control/modelserve/updates_test.go b/internal/control/modelserve/updates_test.go index ee3b22ff..81c94303 100644 --- a/internal/control/modelserve/updates_test.go +++ b/internal/control/modelserve/updates_test.go @@ -60,6 +60,8 @@ type updModels struct { mu sync.Mutex models map[string]catalog.Model declared map[string]bool + // onAdd starts what is added, as the reconciler would. + onAdd func(catalog.Model) } func (c *updModels) Declared(n string) bool { return c.declared[n] } @@ -79,6 +81,29 @@ func (c *updModels) SetOrigin(_ context.Context, n string, o *catalog.Origin) er c.models[n] = m return nil } +func (c *updModels) Add(_ context.Context, m catalog.Model, _ string) error { + c.mu.Lock() + defer c.mu.Unlock() + c.models[m.Name] = m + if c.onAdd != nil { + c.onAdd(m) + } + return nil +} +func (c *updModels) Remove(_ context.Context, n string) error { + c.mu.Lock() + defer c.mu.Unlock() + delete(c.models, n) + return nil +} +func (c *updModels) SetCanary(_ context.Context, n string, cn *catalog.Canary, _ string) (catalog.Model, error) { + c.mu.Lock() + defer c.mu.Unlock() + m := c.models[n] + m.Canary = cn + c.models[n] = m + return m, nil +} func (c *updModels) all() []catalog.Model { c.mu.Lock() defer c.mu.Unlock() diff --git a/internal/control/modelsvc/modelsvc.go b/internal/control/modelsvc/modelsvc.go index b45be61d..36fa40bc 100644 --- a/internal/control/modelsvc/modelsvc.go +++ b/internal/control/modelsvc/modelsvc.go @@ -306,6 +306,23 @@ func (s *Service) SetSource(ctx context.Context, name, source string, origin *ca return final, err } +// SetCanary gives a model added at runtime a canary, changes its share, or +// with nil takes it away. The candidate must already be in the catalog. +func (s *Service) SetCanary(ctx context.Context, name string, c *catalog.Canary, by string) (catalog.Model, error) { + final, err := s.update(ctx, name, by, "set its canary", func(m *catalog.Model) error { + m.Canary = c + return nil + }) + if err == nil { + if c == nil { + s.log.Info("model canary removed", "model", name, "by", by) + } else { + s.log.Info("model canary set", "model", name, "canary", c.Model, "percent", c.Percent, "by", by) + } + } + return final, err +} + // SetOrigin records where a model's weights came from without changing them: // for a model added before origins were kept. A model from configuration is // left alone, its origin worked out again when it is needed. diff --git a/internal/notify/notify.go b/internal/notify/notify.go index 8be77a15..0eca9b56 100644 --- a/internal/notify/notify.go +++ b/internal/notify/notify.go @@ -38,10 +38,13 @@ const ( // JudgeUnavailable is a guardrails judge giving no verdict: with // fail: closed, requests are being refused; with open, served unjudged. JudgeUnavailable = "judge_unavailable" + // CanaryRolledBack is a model's canary taken out because it served + // worse than the model it was a candidate to replace. + CanaryRolledBack = "canary_rolled_back" ) // Kinds are every event there is. -var Kinds = []string{ModelParked, NodeLost, BudgetSpent, UpdateAvailable, JudgeUnavailable} +var Kinds = []string{ModelParked, NodeLost, BudgetSpent, UpdateAvailable, JudgeUnavailable, CanaryRolledBack} // Event is one thing that happened. type Event struct {