Files
llmgw/internal/control/adminapi/api.go
T
Ilker ManapandClaude Opus 5.5 7d97c6b74f Install a node's engines on the node, instead of trusting what is there
A node ran whatever llama-server somebody had once unpacked on it. Onboarding
checked an engine only when told its path, installed none, and a node with
nothing on it registered as ready. The one time that went wrong it went wrong
invisibly: the binary was there, executable, and could not start for want of
libgomp1, while the node reported healthy.

The control plane now says which engines every node has, at which version —
`engines:` in server.yaml, defaulting to the llama.cpp build this release was
checked against — and each agent converges on it. What only the node can
know, the node works out: the build for its processor and driver (CUDA with
its runtime when an NVIDIA driver is loaded, CPU otherwise), the OS packages
that build needs from its own package sources, and whether the result runs.
Every archive is checked against the sha256 GitHub publishes for it and
unpacked refusing any path or link that leaves the install directory. ldd
names a missing library outright; running `--version` proves the rest. Only
then is a directory marked installed, and it is checked again every ten
minutes, because an unrelated upgrade can remove a library.

An engine is reported only once installed and seen to run, or when the
agent's configuration names it on purpose — never because something happened
to be on PATH. The agent now says when it has probed, so an empty list means
"has none" rather than "too old to say": a node rebuilt from nothing is no
longer a candidate for every model at the moment it can run none of them.
Placement's refusal carries the node's own words, "installing: downloading
llama-b10991-bin-ubuntu-x64.tar.gz (16 MB)", and so do `ctl nodes` and the
cluster page, along with a warning for a CPU without AVX2 — the VM CPU type
that makes everything five times slower and nothing look wrong.

Agents also report their OS, architecture, vector extensions and NVIDIA
driver. Onboarding installs llmgw itself from the CI release for the node's
architecture, checked against the release's checksums, rather than copying
whatever binary the operator's machine had.

vLLM and SGLang are not installed yet: they are Python environments of
several gigabytes, not an archive, and still have to be named in the agent's
configuration.

Also fixes a race on the weights fingerprint, which two heartbeat loops touch
during a failover.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-28 10:59:59 +02:00

728 lines
25 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/inventory"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler"
"gitea.mynodes.xyz/ilker/llmgw/internal/store"
)
// API serves the management endpoints.
type API struct {
models Models
capacity Capacity
usage Usage
ctl *control.Controller
}
// New builds the API over a controller.
func New(ctl *control.Controller) *API { return &API{ctl: ctl} }
// Register mounts the endpoints on a mux under /api/v1.
func (a *API) Register(mux *http.ServeMux) {
mux.HandleFunc("GET /api/v1/summary", a.summary)
// Adding a model is a write, and the only one this API has. It lives here
// rather than on the client-facing listener for the same reason the rest of
// it does: reaching this port already means holding the control plane.
mux.HandleFunc("POST /api/v1/models", a.addModel)
mux.HandleFunc("DELETE /api/v1/models/{name}", a.removeModel)
mux.HandleFunc("POST /api/v1/models/{name}/scale", a.scaleModel)
mux.HandleFunc("PUT /api/v1/models/{name}/capacity", a.putCapacity)
mux.HandleFunc("GET /api/v1/capacity", a.listCapacity)
mux.HandleFunc("GET /api/v1/usage", a.usageByScope)
mux.HandleFunc("GET /api/v1/nodes", a.listNodes)
mux.HandleFunc("GET /api/v1/models", a.listModels)
mux.HandleFunc("GET /api/v1/replicas", a.listReplicas)
// Listing is a read and deleting is a separate verb, so a mistyped
// command cannot remove anything.
mux.HandleFunc("GET /api/v1/weights/orphans", a.listOrphans)
mux.HandleFunc("POST /api/v1/weights/collect", a.collectWeights)
mux.HandleFunc("GET /api/v1/explain-placement", a.explainPlacement)
mux.HandleFunc("POST /api/v1/nodes/{id}/state", a.setNodeState)
}
// --- responses ---
// NodeView is a node as `ctl nodes` shows it.
type NodeView struct {
ID string `json:"id"`
Hostname string `json:"hostname,omitempty"`
MgmtAddr string `json:"mgmt_addr,omitempty"`
State string `json:"state"`
Labels map[string]string `json:"labels,omitempty"`
GPUTotal int `json:"gpu_total"`
GPUFree int `json:"gpu_free"`
// MemoryGB is the machine's RAM, 0 when the agent did not report it. It is
// what decides whether a CPU profile can be placed here, so it belongs
// where someone looks when a model will not schedule.
MemoryGB int `json:"memory_gb,omitempty"`
// 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"`
}
// 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"`
}
// 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()
}
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)
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
}
}
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
}