llmgw/internal/control/adminapi/api.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"
	"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
	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/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"`
	// 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"`
	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"`
	// 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 _, 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, 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 {
			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)
}

// 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})
}