diff --git a/internal/control/catalog/catalog.go b/internal/control/catalog/catalog.go index d3a55a6..e0bce3e 100644 --- a/internal/control/catalog/catalog.go +++ b/internal/control/catalog/catalog.go @@ -14,8 +14,8 @@ import ( // Scaling is how many replicas a model should have. type Scaling struct { - Min int `koanf:"min"` - Max int `koanf:"max"` + Min int `json:"min,omitempty" koanf:"min"` + Max int `json:"max,omitempty" koanf:"max"` } // SLO is the latency target a model is expected to meet. `ctl bench` measures @@ -27,19 +27,19 @@ type SLO struct { // Profile is how a model runs on one GPU class. type Profile struct { - Name string `koanf:"name"` - Engine string `koanf:"engine"` - GPUClass string `koanf:"gpu_class"` + Name string `json:"name,omitempty" koanf:"name"` + Engine string `json:"engine,omitempty" koanf:"engine"` + GPUClass string `json:"gpu_class,omitempty" koanf:"gpu_class"` - TP int `koanf:"tp"` - PP int `koanf:"pp"` + TP int `json:"tp,omitempty" koanf:"tp"` + PP int `json:"pp,omitempty" koanf:"pp"` // GPUsPerNode defaults to TP for a single-node profile. - GPUsPerNode int `koanf:"gpus_per_node"` + GPUsPerNode int `json:"gpus_per_node,omitempty" koanf:"gpus_per_node"` // Nodes above 1 makes this a multi-node profile, which requires a fabric. - Nodes int `koanf:"nodes"` + Nodes int `json:"nodes,omitempty" koanf:"nodes"` - Args []string `koanf:"args"` - Env map[string]string `koanf:"env"` + Args []string `json:"args,omitempty" koanf:"args"` + Env map[string]string `json:"env,omitempty" koanf:"env"` // MinMemoryGB is the host RAM this profile needs to run. // @@ -53,7 +53,7 @@ type Profile struct { // this. A node that does not report its memory is not excluded either — // an older agent must not have its nodes quietly emptied by an upgrade of // the control plane. - MinMemoryGB int `koanf:"min_memory_gb"` + MinMemoryGB int `json:"min_memory_gb,omitempty" koanf:"min_memory_gb"` // NodeSelector restricts this profile to nodes carrying every one of these // labels, with these values. @@ -63,16 +63,16 @@ type Profile struct { // in the inventory says how much memory a node has, so the operator says // which nodes are meant for a model instead of the scheduler inferring it. // Agents have reported labels since the beginning; this is what reads them. - NodeSelector map[string]string `koanf:"node_selector"` + NodeSelector map[string]string `json:"node_selector,omitempty" koanf:"node_selector"` // Weight is the router's capacity weight. It should come from ctl bench // rather than a guess (plan §6.2.1). - Weight float64 `koanf:"weight"` + Weight float64 `json:"weight,omitempty" koanf:"weight"` // Priority orders profiles during placement; highest that fits wins. - Priority int `koanf:"priority"` + Priority int `json:"priority,omitempty" koanf:"priority"` // StartupTimeout bounds model loading. A 70B model across two nodes takes // tens of minutes, so this is per profile rather than global. - StartupTimeout time.Duration `koanf:"startup_timeout"` + StartupTimeout time.Duration `json:"startup_timeout,omitempty" koanf:"startup_timeout"` } // CPUOnly reports whether this profile needs no GPUs. @@ -91,16 +91,16 @@ func cpuCapable(engine string) bool { return engine == "llamacpp" } // Model is a catalog entry as clients see it. type Model struct { - Name string `koanf:"name"` - Aliases []string `koanf:"aliases"` + Name string `json:"name,omitempty" koanf:"name"` + Aliases []string `json:"aliases,omitempty" koanf:"aliases"` // Source is the weights path or a HuggingFace id. - Source string `koanf:"source"` - Profiles []Profile `koanf:"profiles"` - Scaling Scaling `koanf:"scaling"` - Fallbacks []string `koanf:"fallbacks"` + Source string `json:"source,omitempty" koanf:"source"` + Profiles []Profile `json:"profiles,omitempty" koanf:"profiles"` + Scaling Scaling `json:"scaling,omitempty" koanf:"scaling"` + Fallbacks []string `json:"fallbacks,omitempty" koanf:"fallbacks"` // RoutingProfile selects how the router scores endpoints for this model. - RoutingProfile string `koanf:"routing_profile"` - SLO SLO `koanf:"slo"` + RoutingProfile string `json:"routing_profile,omitempty" koanf:"routing_profile"` + SLO SLO `json:"slo,omitempty" koanf:"slo"` // ContextLength is how many tokens one request may use, prompt and // completion together, and MaxOutput bounds the completion alone. Zero @@ -112,11 +112,11 @@ type Model struct { // the one an operator writes in the engine's arguments — llama.cpp divides // --ctx-size across its parallel slots, so a pool of 8192 across four slots // is 2048 a request, and nothing said so. - ContextLength int `koanf:"context_length"` - MaxOutput int `koanf:"max_output"` + ContextLength int `json:"context_length,omitempty" koanf:"context_length"` + MaxOutput int `json:"max_output,omitempty" koanf:"max_output"` // Capabilities is what this model can do beyond plain completion. - Capabilities Capabilities `koanf:"capabilities"` + Capabilities Capabilities `json:"capabilities,omitempty" koanf:"capabilities"` } // Capabilities describes what a client may ask a model for. @@ -131,9 +131,9 @@ type Model struct { // then behaves as though the model is broken, and the only way to find out was // to try it. That cost an afternoon. type Capabilities struct { - Tools bool `koanf:"tools"` - Vision bool `koanf:"vision"` - Embeddings bool `koanf:"embeddings"` + Tools bool `json:"tools,omitempty" koanf:"tools"` + Vision bool `json:"vision,omitempty" koanf:"vision"` + Embeddings bool `json:"embeddings,omitempty" koanf:"embeddings"` } // Catalog is the set of models the cluster should serve. diff --git a/internal/control/modelsvc/modelsvc.go b/internal/control/modelsvc/modelsvc.go new file mode 100644 index 0000000..2b09ed0 --- /dev/null +++ b/internal/control/modelsvc/modelsvc.go @@ -0,0 +1,218 @@ +// Package modelsvc owns the catalog that is in force: the models declared in +// configuration, plus the ones added while the cluster is running. +// +// Two sources rather than one, deliberately. Configuration stays declarative — +// a model in the YAML is in the catalog on every start, whatever the database +// says — because that is how a team keeps its production set under review and +// in version control. The database holds what someone added afterwards, which +// is the case this exists for: trying a model should not mean a config change, +// a restart and an outage. +// +// Configuration wins every conflict, and the surprising direction is the +// dangerous one: a request for a reviewed production model must never be +// answered by something added from a laptop. A stored model that collides with +// a declared name is refused when it is added and skipped if it somehow +// survives into the database. +package modelsvc + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "sync" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" + "gitea.mynodes.xyz/ilker/llmgw/internal/store" +) + +// Store is the persistence this needs. +type Store interface { + PutModel(ctx context.Context, m store.StoredModel) error + DeleteModel(ctx context.Context, name string) error + Models(ctx context.Context) ([]store.StoredModel, error) +} + +// Applier receives the rebuilt catalog. The controller implements it. +type Applier interface { + ReplaceCatalog(ctx context.Context, c *catalog.Catalog) +} + +// Service adds and removes models without a restart. +type Service struct { + declared []catalog.Model + store Store + apply Applier + log *slog.Logger + + // mu serialises rebuilds. They are rare — a person adding a model — and two + // at once could apply catalogs in either order, leaving the one in force + // disagreeing with the database that outlives it. + mu sync.Mutex +} + +// New builds the service around the models configuration declared. +func New(declared []catalog.Model, st Store, apply Applier, log *slog.Logger) *Service { + if log == nil { + log = slog.Default() + } + return &Service{declared: declared, store: st, apply: apply, log: log} +} + +// Declared reports whether a name comes from configuration. +func (s *Service) Declared(name string) bool { + for _, d := range s.declared { + if d.Name == name { + return true + } + } + return false +} + +// Add puts a model in the catalog and keeps it across restarts. +// +// Validated by building the whole catalog before anything is written. A model +// that collides with an existing name, names a fallback that does not exist, or +// is malformed in any of the ways catalog.New refuses is rejected here rather +// than persisted and then skipped on every future start — which would be a +// model that looks added and never appears. +func (s *Service) Add(ctx context.Context, m catalog.Model, by string) error { + s.mu.Lock() + defer s.mu.Unlock() + + if s.Declared(m.Name) { + return fmt.Errorf("model %q is declared in configuration; edit that rather than adding a second definition", m.Name) + } + + stored, err := s.storedModels(ctx) + if err != nil { + return err + } + next := append([]catalog.Model{}, s.declared...) + for _, e := range stored { + if e.Name == m.Name { + continue // this call replaces it + } + next = append(next, e) + } + next = append(next, m) + + built, err := catalog.New(next) + if err != nil { + return fmt.Errorf("model %q would make the catalog invalid: %w", m.Name, err) + } + + def, err := json.Marshal(m) + if err != nil { + return err + } + if err := s.store.PutModel(ctx, store.StoredModel{ + Name: m.Name, Definition: def, CreatedAt: time.Now(), CreatedBy: by, + }); err != nil { + return err + } + + s.apply.ReplaceCatalog(ctx, built) + s.log.Info("model added", "model", m.Name, "by", by) + return nil +} + +// Remove takes a model out of the catalog. +// +// A model declared in configuration cannot be removed this way: it would come +// back on the next restart, so the cluster would disagree with its own config +// file until then. Refusing and saying where to edit is the smaller surprise. +func (s *Service) Remove(ctx context.Context, name string) error { + s.mu.Lock() + defer s.mu.Unlock() + + if s.Declared(name) { + return fmt.Errorf("model %q is declared in configuration; remove it there, or it comes back on the next restart", name) + } + + stored, err := s.storedModels(ctx) + if err != nil { + return err + } + next := append([]catalog.Model{}, s.declared...) + found := false + for _, e := range stored { + if e.Name == name { + found = true + continue + } + next = append(next, e) + } + if !found { + return fmt.Errorf("model %q was not added at runtime", name) + } + + built, err := catalog.New(next) + if err != nil { + // Removing a model can invalidate a catalog: another may name it as a + // fallback. Refusing beats leaving a fallback pointing at nothing, + // which is discovered during the outage it was meant to cover. + return fmt.Errorf("removing %q would make the catalog invalid: %w", name, err) + } + if err := s.store.DeleteModel(ctx, name); err != nil { + return err + } + + s.apply.ReplaceCatalog(ctx, built) + s.log.Info("model removed", "model", name) + return 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 +// field changed meaning — is reported and skipped. Refusing to start over one +// bad row would let a model somebody added from a laptop take down the cluster, +// and the models in configuration are precisely the ones that must survive it. +func (s *Service) Load(ctx context.Context) (*catalog.Catalog, error) { + s.mu.Lock() + defer s.mu.Unlock() + + stored, err := s.storedModels(ctx) + if err != nil { + return nil, err + } + + next := append([]catalog.Model{}, s.declared...) + for _, m := range stored { + // Checked one at a time and against what is already accepted, so one + // bad model cannot take a good one with it. + candidate := append(append([]catalog.Model{}, next...), m) + if _, err := catalog.New(candidate); err != nil { + s.log.Warn("a stored model was skipped; it does not make a valid catalog", + "model", m.Name, "err", err) + continue + } + next = append(next, m) + } + return catalog.New(next) +} + +// storedModels reads and decodes the runtime models, skipping any that no +// longer parse rather than failing every operation around them. +func (s *Service) storedModels(ctx context.Context) ([]catalog.Model, error) { + if s.store == nil { + return nil, nil + } + rows, err := s.store.Models(ctx) + if err != nil { + return nil, err + } + out := make([]catalog.Model, 0, len(rows)) + for _, r := range rows { + var m catalog.Model + if err := json.Unmarshal(r.Definition, &m); err != nil { + s.log.Warn("a stored model could not be decoded and was skipped", + "model", r.Name, "err", err) + continue + } + out = append(out, m) + } + return out, nil +} diff --git a/internal/control/modelsvc/modelsvc_test.go b/internal/control/modelsvc/modelsvc_test.go new file mode 100644 index 0000000..33cedc5 --- /dev/null +++ b/internal/control/modelsvc/modelsvc_test.go @@ -0,0 +1,196 @@ +package modelsvc + +import ( + "context" + "encoding/json" + "log/slog" + "strings" + "testing" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" + "gitea.mynodes.xyz/ilker/llmgw/internal/store" +) + +type memStore struct { + rows map[string]store.StoredModel + err error +} + +func newMemStore() *memStore { return &memStore{rows: map[string]store.StoredModel{}} } + +func (m *memStore) PutModel(_ context.Context, s store.StoredModel) error { + if m.err != nil { + return m.err + } + m.rows[s.Name] = s + return nil +} + +func (m *memStore) DeleteModel(_ context.Context, name string) error { + delete(m.rows, name) + return nil +} + +func (m *memStore) Models(context.Context) ([]store.StoredModel, error) { + out := make([]store.StoredModel, 0, len(m.rows)) + for _, r := range m.rows { + out = append(out, r) + } + return out, nil +} + +type recorder struct{ last *catalog.Catalog } + +func (r *recorder) ReplaceCatalog(_ context.Context, c *catalog.Catalog) { r.last = c } + +func model(name string) catalog.Model { + return catalog.Model{ + Name: name, Source: "/models/" + name, + Scaling: catalog.Scaling{Min: 1, Max: 1}, + Profiles: []catalog.Profile{{Name: "p", Engine: "llamacpp", Weight: 1}}, + } +} + +func newService(t *testing.T, declared ...catalog.Model) (*Service, *memStore, *recorder) { + t.Helper() + st, rec := newMemStore(), &recorder{} + return New(declared, st, rec, slog.New(slog.DiscardHandler)), st, rec +} + +func TestAddedModelReachesTheCatalogAndSurvivesARestart(t *testing.T) { + // The whole point: trying a model should not mean a config change, a + // restart and an outage — and it should still be there tomorrow. + s, st, rec := newService(t, model("declared")) + + if err := s.Add(t.Context(), model("added"), "ilker"); err != nil { + t.Fatal(err) + } + if _, ok := rec.last.Resolve("added"); !ok { + t.Error("the new model is not in the catalog that was applied") + } + if _, ok := rec.last.Resolve("declared"); !ok { + t.Error("the declared model was lost") + } + if _, ok := st.rows["added"]; !ok { + t.Error("the model was not persisted") + } + + // A fresh service over the same store, as after a restart. + s2, _, _ := newService(t, model("declared")) + s2.store = st + built, err := s2.Load(t.Context()) + if err != nil { + t.Fatal(err) + } + if _, ok := built.Resolve("added"); !ok { + t.Error("the model did not survive a restart") + } +} + +func TestConfigurationWinsOverTheDatabase(t *testing.T) { + // The dangerous direction: a request for a reviewed production model must + // never be answered by something added from a laptop. + s, _, _ := newService(t, model("shared")) + err := s.Add(t.Context(), model("shared"), "ilker") + if err == nil { + t.Fatal("a stored model was allowed to shadow a declared one") + } + if !strings.Contains(err.Error(), "declared in configuration") { + t.Errorf("the error does not say why: %v", err) + } +} + +func TestADeclaredModelCannotBeRemovedAtRuntime(t *testing.T) { + // It would come back on the next restart, so the cluster would disagree + // with its own config file until then. + s, _, _ := newService(t, model("declared")) + err := s.Remove(t.Context(), "declared") + if err == nil { + t.Fatal("a declared model was removed at runtime") + } + if !strings.Contains(err.Error(), "comes back") { + t.Errorf("the error does not explain the consequence: %v", err) + } +} + +func TestNothingIsPersistedWhenTheCatalogWouldBeInvalid(t *testing.T) { + // A model that is written and then skipped on every start is a model that + // looks added and never appears. + s, st, rec := newService(t) + bad := model("bad") + bad.Fallbacks = []string{"a-model-that-does-not-exist"} + + if err := s.Add(t.Context(), bad, "ilker"); err == nil { + t.Fatal("a model naming a missing fallback was accepted") + } + if len(st.rows) != 0 { + t.Errorf("it was persisted anyway: %v", st.rows) + } + if rec.last != nil { + t.Error("a catalog was applied for a model that was rejected") + } +} + +func TestRemovingAModelOthersFallBackToIsRefused(t *testing.T) { + // Leaving a fallback pointing at nothing is discovered during the outage it + // was meant to cover. + s, _, _ := newService(t) + if err := s.Add(t.Context(), model("small"), "ilker"); err != nil { + t.Fatal(err) + } + big := model("big") + big.Fallbacks = []string{"small"} + if err := s.Add(t.Context(), big, "ilker"); err != nil { + t.Fatal(err) + } + + if err := s.Remove(t.Context(), "small"); err == nil { + t.Fatal("removed a model another falls back to") + } +} + +func TestOneUnreadableStoredModelDoesNotStopTheCluster(t *testing.T) { + // Refusing to start over one bad row would let a model somebody added from + // a laptop take the cluster down, and the declared models are precisely the + // ones that must survive that. + s, st, _ := newService(t, model("declared")) + good, _ := json.Marshal(model("good")) + st.rows["good"] = store.StoredModel{Name: "good", Definition: good, CreatedAt: time.Now()} + st.rows["broken"] = store.StoredModel{ + Name: "broken", Definition: []byte(`{"name":`), CreatedAt: time.Now(), + } + + built, err := s.Load(t.Context()) + if err != nil { + t.Fatalf("one bad row stopped the catalog from loading: %v", err) + } + for _, name := range []string{"declared", "good"} { + if _, ok := built.Resolve(name); !ok { + t.Errorf("%s was lost alongside the bad row", name) + } + } + if _, ok := built.Resolve("broken"); ok { + t.Error("the unreadable model was loaded anyway") + } +} + +func TestAStoredModelThatNoLongerValidatesIsSkippedAlone(t *testing.T) { + // Checked one at a time and against what is already accepted, so one bad + // model cannot take a good one with it. + s, st, _ := newService(t) + bad := model("bad") + bad.Profiles = nil // a model with no profile cannot be placed + badJSON, _ := json.Marshal(bad) + goodJSON, _ := json.Marshal(model("good")) + st.rows["bad"] = store.StoredModel{Name: "bad", Definition: badJSON, CreatedAt: time.Now()} + st.rows["good"] = store.StoredModel{Name: "good", Definition: goodJSON, CreatedAt: time.Now().Add(time.Second)} + + built, err := s.Load(t.Context()) + if err != nil { + t.Fatalf("load failed: %v", err) + } + if _, ok := built.Resolve("good"); !ok { + t.Error("a valid stored model was dropped because another was invalid") + } +} diff --git a/internal/store/migrations/sqlite/0005_models.sql b/internal/store/migrations/sqlite/0005_models.sql new file mode 100644 index 0000000..b928fd1 --- /dev/null +++ b/internal/store/migrations/sqlite/0005_models.sql @@ -0,0 +1,25 @@ +-- Models added while the cluster is running. +-- +-- The catalog came from a YAML file read once at startup, so adding a model +-- meant editing a file on the control plane and restarting it: a settling +-- window, adoption churn, and on one occasion two minutes of outage. That is a +-- reasonable way to manage a fixed production set and a bad one for a platform +-- whose promise is that people can try things. +-- +-- These live alongside the ones in configuration rather than replacing them. +-- Config-as-code stays declarative — a model in YAML is in the catalog on every +-- start, whatever the database says — and this table holds the ones added +-- afterwards. A model added here may not take a name or alias that +-- configuration already uses, which the catalog's own constructor enforces for +-- free, because it refuses a duplicate whatever the source. +-- +-- The definition is stored as JSON rather than as columns. A profile is a +-- nested, evolving shape, and every field added to it would otherwise be a +-- migration; the catalog already knows how to validate one, and a row that +-- fails validation is reported and skipped rather than blocking startup. +CREATE TABLE models ( + name TEXT PRIMARY KEY, + definition TEXT NOT NULL, + created_at TIMESTAMP NOT NULL, + created_by TEXT NOT NULL DEFAULT '' +); diff --git a/internal/store/models.go b/internal/store/models.go new file mode 100644 index 0000000..9c88d4d --- /dev/null +++ b/internal/store/models.go @@ -0,0 +1,68 @@ +package store + +import ( + "context" + "encoding/json" + "fmt" + "time" +) + +// StoredModel is a model added at runtime rather than declared in configuration. +type StoredModel struct { + Name string + // Definition is the catalog entry, as JSON. Stored whole because a profile + // is a nested and still-growing shape: as columns, every field added to it + // would be a migration, and the catalog already knows how to validate one. + Definition []byte + CreatedAt time.Time + CreatedBy string +} + +// PutModel adds or replaces a runtime model. +func (s *SQLite) PutModel(ctx context.Context, m StoredModel) error { + if m.Name == "" { + return fmt.Errorf("a model needs a name") + } + if !json.Valid(m.Definition) { + return fmt.Errorf("model %q: the definition is not valid JSON", m.Name) + } + if m.CreatedAt.IsZero() { + m.CreatedAt = time.Now() + } + _, err := s.db.ExecContext(ctx, + `INSERT INTO models (name, definition, created_at, created_by) + VALUES (?, ?, ?, ?) + ON CONFLICT(name) DO UPDATE SET definition = excluded.definition`, + m.Name, string(m.Definition), m.CreatedAt, m.CreatedBy) + return err +} + +// DeleteModel removes a runtime model. Removing one that is not there is not an +// error: the caller wanted it gone, and it is. +func (s *SQLite) DeleteModel(ctx context.Context, name string) error { + _, err := s.db.ExecContext(ctx, `DELETE FROM models WHERE name = ?`, name) + return err +} + +// Models lists the runtime models, oldest first so the catalog is built in a +// stable order and two starts produce the same result. +func (s *SQLite) Models(ctx context.Context) ([]StoredModel, error) { + rows, err := s.db.QueryContext(ctx, + `SELECT name, definition, created_at, created_by FROM models ORDER BY created_at, name`) + if err != nil { + return nil, err + } + defer rows.Close() + + var out []StoredModel + for rows.Next() { + var m StoredModel + var def string + if err := rows.Scan(&m.Name, &def, &m.CreatedAt, &m.CreatedBy); err != nil { + return nil, err + } + m.Definition = []byte(def) + out = append(out, m) + } + return out, rows.Err() +}