Placement had no idea how much memory a node has or a model needs. A CPU profile has no card to be free, so eligibleNodes accepted every ready node: Qwen2.5-Coder 14B, whose working set is about 24 GB, would have been placed just as happily on one of the 8 GB nodes, which reports itself perfectly healthy right up until the OOM killer arrives. node_selector was yesterday's stopgap and it is the operator restating by hand something the system can know. Agents now report MemTotal, profiles declare min_memory_gb, and placement filters on it, naming what was short — "has 8 GB of memory, needs 24" is actionable where "no candidates were considered" is not. The label selector stays, because "this node is for coding models" is a real thing to want to say that memory does not express. Two deliberate non-exclusions. A profile with no requirement fits anywhere, which is every profile that exists today. And a node that reports no memory is still usable: an agent older than the field says nothing, and letting a control plane upgrade quietly empty half a fleet is a worse failure than the one this prevents. ctl nodes grows a MEM column showing "-" rather than 0 for a node that did not say, because an old agent is not a small machine. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
315 lines
9.6 KiB
Go
315 lines
9.6 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/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"`
|
|
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()),
|
|
}
|
|
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) 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})
|
|
}
|