// Package adminapi is the control plane's management surface, which `ctl` // consumes (plan §11, phase 3). // // It is mounted on the admin listener rather than the client-facing one. The // two have different audiences: one serves inference to whoever holds an API // key, the other lets a caller cordon nodes and read the cluster's shape. package adminapi import ( "context" "encoding/json" "fmt" "io" "net/http" "slices" "time" "gitea.mynodes.xyz/ilker/llmgw/internal/bench" "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" "gitea.mynodes.xyz/ilker/llmgw/internal/store" ) // API serves the management endpoints. type API struct { models Models capacity Capacity usage Usage ctl *control.Controller } // New builds the API over a controller. 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("POST /api/v1/models/{name}/scale", a.scaleModel) mux.HandleFunc("PUT /api/v1/models/{name}/capacity", a.putCapacity) mux.HandleFunc("GET /api/v1/capacity", a.listCapacity) mux.HandleFunc("GET /api/v1/usage", a.usageByScope) mux.HandleFunc("GET /api/v1/nodes", a.listNodes) mux.HandleFunc("GET /api/v1/models", a.listModels) mux.HandleFunc("GET /api/v1/replicas", a.listReplicas) // Listing is a read and deleting is a separate verb, so a mistyped // command cannot remove anything. mux.HandleFunc("GET /api/v1/weights/orphans", a.listOrphans) mux.HandleFunc("POST /api/v1/weights/collect", a.collectWeights) mux.HandleFunc("GET /api/v1/explain-placement", a.explainPlacement) mux.HandleFunc("POST /api/v1/nodes/{id}/state", a.setNodeState) } // --- responses --- // NodeView is a node as `ctl nodes` shows it. type NodeView struct { ID string `json:"id"` Hostname string `json:"hostname,omitempty"` MgmtAddr string `json:"mgmt_addr,omitempty"` State string `json:"state"` Labels map[string]string `json:"labels,omitempty"` GPUTotal int `json:"gpu_total"` GPUFree int `json:"gpu_free"` // MemoryGB is the machine's RAM, 0 when the agent did not report it. It is // what decides whether a CPU profile can be placed here, so it belongs // where someone looks when a model will not schedule. MemoryGB int `json:"memory_gb,omitempty"` // Models are the weights this node can see. Absent means it was not asked // to look, which is not the same as holding none. Models []string `json:"models_on_disk,omitempty"` // DiskFreeGB and DiskTotalGB are the filesystems holding those weights, // summed. Zero means the agent did not report any, which an older agent // and a node with no model directories both do — so it reads as unknown // rather than as a full disk. // // Summed because an operator asking "can this node take another model" // wants one number; the per-filesystem detail is in the node's own record // for anyone who needs it. Shared filesystems are left out of the sum: the // space on a filer is not this node's to spend, and adding it would let a // fleet of twenty nodes each believe it has the same free terabyte. DiskFreeGB int `json:"disk_free_gb,omitempty"` DiskTotalGB int `json:"disk_total_gb,omitempty"` Classes []string `json:"gpu_classes,omitempty"` Fabrics []string `json:"fabrics,omitempty"` // Unhealthy lists GPUs kept out of placement and why, which is usually the // answer when a node looks idle but nothing will schedule onto it. Unhealthy []string `json:"unhealthy_gpus,omitempty"` LastSeenAgo string `json:"last_seen_ago"` AgentVersion string `json:"agent_version,omitempty"` Replicas int `json:"replicas"` } // ModelView is a model as `ctl models` shows it. type ModelView struct { Name string `json:"name"` Aliases []string `json:"aliases,omitempty"` Source string `json:"source"` Min int `json:"min_replicas"` Max int `json:"max_replicas"` Ready int `json:"ready_replicas"` Total int `json:"total_replicas"` Fallbacks []string `json:"fallbacks,omitempty"` Profiles []string `json:"profiles"` } // ReplicaView is a replica as `ctl replicas` shows it. type ReplicaView struct { ID string `json:"id"` Model string `json:"model"` Profile string `json:"profile"` State string `json:"state"` Pool string `json:"pool,omitempty"` Fabric string `json:"fabric,omitempty"` Endpoint string `json:"endpoint,omitempty"` AgeSec int64 `json:"age_seconds"` // IdleSec is how long since a request was routed here. Negative means the // router has not reported any use of this replica, which is not the same // as idle — it is what a replica looks like before anything has asked for // it, and after a router restart. IdleSec int64 `json:"idle_seconds"` Err string `json:"error,omitempty"` Shards []ShardView `json:"shards"` } // ShardView is one shard of a replica. type ShardView struct { ID string `json:"id"` Node string `json:"node"` Rank int `json:"rank"` GPUs []int `json:"gpus"` State string `json:"state"` Err string `json:"error,omitempty"` // Fetch is the download this shard is waiting on, absent when it is not // waiting on one. It is what tells a replica pulling a hundred gigabytes // apart from one that is wedged — which look identical without it. Fetch *FetchView `json:"fetch,omitempty"` } // FetchView is a download in flight. type FetchView struct { Source string `json:"source"` BytesDone int64 `json:"bytes_done"` // BytesTotal is 0 when the server did not say how large the file is. BytesTotal int64 `json:"bytes_total,omitempty"` // Percent is absent rather than zero when the total is unknown: an // invented percentage that stops moving is worse than a byte count that // keeps rising. Percent *int `json:"percent,omitempty"` } // ExplainView is why a model could not be placed. type ExplainView struct { Model string `json:"model"` // Placed is true when the model has replicas, in which case there is // nothing to explain. Placed bool `json:"placed"` // Rejections is empty when placement was never attempted, which is itself // the answer: the model is satisfied, or the reconciler has not run yet. Rejections []RejectionView `json:"rejections,omitempty"` } // RejectionView is one rejected candidate. type RejectionView struct { Profile string `json:"profile"` Pool string `json:"pool,omitempty"` Node string `json:"node,omitempty"` Reason string `json:"reason"` } // --- handlers --- func (a *API) listNodes(w http.ResponseWriter, _ *http.Request) { writeJSON(w, http.StatusOK, a.Nodes()) } // Nodes builds the node view. // // Exported because the dashboard renders the same data server-side. Going back // out through HTTP to fetch what is already in this process would add a way for // the two to disagree. func (a *API) Nodes() []NodeView { replicasOn := map[string]int{} for _, rep := range a.ctl.Reconciler().Replicas() { for _, n := range rep.NodeIDs() { replicasOn[n]++ } } inv := a.ctl.Inventory() out := make([]NodeView, 0) for _, n := range inv.Nodes() { v := NodeView{ ID: n.ID, Hostname: n.Hostname, MgmtAddr: n.MgmtAddr, State: n.State.String(), Labels: n.Labels, GPUTotal: len(n.GPUs), GPUFree: len(inv.FreeGPUs(n.ID)), MemoryGB: n.MemoryTotalMB / 1024, Fabrics: n.FabricNames(), AgentVersion: n.AgentVersion, Replicas: replicasOn[n.ID], } if free, total := n.LocalDisk(); total > 0 { v.DiskFreeGB = int(free / (1 << 30)) v.DiskTotalGB = int(total / (1 << 30)) } for _, w := range n.Weights { v.Models = append(v.Models, w.Path) } seen := map[string]struct{}{} for _, g := range n.GPUs { if _, dup := seen[g.Class()]; !dup { seen[g.Class()] = struct{}{} v.Classes = append(v.Classes, g.Class()) } if !g.Health.OK() { v.Unhealthy = append(v.Unhealthy, fmt.Sprintf("gpu %d: %s", g.Index, g.Health.Problems[0])) } } if !n.LastSeen.IsZero() { v.LastSeenAgo = time.Since(n.LastSeen).Round(time.Second).String() } out = append(out, v) } return out } func (a *API) listModels(w http.ResponseWriter, _ *http.Request) { writeJSON(w, http.StatusOK, a.Models()) } // Models builds the catalog view. func (a *API) Models() []ModelView { ready, total := map[string]int{}, map[string]int{} for _, rep := range a.ctl.Reconciler().Replicas() { total[rep.Model]++ if rep.State == reconciler.Ready { ready[rep.Model]++ } } out := make([]ModelView, 0) for _, m := range a.ctl.Catalog().Models() { v := ModelView{ Name: m.Name, Aliases: m.Aliases, Source: m.Source, Min: m.Scaling.Min, Max: m.Scaling.Max, Ready: ready[m.Name], Total: total[m.Name], Fallbacks: m.Fallbacks, } for _, p := range m.Profiles { v.Profiles = append(v.Profiles, p.Name) } out = append(out, v) } return out } func (a *API) listReplicas(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, a.Replicas(r.URL.Query().Get("model"))) } // Replicas builds the replica view, optionally filtered to one model. func (a *API) Replicas(model string) []ReplicaView { out := make([]ReplicaView, 0) for _, rep := range a.ctl.Reconciler().Replicas() { if model != "" && rep.Model != model { continue } v := ReplicaView{ ID: rep.ID, Model: rep.Model, Profile: rep.Profile, State: rep.State.String(), Pool: rep.Pool, Fabric: rep.Fabric, Endpoint: rep.Endpoint, Err: rep.Err, AgeSec: int64(time.Since(rep.CreatedAt).Seconds()), IdleSec: -1, } if rep.LastUsedAt.IsZero() { v.IdleSec = -1 } else { v.IdleSec = int64(time.Since(rep.LastUsedAt).Seconds()) } for _, s := range rep.Shards { sv := ShardView{ ID: s.ID, Node: s.NodeID, Rank: s.Rank, GPUs: s.GPUIndices, State: s.State.String(), Err: s.Err, } if f := s.Fetch; f != nil { sv.Fetch = &FetchView{ Source: f.Source, BytesDone: f.BytesDone, BytesTotal: f.BytesTotal, } if p, ok := f.Percent(); ok { sv.Fetch.Percent = &p } } v.Shards = append(v.Shards, sv) } out = append(out, v) } 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 Scale(ctx context.Context, name string, s catalog.Scaling, by string) (catalog.Model, error) } // Usage reports what each group of the tenancy tree spent. The store // implements it. type Usage interface { UsageByScope(ctx context.Context, scope store.Scope, since time.Time) ([]store.ScopeUsage, error) } // SetUsage supplies the usage reporting source. func (a *API) SetUsage(u Usage) { a.usage = u } // ScopeUsageView is one group's spend, with its quota when it has one. type ScopeUsageView struct { Scope string `json:"scope"` ID string `json:"id"` Name string `json:"name"` Keys int64 `json:"keys"` Requests int64 `json:"requests"` Prompt int64 `json:"prompt_tokens"` Completion int64 `json:"completion_tokens"` CostMicros int64 `json:"cost_micros"` Errors int64 `json:"errors"` LastAt time.Time `json:"last_at,omitzero"` // Budget is what this level sets, absent when it sets none. A group bounded // only by its parent has no budget of its own, and reporting the parent's // against each child would read as though each had that much. Budget *int64 `json:"budget_micros,omitempty"` BudgetPeriod string `json:"budget_period,omitempty"` // Fraction is spend over budget, absent for the same reason. Fraction *float64 `json:"fraction,omitempty"` } func (a *API) usageByScope(w http.ResponseWriter, r *http.Request) { if a.usage == nil { writeError(w, http.StatusNotImplemented, "this control plane has no store, so nothing is recorded") return } scope := store.Scope(r.URL.Query().Get("by")) if scope == "" { scope = store.ScopeTeam } since := 24 * time.Hour if v := r.URL.Query().Get("since"); v != "" { d, err := time.ParseDuration(v) if err != nil || d <= 0 { writeError(w, http.StatusBadRequest, "since must be a positive duration, like 24h") return } since = d } rows, err := a.usage.UsageByScope(r.Context(), scope, time.Now().Add(-since)) if err != nil { // The only failure a caller can act on is naming a level that is not // one, which is what this mostly is. writeError(w, http.StatusBadRequest, err.Error()) return } out := make([]ScopeUsageView, 0, len(rows)) for _, u := range rows { v := ScopeUsageView{ Scope: string(u.Scope), ID: u.ID, Name: u.Name, Keys: u.Keys, Requests: u.Requests, Prompt: u.Prompt, Completion: u.Completion, CostMicros: u.CostMicros, Errors: u.Errors, LastAt: u.LastAt, Budget: u.Budget, BudgetPeriod: string(u.BudgetPeriod), } if f, ok := u.Spent(); ok { v.Fraction = &f } out = append(out, v) } writeJSON(w, http.StatusOK, out) } // Capacity is where measured capacity is kept. The store implements it. type Capacity interface { PutCapacity(ctx context.Context, c store.StoredCapacity) error Capacities(ctx context.Context) ([]store.StoredCapacity, error) } // SetCapacity supplies somewhere to keep benchmark results. func (a *API) SetCapacity(c Capacity) { a.capacity = c } // putCapacity records what a replica of a model was measured to serve. // // A PUT rather than a POST: re-running a benchmark replaces what was known // about that model on that profile. It is a better answer to the same question, // not another data point. func (a *API) putCapacity(w http.ResponseWriter, r *http.Request) { if a.capacity == nil { writeError(w, http.StatusNotImplemented, "this control plane has no store, so a measurement recorded here would not survive a restart") return } var c bench.Capacity if err := json.NewDecoder(io.LimitReader(r.Body, 1<<20)).Decode(&c); err != nil { writeError(w, http.StatusBadRequest, "the measurement is not valid JSON: "+err.Error()) return } c.Model = r.PathValue("name") if c.Profile == "" { writeError(w, http.StatusBadRequest, "a measurement needs the profile it was taken on: the same model on CPU and on a GPU are different numbers") return } // The model has to exist now, which catches a typo in the name — a // measurement filed under a misspelling is worse than none, because it // looks like the model was never benchmarked. It is not deleted when the // model is, though: a benchmark costs real GPU time and outliving the // catalog entry is the point. m, ok := a.ctl.Catalog().Resolve(c.Model) if !ok { writeError(w, http.StatusNotFound, "no model named "+c.Model+" is in the catalog") return } if !slices.ContainsFunc(m.Profiles, func(p catalog.Profile) bool { return p.Name == c.Profile }) { writeError(w, http.StatusBadRequest, "model "+c.Model+" has no profile named "+c.Profile) return } body, err := json.Marshal(c) if err != nil { writeError(w, http.StatusInternalServerError, err.Error()) return } if err := a.capacity.PutCapacity(r.Context(), store.StoredCapacity{ Model: c.Model, Profile: c.Profile, Measurement: body, MeasuredAt: c.MeasuredAt, MeasuredBy: r.Header.Get("X-Llmgw-Actor"), }); err != nil { writeError(w, http.StatusInternalServerError, err.Error()) return } writeJSON(w, http.StatusOK, c) } // CapacityView is one stored measurement, with who recorded it. type CapacityView struct { bench.Capacity MeasuredBy string `json:"measured_by,omitempty"` } func (a *API) listCapacity(w http.ResponseWriter, r *http.Request) { if a.capacity == nil { writeJSON(w, http.StatusOK, []CapacityView{}) return } rows, err := a.capacity.Capacities(r.Context()) if err != nil { writeError(w, http.StatusInternalServerError, err.Error()) return } out := make([]CapacityView, 0, len(rows)) for _, row := range rows { var c bench.Capacity if err := json.Unmarshal(row.Measurement, &c); err != nil { // Skipped rather than failing the listing: one row written by an // older shape must not hide every other measurement. continue } out = append(out, CapacityView{Capacity: c, MeasuredBy: row.MeasuredBy}) } writeJSON(w, http.StatusOK, out) } // 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}) } // ScaleRequest is how many replicas a model should run. type ScaleRequest struct { Min int `json:"min"` // Max omitted follows Min, which is what the catalog does for a model that // declares only a minimum. Max int `json:"max,omitempty"` IdleTimeout time.Duration `json:"idle_timeout,omitempty"` } // ScaleResponse reports the scaling that is now in force. // // What is in force rather than what was asked for: the catalog fills in a max // that was left out, and a caller that echoed its own request would print a // number the cluster is not using. type ScaleResponse struct { Model string `json:"model"` Min int `json:"min"` Max int `json:"max"` IdleTimeout time.Duration `json:"idle_timeout,omitempty"` OnDemand bool `json:"on_demand"` } func (a *API) scaleModel(w http.ResponseWriter, r *http.Request) { if a.models == nil { writeError(w, http.StatusNotImplemented, "this control plane has no store") return } var req ScaleRequest if err := json.NewDecoder(io.LimitReader(r.Body, 1<<16)).Decode(&req); err != nil { writeError(w, http.StatusBadRequest, "the scaling request is not valid JSON: "+err.Error()) return } if req.Min < 0 || req.Max < 0 { writeError(w, http.StatusBadRequest, "a replica count cannot be negative") return } name := r.PathValue("name") m, err := a.models.Scale(r.Context(), name, catalog.Scaling{Min: req.Min, Max: req.Max, IdleTimeout: req.IdleTimeout}, r.Header.Get("X-Llmgw-Actor")) if err != nil { // Like addModel: every refusal here is the request being wrong — a // model declared in configuration, one that was never added, a max // below a min. None are server faults. writeError(w, http.StatusBadRequest, err.Error()) return } writeJSON(w, http.StatusOK, ScaleResponse{ Model: m.Name, Min: m.Scaling.Min, Max: m.Scaling.Max, IdleTimeout: m.Scaling.IdleTimeout, OnDemand: m.Scaling.OnDemand(), }) } func (a *API) removeModel(w http.ResponseWriter, r *http.Request) { if a.models == nil { writeError(w, http.StatusNotImplemented, "this control plane has no store") 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()) } func (a *API) explainPlacement(w http.ResponseWriter, r *http.Request) { model := r.URL.Query().Get("model") if model == "" { writeError(w, http.StatusBadRequest, "model is required") return } if _, ok := a.ctl.Catalog().Resolve(model); !ok { writeError(w, http.StatusNotFound, fmt.Sprintf("model %q is not in the catalog", model)) return } writeJSON(w, http.StatusOK, a.Explain(model)) } // Explain builds the placement explanation for one model. // // It does not check that the model exists; the handler above does that so it // can answer 404, while the dashboard only ever asks about models it just read // out of the catalog. func (a *API) Explain(model string) ExplainView { v := ExplainView{Model: model} for _, rep := range a.ctl.Reconciler().Replicas() { if rep.Model == model && rep.State.Counts() { v.Placed = true break } } if res, ok := a.ctl.Reconciler().Explain(model); ok { for _, rej := range res.Rejections { v.Rejections = append(v.Rejections, RejectionView{ Profile: rej.Profile, Pool: rej.Pool, Node: rej.Node, Reason: rej.Reason, }) } } return v } // setNodeState implements cordon, uncordon and drain. // // One endpoint rather than three verbs, because they are the same operation // with different targets and the difference matters: cordon stops new // placement, drain also moves existing work off. func (a *API) setNodeState(w http.ResponseWriter, r *http.Request) { id := r.PathValue("id") var body struct { State string `json:"state"` } if err := json.NewDecoder(r.Body).Decode(&body); err != nil { writeError(w, http.StatusBadRequest, err.Error()) return } var st inventory.NodeState switch body.State { case "ready", "uncordon": st = inventory.Ready case "cordoned", "cordon": st = inventory.Cordoned case "draining", "drain": st = inventory.Draining default: writeError(w, http.StatusBadRequest, fmt.Sprintf("unknown state %q (want ready, cordoned or draining)", body.State)) return } if !a.ctl.Inventory().SetState(id, st) { writeError(w, http.StatusNotFound, fmt.Sprintf("no node %q", id)) return } // Act on it now rather than at the next tick: an operator draining a node // is usually about to do something to it. a.ctl.Reconciler().Notify() writeJSON(w, http.StatusOK, map[string]string{"id": id, "state": st.String()}) } func writeJSON(w http.ResponseWriter, code int, v any) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(code) _ = json.NewEncoder(w).Encode(v) } func writeError(w http.ResponseWriter, code int, msg string) { writeJSON(w, code, map[string]string{"error": msg}) }