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