Files
llmgw/internal/control/adminapi/api.go
T
Ilker ManapandClaude Opus 5.5 23648fa19a Watch each fabric's subnet managers, and keep a fabric one through a failover
A fabric's identity was its master subnet manager, which changes on a
failover: for the minute its ports took to agree, one fabric read as
two, and a name covering it as two separate fabrics. Each agent now
also lists every subnet manager on a fabric (saquery SMIR, through one
port per master, once a minute); managers listed together are one
fabric, named by the lowest GUID among them, so the identity survives
the master changing.

The same list says what a fabric can survive. The cluster page's new
subnet manager table, /api/v1/fabrics, doctor and metrics
(llmgw_fabric_master_sm, llmgw_fabric_subnet_managers,
llmgw_fabric_problems) show each fabric's master, priority and
standbys, and say when a fabric has no standby, its ports disagree on
the master, ports are up with no subnet manager (INIT), or the master
changed in the last hour. Every change of master a port reports is
kept, and a fabric's last day of them is listed.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-29 07:47:03 +02:00

842 lines
29 KiB
Go

// 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"
"strings"
"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/fabrics"
"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/fabrics", func(w http.ResponseWriter, _ *http.Request) {
writeJSON(w, http.StatusOK, a.Fabrics())
})
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("GET /api/v1/nodes/{id}/logs", a.nodeLogs)
}
// --- 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"`
// 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"`
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"`
}
// 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"`
}
// 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()),
}
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"`
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"`
// 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"`
}
// 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 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())
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]++
}
}
out := make([]ModelView, 0)
for _, m := range a.ctl.Catalog().Models() {
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,
}
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)
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,
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)
}
// 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})
}
// 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())
}