// 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" "gitea.mynodes.xyz/ilker/llmgw/internal/control/modelserve" "gitea.mynodes.xyz/ilker/llmgw/internal/release" "io" "net/http" "slices" "strconv" "strings" "time" "unicode" "gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery" "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/fabrics" "gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory" "gitea.mynodes.xyz/ilker/llmgw/internal/control/nvlink" "gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler" "gitea.mynodes.xyz/ilker/llmgw/internal/control/topology" "gitea.mynodes.xyz/ilker/llmgw/internal/store" ) // API serves the management endpoints. type API struct { // releases is where release binaries are fetched from, for upgrades. releases string // serve runs registry-to-replica jobs; nil without a mirror and a store. serve *modelserve.Server models Models capacity Capacity usage Usage quotas Quotas 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("POST /api/v1/models/{name}/resources", a.setResources) a.registerQuotas(mux) mux.HandleFunc("GET /api/v1/models/{name}/target", a.target) mux.HandleFunc("POST /api/v1/models/{name}/class", func(w http.ResponseWriter, r *http.Request) { if a.models == nil { writeError(w, http.StatusNotImplemented, "this control plane has no store") return } var req struct { Class string `json:"class"` } if err := json.NewDecoder(io.LimitReader(r.Body, 1<<12)).Decode(&req); err != nil { writeError(w, http.StatusBadRequest, "the request is not valid JSON: "+err.Error()) return } m, err := a.models.SetClass(r.Context(), r.PathValue("name"), req.Class, actorOf(r)) if err != nil { writeError(w, http.StatusBadRequest, err.Error()) return } writeJSON(w, http.StatusOK, map[string]string{"model": m.Name, "class": m.EffectiveClass()}) }) 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/fabrics", func(w http.ResponseWriter, _ *http.Request) { writeJSON(w, http.StatusOK, a.Fabrics()) }) mux.HandleFunc("GET /api/v1/nvlink", func(w http.ResponseWriter, _ *http.Request) { writeJSON(w, http.StatusOK, a.NVLink()) }) mux.HandleFunc("GET /api/v1/fabrics/{id}/topology", func(w http.ResponseWriter, r *http.Request) { t, ok := a.Topology(r.PathValue("id")) if !ok { http.NotFound(w, r) return } writeJSON(w, http.StatusOK, t) }) 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) mux.HandleFunc("POST /api/v1/nodes/{id}/upgrade", a.upgradeNode) mux.HandleFunc("DELETE /api/v1/nodes/{id}", a.removeNode) mux.HandleFunc("POST /api/v1/fabric-test", a.fabricTest) mux.HandleFunc("GET /api/v1/nodes/{id}/logs", a.nodeLogs) mux.HandleFunc("POST /api/v1/restart", a.restart) mux.HandleFunc("POST /api/v1/models/serve", a.serveModel) mux.HandleFunc("GET /api/v1/jobs/{id}", a.job) mux.HandleFunc("GET /api/v1/requests/{id}", a.request) mux.HandleFunc("POST /api/v1/models/{name}/retry", a.retryModel) mux.HandleFunc("GET /api/v1/models/{name}/headroom", func(w http.ResponseWriter, r *http.Request) { limit, _ := strconv.Atoi(r.URL.Query().Get("limit")) v, ok := a.Headroom(r.PathValue("name"), limit) if !ok { http.Error(w, "no such model", http.StatusNotFound) return } writeJSON(w, http.StatusOK, v) }) } // --- 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"` // CPUCount is the machine's CPUs, 0 when the agent did not report them. CPUCount int `json:"cpu_count,omitempty"` // Engines are the inference engines installed here, as the agent found // them. Absent means the agent is older than the probe and said nothing, // which is not the same as having none — placement treats it that way too, // or a control plane upgrade would empty a fleet. Engines []string `json:"engines,omitempty"` // Provision is where installing each wanted engine stands. An engine is in // Engines only once its installation here is ready. Provision []inventory.EngineInstall `json:"provision,omitempty"` // OS is the operating system and architecture, as the agent read them. OS string `json:"os,omitempty"` // Warnings are facts about the machine that make it slow or unusable // without making it unhealthy: a VM with no vector instructions runs every // model, at a fifth of the speed. Warnings []string `json:"warnings,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"` // Features are the requests the agent answers beyond the usual: // "fabric-test". Absent is an agent too old to say. Features []string `json:"features,omitempty"` Replicas int `json:"replicas"` // Load is how busy the machine is, absent when its agent has not // measured it lately — which is unknown, not idle. Load *LoadView `json:"load,omitempty"` // IBPorts are the node's InfiniBand and RoCE ports at the last heartbeat. IBPorts []IBPortView `json:"ib_ports,omitempty"` // GPUs are the cards, with how they are wired: NVLink peers, the // adapter nearest each, and the NVLink domain each has joined. GPUs []GPUView `json:"gpus,omitempty"` // NVLinkDomain is the multi-node domain every card has joined, if any. NVLinkDomain string `json:"nvlink_domain,omitempty"` IMEX string `json:"imex,omitempty"` IMEXChannels int `json:"imex_channels,omitempty"` } // GPUView is one card as a node's page shows it. type GPUView struct { Index int `json:"index"` Name string `json:"name"` NUMA int `json:"numa"` NVLinkPeers []int `json:"nvlink_peers,omitempty"` // NearestNIC is the adapter closest to the card, and how close. NearestNIC string `json:"nearest_nic,omitempty"` NICPath string `json:"nic_path,omitempty"` FabricState string `json:"fabric_state,omitempty"` Domain string `json:"domain,omitempty"` Bandwidth string `json:"bandwidth,omitempty"` } // IBPortView is one port as a page shows it. type IBPortView struct { HCA string `json:"hca"` Port int `json:"port"` State string `json:"state"` PhysState string `json:"phys_state,omitempty"` RateGbps int `json:"rate_gbps"` LinkLayer string `json:"link_layer,omitempty"` Active bool `json:"active"` // Fabric is the declared fabric this port's adapter carries, if any. Fabric string `json:"fabric,omitempty"` // Physical is the fabric the port is physically on: its subnet manager, // or its RoCE network. SMError says why that is not known. Physical string `json:"physical,omitempty"` SMError string `json:"sm_error,omitempty"` // Errors are the error counters that are not zero, by name. Errors map[string]uint64 `json:"errors,omitempty"` TxBytes uint64 `json:"tx_bytes,omitempty"` RxBytes uint64 `json:"rx_bytes,omitempty"` } // LoadView is a node's CPU and memory in use, at its last heartbeat. type LoadView struct { CPUPercent float64 `json:"cpu_percent"` CPUs int `json:"cpus,omitempty"` Load1 float64 `json:"load1"` Load5 float64 `json:"load5"` Load15 float64 `json:"load15"` MemUsed int64 `json:"memory_used_bytes"` MemTotal int64 `json:"memory_total_bytes"` // MemPercent is used over used plus available. MemPercent float64 `json:"memory_percent"` AgeSec int64 `json:"age_seconds"` // What the node's sensors say; zero is not measured. CPUTempC float64 `json:"cpu_temp_c,omitempty"` CPUPowerW float64 `json:"cpu_power_w,omitempty"` InletTempC float64 `json:"inlet_temp_c,omitempty"` GPUs []inventory.GPUSensor `json:"gpu_sensors,omitempty"` } // loadFresh is how old a measurement may be and still be shown as current. const loadFresh = time.Minute func loadView(l *inventory.Load, now time.Time) *LoadView { if l == nil || now.Sub(l.At) > loadFresh { return nil } v := &LoadView{ CPUPercent: l.CPUPercent, CPUs: l.CPUCount, Load1: l.Load1, Load5: l.Load5, Load15: l.Load15, MemUsed: l.MemUsedBytes, MemTotal: l.MemUsedBytes + l.MemAvailBytes, AgeSec: int64(now.Sub(l.At).Seconds()), CPUTempC: l.CPUTempC, CPUPowerW: l.CPUPowerW, InletTempC: l.InletTempC, GPUs: l.GPUs, } if v.MemTotal > 0 { v.MemPercent = float64(v.MemUsed) * 100 / float64(v.MemTotal) } return v } // ModelView is a model as `ctl models` shows it. type ModelView struct { Name string `json:"name"` // Description is what the model is for, when somebody said. Description string `json:"description,omitempty"` 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"` // Desired is how many it should have now: Min, or for an on-demand // model one while it is being asked for and none otherwise. Desired int `json:"desired_replicas"` // Failures counts replicas that failed to start in a row. Failures int `json:"failures,omitempty"` // Parked is set once Failures reached the limit; the model is not retried // until its definition changes or `ctl retry` is run. Reason is the last // error. Parked bool `json:"parked,omitempty"` Reason string `json:"reason,omitempty"` // Unplaceable is set when it is short and the last attempt found no node. Unplaceable bool `json:"unplaceable,omitempty"` Fallbacks []string `json:"fallbacks,omitempty"` Profiles []string `json:"profiles"` // MemoryGB and CPUCores are what one replica reserves, from its first // profile; MeasuredMemoryGB is the most one of its replicas uses now, 0 // when none is measured. MemoryGB int `json:"memory_gb,omitempty"` CPUCores int `json:"cpu_cores,omitempty"` MeasuredMemoryGB float64 `json:"measured_memory_gb,omitempty"` // Warnings are what makes its reservation wrong: nothing declared where // placement needs a figure, or less declared than it uses. Warnings []string `json:"warnings,omitempty"` // Class is its capacity class; Preempted the replicas it lost to a higher // class; HeldFor the model it is giving room to now, if any. Class string `json:"class"` Owner string `json:"owner,omitempty"` Preempted int `json:"preempted,omitempty"` HeldFor string `json:"held_for,omitempty"` } // ReplicaView is a replica as `ctl replicas` shows it. type ReplicaView struct { ID string `json:"id"` // EngineKey is the key this replica's engine demands, for `ctl bench` // and anyone else holding the admin listener who needs to reach an engine // directly. The client-facing pages never render it. EngineKey string `json:"engine_key,omitempty"` // Restart is "queued" or "replacing" while an operator's restart of this // replica is under way. Restart string `json:"restart,omitempty"` Model string `json:"model"` Profile string `json:"profile"` // Stale is true when the profile has been edited since this replica was // placed, so it is running a definition the catalog no longer holds. // // Reported rather than acted on. Moving a replica that is serving costs // more than the imperfection, and the operator who made the edit is the // one who should decide whether it is worth a restart. Stale bool `json:"stale,omitempty"` 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"` // CPUCores and MemoryBytes are the engine's use of its node, from the // node's last heartbeat; zero when not measured. CPUCores float64 `json:"cpu_cores,omitempty"` MemoryBytes int64 `json:"memory_bytes,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"` } // HeadroomView is how many more replicas of a model fit now, and where. type HeadroomView struct { Model string `json:"model"` // Current is the replicas holding capacity: starting or ready. Current int `json:"current"` // More is how many more placement would find room for now. More int `json:"more"` // Nodes is where each of those would go. Nodes [][]string `json:"nodes,omitempty"` // Limited is set when More stopped at the limit asked rather than at the // cluster's edge: there may be room for more. Limited bool `json:"limited,omitempty"` // Blocked says why one more does not fit, a line per node considered. Blocked []string `json:"blocked,omitempty"` // TokPerSec is what one more replica was measured to serve; 0 unmeasured. TokPerSec float64 `json:"tok_per_sec,omitempty"` // ByPreempting is how many more would fit by stopping replicas this // model's class may take; Preempts the models those belong to. ByPreempting int `json:"by_preempting,omitempty"` Preempts []string `json:"preempts,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, CPUCount: n.CPUCount, Fabrics: n.FabricNames(), AgentVersion: n.AgentVersion, Features: n.Features, Replicas: replicasOn[n.ID], } if free, total := n.LocalDisk(); total > 0 { v.DiskFreeGB = int(free / (1 << 30)) v.DiskTotalGB = int(total / (1 << 30)) } for name := range n.EngineVersions { v.Engines = append(v.Engines, name) } slices.Sort(v.Engines) v.Provision = n.Provision if h := n.Host; h.OSID != "" { v.OS = strings.TrimSpace(h.OSID + " " + h.OSVersion + " " + h.Arch) if h.Arch == "amd64" && !h.HasCPUFlag("avx2") { v.Warnings = append(v.Warnings, "no AVX2: inference runs several times slower "+ "(a VM needs CPU type \"host\")") } } 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() } v.Load = loadView(n.Load, time.Now()) v.NVLinkDomain, v.IMEX, v.IMEXChannels = n.NVLinkDomain(), n.IMEX.State, n.IMEX.Channels for _, g := range n.GPUs { gv := GPUView{Index: g.Index, Name: g.Name, NUMA: g.NUMANode, NVLinkPeers: g.NVLinkPeers} for hca, rel := range g.NICPaths { if gv.NearestNIC == "" || discovery.PathRank(rel) < discovery.PathRank(gv.NICPath) || (discovery.PathRank(rel) == discovery.PathRank(gv.NICPath) && hca < gv.NearestNIC) { gv.NearestNIC, gv.NICPath = hca, rel } } if f := g.Fabric; f != nil { gv.FabricState, gv.Domain, gv.Bandwidth = f.State, f.Domain(), f.Bandwidth } v.GPUs = append(v.GPUs, gv) } fabricOf := map[string]string{} for _, f := range n.Fabrics { for _, h := range f.HCAs { fabricOf[h] = f.Fabric } } for _, p := range n.IBPorts { pv := IBPortView{HCA: p.HCA, Port: p.Port, State: p.State, PhysState: p.PhysState, RateGbps: p.RateGbps, LinkLayer: p.LinkLayer, Active: p.Active(), Fabric: fabricOf[p.HCA], TxBytes: p.Counters["port_xmit_data"], RxBytes: p.Counters["port_rcv_data"], SMError: p.SMError} switch { case p.RoCE() && p.Network != "": pv.Physical = "RoCE " + p.Network case p.SMGUID != "": pv.Physical = fmt.Sprintf("SM %s (LID %d)", p.SMGUID, p.SMLID) } for name, c := range p.Counters { if c > 0 && name != "port_xmit_data" && name != "port_rcv_data" { if pv.Errors == nil { pv.Errors = map[string]uint64{} } pv.Errors[name] = c } } v.IBPorts = append(v.IBPorts, pv) } 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]++ } } health := map[string]reconciler.ModelHealth{} for _, h := range a.ctl.Reconciler().ModelHealth() { health[h.Model] = h } // The most one replica of each model uses, summed over its shards. measured := map[string]int64{} for _, rep := range a.Replicas("") { var b int64 for _, sh := range rep.Shards { b += sh.MemoryBytes } measured[rep.Model] = max(measured[rep.Model], b) } out := make([]ModelView, 0) for _, m := range a.ctl.Catalog().Models() { h := health[m.Name] v := ModelView{ Name: m.Name, Description: m.Description, 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, Desired: h.Desired, Failures: h.Failures, Parked: h.Parked, Reason: h.Reason, Unplaceable: h.Unplaceable, Class: m.EffectiveClass(), Owner: m.Owner, Preempted: h.Preempted, HeldFor: h.HeldFor, } for _, p := range m.Profiles { v.Profiles = append(v.Profiles, p.Name) } if len(m.Profiles) > 0 { v.MemoryGB, v.CPUCores = m.Profiles[0].MinMemoryGB, m.Profiles[0].CPUCores } v.MeasuredMemoryGB = float64(measured[m.Name]) / (1 << 30) v.Warnings = reservationWarnings(m, v.MeasuredMemoryGB) 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) loads := map[string]*inventory.Load{} for _, n := range a.ctl.Inventory().Nodes() { if n.Load != nil && time.Since(n.Load.At) <= loadFresh { loads[n.ID] = n.Load } } 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, EngineKey: a.ctl.EngineKey(rep.ID), Restart: a.ctl.Reconciler().RestartState(rep.ID), Stale: a.profileChanged(rep.Model, rep.Profile, rep.ProfileFP), 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 } } if l := loads[s.NodeID]; l != nil { if use, ok := l.Shards[s.ID]; ok { sv.CPUCores, sv.MemoryBytes = use.CPUCores, use.RSSBytes } } 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) 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) } // ResourcesRequest sets what a model's profiles reserve of a node. type ResourcesRequest struct { // Profile is one profile's name; empty is every profile. Profile string `json:"profile,omitempty"` MemoryGB int `json:"memory_gb"` CPUCores int `json:"cpu_cores"` } // 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: actorOf(r), }); 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, actorOf(r)); 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}, actorOf(r)) 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(), }) } // 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) { if a.models == nil { writeError(w, http.StatusNotImplemented, "this control plane has no store") return } var req ResourcesRequest if err := json.NewDecoder(io.LimitReader(r.Body, 1<<16)).Decode(&req); err != nil { writeError(w, http.StatusBadRequest, "the request is not valid JSON: "+err.Error()) return } m, err := a.models.Resources(r.Context(), r.PathValue("name"), req.Profile, req.MemoryGB, req.CPUCores, actorOf(r)) if err != nil { writeError(w, http.StatusBadRequest, err.Error()) return } writeJSON(w, http.StatusOK, m) } 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. // Headroom tries placing up to limit more replicas of a model, on a copy of // the inventory, and says how many fit. A limit of zero or less takes 16. func (a *API) Headroom(model string, limit int) (HeadroomView, bool) { if limit <= 0 || limit > 64 { limit = 16 } h, ok := a.ctl.Reconciler().Headroom(model, limit) if !ok { return HeadroomView{}, false } return HeadroomView{Model: model, Current: h.Current, More: h.More, Nodes: h.Nodes, Limited: h.Limited, Blocked: h.Blocked, TokPerSec: h.TokPerSec, ByPreempting: h.ByPreempting, Preempts: h.Preempts}, true } 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. // removeNode forgets a node that has gone for good: a machine retired or // rebuilt under another name, which otherwise stays listed as lost until the // control plane restarts. // // Only one that is not there: lost, or unreachable with nothing placed on // it. A connected node removed here would register again on its next // heartbeat, and one with replicas would take them out of the reconciler's // sight while they may still be running. func (a *API) removeNode(w http.ResponseWriter, r *http.Request) { id := r.PathValue("id") inv := a.ctl.Inventory() n, ok := inv.Get(id) if !ok { writeError(w, http.StatusNotFound, fmt.Sprintf("no node %q", id)) return } switch { case n.State == inventory.Lost: case n.State == inventory.Unreachable && len(inv.Allocations(id)) == 0: default: writeError(w, http.StatusConflict, fmt.Sprintf( "%s is %s: stop its agent (systemctl stop llmgw-agent) and remove it once it is lost, a minute later", id, n.State)) return } inv.Remove(id) a.ctl.Reconciler().Notify() writeJSON(w, http.StatusOK, map[string]string{"id": id, "removed": "true"}) } 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}) } // profileChanged reports whether a replica is running a profile the catalog no // longer holds. // // A replica with no recorded fingerprint is not stale: that is every replica // adopted from an agent older than the field, and calling those stale would // mark a whole fleet as wrong on the first upgrade. Nor is one whose profile // has been deleted outright — the model or the profile going away is a // different thing, and the replica's own state already says it. func (a *API) profileChanged(model, profile, was string) bool { if was == "" { return false } m, ok := a.ctl.Catalog().Resolve(model) if !ok { return false } for _, p := range m.Profiles { if p.Name == profile { return p.Fingerprint() != was } } return false } // Fabrics is the physical fabrics the adapters are on, against the names // the nodes gave them. func (a *API) Fabrics() fabrics.Report { inv := a.ctl.Inventory() return fabrics.Analyze(inv.Nodes(), inv.SMChanges(), time.Now()) } // TopologyView is one fabric's map, with the cluster's own adapters on it // live: their state, rate and traffic from the last heartbeat. type TopologyView struct { ID string `json:"id"` Key string `json:"key"` Label string `json:"label"` Declared []string `json:"declared"` ScannedAt time.Time `json:"scanned_at"` ScannedBy string `json:"scanned_by,omitempty"` TookMs int64 `json:"took_ms"` Err string `json:"error,omitempty"` Map *topology.View `json:"map,omitempty"` Live map[string]Live `json:"live"` // Problems are the fabric's own, from the analysis of its identity and // subnet managers. Problems []fabrics.Problem `json:"problems,omitempty"` SM *fabrics.SM `json:"sm,omitempty"` SMs []fabrics.SM `json:"sms,omitempty"` } // Live is one of the cluster's adapters as its node last reported it. type Live struct { Node string `json:"node"` HCA string `json:"hca"` State string `json:"state"` Phys string `json:"phys_state"` Gbps int `json:"gbps"` TxBps float64 `json:"tx_bps"` RxBps float64 `json:"rx_bps"` Errors map[string]uint64 `json:"errors,omitempty"` } // FabricID is a physical fabric's key as an address can carry it. func FabricID(key string) string { return strings.NewReplacer(":", "-", "/", "_").Replace(key) } // Topologies lists every physical fabric with what is known of its map. func (a *API) Topologies() []TopologyView { rep := a.Fabrics() scans := a.ctl.Inventory().Scans() out := make([]TopologyView, 0, len(rep.Physical)) for _, ph := range rep.Physical { t := TopologyView{ID: FabricID(ph.Key), Key: ph.Key, Label: ph.Label, Declared: ph.Declared} if sc, ok := scans[ph.Key]; ok { t.ScannedAt, t.ScannedBy, t.TookMs, t.Err = sc.At, sc.Node+" "+sc.Port, sc.Took.Milliseconds(), sc.Err } out = append(out, t) } return out } // Topology is one fabric's map, by the id Topologies gives it. func (a *API) Topology(id string) (TopologyView, bool) { rep := a.Fabrics() var ph *fabrics.Physical for i := range rep.Physical { if FabricID(rep.Physical[i].Key) == id { ph = &rep.Physical[i] } } if ph == nil { return TopologyView{}, false } t := TopologyView{ID: id, Key: ph.Key, Label: ph.Label, Declared: ph.Declared, Live: map[string]Live{}, SM: ph.SM, SMs: ph.SMs} for _, p := range rep.Problems { for _, d := range ph.Declared { if p.Fabric == d { t.Problems = append(t.Problems, p) break } } } hosts := map[string]topology.Host{} for _, n := range a.ctl.Inventory().Nodes() { for _, p := range n.IBPorts { if p.NodeGUID == "" { continue } hosts[p.NodeGUID] = topology.Host{Node: n.ID, HCA: p.HCA} l := Live{Node: n.ID, HCA: p.HCA, State: p.State, Phys: p.PhysState, Gbps: p.RateGbps, TxBps: p.TxBps, RxBps: p.RxBps} for name, c := range p.Counters { if c > 0 && name != "port_xmit_data" && name != "port_rcv_data" { if l.Errors == nil { l.Errors = map[string]uint64{} } l.Errors[name] = c } } t.Live[p.NodeGUID] = l } } sc, ok := a.ctl.Inventory().Scans()[ph.Key] if !ok { return t, true } t.ScannedAt, t.ScannedBy, t.TookMs, t.Err = sc.At, sc.Node+" "+sc.Port, sc.Took.Milliseconds(), sc.Err if sc.Output != "" { f, err := topology.Parse(sc.Output) if err != nil { t.Err = "the map could not be read: " + err.Error() return t, true } v := topology.Build(f, hosts) t.Map = &v } return t, true } // NVLink is the cluster's multi-node NVLink domains, and what keeps any of // their nodes out. func (a *API) NVLink() nvlink.Report { return nvlink.Analyze(a.ctl.Inventory().Nodes()) } // RestartRequest names what to restart: replicas by id, every replica of some // models, or — All — every ready replica in the cluster. type RestartRequest struct { Replicas []string `json:"replicas,omitempty"` Models []string `json:"models,omitempty"` All bool `json:"all,omitempty"` } // RestartResult lists the replicas queued. type RestartResult struct { Queued []string `json:"queued"` } func (a *API) restart(w http.ResponseWriter, r *http.Request) { var req RestartRequest if err := json.NewDecoder(io.LimitReader(r.Body, 1<<16)).Decode(&req); err != nil { writeError(w, http.StatusBadRequest, "the restart request is not valid JSON: "+err.Error()) return } ids := slices.Clone(req.Replicas) for _, rep := range a.ctl.Reconciler().ReadyReplicas() { if req.All || slices.Contains(req.Models, rep.Model) { ids = append(ids, rep.ID) } } if len(ids) == 0 { writeError(w, http.StatusBadRequest, "nothing ready matches; name replicas, models, or all") return } if err := a.ctl.Reconciler().Restart(ids...); err != nil { writeError(w, http.StatusNotFound, err.Error()) return } writeJSON(w, http.StatusAccepted, RestartResult{Queued: ids}) } func (a *API) retryModel(w http.ResponseWriter, r *http.Request) { name := r.PathValue("name") if _, ok := a.ctl.Catalog().Resolve(name); !ok { writeError(w, http.StatusNotFound, "no model called "+name) return } was := a.ctl.Reconciler().Retry(name) writeJSON(w, http.StatusOK, map[string]bool{"was_parked": was}) } // requestFinder looks a request up by the id it was given. Asked of the usage // source by type assertion, so one without it answers 501 rather than every // source having to grow the method. type requestFinder interface { UsageByRequest(ctx context.Context, requestID string) (store.Usage, bool, error) } // RequestView is what one request went through. type RequestView struct { RequestID string `json:"request_id"` At time.Time `json:"at"` KeyID string `json:"key_id"` UserID string `json:"user_id,omitempty"` ProjectID string `json:"project_id"` Model string `json:"model"` ServedModel string `json:"served_model"` Replica string `json:"replica,omitempty"` Status int `json:"status"` LatencyMs int64 `json:"latency_ms"` QueueMs int64 `json:"queue_ms"` TTFTMs int64 `json:"ttft_ms"` PromptTokens int64 `json:"prompt_tokens"` CompletionTokens int64 `json:"completion_tokens"` CostMicros int64 `json:"cost_micros"` } func (a *API) request(w http.ResponseWriter, r *http.Request) { f, ok := a.usage.(requestFinder) if !ok { writeError(w, http.StatusNotImplemented, "this control plane keeps no usage records") return } u, found, err := f.UsageByRequest(r.Context(), r.PathValue("id")) switch { case err != nil: writeError(w, http.StatusInternalServerError, err.Error()) return case !found: writeError(w, http.StatusNotFound, "no request with that id; records are written a few seconds after a request ends") return } writeJSON(w, http.StatusOK, RequestView{ RequestID: u.RequestID, At: u.At, KeyID: u.KeyID, UserID: u.UserID, ProjectID: u.ProjectID, Model: u.Model, ServedModel: u.ServedModel, Replica: u.Replica, Status: u.Status, LatencyMs: u.LatencyMs, QueueMs: u.QueueMs, TTFTMs: u.TTFTMs, PromptTokens: u.PromptTokens, CompletionTokens: u.CompletionTokens, CostMicros: u.CostMicros, }) } // SetReleases sets where upgrades fetch release binaries from. func (a *API) SetReleases(base string) { a.releases = base } // UpgradeRequest names the release a node should run. type UpgradeRequest struct { Version string `json:"version"` } func (a *API) upgradeNode(w http.ResponseWriter, r *http.Request) { var req UpgradeRequest if err := json.NewDecoder(io.LimitReader(r.Body, 1<<12)).Decode(&req); err != nil { writeError(w, http.StatusBadRequest, "the upgrade request is not valid JSON: "+err.Error()) return } if !release.Tag.MatchString(req.Version) { writeError(w, http.StatusBadRequest, req.Version+" is not a release version") return } u := a.ctl.Upgrader() if u == nil { writeError(w, http.StatusNotImplemented, "this control plane cannot upgrade nodes") return } base := a.releases if base == "" { base = release.DefaultBase } if err := u.Upgrade(r.Context(), r.PathValue("id"), req.Version, base); err != nil { writeError(w, http.StatusBadGateway, err.Error()) return } writeJSON(w, http.StatusOK, map[string]string{"node": r.PathValue("id"), "version": req.Version}) } // SetServe supplies what serves a model from the registry. func (a *API) SetServe(s *modelserve.Server) { a.serve = s } func (a *API) serveModel(w http.ResponseWriter, r *http.Request) { if a.serve == nil { writeError(w, http.StatusNotImplemented, "serving from the registry needs a mirror and a store, and this control plane lacks one") return } var req modelserve.Request if err := json.NewDecoder(io.LimitReader(r.Body, 1<<16)).Decode(&req); err != nil { writeError(w, http.StatusBadRequest, "the request is not valid JSON: "+err.Error()) return } req.By = actorOf(r) job, err := a.serve.Start(req) if err != nil { writeError(w, http.StatusBadRequest, err.Error()) return } writeJSON(w, http.StatusAccepted, job) } func (a *API) job(w http.ResponseWriter, r *http.Request) { if a.serve == nil { writeError(w, http.StatusNotFound, "no such job") return } job, ok := a.serve.Get(r.PathValue("id")) if !ok { writeError(w, http.StatusNotFound, "no such job; jobs are kept only until the control plane restarts") return } writeJSON(w, http.StatusOK, job) } // actorOf is who a change is recorded as made by: what the caller says, since // the admin API answers only this machine and trusts whoever reaches it. Kept // to a short printable name, so a claim cannot pass for more than one — no // line breaks to forge a second log line, no length to bury the rest. func actorOf(r *http.Request) string { claimed := r.Header.Get("X-Llmgw-Actor") var b strings.Builder for _, c := range claimed { if b.Len() >= 64 { break } if unicode.IsLetter(c) || unicode.IsDigit(c) || strings.ContainsRune("._-@", c) { b.WriteRune(c) } } if b.Len() == 0 { return "unknown" } return b.String() } // NodeCPUs is the CPU counts of ready nodes with at least this much memory: // where a replica needing it could go. func (a *API) NodeCPUs(minMemoryGB int) []int { var out []int for _, n := range a.ctl.Inventory().Nodes() { if n.State.String() != "ready" || n.CPUCount == 0 { continue } if n.MemoryTotalMB > 0 && n.MemoryTotalMB/1024 < minMemoryGB { continue } out = append(out, n.CPUCount) } return out } // reservationWarnings says what is wrong with what a model reserves. // // Resident memory includes the weights mapped from disk, which the kernel can // take back, so it overstates what a model needs by the part of the weights // not in use. A tenth over is let pass for that; more is a figure too low for // placement to be trusted with. func reservationWarnings(m catalog.Model, measuredGB float64) []string { var out []string for _, p := range m.Profiles { if p.CPUOnly() && p.MinMemoryGB <= 0 { out = append(out, fmt.Sprintf("profile %s declares no memory: placement cannot tell when a node is full "+ "(llmgw ctl model resources %s --memory GB)", p.Name, m.Name)) } } if len(m.Profiles) == 1 && m.Profiles[0].MinMemoryGB > 0 && measuredGB > float64(m.Profiles[0].MinMemoryGB)*1.1 { out = append(out, fmt.Sprintf("uses %.1f GB and reserves %d: raise it, or nodes will be filled past their memory", measuredGB, m.Profiles[0].MinMemoryGB)) } return out }