diff --git a/CHANGELOG.md b/CHANGELOG.md index f0e679a6..259d5887 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -27,6 +27,14 @@ APIs change between minors without a deprecation period. ## Unreleased +### Added + +- **A model stopped and kept, and a model removed with its files.** `llmgw ctl model stop|start MODEL`: none of its + replicas run, it stays in the catalog with its files, and start runs it again scaled as it was; requests are told an + administrator stopped it. `llmgw ctl model remove MODEL --purge` (`DELETE /api/v1/models/{name}?purge=true`) also + deletes every version of it on the mirror, a second mirror following, and the nodes' copies once its replicas have + stopped; what another model uses stays. Both are in the console's model menu. + ### Fixed - **`llmgw mirror add-peer` on a mirror with a machine of its own.** It set the second mirror up and then diff --git a/docs/admin-api.openapi.json b/docs/admin-api.openapi.json index a38c6d6e..71fe60a3 100644 --- a/docs/admin-api.openapi.json +++ b/docs/admin-api.openapi.json @@ -2387,7 +2387,7 @@ "delete": { "operationId": "deleteModelsName", "summary": "Remove a model added at runtime", - "description": "A model declared in the cluster's settings is refused: it is changed there.", + "description": "A model declared in the cluster's settings is refused: it is changed there. With purge, its files go too: every version of it the mirror holds (a peer mirror follows), and the nodes' copies once its replicas have stopped, waiting up to a minute for them; what another model uses stays. The answer is then a PurgeView.", "tags": [ "models" ], @@ -2401,6 +2401,14 @@ "type": "string" } }, + { + "name": "purge", + "in": "query", + "description": "Delete its files from the mirror and the nodes as well.", + "schema": { + "type": "boolean" + } + }, { "$ref": "#/components/parameters/Actor" } @@ -3400,6 +3408,89 @@ } } }, + "/api/v1/models/{name}/start": { + "post": { + "operationId": "postModelsNameStart", + "summary": "Run a stopped model again, scaled as it was", + "tags": [ + "models" + ], + "parameters": [ + { + "name": "name", + "in": "path", + "required": true, + "description": "The model's name in the catalog.", + "schema": { + "type": "string" + } + }, + { + "$ref": "#/components/parameters/Actor" + } + ], + "responses": { + "200": { + "description": "OK", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ScaleResponse" + } + } + } + }, + "401": { + "$ref": "#/components/responses/Unauthorized" + }, + "default": { + "$ref": "#/components/responses/Error" + } + } + } + }, + "/api/v1/models/{name}/stop": { + "post": { + "operationId": "postModelsNameStop", + "summary": "Take a model out of service and keep it", + "description": "None of its replicas run, and requests are told an administrator stopped it; what it was scaled to is kept for start.", + "tags": [ + "models" + ], + "parameters": [ + { + "name": "name", + "in": "path", + "required": true, + "description": "The model's name in the catalog.", + "schema": { + "type": "string" + } + }, + { + "$ref": "#/components/parameters/Actor" + } + ], + "responses": { + "200": { + "description": "OK", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ScaleResponse" + } + } + } + }, + "401": { + "$ref": "#/components/responses/Unauthorized" + }, + "default": { + "$ref": "#/components/responses/Error" + } + } + } + }, "/api/v1/models/{name}/target": { "get": { "operationId": "getModelsNameTarget", @@ -7859,6 +7950,10 @@ "max_output": { "type": "integer" }, + "stopped": { + "type": "boolean", + "description": "Stopped is a model an administrator took out of service, kept." + }, "canary": { "$ref": "#/components/schemas/catalog.Canary", "description": "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." @@ -8939,6 +9034,10 @@ }, "on_demand": { "type": "boolean" + }, + "stopped": { + "type": "boolean", + "description": "Stopped is a model taken out of service, kept, until it is started." } } }, @@ -10064,6 +10163,10 @@ "scaling": { "$ref": "#/components/schemas/catalog.Scaling" }, + "stopped": { + "$ref": "#/components/schemas/catalog.Scaling", + "description": "Stopped is the scaling a model had when an administrator stopped it: kept in the catalog with none running (Scaling 0/0) until started again with this. Nil for a model not stopped." + }, "class": { "type": "string", "description": "Class is how much this model's capacity is worth against another's, when there is not enough for both: critical, standard or opportunistic. Empty is standard, or opportunistic for a model loaded on demand. See EffectiveClass." diff --git a/docs/install.md b/docs/install.md index d107c4f9..af8b8249 100644 --- a/docs/install.md +++ b/docs/install.md @@ -1629,6 +1629,30 @@ Not yet: models declared in server.yaml cannot name a peer's model as a fallback (the catalog in the file is checked before the peers are read), and a peer behind a private CA needs that CA in the system's trust store. +### Stopping a model, and removing one with its files + +``` +llmgw ctl model stop qwen-7b # none of its replicas run; it is kept +llmgw ctl model start qwen-7b # runs again, scaled as it was +llmgw ctl model remove qwen-7b # leaves the catalog; its files stay +llmgw ctl model remove qwen-7b --purge # its files on the mirror and the nodes too +``` + +- **Stop** keeps the model in the catalog with its files and what it was + scaled to; a request for it is told an administrator stopped it. `ctl + models` shows it as stopped. A `ctl scale` while it is stopped starts it at + that number. +- **Remove** leaves its files on the mirror and the nodes, for adding it + again without fetching it. `ctl weights` lists such files on the nodes. +- **Remove --purge** deletes every version of it on the mirror (a second + mirror follows), its adapters and the weights its last update replaced, + and the nodes' copies once its replicas have stopped. What another model + uses — a canary's weights, a shared adapter — stays. It cannot be undone: + adding it again fetches it again. + +The console has the same in a model's menu: Stop… or Start, Remove…, and +Remove with its files…. + ### Changing a served model's context How much one request may use is set when a model is served. To change it diff --git a/docs/reference.md b/docs/reference.md index b75371ca..fa3470d1 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -192,6 +192,7 @@ rather than repeated. | `allow_pickle` | `bool` | [Model](#model) | | `profiles` | `[]Profile` | [Model](#model) | | `scaling` | `Scaling` | [Model](#model) | +| `stopped` | `*Scaling` | [Model](#model) | | `class` | `string` | [Model](#model) | | `owner` | `string` | [Model](#model) | | `external` | `*External` | [Model](#model) | @@ -1824,6 +1825,16 @@ no safetensors or GGUF form and whose files were checked. Takes a [Scaling](#scaling) block. +### `stopped` + +`*Scaling` + +Stopped is the scaling a model had when an administrator stopped it: +kept in the catalog with none running (Scaling 0/0) until started +again with this. Nil for a model not stopped. + +Takes a [Scaling](#scaling) block. + ### `class` `string` diff --git a/internal/cli/ctl.go b/internal/cli/ctl.go index 27cd76d0..70a3330b 100644 --- a/internal/cli/ctl.go +++ b/internal/cli/ctl.go @@ -321,6 +321,8 @@ func locationColumn(n adminapi.NodeView) string { // actually served, never what is merely catalogued. func modelStatus(m adminapi.ModelView) string { switch { + case m.Stopped: + return "stopped (llmgw ctl model start)" case m.Parked: return "parked" case m.External != "": diff --git a/internal/cli/model.go b/internal/cli/model.go index 602d10d5..c9ef8c95 100644 --- a/internal/cli/model.go +++ b/internal/cli/model.go @@ -34,7 +34,7 @@ func newModelCmd(client func() *ctlClient, asJSON *bool) *cobra.Command { Use: "model", Short: "Add or remove models while the cluster is running", } - c.AddCommand(newModelContextCmd(client), newModelGPUsCmd(client), newModelAddCmd(client), newModelPresetsCmd(client, asJSON), newModelRemoveCmd(client), newModelPlanCmd(client), + c.AddCommand(newModelStopCmd(client), newModelStartCmd(client), newModelContextCmd(client), newModelGPUsCmd(client), newModelAddCmd(client), newModelPresetsCmd(client, asJSON), newModelRemoveCmd(client), newModelPlanCmd(client), newResourcesCmd(client), newClassCmd(client), newOwnerCmd(client), newTargetCmd(client, asJSON), newAdaptersCmd(client), newModelServeCmd(client), newModelUpdatesCmd(client, asJSON), newModelUpdateCmd(client), @@ -163,21 +163,104 @@ func readModelFile(path string) (catalog.Model, error) { } func newModelRemoveCmd(client func() *ctlClient) *cobra.Command { - return &cobra.Command{ + var purge bool + c := &cobra.Command{ Use: "remove ", - Short: "Remove a model that was added at runtime", + Short: "Remove a model that was added at runtime, and with --purge its files", 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.", + "withdrawn, so traffic stops before the engine does.\n\n" + + "Its files stay on the mirror and the nodes, for adding it again. With\n" + + "--purge they go too: every version of it on the mirror (a mirror kept in\n" + + "step follows), and the nodes' copies once its replicas have stopped.\n" + + "What another model uses stays. To take a model out of service and keep\n" + + "it, use llmgw ctl model stop.", + Example: " llmgw ctl model remove qwen-7b\n" + + " llmgw ctl model remove qwen-7b --purge # its files on the mirror and the nodes too", 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 { + out := cmd.OutOrStdout() + if !purge { + var res map[string]string + if err := client().delete(cmd.Context(), "/api/v1/models/"+args[0], &res); err != nil { + return err + } + fmt.Fprintf(out, "%s removed. Its files stay on the mirror and the nodes; --purge deletes them.\n", args[0]) + return nil + } + var v adminapi.PurgeView + if err := client().patient(3*time.Minute).delete(cmd.Context(), "/api/v1/models/"+args[0]+"?purge=true", &v); err != nil { return err } - fmt.Fprintf(cmd.OutOrStdout(), "%s removed.\n", args[0]) + fmt.Fprintf(out, "%s removed.\n", v.Removed) + for _, m := range v.Mirror { + fmt.Fprintf(out, " mirror %s deleted\n", m) + } + for _, m := range v.Kept { + fmt.Fprintf(out, " mirror %s kept: another model uses it\n", m) + } + var freed int64 + for _, n := range v.Nodes { + if n.Error != "" { + fmt.Fprintf(out, " node %s: %s\n", n.Node, n.Error) + continue + } + freed += n.Bytes + fmt.Fprintf(out, " node %s: %d file(s), %s\n", n.Node, n.Files, humanBytes(n.Bytes)) + } + if len(v.Mirror) == 0 && len(v.Nodes) == 0 && v.Note == "" { + fmt.Fprintln(out, " it had no files of its own on the mirror or the nodes") + } + if freed > 0 { + fmt.Fprintf(out, " %s freed on the nodes\n", humanBytes(freed)) + } + if v.Note != "" { + fmt.Fprintln(out, " "+v.Note) + } + return nil + }, + } + c.Flags().BoolVar(&purge, "purge", false, "delete its files from the mirror and the nodes as well") + return c +} + +func newModelStopCmd(client func() *ctlClient) *cobra.Command { + return &cobra.Command{ + Use: "stop ", + Short: "Take a model out of service and keep it: none of its replicas run until it is started", + Long: "Its replicas stop; it stays in the catalog with its files, and a request\n" + + "for it is told an administrator stopped it. What it was scaled to is kept:\n" + + "llmgw ctl model start runs it again as it was.", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + var v adminapi.ScaleResponse + if err := client().post(cmd.Context(), "/api/v1/models/"+args[0]+"/stop", nil, &v); err != nil { + return err + } + fmt.Fprintf(cmd.OutOrStdout(), "%s stopped: its replicas stop on the next pass. It is kept; "+ + "llmgw ctl model start %s runs it again.\n", v.Model, v.Model) + return nil + }, + } +} + +func newModelStartCmd(client func() *ctlClient) *cobra.Command { + return &cobra.Command{ + Use: "start ", + Short: "Run a stopped model again, scaled as it was when it was stopped", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + var v adminapi.ScaleResponse + if err := client().post(cmd.Context(), "/api/v1/models/"+args[0]+"/start", nil, &v); err != nil { + return err + } + how := fmt.Sprintf("%d replica(s)", v.Min) + if v.OnDemand { + how = "on demand" + } + fmt.Fprintf(cmd.OutOrStdout(), "%s started: %s, max %d. Watch it with llmgw ctl models.\n", v.Model, how, v.Max) return nil }, } diff --git a/internal/cli/server.go b/internal/cli/server.go index e0abc232..8f9d2bd7 100644 --- a/internal/cli/server.go +++ b/internal/cli/server.go @@ -583,6 +583,10 @@ func newServerCmd() *cobra.Command { agentsvc.SetSourceRewrite(src.Rewrite) } rt.SetMirror(mc) + if adminAPI != nil { + // A model removed with its files is removed from it too. + adminAPI.SetPurgeMirror(mc) + } if fwPub != nil { fwPub.setMirror(mc) } diff --git a/internal/cli/tui_actions.go b/internal/cli/tui_actions.go index aa255e57..00367d9f 100644 --- a/internal/cli/tui_actions.go +++ b/internal/cli/tui_actions.go @@ -97,10 +97,21 @@ func (m *consoleModel) openForm(f *formDialog) { // call runs one admin API request in the background and reports it. func (m *consoleModel) call(title, method, path string, body any, say func(raw json.RawMessage) string) tea.Cmd { + return m.callWaiting(title, method, path, body, say, 60*time.Second) +} + +// callWaiting is call for a change the server may spend longer on than the +// console's usual wait. +func (m *consoleModel) callWaiting(title, method, path string, body any, say func(raw json.RawMessage) string, + wait time.Duration, +) tea.Cmd { m.busy = title c := m.client + if wait > 60*time.Second { + c = c.patient(wait) + } return func() tea.Msg { - ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + ctx, cancel := context.WithTimeout(context.Background(), wait) defer cancel() var raw json.RawMessage // A change may answer with no body at all. @@ -479,13 +490,40 @@ func (m consoleModel) modelMenu(md adminapi.ModelView) *menuDialog { return m.call("Retry "+md.Name, http.MethodPost, path+"/retry", map[string]string{}, nil) }}) } + if md.Stopped { + items = append(items, menuItem{"Start", func(m *consoleModel) tea.Cmd { + return m.call("Start "+md.Name, http.MethodPost, path+"/start", nil, func(json.RawMessage) string { + return md.Name + " runs again, scaled as it was when it was stopped." + }) + }}) + } else { + items = append(items, menuItem{"Stop…", func(m *consoleModel) tea.Cmd { + m.confirmD = &confirmDialog{title: "Stop " + md.Name, + text: "Its replicas stop; it stays in the catalog with its files, and Start runs it again as it was.", + yes: func(m *consoleModel) tea.Cmd { + return m.call("Stop "+md.Name, http.MethodPost, path+"/stop", nil, func(json.RawMessage) string { + return md.Name + " is stopped and kept." + }) + }} + return nil + }}) + } items = append(items, menuItem{"Remove…", func(m *consoleModel) tea.Cmd { m.confirmD = &confirmDialog{title: "Remove " + md.Name, - text: "Its replicas stop and it leaves the catalog. A model from server.yaml cannot be removed here.", + text: "Its replicas stop and it leaves the catalog; its files stay on the mirror and the nodes. " + + "A model from server.yaml cannot be removed here.", yes: func(m *consoleModel) tea.Cmd { return m.call("Remove "+md.Name, http.MethodDelete, path, nil, nil) }} return nil + }}, menuItem{"Remove with its files…", func(m *consoleModel) tea.Cmd { + m.confirmD = &confirmDialog{title: "Remove " + md.Name + " and its files", + text: "It leaves the catalog, every version of it is deleted from the mirror, and the nodes' copies once " + + "its replicas have stopped. What another model uses stays. This cannot be undone: adding it again fetches it again.", + yes: func(m *consoleModel) tea.Cmd { + return m.callWaiting("Remove "+md.Name, http.MethodDelete, path+"?purge=true", nil, purgeSays, 3*time.Minute) + }} + return nil }}) return &menuDialog{title: md.Name, items: items} } @@ -990,3 +1028,31 @@ func gpusSays(raw json.RawMessage) string { } return s } + +// purgeSays is a model removed with its files, in a few lines. +func purgeSays(raw json.RawMessage) string { + var v adminapi.PurgeView + if json.Unmarshal(raw, &v) != nil { + return "Done." + } + s := v.Removed + " removed." + if len(v.Mirror) > 0 { + s += fmt.Sprintf("\nMirror: %d version(s) deleted.", len(v.Mirror)) + } + if len(v.Kept) > 0 { + s += "\nKept on the mirror, another model uses them: " + strings.Join(v.Kept, ", ") + } + var files int + var freed int64 + for _, n := range v.Nodes { + files += n.Files + freed += n.Bytes + } + if files > 0 { + s += fmt.Sprintf("\nNodes: %d file(s), %s freed.", files, humanBytes(freed)) + } + if v.Note != "" { + s += "\n" + v.Note + } + return s +} diff --git a/internal/control/adminapi/api.go b/internal/control/adminapi/api.go index 38aacc1e..0b600d37 100644 --- a/internal/control/adminapi/api.go +++ b/internal/control/adminapi/api.go @@ -60,6 +60,9 @@ type API struct { controlPlanes func() ControlPlanes // serve runs registry-to-replica jobs; nil without a mirror and a store. serve *modelserve.Server + // purgeMirror is the mirror a model removed with its files is removed + // from; nil without one. + purgeMirror PurgeMirror // history is the Prometheus the console's charts read from. history History // audit reads the audit log. @@ -123,6 +126,8 @@ func (a *API) Register(mux Mux) { mux.HandleFunc("POST /api/v1/models", a.addModel) mux.HandleFunc("GET /api/v1/models/{name}", a.getModel) mux.HandleFunc("DELETE /api/v1/models/{name}", a.removeModel) + mux.HandleFunc("POST /api/v1/models/{name}/stop", a.stopModel) + mux.HandleFunc("POST /api/v1/models/{name}/start", a.startModel) mux.HandleFunc("POST /api/v1/models/{name}/scale", a.scaleModel) mux.HandleFunc("POST /api/v1/models/{name}/resources", a.setResources) a.registerQuotas(mux) @@ -451,6 +456,8 @@ type ModelView struct { // answer; 0 when the catalog does not say. ContextLength int `json:"context_length,omitempty"` MaxOutput int `json:"max_output,omitempty"` + // Stopped is a model an administrator took out of service, kept. + Stopped bool `json:"stopped,omitempty"` // 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"` @@ -843,7 +850,7 @@ func (a *API) Models() []ModelView { h := health[m.Name] v := ModelView{ Name: m.Name, Description: m.Description, Aliases: m.Aliases, Source: m.Source, - ContextLength: m.ContextLength, MaxOutput: m.MaxOutput, + ContextLength: m.ContextLength, MaxOutput: m.MaxOutput, Stopped: m.Stopped != nil, Min: m.Scaling.Min, Max: m.Scaling.Max, Ready: ready[m.Name], Total: total[m.Name], Fallbacks: m.Fallbacks, @@ -945,6 +952,8 @@ 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) + Stop(ctx context.Context, name, by string) (catalog.Model, error) + Start(ctx context.Context, name, by string) (catalog.Model, error) Resources(ctx context.Context, name, profile string, memoryGB, cpuCores int, by string) (catalog.Model, error) SetClass(ctx context.Context, name, class, by string) (catalog.Model, error) SetOwner(ctx context.Context, name, owner, by string) (catalog.Model, error) @@ -1194,6 +1203,8 @@ type ScaleResponse struct { Max int `json:"max"` IdleTimeout time.Duration `json:"idle_timeout,omitempty"` OnDemand bool `json:"on_demand"` + // Stopped is a model taken out of service, kept, until it is started. + Stopped bool `json:"stopped,omitempty"` } func (a *API) scaleModel(w http.ResponseWriter, r *http.Request) { @@ -1227,6 +1238,32 @@ func (a *API) scaleModel(w http.ResponseWriter, r *http.Request) { }) } +// stopModel takes a model out of service, kept. +func (a *API) stopModel(w http.ResponseWriter, r *http.Request) { + a.stopOrStart(w, r, a.models.Stop) +} + +// startModel runs a stopped model again. +func (a *API) startModel(w http.ResponseWriter, r *http.Request) { + a.stopOrStart(w, r, a.models.Start) +} + +func (a *API) stopOrStart(w http.ResponseWriter, r *http.Request, f func(context.Context, string, string) (catalog.Model, error)) { + if a.models == nil { + writeError(w, http.StatusNotImplemented, "this control plane has no store") + return + } + m, err := f(r.Context(), r.PathValue("name"), actorOf(r)) + if err != nil { + 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(), Stopped: m.Stopped != nil, + }) +} + // setResources sets a model's reserved memory and cores. Placed replicas keep // what they were started with until they are restarted. func (a *API) setResources(w http.ResponseWriter, r *http.Request) { @@ -1253,6 +1290,10 @@ func (a *API) removeModel(w http.ResponseWriter, r *http.Request) { return } name := r.PathValue("name") + if r.URL.Query().Get("purge") == "true" { + a.purgeModel(w, r, name) + return + } if err := a.models.Remove(r.Context(), name); err != nil { writeError(w, http.StatusBadRequest, err.Error()) return diff --git a/internal/control/adminapi/api_test.go b/internal/control/adminapi/api_test.go index 69317858..1ce542ee 100644 --- a/internal/control/adminapi/api_test.go +++ b/internal/control/adminapi/api_test.go @@ -749,3 +749,11 @@ func TestANodesCardInTheWrongSlotIsAWarning(t *testing.T) { t.Errorf("warnings %q", warned) } } + +func (f *fakeModels) Stop(_ context.Context, name, _ string) (catalog.Model, error) { + return catalog.Model{Name: name, Stopped: &catalog.Scaling{Min: 1, Max: 1}}, f.err +} + +func (f *fakeModels) Start(_ context.Context, name, _ string) (catalog.Model, error) { + return catalog.Model{Name: name, Scaling: catalog.Scaling{Min: 1, Max: 1}}, f.err +} diff --git a/internal/control/adminapi/purge.go b/internal/control/adminapi/purge.go new file mode 100644 index 00000000..c4021d83 --- /dev/null +++ b/internal/control/adminapi/purge.go @@ -0,0 +1,230 @@ +package adminapi + +import ( + "context" + "log/slog" + "net/http" + "net/url" + "path/filepath" + "slices" + "strings" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/agent/weights" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" + "gitea.mynodes.xyz/ilker/llmgw/internal/mirror" +) + +// A model removed with its files: the catalog entry, every version of it the +// mirror holds (and so, carried by its removal, a peer mirror's), and the +// copies the nodes fetched, once its replicas have stopped. What another +// model still uses stays. + +// PurgeMirror is what a purge asks of the mirror. +type PurgeMirror interface { + List(ctx context.Context) (mirror.ModelsResponse, error) + Remove(ctx context.Context, model, version string) ([]string, error) +} + +// SetPurgeMirror gives removing a model with its files the mirror. +func (a *API) SetPurgeMirror(m PurgeMirror) { a.purgeMirror = m } + +// purgeWait is how long a purge waits for the model's replicas to stop +// before deleting the nodes' copies; after it, they are left for llmgw ctl +// weights --collect. +var purgeWait = 60 * time.Second + +// PurgeView is what a model removed with its files took with it. +type PurgeView struct { + Removed string `json:"removed"` + // Mirror are the versions removed from the mirror, model/version. + Mirror []string `json:"mirror,omitempty"` + // Kept are versions another model still uses, left on the mirror. + Kept []string `json:"kept,omitempty"` + Nodes []CollectResult `json:"nodes,omitempty"` + Note string `json:"note,omitempty"` +} + +// purgeModel removes a model from the catalog and its files from the mirror +// and the nodes. +func (a *API) purgeModel(w http.ResponseWriter, r *http.Request, name string) { + m, ok := a.ctl.Catalog().Resolve(name) + if !ok { + writeError(w, http.StatusNotFound, "no model "+name+" in the catalog") + return + } + name = m.Name + others := []catalog.Model{} + for _, o := range a.ctl.Catalog().Models() { + if o.Name != name { + others = append(others, o) + } + } + if err := a.models.Remove(r.Context(), name); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + a.presetDeclined(r.Context(), name, actorOf(r)) + out := PurgeView{Removed: name} + + sources := purgeSources(m) + if a.purgeMirror == nil { + out.Note = "this control plane has no mirror; its files there, if any, are left" + } else { + out.Mirror, out.Kept, out.Note = purgeFromMirror(r.Context(), a.purgeMirror, sources, others) + } + + // The nodes' copies, once nothing runs on them. + deadline := time.Now().Add(purgeWait) + for a.runs(name) && time.Now().Before(deadline) { + select { + case <-r.Context().Done(): + return + case <-time.After(time.Second): + } + } + if a.runs(name) { + out.Note = strings.TrimSpace(out.Note + " Its replicas had not stopped yet: its copies on the nodes are " + + "left, and llmgw ctl weights --collect deletes them.") + writeJSON(w, http.StatusOK, out) + return + } + out.Nodes = a.purgeFromNodes(r.Context(), sources, others) + writeJSON(w, http.StatusOK, out) +} + +// runs says whether any replica of a model is still on a node. +func (a *API) runs(model string) bool { + for _, rep := range a.ctl.Reconciler().Replicas() { + if rep.Model == model { + return true + } + } + return false +} + +// purgeSources are the files a model fetched: its weights, its adapters, +// and the weights its last update replaced. +func purgeSources(m catalog.Model) []string { + out := m.WeightSources() + for _, ad := range m.Adapters { + out = append(out, ad.Source) + } + if m.Origin != nil && m.Origin.Previous != nil && m.Origin.Previous.Source != "" { + out = append(out, m.Origin.Previous.Source) + } + return out +} + +// inUse is every file the other models fetch. +func inUse(others []catalog.Model) []string { + var out []string + for _, o := range others { + out = append(out, purgeSources(o)...) + } + return out +} + +// purgeFromMirror removes from the mirror every version of the model +// directories the sources are in, but those another model uses. +func purgeFromMirror(ctx context.Context, mc PurgeMirror, sources []string, others []catalog.Model) (removed, kept []string, note string) { + held, err := mc.List(ctx) + if err != nil { + return nil, nil, "the mirror could not be read (" + err.Error() + "): its files there are left" + } + dirs := map[string]bool{} + for _, s := range sources { + if model, _, ok := mirrorPlace(s); ok { + dirs[model] = true + } + } + used := map[string]bool{} + for _, s := range inUse(others) { + if model, version, ok := mirrorPlace(s); ok { + used[model+"/"+version] = true + } + } + seen := map[string]bool{} + for _, st := range held.Models { + key := st.Model + "/" + st.Version + if !dirs[st.Model] || seen[key] { + continue + } + seen[key] = true + if used[key] { + kept = append(kept, key) + continue + } + if _, err := mc.Remove(ctx, st.Model, st.Version); err != nil { + note = strings.TrimSpace(note + " " + key + " could not be removed from the mirror: " + err.Error() + ".") + continue + } + removed = append(removed, key) + } + slices.Sort(removed) + slices.Sort(kept) + return removed, kept, note +} + +// mirrorPlace is the model directory and version a mirror URL names: +// BASE/MODEL/VERSION/FILE. +func mirrorPlace(source string) (model, version string, ok bool) { + u, err := url.Parse(source) + if err != nil || (u.Scheme != "http" && u.Scheme != "https") { + return "", "", false + } + // A mirror serves from its root: a registry's address is longer. + parts := strings.Split(strings.Trim(u.Path, "/"), "/") + if len(parts) != 3 { + return "", "", false + } + model, version = parts[0], parts[1] + if mirror.ValidName(model) != nil || mirror.ValidName(version) != nil { + return "", "", false + } + return model, version, true +} + +// purgeFromNodes deletes the nodes' copies of the sources: the files llmgw +// fetched for them under any mirror's address, that no other model wants. +func (a *API) purgeFromNodes(ctx context.Context, sources []string, others []catalog.Model) []CollectResult { + rm := a.ctl.WeightRemover() + if rm == nil { + return nil + } + names := map[string]bool{} + for _, s := range sources { + names[filepath.Base(s)] = true + } + wanted := inUse(others) + var out []CollectResult + for _, n := range a.ctl.Inventory().Nodes() { + size := map[string]int64{} + have := make([]string, 0, len(n.Weights)) + for _, w := range n.Weights { + have = append(have, w.Path) + size[w.Path] = w.SizeBytes + } + var paths []string + for _, p := range weights.Orphans(have, wanted) { + // The cache name is a hash of the address, then the file's own + // name: the address may be either mirror's, the name is the same. + if _, file, ok := strings.Cut(filepath.Base(p), "-"); ok && names[file] { + paths = append(paths, p) + } + } + if len(paths) == 0 { + continue + } + res := CollectResult{Node: n.ID, Files: len(paths)} + for _, p := range paths { + res.Bytes += size[p] + } + if err := rm.RemoveWeights(ctx, n.ID, paths); err != nil { + res.Error, res.Files, res.Bytes = err.Error(), 0, 0 + slog.Warn("a purged model's copies could not be deleted on a node", "node", n.ID, "err", err) + } + out = append(out, res) + } + return out +} diff --git a/internal/control/adminapi/purge_test.go b/internal/control/adminapi/purge_test.go new file mode 100644 index 00000000..98b9edff --- /dev/null +++ b/internal/control/adminapi/purge_test.go @@ -0,0 +1,68 @@ +package adminapi + +import ( + "context" + "slices" + "testing" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" + "gitea.mynodes.xyz/ilker/llmgw/internal/mirror" +) + +type purgeMirrorFake struct { + held []mirror.Status + removed []string +} + +func (f *purgeMirrorFake) List(context.Context) (mirror.ModelsResponse, error) { + return mirror.ModelsResponse{Base: "http://m:8081", Models: f.held}, nil +} + +func (f *purgeMirrorFake) Remove(_ context.Context, model, version string) ([]string, error) { + f.removed = append(f.removed, model+"/"+version) + return nil, nil +} + +func TestAPurgeRemovesEveryVersionOfTheModelButWhatAnotherUses(t *testing.T) { + f := &purgeMirrorFake{held: []mirror.Status{ + {Model: "qwen", Version: "v2", File: "a.gguf"}, + {Model: "qwen", Version: "v2", File: "b.gguf"}, + {Model: "qwen", Version: "v1", File: "a.gguf"}, + {Model: "qwen", Version: "v3", File: "a.gguf"}, + {Model: "other", Version: "v1", File: "o.gguf"}, + {Model: "lora", Version: "t1", File: "l.gguf"}, + }} + m := catalog.Model{Name: "qwen", Source: "http://m:8081/qwen/v2/a.gguf", + Adapters: []catalog.Adapter{{Name: "tone", Source: "http://m:8081/lora/t1/l.gguf"}}} + // The canary runs v3: it stays. + others := []catalog.Model{{Name: "qwen-canary", Source: "http://other-mirror:8081/qwen/v3/a.gguf"}, + {Name: "other", Source: "http://m:8081/other/v1/o.gguf"}} + removed, kept, note := purgeFromMirror(t.Context(), f, purgeSources(m), others) + if note != "" { + t.Fatal(note) + } + if !slices.Equal(removed, []string{"lora/t1", "qwen/v1", "qwen/v2"}) || !slices.Equal(kept, []string{"qwen/v3"}) { + t.Errorf("removed %v, kept %v", removed, kept) + } + if slices.Contains(f.removed, "other/v1") || len(f.removed) != 3 { + t.Errorf("asked to remove %v", f.removed) + } +} + +func TestOnlyAMirrorsAddressNamesAPlaceOnIt(t *testing.T) { + for src, want := range map[string]string{ + "http://10.0.0.146:8081/qwen/abc/q4.gguf": "qwen/abc", + "https://huggingface.co/o/r/resolve/main/q4.gguf": "", + "/srv/models/q4.gguf": "", + "Qwen/Qwen2.5-7B": "", + } { + model, version, ok := mirrorPlace(src) + got := "" + if ok { + got = model + "/" + version + } + if got != want { + t.Errorf("%s: %q, want %q", src, got, want) + } + } +} diff --git a/internal/control/adminapi/routes.go b/internal/control/adminapi/routes.go index 95688baf..d4382499 100644 --- a/internal/control/adminapi/routes.go +++ b/internal/control/adminapi/routes.go @@ -155,8 +155,16 @@ var Routes = []Route{ Description: "With the defaults the catalog filled in. By its name only, not an alias.", Response: catalog.Model{}}, {Method: "DELETE", Path: "/api/v1/models/{name}", Summary: "Remove a model added at runtime", - Description: "A model declared in the cluster's settings is refused: it is changed there.", - Response: Removed{}}, + Description: "A model declared in the cluster's settings is refused: it is changed there. With purge, its " + + "files go too: every version of it the mirror holds (a peer mirror follows), and the nodes' copies once its " + + "replicas have stopped, waiting up to a minute for them; what another model uses stays. The answer is then a PurgeView.", + Query: []Param{{Name: "purge", Type: "boolean", Description: "Delete its files from the mirror and the nodes as well."}}, + Response: Removed{}}, + {Method: "POST", Path: "/api/v1/models/{name}/stop", Summary: "Take a model out of service and keep it", + Description: "None of its replicas run, and requests are told an administrator stopped it; what it was scaled to is kept for start.", + Response: ScaleResponse{}}, + {Method: "POST", Path: "/api/v1/models/{name}/start", Summary: "Run a stopped model again, scaled as it was", + Response: ScaleResponse{}}, {Method: "POST", Path: "/api/v1/models/{name}/scale", Summary: "Set how many replicas a model runs", Request: ScaleRequest{}, Response: ScaleResponse{}}, {Method: "POST", Path: "/api/v1/models/{name}/resources", Summary: "Set what a model's profiles reserve of a node", diff --git a/internal/control/catalog/catalog.go b/internal/control/catalog/catalog.go index 28e899ca..23177b09 100644 --- a/internal/control/catalog/catalog.go +++ b/internal/control/catalog/catalog.go @@ -332,6 +332,10 @@ type Model struct { AllowPickle bool `json:"allow_pickle,omitempty" koanf:"allow_pickle"` Profiles []Profile `json:"profiles,omitempty" koanf:"profiles"` Scaling Scaling `json:"scaling,omitempty" koanf:"scaling"` + // Stopped is the scaling a model had when an administrator stopped it: + // kept in the catalog with none running (Scaling 0/0) until started + // again with this. Nil for a model not stopped. + Stopped *Scaling `json:"stopped,omitempty" koanf:"stopped"` // Class is how much this model's capacity is worth against another's, // when there is not enough for both: critical, standard or opportunistic. // Empty is standard, or opportunistic for a model loaded on demand. See diff --git a/internal/control/modelsvc/modelsvc.go b/internal/control/modelsvc/modelsvc.go index 1526cd9f..9496bbf9 100644 --- a/internal/control/modelsvc/modelsvc.go +++ b/internal/control/modelsvc/modelsvc.go @@ -228,6 +228,8 @@ func (s *Service) Remove(ctx context.Context, name string) error { func (s *Service) Scale(ctx context.Context, name string, sc catalog.Scaling, by string) (catalog.Model, error) { final, err := s.update(ctx, name, by, "change its scaling", func(m *catalog.Model) error { m.Scaling = sc + // A number set is what it runs now: a stopped model is started by it. + m.Stopped = nil return nil }) if err == nil { @@ -237,6 +239,39 @@ func (s *Service) Scale(ctx context.Context, name string, sc catalog.Scaling, by return final, err } +// Stop takes a model out of service and keeps it: none of its replicas run, +// and what it was scaled to is kept for Start. +func (s *Service) Stop(ctx context.Context, name, by string) (catalog.Model, error) { + final, err := s.update(ctx, name, by, "stop it", func(m *catalog.Model) error { + if m.Stopped != nil { + return fmt.Errorf("model %q is already stopped; llmgw ctl model start %s runs it again", name, name) + } + was := m.Scaling + m.Stopped = &was + m.Scaling = catalog.Scaling{IdleTimeout: was.IdleTimeout} + return nil + }) + if err == nil { + s.log.Info("model stopped", "model", name, "by", by) + } + return final, err +} + +// Start runs a stopped model again, scaled as it was when it was stopped. +func (s *Service) Start(ctx context.Context, name, by string) (catalog.Model, error) { + final, err := s.update(ctx, name, by, "start it", func(m *catalog.Model) error { + if m.Stopped == nil { + return fmt.Errorf("model %q is not stopped; llmgw ctl scale changes how many replicas it runs", name) + } + m.Scaling, m.Stopped = *m.Stopped, nil + return nil + }) + if err == nil { + s.log.Info("model started", "model", name, "min", final.Scaling.Min, "max", final.Scaling.Max, "by", by) + } + return final, err +} + // Resources sets what a model's profile reserves of a node: host memory, and // the CPU cores a replica without a GPU computes on. An empty profile name // sets every profile. diff --git a/internal/control/modelsvc/modelsvc_test.go b/internal/control/modelsvc/modelsvc_test.go index 810806e8..77b5a9c7 100644 --- a/internal/control/modelsvc/modelsvc_test.go +++ b/internal/control/modelsvc/modelsvc_test.go @@ -400,3 +400,41 @@ func TestAddNewLeavesAModelAddedMeanwhile(t *testing.T) { t.Errorf("a new model: %v", err) } } + +func TestAStoppedModelRunsNoneAndStartsAsItWas(t *testing.T) { + s, _, rec := 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: 2, Max: 4, IdleTimeout: 20 * time.Minute}, "ilker"); err != nil { + t.Fatal(err) + } + got, err := s.Stop(t.Context(), "added", "ilker") + if err != nil { + t.Fatal(err) + } + if got.Scaling.Min != 0 || got.Scaling.Max != 0 || got.Stopped == nil || got.Stopped.Min != 2 { + t.Fatalf("stopped: %+v %+v", got.Scaling, got.Stopped) + } + if m, _ := rec.last.Resolve("added"); m.Scaling.Max != 0 || m.Scaling.OnDemand() { + t.Errorf("the catalog in force runs it: %+v", m.Scaling) + } + if _, err := s.Stop(t.Context(), "added", "ilker"); err == nil { + t.Error("stopped twice") + } + got, err = s.Start(t.Context(), "added", "ilker") + if err != nil { + t.Fatal(err) + } + if got.Scaling.Min != 2 || got.Scaling.Max != 4 || got.Scaling.IdleTimeout != 20*time.Minute || got.Stopped != nil { + t.Errorf("started: %+v %+v", got.Scaling, got.Stopped) + } + if _, err := s.Start(t.Context(), "added", "ilker"); err == nil { + t.Error("started what was not stopped") + } + // A scale while stopped starts it at that number. + _, _ = s.Stop(t.Context(), "added", "ilker") + if got, _ := s.Scale(t.Context(), "added", catalog.Scaling{Min: 1}, "ilker"); got.Stopped != nil { + t.Error("still stopped after a scale") + } +} diff --git a/internal/docgen/reference_gen.go b/internal/docgen/reference_gen.go index b36d0b7c..f6fa2033 100644 --- a/internal/docgen/reference_gen.go +++ b/internal/docgen/reference_gen.go @@ -244,6 +244,7 @@ var Reference = []Block{ {Key: "allow_pickle", Type: "bool", Doc: "AllowPickle lets the nodes fetch this model's weights, or its\nadapters', in a pickle format (.bin, .pt, .pth, .ckpt), which run code\nas they are loaded. Off, such a source is refused when the model is\nadded and by the node before it fetches anything. For a model that has\nno safetensors or GGUF form and whose files were checked.", Nested: ""}, {Key: "profiles", Type: "[]Profile", Doc: "", Nested: ""}, {Key: "scaling", Type: "Scaling", Doc: "", Nested: "catalog.Scaling"}, + {Key: "stopped", Type: "*Scaling", Doc: "Stopped is the scaling a model had when an administrator stopped it:\nkept in the catalog with none running (Scaling 0/0) until started\nagain with this. Nil for a model not stopped.", Nested: "catalog.Scaling"}, {Key: "class", Type: "string", Doc: "Class is how much this model's capacity is worth against another's,\nwhen there is not enough for both: critical, standard or opportunistic.\nEmpty is standard, or opportunistic for a model loaded on demand. See\nEffectiveClass.", Nested: ""}, {Key: "owner", Type: "string", Doc: "Owner is the part of the tenancy tree whose capacity quota this model\ncounts against, written org:ID, team:ID or project:ID. Empty is shared:\nheld by the cluster, counted against nobody.", Nested: ""}, {Key: "external", Type: "*External", Doc: "External is a model served by someone else — an OpenAI-compatible API\noutside the cluster — rather than by replicas placed here. Such a model\nhas no profiles and nothing is placed for it; the router sends its\nrequests to the provider. Most useful named in another model's\nfallbacks: used only when the cluster has no capacity left. Its\nprompts leave the cluster, so which keys may name it is worth an\nallowlist.", Nested: "catalog.External"}, diff --git a/internal/router/chat_load.go b/internal/router/chat_load.go index afe0d4a9..f8c191e4 100644 --- a/internal/router/chat_load.go +++ b/internal/router/chat_load.go @@ -68,9 +68,13 @@ func (r *Router) loadStatus(model string) modelStatus { if best != nil { return *best } - if _, m := r.resolve(model); m.Scaling.OnDemand() { + _, m := r.resolve(model) + if m.Scaling.OnDemand() { return modelStatus{State: "waiting", Message: "waiting for a node with room for it"} } + if m.Stopped != nil { + return modelStatus{State: "stopped", Message: "an administrator stopped this model; it is kept, and runs again when started"} + } return modelStatus{State: "stopped", Message: "this model is not running"} } diff --git a/internal/web/admin_serve_test.go b/internal/web/admin_serve_test.go index 80acfff1..e3d168bf 100644 --- a/internal/web/admin_serve_test.go +++ b/internal/web/admin_serve_test.go @@ -546,3 +546,11 @@ func TestTheCPUWithGPUsFreeIsAChoiceMadeOnPurpose(t *testing.T) { t.Error("with no GPU free the CPU is the only choice and should not be questioned") } } + +func (f *fakeCatalog) Stop(_ context.Context, name, _ string) (catalog.Model, error) { + return catalog.Model{Name: name, Stopped: &catalog.Scaling{Min: 1, Max: 1}}, f.err +} + +func (f *fakeCatalog) Start(_ context.Context, name, _ string) (catalog.Model, error) { + return catalog.Model{Name: name, Scaling: catalog.Scaling{Min: 1, Max: 1}}, f.err +}