CI / test (push) Failing after 50s
ctl talks to a management API on the admin listener, not 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. - nodes surfaces unhealthy GPUs and why they are excluded. That is usually the answer when a node looks idle but nothing will schedule onto it, and it was invisible before. - explain-placement prints every candidate considered and why it was rejected. - replicas shows the shards a replica is made of, with per-shard errors. - cordon, uncordon and drain are one endpoint with different targets, because the difference between cordon and drain is exactly what an operator needs to be sure of before running either. Draining had to be fixed first to make ctl drain worth having. A drained replica was stopped on the next pass, which cuts in-flight requests — the thing draining exists to avoid. Now a node marked Draining has its replicas drained, the model falls below its desired count, a replacement is placed elsewhere, and the original is stopped only once the replacement is serving. A grace period bounds it, because draining the only node that can host a model would otherwise never finish and the node could never be taken out of service. Verified against a two-node cluster of real binaries: draining moved a replica to the surviving node and traffic kept returning 200 throughout. ctl logs is deliberately absent. It needs a request/response pair the agent protocol does not have, and the failure path already ships an engine log tail with every status report — so the case that matters most is covered, and reading a healthy shard's output on demand is not worth a protocol change yet. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
284 lines
8.5 KiB
Go
284 lines
8.5 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"`
|
|
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) {
|
|
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)),
|
|
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)
|
|
}
|
|
writeJSON(w, http.StatusOK, out)
|
|
}
|
|
|
|
func (a *API) listModels(w http.ResponseWriter, _ *http.Request) {
|
|
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)
|
|
}
|
|
writeJSON(w, http.StatusOK, out)
|
|
}
|
|
|
|
func (a *API) listReplicas(w http.ResponseWriter, r *http.Request) {
|
|
model := r.URL.Query().Get("model")
|
|
|
|
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)
|
|
}
|
|
writeJSON(w, http.StatusOK, 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
|
|
}
|
|
|
|
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,
|
|
})
|
|
}
|
|
}
|
|
writeJSON(w, http.StatusOK, 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})
|
|
}
|