From 331de9b495dd4133bdcd4d22ac0846aba8fb2dd1 Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Thu, 17 Sep 2026 06:27:53 +0200 Subject: [PATCH] Add and remove models from the command line MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The last piece of not restarting a control plane to try a model: llmgw ctl model add -f qwen-coder-32b.yaml llmgw ctl model remove qwen-coder-32b The file is the same YAML a model takes in server.yaml, read by the same koanf parser through the same tags, so an entry moves between the two by being copied and a field that works in one cannot silently mean nothing in the other. Wired only when there is a store. A control plane without one says so rather than accepting a model that vanishes on the next restart, which is the kind of success that wastes an afternoon. At startup the catalog is rebuilt from configuration plus what was stored, so models added before a restart are still there after it — without that the whole feature would be a slower way to lose work. Every failure adding a model is the definition being wrong — a collision, a missing fallback, a profile that cannot be placed — so they are 400s. A 500 would send someone to the server's logs for a mistake in their own file. The actor comes from the caller and is kept with the row, so a catalog that has grown over months can say where each entry came from. Co-Authored-By: Claude Opus 5 (1M context) --- internal/cli/ctl.go | 14 ++++ internal/cli/model.go | 110 ++++++++++++++++++++++++++ internal/cli/server.go | 17 ++++ internal/control/adminapi/api.go | 57 ++++++++++++- internal/control/adminapi/api_test.go | 107 +++++++++++++++++++++++++ 5 files changed, 304 insertions(+), 1 deletion(-) create mode 100644 internal/cli/model.go diff --git a/internal/cli/ctl.go b/internal/cli/ctl.go index 31bcefca..c88b3e3b 100644 --- a/internal/cli/ctl.go +++ b/internal/cli/ctl.go @@ -37,6 +37,16 @@ func (c *ctlClient) get(ctx context.Context, path string, out any) error { return c.do(ctx, http.MethodGet, path, nil, out) } +// post sends an already-encoded body, so a caller that has marshalled a typed +// value keeps its field names rather than having them re-derived here. +func (c *ctlClient) post(ctx context.Context, path string, body []byte, out any) error { + return c.do(ctx, http.MethodPost, path, json.RawMessage(body), out) +} + +func (c *ctlClient) delete(ctx context.Context, path string, out any) error { + return c.do(ctx, http.MethodDelete, path, nil, out) +} + func (c *ctlClient) do(ctx context.Context, method, path string, body, out any) error { var rdr io.Reader if body != nil { @@ -54,6 +64,9 @@ func (c *ctlClient) do(ctx context.Context, method, path string, body, out any) if body != nil { req.Header.Set("Content-Type", "application/json") } + // Who made the change, kept with a stored model so a catalog that grew + // over months can say where each entry came from. + req.Header.Set("X-Llmgw-Actor", actor()) resp, err := c.http.Do(req) if err != nil { @@ -110,6 +123,7 @@ func newCtlCmd() *cobra.Command { // apply to it; it lives here because this is where operators look. newBenchCmd(&asJSON), newStatusCmd(client, &asJSON), + newModelCmd(client), newEngineCheckCmd(), newToolCheckCmd(), newCertCmd(), diff --git a/internal/cli/model.go b/internal/cli/model.go new file mode 100644 index 00000000..374d1bd7 --- /dev/null +++ b/internal/cli/model.go @@ -0,0 +1,110 @@ +package cli + +import ( + "encoding/json" + "fmt" + "os/user" + + "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/catalog" +) + +// newModelCmd adds and removes models without a restart. +// +// Adding a model used to mean editing a file on the control plane and +// restarting it: a settling window, adoption churn, and on one occasion two +// minutes of outage. For a fixed production set that is a reasonable trade; for +// a platform whose promise is that people can try things, it is the difference +// between trying one and not bothering. +func newModelCmd(client func() *ctlClient) *cobra.Command { + c := &cobra.Command{ + Use: "model", + Short: "Add or remove models while the cluster is running", + } + c.AddCommand(newModelAddCmd(client), newModelRemoveCmd(client)) + return c +} + +func newModelAddCmd(client func() *ctlClient) *cobra.Command { + var path string + c := &cobra.Command{ + Use: "add -f ", + Short: "Add a model from a definition file", + Long: "The definition is the same YAML a model takes in server.yaml, so an\n" + + "entry can be moved between the two by copying it.\n\n" + + "The catalog is rebuilt and validated before anything is written, so a\n" + + "definition that collides with an existing name, or names a fallback that\n" + + "does not exist, is refused rather than stored and then ignored on every\n" + + "start. Once accepted it takes effect on the reconciler's next pass — no\n" + + "restart — and survives one.", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + if path == "" { + return fmt.Errorf("--file is required: the model definition, as YAML") + } + // Read by koanf, the same parser and the same tags that read + // server.yaml — so an entry can be moved between the two by copying + // it, and a field that works in one cannot silently mean nothing in + // the other. + k := koanf.New(".") + if err := k.Load(file.Provider(path), yaml.Parser()); err != nil { + return fmt.Errorf("%s: %w", path, err) + } + var m catalog.Model + if err := k.UnmarshalWithConf("", &m, koanf.UnmarshalConf{Tag: "koanf"}); err != nil { + return fmt.Errorf("%s: %w", path, err) + } + if m.Name == "" { + return fmt.Errorf("%s: the definition has no name", path) + } + body, err := json.Marshal(m) + if err != nil { + return err + } + + var res map[string]string + if err := client().post(cmd.Context(), "/api/v1/models", body, &res); err != nil { + return err + } + fmt.Fprintf(cmd.OutOrStdout(), + "%s added. It will be placed on the next reconcile pass; watch it with:\n"+ + " llmgw ctl models\n", m.Name) + return nil + }, + } + c.Flags().StringVarP(&path, "file", "f", "", "YAML file holding the model definition") + return c +} + +func newModelRemoveCmd(client func() *ctlClient) *cobra.Command { + return &cobra.Command{ + Use: "remove ", + Short: "Remove a model that was added at runtime", + Long: "Only models added at runtime. One declared in server.yaml would come\n" + + "back on the next restart, so removing it here would leave the cluster\n" + + "disagreeing with its own configuration until then.\n\n" + + "Its replicas are stopped on the next pass, after the endpoint has been\n" + + "withdrawn, so traffic stops before the engine does.", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + var res map[string]string + if err := client().delete(cmd.Context(), "/api/v1/models/"+args[0], &res); err != nil { + return err + } + fmt.Fprintf(cmd.OutOrStdout(), "%s removed.\n", args[0]) + return nil + }, + } +} + +// actor names who made a change, for the audit trail on a stored model. +func actor() string { + if u, err := user.Current(); err == nil && u.Username != "" { + return u.Username + } + return "unknown" +} diff --git a/internal/cli/server.go b/internal/cli/server.go index 432e16b9..89184a31 100644 --- a/internal/cli/server.go +++ b/internal/cli/server.go @@ -24,6 +24,7 @@ import ( "gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi" "gitea.mynodes.xyz/ilker/llmgw/internal/control/agentsvc" "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/modelsvc" "gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler" "gitea.mynodes.xyz/ilker/llmgw/internal/router" "gitea.mynodes.xyz/ilker/llmgw/internal/version" @@ -121,6 +122,22 @@ func newServerCmd() *cobra.Command { } if admin != nil { api := adminapi.New(admin) + // Catalog changes need somewhere to persist, so this is wired + // only when there is a store. Without one, the endpoints say so + // rather than accepting a model that vanishes on restart. + if st := rt.Store(); st != nil { + svc := modelsvc.New(cfg.Models, st, admin, log) + // Rebuild the catalog from configuration plus what was + // stored, and put that in force — otherwise models added + // before the last restart would be silently absent until + // someone added them again. + if built, err := svc.Load(ctx); err != nil { + log.Error("could not load stored models; continuing with configuration only", "err", err) + } else { + admin.ReplaceCatalog(ctx, built) + } + api.SetModels(svc) + } api.Register(adminMux) // The dashboard lives here and only here. This listener is diff --git a/internal/control/adminapi/api.go b/internal/control/adminapi/api.go index 75791003..d96bbc60 100644 --- a/internal/control/adminapi/api.go +++ b/internal/control/adminapi/api.go @@ -7,19 +7,23 @@ package adminapi import ( + "context" "encoding/json" "fmt" + "io" "net/http" "time" "gitea.mynodes.xyz/ilker/llmgw/internal/control" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" "gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory" "gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler" ) // API serves the management endpoints. type API struct { - ctl *control.Controller + models Models + ctl *control.Controller } // New builds the API over a controller. @@ -28,6 +32,11 @@ func New(ctl *control.Controller) *API { return &API{ctl: ctl} } // Register mounts the endpoints on a mux under /api/v1. func (a *API) Register(mux *http.ServeMux) { mux.HandleFunc("GET /api/v1/summary", a.summary) + // Adding a model is a write, and the only one this API has. It lives here + // rather than on the client-facing listener for the same reason the rest of + // 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("GET /api/v1/nodes", a.listNodes) mux.HandleFunc("GET /api/v1/models", a.listModels) mux.HandleFunc("GET /api/v1/replicas", a.listReplicas) @@ -237,6 +246,52 @@ func (a *API) Replicas(model string) []ReplicaView { return out } +// Models is the service that owns catalog changes, when there is one. +// +// Optional because a control plane with no store cannot persist a model, and +// pretending otherwise would accept a model that vanishes on restart. +type Models interface { + Add(ctx context.Context, m catalog.Model, by string) error + Remove(ctx context.Context, name string) error +} + +// SetModels supplies the catalog-changing service. +func (a *API) SetModels(m Models) { a.models = m } + +func (a *API) addModel(w http.ResponseWriter, r *http.Request) { + if a.models == nil { + writeError(w, http.StatusNotImplemented, + "this control plane has no store, so a model added here would not survive a restart") + return + } + var m catalog.Model + if err := json.NewDecoder(io.LimitReader(r.Body, 1<<20)).Decode(&m); err != nil { + writeError(w, http.StatusBadRequest, "the model definition is not valid JSON: "+err.Error()) + return + } + if err := a.models.Add(r.Context(), m, r.Header.Get("X-Llmgw-Actor")); err != nil { + // Every failure here is the definition being wrong, which is the + // caller's to fix — a collision, a missing fallback, a profile that + // cannot be placed. None of them are server faults. + writeError(w, http.StatusBadRequest, err.Error()) + return + } + writeJSON(w, http.StatusOK, map[string]string{"added": m.Name}) +} + +func (a *API) removeModel(w http.ResponseWriter, r *http.Request) { + if a.models == nil { + writeError(w, http.StatusNotImplemented, "this control plane has no store") + return + } + name := r.PathValue("name") + if err := a.models.Remove(r.Context(), name); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + writeJSON(w, http.StatusOK, map[string]string{"removed": name}) +} + func (a *API) summary(w http.ResponseWriter, _ *http.Request) { writeJSON(w, http.StatusOK, a.Summary()) } diff --git a/internal/control/adminapi/api_test.go b/internal/control/adminapi/api_test.go index a5f27d7e..aaef26c1 100644 --- a/internal/control/adminapi/api_test.go +++ b/internal/control/adminapi/api_test.go @@ -3,6 +3,7 @@ package adminapi import ( "context" "encoding/json" + "errors" "log/slog" "net/http" "net/http/httptest" @@ -246,3 +247,109 @@ func TestEmptyClusterReturnsArraysNotNull(t *testing.T) { } } } + +type fakeModels struct { + added []catalog.Model + actors []string + removed []string + err error +} + +func (f *fakeModels) Add(_ context.Context, m catalog.Model, by string) error { + if f.err != nil { + return f.err + } + f.added = append(f.added, m) + f.actors = append(f.actors, by) + return nil +} + +func (f *fakeModels) Remove(_ context.Context, name string) error { + if f.err != nil { + return f.err + } + f.removed = append(f.removed, name) + return nil +} + +func TestAddingAModelOverTheAPI(t *testing.T) { + a, _ := setup(t, testModels()) + fm := &fakeModels{} + a.SetModels(fm) + mux := http.NewServeMux() + a.Register(mux) + + body := `{"name":"new","source":"/models/new","profiles":[{"name":"p","engine":"llamacpp"}]}` + req := httptest.NewRequest(http.MethodPost, "/api/v1/models", strings.NewReader(body)) + req.Header.Set("X-Llmgw-Actor", "ilker") + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + if len(fm.added) != 1 || fm.added[0].Name != "new" { + t.Fatalf("added = %+v", fm.added) + } + // Who made the change, kept so a catalog that grew over months can say + // where each entry came from. + if len(fm.actors) != 1 || fm.actors[0] != "ilker" { + t.Errorf("actor = %v", fm.actors) + } +} + +func TestABadModelDefinitionIsTheCallersFault(t *testing.T) { + // Every failure here is the definition being wrong — a collision, a missing + // fallback, a profile that cannot be placed. None are server faults, and + // returning 500 for them would send someone looking at the wrong logs. + a, _ := setup(t, testModels()) + a.SetModels(&fakeModels{err: errors.New("model \"x\" collides")}) + mux := http.NewServeMux() + a.Register(mux) + + req := httptest.NewRequest(http.MethodPost, "/api/v1/models", strings.NewReader(`{"name":"x"}`)) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + if rec.Code != http.StatusBadRequest { + t.Errorf("status %d, expected 400", rec.Code) + } + if !strings.Contains(rec.Body.String(), "collides") { + t.Errorf("the reason was lost: %s", rec.Body.String()) + } +} + +func TestWithoutAStoreTheAPIRefusesRatherThanForgets(t *testing.T) { + // Accepting a model that vanishes on the next restart is worse than saying + // it cannot be kept. + a, _ := setup(t, testModels()) // SetModels never called + mux := http.NewServeMux() + a.Register(mux) + + req := httptest.NewRequest(http.MethodPost, "/api/v1/models", strings.NewReader(`{"name":"x"}`)) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + if rec.Code != http.StatusNotImplemented { + t.Errorf("status %d, expected 501", rec.Code) + } + if !strings.Contains(rec.Body.String(), "restart") { + t.Errorf("the reason does not explain the consequence: %s", rec.Body.String()) + } +} + +func TestRemovingAModelOverTheAPI(t *testing.T) { + a, _ := setup(t, testModels()) + fm := &fakeModels{} + a.SetModels(fm) + mux := http.NewServeMux() + a.Register(mux) + + req := httptest.NewRequest(http.MethodDelete, "/api/v1/models/gone", nil) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + if len(fm.removed) != 1 || fm.removed[0] != "gone" { + t.Errorf("removed = %v", fm.removed) + } +}