// 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" "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 { models Models 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("GET /api/v1/nodes", a.listNodes) mux.HandleFunc("GET /api/v1/models", a.listModels) mux.HandleFunc("GET /api/v1/replicas", a.listReplicas) 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"` } // 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 { v.Shards = append(v.Shards, ShardView{ ID: s.ID, Node: s.NodeID, Rank: s.Rank, GPUs: s.GPUIndices, State: s.State.String(), Err: s.Err, }) } 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 } // 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()) } 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}) }