Two things a cluster with many nodes and many GPUs needs that were missing, and they are the same thing seen from two distances. Idle tracking. The control plane knew what was placed and whether it was healthy, never whether anyone wanted it — so on a fleet with more models than cards there was no way to answer the first question anyone asks, which is what can I stop. The router is the only thing that knows, so it now reports it: a timestamp written in Acquire, where a request already meets an endpoint under a lock already held, and collected once a tick as a map. That shape is the point. Per-request notification would be a firehose carrying one bit — the newest timestamp wins — and this costs the same on four replicas as on four hundred. It travels through an interface in the direction opposite EndpointSink, for the same reason that one exists: the control plane decides what should run, the router knows whether anyone wants it, and neither should need to be in the other's process to say so. Unknown is kept distinct from idle throughout. A router that has just restarted knows nothing, and a zero timestamp would be the oldest idle replica in every comparison — so one restart would nominate an entire fleet for eviction. Absent from the map, "-" in the table, negative in the JSON, and IdleFor returns a second value rather than a duration anything could act on by mistake. ctl status. A per-node listing is right for four nodes and wrong for two hundred: how full am I, what is broken, what is holding cards nobody is using are aggregates, and reading them off a long table is a chore, which means it gets skipped. Capacity per GPU class with usage bars, free whole nodes per fabric — the number that decides whether a multi-node replica can land, since free cards on the wrong fabric are not capacity for a model that spans nodes — replica and model counts, models that cannot be placed, and what has gone unasked-for. Derived from the same state the tables show, so it cannot disagree with them. Unhealthy GPUs count as neither used nor free, because counting them free is how a cluster looks like it has room it does not. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
331 lines
10 KiB
Go
331 lines
10 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 (
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"time"
|
|
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/control"
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory"
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler"
|
|
)
|
|
|
|
// API serves the management endpoints.
|
|
type API struct {
|
|
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)
|
|
mux.HandleFunc("GET /api/v1/nodes", a.listNodes)
|
|
mux.HandleFunc("GET /api/v1/models", a.listModels)
|
|
mux.HandleFunc("GET /api/v1/replicas", a.listReplicas)
|
|
mux.HandleFunc("GET /api/v1/explain-placement", a.explainPlacement)
|
|
mux.HandleFunc("POST /api/v1/nodes/{id}/state", a.setNodeState)
|
|
}
|
|
|
|
// --- responses ---
|
|
|
|
// NodeView is a node as `ctl nodes` shows it.
|
|
type NodeView struct {
|
|
ID string `json:"id"`
|
|
Hostname string `json:"hostname,omitempty"`
|
|
MgmtAddr string `json:"mgmt_addr,omitempty"`
|
|
State string `json:"state"`
|
|
Labels map[string]string `json:"labels,omitempty"`
|
|
|
|
GPUTotal int `json:"gpu_total"`
|
|
GPUFree int `json:"gpu_free"`
|
|
// MemoryGB is the machine's RAM, 0 when the agent did not report it. It is
|
|
// what decides whether a CPU profile can be placed here, so it belongs
|
|
// where someone looks when a model will not schedule.
|
|
MemoryGB int `json:"memory_gb,omitempty"`
|
|
Classes []string `json:"gpu_classes,omitempty"`
|
|
Fabrics []string `json:"fabrics,omitempty"`
|
|
|
|
// Unhealthy lists GPUs kept out of placement and why, which is usually the
|
|
// answer when a node looks idle but nothing will schedule onto it.
|
|
Unhealthy []string `json:"unhealthy_gpus,omitempty"`
|
|
|
|
LastSeenAgo string `json:"last_seen_ago"`
|
|
AgentVersion string `json:"agent_version,omitempty"`
|
|
Replicas int `json:"replicas"`
|
|
}
|
|
|
|
// ModelView is a model as `ctl models` shows it.
|
|
type ModelView struct {
|
|
Name string `json:"name"`
|
|
Aliases []string `json:"aliases,omitempty"`
|
|
Source string `json:"source"`
|
|
Min int `json:"min_replicas"`
|
|
Max int `json:"max_replicas"`
|
|
Ready int `json:"ready_replicas"`
|
|
Total int `json:"total_replicas"`
|
|
Fallbacks []string `json:"fallbacks,omitempty"`
|
|
Profiles []string `json:"profiles"`
|
|
}
|
|
|
|
// ReplicaView is a replica as `ctl replicas` shows it.
|
|
type ReplicaView struct {
|
|
ID string `json:"id"`
|
|
Model string `json:"model"`
|
|
Profile string `json:"profile"`
|
|
State string `json:"state"`
|
|
Pool string `json:"pool,omitempty"`
|
|
Fabric string `json:"fabric,omitempty"`
|
|
Endpoint string `json:"endpoint,omitempty"`
|
|
AgeSec int64 `json:"age_seconds"`
|
|
// IdleSec is how long since a request was routed here. Negative means the
|
|
// router has not reported any use of this replica, which is not the same
|
|
// as idle — it is what a replica looks like before anything has asked for
|
|
// it, and after a router restart.
|
|
IdleSec int64 `json:"idle_seconds"`
|
|
Err string `json:"error,omitempty"`
|
|
Shards []ShardView `json:"shards"`
|
|
}
|
|
|
|
// ShardView is one shard of a replica.
|
|
type ShardView struct {
|
|
ID string `json:"id"`
|
|
Node string `json:"node"`
|
|
Rank int `json:"rank"`
|
|
GPUs []int `json:"gpus"`
|
|
State string `json:"state"`
|
|
Err string `json:"error,omitempty"`
|
|
}
|
|
|
|
// ExplainView is why a model could not be placed.
|
|
type ExplainView struct {
|
|
Model string `json:"model"`
|
|
// Placed is true when the model has replicas, in which case there is
|
|
// nothing to explain.
|
|
Placed bool `json:"placed"`
|
|
// Rejections is empty when placement was never attempted, which is itself
|
|
// the answer: the model is satisfied, or the reconciler has not run yet.
|
|
Rejections []RejectionView `json:"rejections,omitempty"`
|
|
}
|
|
|
|
// RejectionView is one rejected candidate.
|
|
type RejectionView struct {
|
|
Profile string `json:"profile"`
|
|
Pool string `json:"pool,omitempty"`
|
|
Node string `json:"node,omitempty"`
|
|
Reason string `json:"reason"`
|
|
}
|
|
|
|
// --- handlers ---
|
|
|
|
func (a *API) listNodes(w http.ResponseWriter, _ *http.Request) {
|
|
writeJSON(w, http.StatusOK, a.Nodes())
|
|
}
|
|
|
|
// Nodes builds the node view.
|
|
//
|
|
// Exported because the dashboard renders the same data server-side. Going back
|
|
// out through HTTP to fetch what is already in this process would add a way for
|
|
// the two to disagree.
|
|
func (a *API) Nodes() []NodeView {
|
|
replicasOn := map[string]int{}
|
|
for _, rep := range a.ctl.Reconciler().Replicas() {
|
|
for _, n := range rep.NodeIDs() {
|
|
replicasOn[n]++
|
|
}
|
|
}
|
|
|
|
inv := a.ctl.Inventory()
|
|
out := make([]NodeView, 0)
|
|
for _, n := range inv.Nodes() {
|
|
v := NodeView{
|
|
ID: n.ID, Hostname: n.Hostname, MgmtAddr: n.MgmtAddr,
|
|
State: n.State.String(), Labels: n.Labels,
|
|
GPUTotal: len(n.GPUs), GPUFree: len(inv.FreeGPUs(n.ID)),
|
|
MemoryGB: n.MemoryTotalMB / 1024,
|
|
Fabrics: n.FabricNames(), AgentVersion: n.AgentVersion,
|
|
Replicas: replicasOn[n.ID],
|
|
}
|
|
seen := map[string]struct{}{}
|
|
for _, g := range n.GPUs {
|
|
if _, dup := seen[g.Class()]; !dup {
|
|
seen[g.Class()] = struct{}{}
|
|
v.Classes = append(v.Classes, g.Class())
|
|
}
|
|
if !g.Health.OK() {
|
|
v.Unhealthy = append(v.Unhealthy,
|
|
fmt.Sprintf("gpu %d: %s", g.Index, g.Health.Problems[0]))
|
|
}
|
|
}
|
|
if !n.LastSeen.IsZero() {
|
|
v.LastSeenAgo = time.Since(n.LastSeen).Round(time.Second).String()
|
|
}
|
|
out = append(out, v)
|
|
}
|
|
return out
|
|
}
|
|
|
|
func (a *API) listModels(w http.ResponseWriter, _ *http.Request) {
|
|
writeJSON(w, http.StatusOK, a.Models())
|
|
}
|
|
|
|
// Models builds the catalog view.
|
|
func (a *API) Models() []ModelView {
|
|
ready, total := map[string]int{}, map[string]int{}
|
|
for _, rep := range a.ctl.Reconciler().Replicas() {
|
|
total[rep.Model]++
|
|
if rep.State == reconciler.Ready {
|
|
ready[rep.Model]++
|
|
}
|
|
}
|
|
|
|
out := make([]ModelView, 0)
|
|
for _, m := range a.ctl.Catalog().Models() {
|
|
v := ModelView{
|
|
Name: m.Name, Aliases: m.Aliases, Source: m.Source,
|
|
Min: m.Scaling.Min, Max: m.Scaling.Max,
|
|
Ready: ready[m.Name], Total: total[m.Name],
|
|
Fallbacks: m.Fallbacks,
|
|
}
|
|
for _, p := range m.Profiles {
|
|
v.Profiles = append(v.Profiles, p.Name)
|
|
}
|
|
out = append(out, v)
|
|
}
|
|
return out
|
|
}
|
|
|
|
func (a *API) listReplicas(w http.ResponseWriter, r *http.Request) {
|
|
writeJSON(w, http.StatusOK, a.Replicas(r.URL.Query().Get("model")))
|
|
}
|
|
|
|
// Replicas builds the replica view, optionally filtered to one model.
|
|
func (a *API) Replicas(model string) []ReplicaView {
|
|
out := make([]ReplicaView, 0)
|
|
for _, rep := range a.ctl.Reconciler().Replicas() {
|
|
if model != "" && rep.Model != model {
|
|
continue
|
|
}
|
|
v := ReplicaView{
|
|
ID: rep.ID, Model: rep.Model, Profile: rep.Profile,
|
|
State: rep.State.String(), Pool: rep.Pool, Fabric: rep.Fabric,
|
|
Endpoint: rep.Endpoint, Err: rep.Err,
|
|
AgeSec: int64(time.Since(rep.CreatedAt).Seconds()),
|
|
IdleSec: -1,
|
|
}
|
|
if rep.LastUsedAt.IsZero() {
|
|
v.IdleSec = -1
|
|
} else {
|
|
v.IdleSec = int64(time.Since(rep.LastUsedAt).Seconds())
|
|
}
|
|
for _, s := range rep.Shards {
|
|
v.Shards = append(v.Shards, ShardView{
|
|
ID: s.ID, Node: s.NodeID, Rank: s.Rank,
|
|
GPUs: s.GPUIndices, State: s.State.String(), Err: s.Err,
|
|
})
|
|
}
|
|
out = append(out, v)
|
|
}
|
|
return out
|
|
}
|
|
|
|
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})
|
|
}
|