diff --git a/internal/cli/ctl.go b/internal/cli/ctl.go index 918989ae..4100c591 100644 --- a/internal/cli/ctl.go +++ b/internal/cli/ctl.go @@ -124,6 +124,7 @@ func newCtlCmd() *cobra.Command { newBenchCmd(&asJSON), newStatusCmd(client, &asJSON), newModelCmd(client), + newScaleCmd(client), newEngineCheckCmd(), newToolCheckCmd(), newWeightsCmd(client, &asJSON), diff --git a/internal/cli/model.go b/internal/cli/model.go index fbbb7bcc..f971f7d8 100644 --- a/internal/cli/model.go +++ b/internal/cli/model.go @@ -4,12 +4,14 @@ import ( "encoding/json" "fmt" "os/user" + "time" "github.com/knadh/koanf/parsers/yaml" "github.com/knadh/koanf/providers/file" "github.com/knadh/koanf/v2" "github.com/spf13/cobra" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi" "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" ) @@ -128,3 +130,58 @@ func actor() string { } return "unknown" } + +func newScaleCmd(client func() *ctlClient) *cobra.Command { + var min, max int + var idle time.Duration + c := &cobra.Command{ + Use: "scale --min N [--max M]", + Short: "Change how many replicas a model runs", + Long: "The replica count is --min. --max is a ceiling, not a target: a model\n" + + "with a minimum runs exactly that many, and nothing scales it up under\n" + + "load. Leaving --max out makes it follow --min.\n\n" + + "--min 0 with a --max above it is the on-demand model: nothing runs until\n" + + "somebody asks, and it stops again after the idle timeout. That loads one\n" + + "replica however high --max is. --min 0 --max 0 is a model kept in the\n" + + "catalog and deliberately not running.\n\n" + + "Only models added at runtime. One declared in server.yaml is read again\n" + + "on every start, so a number set here would hold until the next restart\n" + + "and then revert — change it in the file instead.\n\n" + + "It takes effect on the reconciler's next pass, and survives a restart.", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + if !cmd.Flags().Changed("min") { + return fmt.Errorf("--min is required: it is the replica count") + } + body, err := json.Marshal(adminapi.ScaleRequest{Min: min, Max: max, IdleTimeout: idle}) + if err != nil { + return err + } + var res adminapi.ScaleResponse + if err := client().post(cmd.Context(), + "/api/v1/models/"+args[0]+"/scale", body, &res); err != nil { + return err + } + // What is in force, not what was asked for: the catalog fills in a + // max that was left out, and printing the request back would show a + // number the cluster is not using. + out := cmd.OutOrStdout() + switch { + case res.OnDemand: + fmt.Fprintf(out, "%s is on-demand: loaded when asked for, up to %d, stopped when idle.\n", + res.Model, res.Max) + case res.Min == 0: + fmt.Fprintf(out, "%s will not be placed; it stays in the catalog.\n", res.Model) + default: + fmt.Fprintf(out, "%s: %d replicas (max %d). It takes effect on the next pass; watch it with:\n"+ + " llmgw ctl models\n", res.Model, res.Min, res.Max) + } + return nil + }, + } + c.Flags().IntVar(&min, "min", 0, "replicas to keep running; this is the count") + c.Flags().IntVar(&max, "max", 0, "ceiling for an on-demand model; follows --min when left out") + c.Flags().DurationVar(&idle, "idle-timeout", 0, + "how long an on-demand model stays up unasked-for; zero takes the cluster default") + return c +} diff --git a/internal/control/adminapi/api.go b/internal/control/adminapi/api.go index cae6aa48..68389de5 100644 --- a/internal/control/adminapi/api.go +++ b/internal/control/adminapi/api.go @@ -37,6 +37,7 @@ func (a *API) Register(mux *http.ServeMux) { // it does: reaching this port already means holding the control plane. mux.HandleFunc("POST /api/v1/models", a.addModel) mux.HandleFunc("DELETE /api/v1/models/{name}", a.removeModel) + mux.HandleFunc("POST /api/v1/models/{name}/scale", a.scaleModel) mux.HandleFunc("GET /api/v1/nodes", a.listNodes) mux.HandleFunc("GET /api/v1/models", a.listModels) mux.HandleFunc("GET /api/v1/replicas", a.listReplicas) @@ -304,6 +305,7 @@ func (a *API) Replicas(model string) []ReplicaView { type Models interface { Add(ctx context.Context, m catalog.Model, by string) error Remove(ctx context.Context, name string) error + Scale(ctx context.Context, name string, s catalog.Scaling, by string) (catalog.Model, error) } // SetModels supplies the catalog-changing service. @@ -330,6 +332,59 @@ func (a *API) addModel(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, map[string]string{"added": m.Name}) } +// ScaleRequest is how many replicas a model should run. +type ScaleRequest struct { + Min int `json:"min"` + // Max omitted follows Min, which is what the catalog does for a model that + // declares only a minimum. + Max int `json:"max,omitempty"` + IdleTimeout time.Duration `json:"idle_timeout,omitempty"` +} + +// ScaleResponse reports the scaling that is now in force. +// +// What is in force rather than what was asked for: the catalog fills in a max +// that was left out, and a caller that echoed its own request would print a +// number the cluster is not using. +type ScaleResponse struct { + Model string `json:"model"` + Min int `json:"min"` + Max int `json:"max"` + IdleTimeout time.Duration `json:"idle_timeout,omitempty"` + OnDemand bool `json:"on_demand"` +} + +func (a *API) scaleModel(w http.ResponseWriter, r *http.Request) { + if a.models == nil { + writeError(w, http.StatusNotImplemented, "this control plane has no store") + return + } + var req ScaleRequest + if err := json.NewDecoder(io.LimitReader(r.Body, 1<<16)).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "the scaling request is not valid JSON: "+err.Error()) + return + } + if req.Min < 0 || req.Max < 0 { + writeError(w, http.StatusBadRequest, "a replica count cannot be negative") + return + } + name := r.PathValue("name") + m, err := a.models.Scale(r.Context(), name, + catalog.Scaling{Min: req.Min, Max: req.Max, IdleTimeout: req.IdleTimeout}, + r.Header.Get("X-Llmgw-Actor")) + if err != nil { + // Like addModel: every refusal here is the request being wrong — a + // model declared in configuration, one that was never added, a max + // below a min. None are server faults. + writeError(w, http.StatusBadRequest, err.Error()) + return + } + writeJSON(w, http.StatusOK, ScaleResponse{ + Model: m.Name, Min: m.Scaling.Min, Max: m.Scaling.Max, + IdleTimeout: m.Scaling.IdleTimeout, OnDemand: m.Scaling.OnDemand(), + }) +} + func (a *API) removeModel(w http.ResponseWriter, r *http.Request) { if a.models == nil { writeError(w, http.StatusNotImplemented, "this control plane has no store") diff --git a/internal/control/adminapi/api_test.go b/internal/control/adminapi/api_test.go index aaef26c1..4ccbfcca 100644 --- a/internal/control/adminapi/api_test.go +++ b/internal/control/adminapi/api_test.go @@ -252,9 +252,24 @@ type fakeModels struct { added []catalog.Model actors []string removed []string + scaled []catalog.Scaling err error } +func (f *fakeModels) Scale(_ context.Context, name string, s catalog.Scaling, by string) (catalog.Model, error) { + if f.err != nil { + return catalog.Model{}, f.err + } + f.scaled = append(f.scaled, s) + f.actors = append(f.actors, by) + // Max following min is what the catalog does; the fake has to do it too or + // the handler test would pass while the real answer reported a zero. + if s.Max == 0 && s.Min > 0 { + s.Max = s.Min + } + return catalog.Model{Name: name, Scaling: s}, nil +} + func (f *fakeModels) Add(_ context.Context, m catalog.Model, by string) error { if f.err != nil { return f.err @@ -353,3 +368,53 @@ func TestRemovingAModelOverTheAPI(t *testing.T) { t.Errorf("removed = %v", fm.removed) } } + +func TestScalingAModelOverTheAPI(t *testing.T) { + a, _ := setup(t, testModels()) + fm := &fakeModels{} + a.SetModels(fm) + mux := http.NewServeMux() + a.Register(mux) + + req := httptest.NewRequest(http.MethodPost, "/api/v1/models/m1/scale", + strings.NewReader(`{"min":3}`)) + req.Header.Set("X-Llmgw-Actor", "ilker") + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", rec.Code, rec.Body) + } + var res ScaleResponse + if err := json.Unmarshal(rec.Body.Bytes(), &res); err != nil { + t.Fatal(err) + } + // The scaling in force, not the request echoed back: max follows min when + // it is left out, and a caller printing its own request would show 0. + if res.Min != 3 || res.Max != 3 { + t.Errorf("expected 3/3 in force, got %d/%d", res.Min, res.Max) + } + if len(fm.actors) != 1 || fm.actors[0] != "ilker" { + t.Errorf("the change should be attributed to whoever made it; got %v", fm.actors) + } +} + +func TestANegativeReplicaCountIsRefused(t *testing.T) { + a, _ := setup(t, testModels()) + fm := &fakeModels{} + a.SetModels(fm) + mux := http.NewServeMux() + a.Register(mux) + + req := httptest.NewRequest(http.MethodPost, "/api/v1/models/m1/scale", + strings.NewReader(`{"min":-1}`)) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusBadRequest { + t.Errorf("expected 400, got %d", rec.Code) + } + if len(fm.scaled) != 0 { + t.Error("nothing should have reached the service") + } +} diff --git a/internal/control/modelsvc/modelsvc.go b/internal/control/modelsvc/modelsvc.go index 2b09ed0d..37b9a1b8 100644 --- a/internal/control/modelsvc/modelsvc.go +++ b/internal/control/modelsvc/modelsvc.go @@ -164,6 +164,73 @@ func (s *Service) Remove(ctx context.Context, name string) error { return nil } +// Scale changes how many replicas a model runs, without a restart. +// +// Only the scaling block moves. Everything else about the model — its source, +// its profiles, the engine arguments somebody tuned — is read back from what +// was stored and written out again untouched, so changing a number cannot +// quietly become a redefinition. +// +// A model declared in configuration is refused, for the same reason Remove +// refuses one: the YAML is read on every start, so the new number would hold +// until the next restart and then silently revert. A cluster that disagrees +// with its own config file is worse than an error message saying where to edit. +func (s *Service) Scale(ctx context.Context, name string, sc catalog.Scaling, by string) (catalog.Model, error) { + s.mu.Lock() + defer s.mu.Unlock() + + if s.Declared(name) { + return catalog.Model{}, fmt.Errorf( + "model %q is declared in configuration; change its scaling there, or it reverts on the next restart", name) + } + + stored, err := s.storedModels(ctx) + if err != nil { + return catalog.Model{}, err + } + + next := append([]catalog.Model{}, s.declared...) + found := false + for _, e := range stored { + if e.Name == name { + e.Scaling = sc + found = true + } + next = append(next, e) + } + if !found { + return catalog.Model{}, fmt.Errorf("model %q was not added at runtime", name) + } + + // Built before anything is written, like Add: catalog.New is what applies + // the defaults (max follows min when it is left out) and what refuses a + // max below a min, so the model that comes back is the one that will run + // rather than the one that was asked for. + built, err := catalog.New(next) + if err != nil { + return catalog.Model{}, fmt.Errorf("scaling %q that way would make the catalog invalid: %w", name, err) + } + final, ok := built.Resolve(name) + if !ok { + return catalog.Model{}, fmt.Errorf("model %q vanished from the catalog it was just built into", name) + } + + def, err := json.Marshal(final) + if err != nil { + return catalog.Model{}, err + } + if err := s.store.PutModel(ctx, store.StoredModel{ + Name: name, Definition: def, CreatedAt: time.Now(), CreatedBy: by, + }); err != nil { + return catalog.Model{}, err + } + + s.apply.ReplaceCatalog(ctx, built) + s.log.Info("model scaled", "model", name, + "min", final.Scaling.Min, "max", final.Scaling.Max, "by", by) + return final, nil +} + // Load builds the catalog at startup: configuration plus what was stored. // // A stored model that no longer validates — its engine left the build, a diff --git a/internal/control/modelsvc/modelsvc_test.go b/internal/control/modelsvc/modelsvc_test.go index 33cedc5e..63b49ddd 100644 --- a/internal/control/modelsvc/modelsvc_test.go +++ b/internal/control/modelsvc/modelsvc_test.go @@ -194,3 +194,146 @@ func TestAStoredModelThatNoLongerValidatesIsSkippedAlone(t *testing.T) { t.Error("a valid stored model was dropped because another was invalid") } } + +func TestScalingChangesTheCountAndSurvivesARestart(t *testing.T) { + s, st, rec := newService(t) + if err := s.Add(t.Context(), model("added"), "ilker"); err != nil { + t.Fatal(err) + } + + got, err := s.Scale(t.Context(), "added", catalog.Scaling{Min: 3}, "ilker") + if err != nil { + t.Fatal(err) + } + // Max follows min when it is left out, and the answer reports what the + // catalog settled on rather than what was asked for — a caller echoing its + // own request would print a max of zero. + if got.Scaling.Min != 3 || got.Scaling.Max != 3 { + t.Errorf("expected 3/3 in force, got %d/%d", got.Scaling.Min, got.Scaling.Max) + } + if m, _ := rec.last.Resolve("added"); m.Scaling.Min != 3 { + t.Errorf("the applied catalog still says %d", m.Scaling.Min) + } + + s2, _, _ := newService(t) + s2.store = st + built, err := s2.Load(t.Context()) + if err != nil { + t.Fatal(err) + } + if m, _ := built.Resolve("added"); m.Scaling.Min != 3 { + t.Errorf("the new count did not survive a restart; got %d", m.Scaling.Min) + } +} + +func TestScalingChangesNothingButTheCount(t *testing.T) { + s, _, rec := newService(t) + m := model("added") + m.Aliases = []string{"a"} + m.Profiles[0].Args = []string{"--ctx-size", "16384"} + if err := s.Add(t.Context(), m, "ilker"); err != nil { + t.Fatal(err) + } + + if _, err := s.Scale(t.Context(), "added", catalog.Scaling{Min: 2}, "ilker"); err != nil { + t.Fatal(err) + } + + // Changing a number must not quietly become a redefinition: the source and + // the engine arguments somebody tuned are read back and written out again. + got, _ := rec.last.Resolve("added") + if got.Source != "/models/added" { + t.Errorf("the source changed to %q", got.Source) + } + if len(got.Profiles) != 1 || strings.Join(got.Profiles[0].Args, " ") != "--ctx-size 16384" { + t.Errorf("the profile arguments were lost: %+v", got.Profiles) + } + if len(got.Aliases) != 1 || got.Aliases[0] != "a" { + t.Errorf("the aliases were lost: %v", got.Aliases) + } +} + +func TestADeclaredModelCannotBeScaledHere(t *testing.T) { + s, _, _ := newService(t, model("declared")) + + _, err := s.Scale(t.Context(), "declared", catalog.Scaling{Min: 5}, "ilker") + + // The YAML is read on every start, so the number would hold until the next + // restart and then silently revert. Refusing and saying where to edit is + // the smaller surprise — the same call Remove makes. + if err == nil { + t.Fatal("expected scaling a declared model to be refused") + } + if !strings.Contains(err.Error(), "configuration") { + t.Errorf("the error should say where to change it; got %q", err) + } +} + +func TestScalingAModelThatWasNeverAddedIsAnError(t *testing.T) { + s, _, _ := newService(t) + if _, err := s.Scale(t.Context(), "ghost", catalog.Scaling{Min: 1}, "ilker"); err == nil { + t.Error("expected an error for a model that is not there") + } +} + +func TestAMaxBelowAMinIsRefusedBeforeAnythingIsWritten(t *testing.T) { + s, st, _ := newService(t) + if err := s.Add(t.Context(), model("added"), "ilker"); err != nil { + t.Fatal(err) + } + + if _, err := s.Scale(t.Context(), "added", catalog.Scaling{Min: 3, Max: 1}, "ilker"); err == nil { + t.Fatal("expected a max below a min to be refused") + } + + // Refused before the write, so the stored definition is untouched: a + // rejected change that had already been persisted would come back on the + // next restart as a catalog that does not build. + var stored catalog.Model + if err := json.Unmarshal(st.rows["added"].Definition, &stored); err != nil { + t.Fatal(err) + } + if stored.Scaling.Min != 1 { + t.Errorf("the stored definition was changed to %d despite the refusal", stored.Scaling.Min) + } +} + +func TestScalingToZeroKeepsTheModelInTheCatalog(t *testing.T) { + s, _, rec := newService(t) + if err := s.Add(t.Context(), model("added"), "ilker"); err != nil { + t.Fatal(err) + } + + got, err := s.Scale(t.Context(), "added", catalog.Scaling{Min: 0, Max: 0}, "ilker") + if err != nil { + t.Fatal(err) + } + // Configured and deliberately not running is a different and equally + // useful thing from removed: the entry stays so it can be brought back + // with one number. + if got.Scaling.OnDemand() { + t.Error("min 0 max 0 is not on-demand; it is off") + } + if _, ok := rec.last.Resolve("added"); !ok { + t.Error("scaling to zero must not remove the model") + } +} + +func TestMinZeroWithAMaxIsOnDemand(t *testing.T) { + s, _, _ := newService(t) + if err := s.Add(t.Context(), model("added"), "ilker"); err != nil { + t.Fatal(err) + } + + got, err := s.Scale(t.Context(), "added", + catalog.Scaling{Min: 0, Max: 2, IdleTimeout: 10 * time.Minute}, "ilker") + if err != nil { + t.Fatal(err) + } + if !got.Scaling.OnDemand() { + t.Error("min 0 with a max above it should be on-demand") + } + if got.Scaling.IdleTimeout != 10*time.Minute { + t.Errorf("the idle timeout was lost; got %v", got.Scaling.IdleTimeout) + } +}