Ilker ManapandClaude Opus 5.5
57739d736a
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
1810 lines
65 KiB
Go
1810 lines
65 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"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"net/url"
|
|
"slices"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
"unicode"
|
|
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/control/modelserve"
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/gguf"
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/modelplan"
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/release"
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/safedial"
|
|
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery"
|
|
"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/fabrics"
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory"
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/control/nvlink"
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler"
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/control/topology"
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/store"
|
|
)
|
|
|
|
// API serves the management endpoints.
|
|
type API struct {
|
|
// releases is where release binaries are fetched from, for upgrades.
|
|
releases string
|
|
// storeStatus reads a replicated store's members; nil for one database.
|
|
storeStatus func() ([]StoreMember, error)
|
|
// serve runs registry-to-replica jobs; nil without a mirror and a store.
|
|
serve *modelserve.Server
|
|
// history is the Prometheus the console's charts read from.
|
|
history History
|
|
// audit reads the audit log.
|
|
audit AuditReader
|
|
// guard is the guardrail policies' library and attachments.
|
|
guard *GuardAdmin
|
|
// revoke records a removed node, so its certificate is refused.
|
|
revoke func(ctx context.Context, nodeID string, at time.Time) error
|
|
|
|
models Models
|
|
capacity Capacity
|
|
usage Usage
|
|
quotas Quotas
|
|
ctl *control.Controller
|
|
}
|
|
|
|
// New builds the API over a controller.
|
|
func New(ctl *control.Controller) *API { return &API{ctl: ctl} }
|
|
|
|
// SetRevoker has removing a node revoke its certificate.
|
|
func (a *API) SetRevoker(f func(ctx context.Context, nodeID string, at time.Time) error) {
|
|
a.revoke = f
|
|
}
|
|
|
|
// 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("POST /api/v1/models/{name}/resources", a.setResources)
|
|
a.registerQuotas(mux)
|
|
a.registerGuard(mux)
|
|
mux.HandleFunc("GET /api/v1/audit", a.auditEvents)
|
|
mux.HandleFunc("GET /api/v1/usage/export", a.usageExport)
|
|
mux.HandleFunc("POST /api/v1/models/{name}/adapters", a.setAdapters)
|
|
mux.HandleFunc("GET /api/v1/models/{name}/target", a.target)
|
|
mux.HandleFunc("POST /api/v1/models/{name}/class", func(w http.ResponseWriter, r *http.Request) {
|
|
if a.models == nil {
|
|
writeError(w, http.StatusNotImplemented, "this control plane has no store")
|
|
return
|
|
}
|
|
var req struct {
|
|
Class string `json:"class"`
|
|
}
|
|
if err := json.NewDecoder(io.LimitReader(r.Body, 1<<12)).Decode(&req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "the request is not valid JSON: "+err.Error())
|
|
return
|
|
}
|
|
m, err := a.models.SetClass(r.Context(), r.PathValue("name"), req.Class, actorOf(r))
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, err.Error())
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]string{"model": m.Name, "class": m.EffectiveClass()})
|
|
})
|
|
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/fabrics", func(w http.ResponseWriter, _ *http.Request) {
|
|
writeJSON(w, http.StatusOK, a.Fabrics())
|
|
})
|
|
mux.HandleFunc("GET /api/v1/nvlink", func(w http.ResponseWriter, _ *http.Request) {
|
|
writeJSON(w, http.StatusOK, a.NVLink())
|
|
})
|
|
mux.HandleFunc("GET /api/v1/fabrics/{id}/topology", func(w http.ResponseWriter, r *http.Request) {
|
|
t, ok := a.Topology(r.PathValue("id"))
|
|
if !ok {
|
|
http.NotFound(w, r)
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, t)
|
|
})
|
|
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)
|
|
mux.HandleFunc("POST /api/v1/nodes/{id}/upgrade", a.upgradeNode)
|
|
mux.HandleFunc("POST /api/v1/nodes/{id}/engines", a.setNodeEngine)
|
|
mux.HandleFunc("DELETE /api/v1/nodes/{id}", a.removeNode)
|
|
mux.HandleFunc("POST /api/v1/fabric-test", a.fabricTest)
|
|
mux.HandleFunc("GET /api/v1/nodes/{id}/logs", a.nodeLogs)
|
|
mux.HandleFunc("POST /api/v1/restart", a.restart)
|
|
mux.HandleFunc("POST /api/v1/models/serve", a.serveModel)
|
|
mux.HandleFunc("GET /api/v1/jobs/{id}", a.job)
|
|
mux.HandleFunc("GET /api/v1/jobs", func(w http.ResponseWriter, _ *http.Request) {
|
|
out := []modelserve.Job{}
|
|
if a.serve != nil {
|
|
out = a.serve.Jobs()
|
|
}
|
|
writeJSON(w, http.StatusOK, out)
|
|
})
|
|
mux.HandleFunc("GET /api/v1/updates", a.listUpdates)
|
|
mux.HandleFunc("POST /api/v1/updates/check", a.checkUpdates)
|
|
mux.HandleFunc("POST /api/v1/models/{name}/update", a.updateModel)
|
|
mux.HandleFunc("POST /api/v1/models/{name}/rollback", a.rollbackModel)
|
|
mux.HandleFunc("POST /api/v1/models/{name}/origin", a.setOrigin)
|
|
mux.HandleFunc("GET /api/v1/plan", a.planModel)
|
|
mux.HandleFunc("POST /api/v1/metrics/range", a.metricsBatch)
|
|
mux.HandleFunc("GET /api/v1/requests/{id}", a.request)
|
|
mux.HandleFunc("POST /api/v1/models/{name}/retry", a.retryModel)
|
|
mux.HandleFunc("GET /api/v1/models/{name}/headroom", func(w http.ResponseWriter, r *http.Request) {
|
|
limit, _ := strconv.Atoi(r.URL.Query().Get("limit"))
|
|
v, ok := a.Headroom(r.PathValue("name"), limit)
|
|
if !ok {
|
|
http.Error(w, "no such model", http.StatusNotFound)
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, v)
|
|
})
|
|
}
|
|
|
|
// --- 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"`
|
|
// CPUCount is the machine's CPUs, 0 when the agent did not report them.
|
|
CPUCount int `json:"cpu_count,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"`
|
|
// EngineBuilds are the builds chosen for this node, by engine: "auto"
|
|
// (its hardware decides), a build ("cu129"), or "none". Only on a node
|
|
// with GPUs, the only kind with a choice to make.
|
|
EngineBuilds map[string]string `json:"engine_builds,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"`
|
|
// Features are the requests the agent answers beyond the usual:
|
|
// "fabric-test". Absent is an agent too old to say.
|
|
Features []string `json:"features,omitempty"`
|
|
Replicas int `json:"replicas"`
|
|
|
|
// Load is how busy the machine is, absent when its agent has not
|
|
// measured it lately — which is unknown, not idle.
|
|
Load *LoadView `json:"load,omitempty"`
|
|
// IBPorts are the node's InfiniBand and RoCE ports at the last heartbeat.
|
|
IBPorts []IBPortView `json:"ib_ports,omitempty"`
|
|
// GPUs are the cards, with how they are wired: NVLink peers, the
|
|
// adapter nearest each, and the NVLink domain each has joined.
|
|
GPUs []GPUView `json:"gpus,omitempty"`
|
|
// NVLinkDomain is the multi-node domain every card has joined, if any.
|
|
NVLinkDomain string `json:"nvlink_domain,omitempty"`
|
|
IMEX string `json:"imex,omitempty"`
|
|
IMEXChannels int `json:"imex_channels,omitempty"`
|
|
}
|
|
|
|
// GPUView is one card as a node's page shows it.
|
|
type GPUView struct {
|
|
Index int `json:"index"`
|
|
Name string `json:"name"`
|
|
// ComputeCapability is what decides which engine builds run on the card:
|
|
// CUDA 13 builds need 7.5, so a V100 (7.0) runs only older ones.
|
|
ComputeCapability string `json:"compute_capability,omitempty"`
|
|
NUMA int `json:"numa"`
|
|
NVLinkPeers []int `json:"nvlink_peers,omitempty"`
|
|
// NearestNIC is the adapter closest to the card, and how close.
|
|
NearestNIC string `json:"nearest_nic,omitempty"`
|
|
NICPath string `json:"nic_path,omitempty"`
|
|
FabricState string `json:"fabric_state,omitempty"`
|
|
Domain string `json:"domain,omitempty"`
|
|
Bandwidth string `json:"bandwidth,omitempty"`
|
|
}
|
|
|
|
// IBPortView is one port as a page shows it.
|
|
type IBPortView struct {
|
|
HCA string `json:"hca"`
|
|
Port int `json:"port"`
|
|
State string `json:"state"`
|
|
PhysState string `json:"phys_state,omitempty"`
|
|
RateGbps int `json:"rate_gbps"`
|
|
LinkLayer string `json:"link_layer,omitempty"`
|
|
Active bool `json:"active"`
|
|
// Fabric is the declared fabric this port's adapter carries, if any.
|
|
Fabric string `json:"fabric,omitempty"`
|
|
// Physical is the fabric the port is physically on: its subnet manager,
|
|
// or its RoCE network. SMError says why that is not known.
|
|
Physical string `json:"physical,omitempty"`
|
|
SMError string `json:"sm_error,omitempty"`
|
|
// Errors are the error counters that are not zero, by name.
|
|
Errors map[string]uint64 `json:"errors,omitempty"`
|
|
TxBytes uint64 `json:"tx_bytes,omitempty"`
|
|
RxBytes uint64 `json:"rx_bytes,omitempty"`
|
|
}
|
|
|
|
// LoadView is a node's CPU and memory in use, at its last heartbeat.
|
|
type LoadView struct {
|
|
CPUPercent float64 `json:"cpu_percent"`
|
|
CPUs int `json:"cpus,omitempty"`
|
|
Load1 float64 `json:"load1"`
|
|
Load5 float64 `json:"load5"`
|
|
Load15 float64 `json:"load15"`
|
|
MemUsed int64 `json:"memory_used_bytes"`
|
|
MemTotal int64 `json:"memory_total_bytes"`
|
|
// MemPercent is used over used plus available.
|
|
MemPercent float64 `json:"memory_percent"`
|
|
AgeSec int64 `json:"age_seconds"`
|
|
// What the node's sensors say; zero is not measured.
|
|
CPUTempC float64 `json:"cpu_temp_c,omitempty"`
|
|
CPUPowerW float64 `json:"cpu_power_w,omitempty"`
|
|
InletTempC float64 `json:"inlet_temp_c,omitempty"`
|
|
GPUs []inventory.GPUSensor `json:"gpu_sensors,omitempty"`
|
|
}
|
|
|
|
// loadFresh is how old a measurement may be and still be shown as current.
|
|
const loadFresh = time.Minute
|
|
|
|
func loadView(l *inventory.Load, now time.Time) *LoadView {
|
|
if l == nil || now.Sub(l.At) > loadFresh {
|
|
return nil
|
|
}
|
|
v := &LoadView{
|
|
CPUPercent: l.CPUPercent, CPUs: l.CPUCount,
|
|
Load1: l.Load1, Load5: l.Load5, Load15: l.Load15,
|
|
MemUsed: l.MemUsedBytes, MemTotal: l.MemUsedBytes + l.MemAvailBytes,
|
|
AgeSec: int64(now.Sub(l.At).Seconds()),
|
|
CPUTempC: l.CPUTempC, CPUPowerW: l.CPUPowerW, InletTempC: l.InletTempC, GPUs: l.GPUs,
|
|
}
|
|
if v.MemTotal > 0 {
|
|
v.MemPercent = float64(v.MemUsed) * 100 / float64(v.MemTotal)
|
|
}
|
|
return v
|
|
}
|
|
|
|
// 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"`
|
|
// Desired is how many it should have now: Min, or for an on-demand
|
|
// model one while it is being asked for and none otherwise.
|
|
Desired int `json:"desired_replicas"`
|
|
// Failures counts replicas that failed to start in a row.
|
|
Failures int `json:"failures,omitempty"`
|
|
// Parked is set once Failures reached the limit; the model is not retried
|
|
// until its definition changes or `ctl retry` is run. Reason is the last
|
|
// error.
|
|
Parked bool `json:"parked,omitempty"`
|
|
Reason string `json:"reason,omitempty"`
|
|
// Unplaceable is set when it is short and the last attempt found no node.
|
|
Unplaceable bool `json:"unplaceable,omitempty"`
|
|
Fallbacks []string `json:"fallbacks,omitempty"`
|
|
Profiles []string `json:"profiles"`
|
|
// MemoryGB and CPUCores are what one replica reserves, from its first
|
|
// profile; MeasuredMemoryGB is the most one of its replicas uses now, 0
|
|
// when none is measured.
|
|
MemoryGB int `json:"memory_gb,omitempty"`
|
|
CPUCores int `json:"cpu_cores,omitempty"`
|
|
MeasuredMemoryGB float64 `json:"measured_memory_gb,omitempty"`
|
|
// Warnings are what makes its reservation wrong: nothing declared where
|
|
// placement needs a figure, or less declared than it uses.
|
|
Warnings []string `json:"warnings,omitempty"`
|
|
// Class is its capacity class; Preempted the replicas it lost to a higher
|
|
// class; HeldFor the model it is giving room to now, if any.
|
|
Class string `json:"class"`
|
|
Owner string `json:"owner,omitempty"`
|
|
// External is where a model served outside the cluster goes, provider
|
|
// host and its name for the model; empty for one served here.
|
|
External string `json:"external,omitempty"`
|
|
// Adapters are the LoRA adapters it serves.
|
|
Adapters []catalog.Adapter `json:"adapters,omitempty"`
|
|
Preempted int `json:"preempted,omitempty"`
|
|
HeldFor string `json:"held_for,omitempty"`
|
|
// Origin is the registry repository its weights came from, and at which
|
|
// commit; Update what that repository had when it was last asked.
|
|
Origin *OriginView `json:"origin,omitempty"`
|
|
Update *modelserve.UpdateInfo `json:"update,omitempty"`
|
|
}
|
|
|
|
// OriginView is where a model's weights were published.
|
|
type OriginView struct {
|
|
Repo string `json:"repo"`
|
|
Revision string `json:"revision"`
|
|
// Previous is the commit the last update replaced, which it can go back
|
|
// to.
|
|
Previous string `json:"previous,omitempty"`
|
|
}
|
|
|
|
// ReplicaView is a replica as `ctl replicas` shows it.
|
|
type ReplicaView struct {
|
|
ID string `json:"id"`
|
|
// EngineKey is the key this replica's engine demands, for `ctl bench`
|
|
// and anyone else holding the admin listener who needs to reach an engine
|
|
// directly. The client-facing pages never render it.
|
|
EngineKey string `json:"engine_key,omitempty"`
|
|
// Restart is "queued" or "replacing" while an operator's restart of this
|
|
// replica is under way.
|
|
Restart string `json:"restart,omitempty"`
|
|
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"`
|
|
// CPUCores and MemoryBytes are the engine's use of its node, from the
|
|
// node's last heartbeat; zero when not measured.
|
|
CPUCores float64 `json:"cpu_cores,omitempty"`
|
|
MemoryBytes int64 `json:"memory_bytes,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"`
|
|
}
|
|
|
|
// HeadroomView is how many more replicas of a model fit now, and where.
|
|
type HeadroomView struct {
|
|
Model string `json:"model"`
|
|
// Current is the replicas holding capacity: starting or ready.
|
|
Current int `json:"current"`
|
|
// More is how many more placement would find room for now.
|
|
More int `json:"more"`
|
|
// Nodes is where each of those would go.
|
|
Nodes [][]string `json:"nodes,omitempty"`
|
|
// Limited is set when More stopped at the limit asked rather than at the
|
|
// cluster's edge: there may be room for more.
|
|
Limited bool `json:"limited,omitempty"`
|
|
// Blocked says why one more does not fit, a line per node considered.
|
|
Blocked []string `json:"blocked,omitempty"`
|
|
// TokPerSec is what one more replica was measured to serve; 0 unmeasured.
|
|
TokPerSec float64 `json:"tok_per_sec,omitempty"`
|
|
// ByPreempting is how many more would fit by stopping replicas this
|
|
// model's class may take; Preempts the models those belong to.
|
|
ByPreempting int `json:"by_preempting,omitempty"`
|
|
Preempts []string `json:"preempts,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, CPUCount: n.CPUCount,
|
|
Fabrics: n.FabricNames(), AgentVersion: n.AgentVersion, Features: n.Features,
|
|
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 ch := a.ctl.EngineChoices(); ch != nil && len(n.GPUs) > 0 {
|
|
chosen := ch.For(n.ID)
|
|
v.EngineBuilds = map[string]string{"vllm": "auto"}
|
|
for e, b := range chosen {
|
|
v.EngineBuilds[e] = b
|
|
}
|
|
}
|
|
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()
|
|
}
|
|
v.Load = loadView(n.Load, time.Now())
|
|
v.NVLinkDomain, v.IMEX, v.IMEXChannels = n.NVLinkDomain(), n.IMEX.State, n.IMEX.Channels
|
|
for _, g := range n.GPUs {
|
|
gv := GPUView{Index: g.Index, Name: g.Name, ComputeCapability: g.ComputeCapability,
|
|
NUMA: g.NUMANode, NVLinkPeers: g.NVLinkPeers}
|
|
for hca, rel := range g.NICPaths {
|
|
if gv.NearestNIC == "" || discovery.PathRank(rel) < discovery.PathRank(gv.NICPath) ||
|
|
(discovery.PathRank(rel) == discovery.PathRank(gv.NICPath) && hca < gv.NearestNIC) {
|
|
gv.NearestNIC, gv.NICPath = hca, rel
|
|
}
|
|
}
|
|
if f := g.Fabric; f != nil {
|
|
gv.FabricState, gv.Domain, gv.Bandwidth = f.State, f.Domain(), f.Bandwidth
|
|
}
|
|
v.GPUs = append(v.GPUs, gv)
|
|
}
|
|
fabricOf := map[string]string{}
|
|
for _, f := range n.Fabrics {
|
|
for _, h := range f.HCAs {
|
|
fabricOf[h] = f.Fabric
|
|
}
|
|
}
|
|
for _, p := range n.IBPorts {
|
|
pv := IBPortView{HCA: p.HCA, Port: p.Port, State: p.State, PhysState: p.PhysState,
|
|
RateGbps: p.RateGbps, LinkLayer: p.LinkLayer, Active: p.Active(), Fabric: fabricOf[p.HCA],
|
|
TxBytes: p.Counters["port_xmit_data"], RxBytes: p.Counters["port_rcv_data"], SMError: p.SMError}
|
|
switch {
|
|
case p.RoCE() && p.Network != "":
|
|
pv.Physical = "RoCE " + p.Network
|
|
case p.SMGUID != "":
|
|
pv.Physical = fmt.Sprintf("SM %s (LID %d)", p.SMGUID, p.SMLID)
|
|
}
|
|
for name, c := range p.Counters {
|
|
if c > 0 && name != "port_xmit_data" && name != "port_rcv_data" {
|
|
if pv.Errors == nil {
|
|
pv.Errors = map[string]uint64{}
|
|
}
|
|
pv.Errors[name] = c
|
|
}
|
|
}
|
|
v.IBPorts = append(v.IBPorts, pv)
|
|
}
|
|
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]++
|
|
}
|
|
}
|
|
|
|
health := map[string]reconciler.ModelHealth{}
|
|
for _, h := range a.ctl.Reconciler().ModelHealth() {
|
|
health[h.Model] = h
|
|
}
|
|
// The most one replica of each model uses, summed over its shards.
|
|
measured := map[string]int64{}
|
|
for _, rep := range a.Replicas("") {
|
|
var b int64
|
|
for _, sh := range rep.Shards {
|
|
b += sh.MemoryBytes
|
|
}
|
|
measured[rep.Model] = max(measured[rep.Model], b)
|
|
}
|
|
var updates map[string]modelserve.UpdateInfo
|
|
if a.serve != nil {
|
|
updates = a.serve.Updates()
|
|
}
|
|
out := make([]ModelView, 0)
|
|
for _, m := range a.ctl.Catalog().Models() {
|
|
h := health[m.Name]
|
|
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,
|
|
Desired: h.Desired, Failures: h.Failures,
|
|
Parked: h.Parked, Reason: h.Reason, Unplaceable: h.Unplaceable,
|
|
Class: m.EffectiveClass(), Owner: m.Owner, Preempted: h.Preempted, HeldFor: h.HeldFor,
|
|
External: externalLabel(m), Adapters: m.Adapters,
|
|
}
|
|
for _, p := range m.Profiles {
|
|
v.Profiles = append(v.Profiles, p.Name)
|
|
}
|
|
if len(m.Profiles) > 0 {
|
|
v.MemoryGB, v.CPUCores = m.Profiles[0].MinMemoryGB, m.Profiles[0].CPUCores
|
|
}
|
|
if o := m.Origin; o != nil {
|
|
v.Origin = &OriginView{Repo: o.Repo, Revision: o.Revision}
|
|
if o.Previous != nil {
|
|
v.Origin.Previous = o.Previous.Revision
|
|
}
|
|
}
|
|
if u, ok := updates[m.Name]; ok {
|
|
v.Update = &u
|
|
}
|
|
v.MeasuredMemoryGB = float64(measured[m.Name]) / (1 << 30)
|
|
v.Warnings = reservationWarnings(m, v.MeasuredMemoryGB)
|
|
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)
|
|
loads := map[string]*inventory.Load{}
|
|
for _, n := range a.ctl.Inventory().Nodes() {
|
|
if n.Load != nil && time.Since(n.Load.At) <= loadFresh {
|
|
loads[n.ID] = n.Load
|
|
}
|
|
}
|
|
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,
|
|
EngineKey: a.ctl.EngineKey(rep.ID),
|
|
Restart: a.ctl.Reconciler().RestartState(rep.ID),
|
|
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
|
|
}
|
|
}
|
|
if l := loads[s.NodeID]; l != nil {
|
|
if use, ok := l.Shards[s.ID]; ok {
|
|
sv.CPUCores, sv.MemoryBytes = use.CPUCores, use.RSSBytes
|
|
}
|
|
}
|
|
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)
|
|
Resources(ctx context.Context, name, profile string, memoryGB, cpuCores int, by string) (catalog.Model, error)
|
|
SetClass(ctx context.Context, name, class, by string) (catalog.Model, error)
|
|
SetOwner(ctx context.Context, name, owner, by string) (catalog.Model, error)
|
|
SetAdapters(ctx context.Context, name string, adapters []catalog.Adapter, by string) (catalog.Model, error)
|
|
}
|
|
|
|
// ResourcesRequest sets what a model's profiles reserve of a node.
|
|
type ResourcesRequest struct {
|
|
// Profile is one profile's name; empty is every profile.
|
|
Profile string `json:"profile,omitempty"`
|
|
MemoryGB int `json:"memory_gb"`
|
|
CPUCores int `json:"cpu_cores"`
|
|
}
|
|
|
|
// 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 }
|
|
|
|
// SetStoreStatus supplies the state of a replicated store, for the summary.
|
|
func (a *API) SetStoreStatus(f func() ([]StoreMember, error)) { a.storeStatus = f }
|
|
|
|
// 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: actorOf(r),
|
|
}); 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, actorOf(r)); 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},
|
|
actorOf(r))
|
|
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(),
|
|
})
|
|
}
|
|
|
|
// setResources sets a model's reserved memory and cores. Placed replicas keep
|
|
// what they were started with until they are restarted.
|
|
func (a *API) setResources(w http.ResponseWriter, r *http.Request) {
|
|
if a.models == nil {
|
|
writeError(w, http.StatusNotImplemented, "this control plane has no store")
|
|
return
|
|
}
|
|
var req ResourcesRequest
|
|
if err := json.NewDecoder(io.LimitReader(r.Body, 1<<16)).Decode(&req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "the request is not valid JSON: "+err.Error())
|
|
return
|
|
}
|
|
m, err := a.models.Resources(r.Context(), r.PathValue("name"), req.Profile, req.MemoryGB, req.CPUCores, actorOf(r))
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, err.Error())
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, m)
|
|
}
|
|
|
|
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.
|
|
// Headroom tries placing up to limit more replicas of a model, on a copy of
|
|
// the inventory, and says how many fit. A limit of zero or less takes 16.
|
|
func (a *API) Headroom(model string, limit int) (HeadroomView, bool) {
|
|
if limit <= 0 || limit > 64 {
|
|
limit = 16
|
|
}
|
|
h, ok := a.ctl.Reconciler().Headroom(model, limit)
|
|
if !ok {
|
|
return HeadroomView{}, false
|
|
}
|
|
return HeadroomView{Model: model, Current: h.Current, More: h.More, Nodes: h.Nodes,
|
|
Limited: h.Limited, Blocked: h.Blocked, TokPerSec: h.TokPerSec,
|
|
ByPreempting: h.ByPreempting, Preempts: h.Preempts}, true
|
|
}
|
|
|
|
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.
|
|
// removeNode forgets a node that has gone for good: a machine retired or
|
|
// rebuilt under another name, which otherwise stays listed as lost until the
|
|
// control plane restarts.
|
|
//
|
|
// Only one that is not there: lost, or unreachable with nothing placed on
|
|
// it. A connected node removed here would register again on its next
|
|
// heartbeat, and one with replicas would take them out of the reconciler's
|
|
// sight while they may still be running.
|
|
func (a *API) removeNode(w http.ResponseWriter, r *http.Request) {
|
|
id := r.PathValue("id")
|
|
inv := a.ctl.Inventory()
|
|
n, ok := inv.Get(id)
|
|
if !ok {
|
|
writeError(w, http.StatusNotFound, fmt.Sprintf("no node %q", id))
|
|
return
|
|
}
|
|
switch {
|
|
case n.State == inventory.Lost:
|
|
case n.State == inventory.Unreachable && len(inv.Allocations(id)) == 0:
|
|
default:
|
|
writeError(w, http.StatusConflict, fmt.Sprintf(
|
|
"%s is %s: stop its agent (systemctl stop llmgw-agent) and remove it once it is lost, a minute later", id, n.State))
|
|
return
|
|
}
|
|
// Revoked first: a node removed but still holding a certificate the
|
|
// agent port accepts could connect again and be given work.
|
|
if a.revoke != nil {
|
|
if err := a.revoke(r.Context(), id, time.Now()); err != nil {
|
|
writeError(w, http.StatusInternalServerError, "revoking the node's certificate: "+err.Error())
|
|
return
|
|
}
|
|
}
|
|
inv.Remove(id)
|
|
a.ctl.Reconciler().Notify()
|
|
writeJSON(w, http.StatusOK, map[string]string{"id": id, "removed": "true"})
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
// Fabrics is the physical fabrics the adapters are on, against the names
|
|
// the nodes gave them.
|
|
func (a *API) Fabrics() fabrics.Report {
|
|
inv := a.ctl.Inventory()
|
|
return fabrics.Analyze(inv.Nodes(), inv.SMChanges(), time.Now())
|
|
}
|
|
|
|
// TopologyView is one fabric's map, with the cluster's own adapters on it
|
|
// live: their state, rate and traffic from the last heartbeat.
|
|
type TopologyView struct {
|
|
ID string `json:"id"`
|
|
Key string `json:"key"`
|
|
Label string `json:"label"`
|
|
Declared []string `json:"declared"`
|
|
ScannedAt time.Time `json:"scanned_at"`
|
|
ScannedBy string `json:"scanned_by,omitempty"`
|
|
TookMs int64 `json:"took_ms"`
|
|
Err string `json:"error,omitempty"`
|
|
Map *topology.View `json:"map,omitempty"`
|
|
Live map[string]Live `json:"live"`
|
|
// Problems are the fabric's own, from the analysis of its identity and
|
|
// subnet managers.
|
|
Problems []fabrics.Problem `json:"problems,omitempty"`
|
|
SM *fabrics.SM `json:"sm,omitempty"`
|
|
SMs []fabrics.SM `json:"sms,omitempty"`
|
|
}
|
|
|
|
// Live is one of the cluster's adapters as its node last reported it.
|
|
type Live struct {
|
|
Node string `json:"node"`
|
|
HCA string `json:"hca"`
|
|
State string `json:"state"`
|
|
Phys string `json:"phys_state"`
|
|
Gbps int `json:"gbps"`
|
|
TxBps float64 `json:"tx_bps"`
|
|
RxBps float64 `json:"rx_bps"`
|
|
Errors map[string]uint64 `json:"errors,omitempty"`
|
|
}
|
|
|
|
// FabricID is a physical fabric's key as an address can carry it.
|
|
func FabricID(key string) string {
|
|
return strings.NewReplacer(":", "-", "/", "_").Replace(key)
|
|
}
|
|
|
|
// Topologies lists every physical fabric with what is known of its map.
|
|
func (a *API) Topologies() []TopologyView {
|
|
rep := a.Fabrics()
|
|
scans := a.ctl.Inventory().Scans()
|
|
out := make([]TopologyView, 0, len(rep.Physical))
|
|
for _, ph := range rep.Physical {
|
|
t := TopologyView{ID: FabricID(ph.Key), Key: ph.Key, Label: ph.Label, Declared: ph.Declared}
|
|
if sc, ok := scans[ph.Key]; ok {
|
|
t.ScannedAt, t.ScannedBy, t.TookMs, t.Err = sc.At, sc.Node+" "+sc.Port, sc.Took.Milliseconds(), sc.Err
|
|
}
|
|
out = append(out, t)
|
|
}
|
|
return out
|
|
}
|
|
|
|
// Topology is one fabric's map, by the id Topologies gives it.
|
|
func (a *API) Topology(id string) (TopologyView, bool) {
|
|
rep := a.Fabrics()
|
|
var ph *fabrics.Physical
|
|
for i := range rep.Physical {
|
|
if FabricID(rep.Physical[i].Key) == id {
|
|
ph = &rep.Physical[i]
|
|
}
|
|
}
|
|
if ph == nil {
|
|
return TopologyView{}, false
|
|
}
|
|
t := TopologyView{ID: id, Key: ph.Key, Label: ph.Label, Declared: ph.Declared, Live: map[string]Live{},
|
|
SM: ph.SM, SMs: ph.SMs}
|
|
for _, p := range rep.Problems {
|
|
for _, d := range ph.Declared {
|
|
if p.Fabric == d {
|
|
t.Problems = append(t.Problems, p)
|
|
break
|
|
}
|
|
}
|
|
}
|
|
hosts := map[string]topology.Host{}
|
|
for _, n := range a.ctl.Inventory().Nodes() {
|
|
for _, p := range n.IBPorts {
|
|
if p.NodeGUID == "" {
|
|
continue
|
|
}
|
|
hosts[p.NodeGUID] = topology.Host{Node: n.ID, HCA: p.HCA}
|
|
l := Live{Node: n.ID, HCA: p.HCA, State: p.State, Phys: p.PhysState, Gbps: p.RateGbps,
|
|
TxBps: p.TxBps, RxBps: p.RxBps}
|
|
for name, c := range p.Counters {
|
|
if c > 0 && name != "port_xmit_data" && name != "port_rcv_data" {
|
|
if l.Errors == nil {
|
|
l.Errors = map[string]uint64{}
|
|
}
|
|
l.Errors[name] = c
|
|
}
|
|
}
|
|
t.Live[p.NodeGUID] = l
|
|
}
|
|
}
|
|
sc, ok := a.ctl.Inventory().Scans()[ph.Key]
|
|
if !ok {
|
|
return t, true
|
|
}
|
|
t.ScannedAt, t.ScannedBy, t.TookMs, t.Err = sc.At, sc.Node+" "+sc.Port, sc.Took.Milliseconds(), sc.Err
|
|
if sc.Output != "" {
|
|
f, err := topology.Parse(sc.Output)
|
|
if err != nil {
|
|
t.Err = "the map could not be read: " + err.Error()
|
|
return t, true
|
|
}
|
|
v := topology.Build(f, hosts)
|
|
t.Map = &v
|
|
}
|
|
return t, true
|
|
}
|
|
|
|
// NVLink is the cluster's multi-node NVLink domains, and what keeps any of
|
|
// their nodes out.
|
|
func (a *API) NVLink() nvlink.Report {
|
|
return nvlink.Analyze(a.ctl.Inventory().Nodes())
|
|
}
|
|
|
|
// RestartRequest names what to restart: replicas by id, every replica of some
|
|
// models, or — All — every ready replica in the cluster.
|
|
type RestartRequest struct {
|
|
Replicas []string `json:"replicas,omitempty"`
|
|
Models []string `json:"models,omitempty"`
|
|
All bool `json:"all,omitempty"`
|
|
// InPlace stops each replica before starting its replacement, rather
|
|
// than starting the replacement beside it first.
|
|
InPlace bool `json:"in_place,omitempty"`
|
|
}
|
|
|
|
// RestartResult lists the replicas queued.
|
|
type RestartResult struct {
|
|
Queued []string `json:"queued"`
|
|
}
|
|
|
|
func (a *API) restart(w http.ResponseWriter, r *http.Request) {
|
|
var req RestartRequest
|
|
if err := json.NewDecoder(io.LimitReader(r.Body, 1<<16)).Decode(&req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "the restart request is not valid JSON: "+err.Error())
|
|
return
|
|
}
|
|
ids := slices.Clone(req.Replicas)
|
|
for _, rep := range a.ctl.Reconciler().ReadyReplicas() {
|
|
if req.All || slices.Contains(req.Models, rep.Model) {
|
|
ids = append(ids, rep.ID)
|
|
}
|
|
}
|
|
if len(ids) == 0 {
|
|
writeError(w, http.StatusBadRequest, "nothing ready matches; name replicas, models, or all")
|
|
return
|
|
}
|
|
restart := a.ctl.Reconciler().Restart
|
|
if req.InPlace {
|
|
restart = a.ctl.Reconciler().RestartInPlace
|
|
}
|
|
if err := restart(ids...); err != nil {
|
|
writeError(w, http.StatusNotFound, err.Error())
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusAccepted, RestartResult{Queued: ids})
|
|
}
|
|
|
|
func (a *API) retryModel(w http.ResponseWriter, r *http.Request) {
|
|
name := r.PathValue("name")
|
|
if _, ok := a.ctl.Catalog().Resolve(name); !ok {
|
|
writeError(w, http.StatusNotFound, "no model called "+name)
|
|
return
|
|
}
|
|
was := a.ctl.Reconciler().Retry(name)
|
|
writeJSON(w, http.StatusOK, map[string]bool{"was_parked": was})
|
|
}
|
|
|
|
// requestFinder looks a request up by the id it was given. Asked of the usage
|
|
// source by type assertion, so one without it answers 501 rather than every
|
|
// source having to grow the method.
|
|
type requestFinder interface {
|
|
UsageByRequest(ctx context.Context, requestID string) (store.Usage, bool, error)
|
|
}
|
|
|
|
// RequestView is what one request went through.
|
|
type RequestView struct {
|
|
RequestID string `json:"request_id"`
|
|
At time.Time `json:"at"`
|
|
KeyID string `json:"key_id"`
|
|
UserID string `json:"user_id,omitempty"`
|
|
ProjectID string `json:"project_id"`
|
|
Model string `json:"model"`
|
|
ServedModel string `json:"served_model"`
|
|
Replica string `json:"replica,omitempty"`
|
|
Status int `json:"status"`
|
|
LatencyMs int64 `json:"latency_ms"`
|
|
QueueMs int64 `json:"queue_ms"`
|
|
TTFTMs int64 `json:"ttft_ms"`
|
|
PromptTokens int64 `json:"prompt_tokens"`
|
|
CompletionTokens int64 `json:"completion_tokens"`
|
|
CostMicros int64 `json:"cost_micros"`
|
|
// What the guardrails did, by name.
|
|
GuardAction string `json:"guard_action,omitempty"`
|
|
GuardPolicy string `json:"guard_policy,omitempty"`
|
|
GuardCategory string `json:"guard_category,omitempty"`
|
|
GuardMs int64 `json:"guard_ms,omitempty"`
|
|
}
|
|
|
|
func (a *API) request(w http.ResponseWriter, r *http.Request) {
|
|
f, ok := a.usage.(requestFinder)
|
|
if !ok {
|
|
writeError(w, http.StatusNotImplemented, "this control plane keeps no usage records")
|
|
return
|
|
}
|
|
u, found, err := f.UsageByRequest(r.Context(), r.PathValue("id"))
|
|
switch {
|
|
case err != nil:
|
|
writeError(w, http.StatusInternalServerError, err.Error())
|
|
return
|
|
case !found:
|
|
writeError(w, http.StatusNotFound, "no request with that id; records are written a few seconds after a request ends")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, RequestView{
|
|
RequestID: u.RequestID, At: u.At, KeyID: u.KeyID, UserID: u.UserID, ProjectID: u.ProjectID,
|
|
Model: u.Model, ServedModel: u.ServedModel, Replica: u.Replica, Status: u.Status,
|
|
LatencyMs: u.LatencyMs, QueueMs: u.QueueMs, TTFTMs: u.TTFTMs,
|
|
PromptTokens: u.PromptTokens, CompletionTokens: u.CompletionTokens, CostMicros: u.CostMicros,
|
|
GuardAction: u.GuardAction, GuardPolicy: u.GuardPolicy, GuardCategory: u.GuardCategory, GuardMs: u.GuardMs,
|
|
})
|
|
}
|
|
|
|
// SetReleases sets where upgrades fetch release binaries from.
|
|
func (a *API) SetReleases(base string) { a.releases = base }
|
|
|
|
// UpgradeRequest names the release a node should run.
|
|
type UpgradeRequest struct {
|
|
Version string `json:"version"`
|
|
}
|
|
|
|
func (a *API) upgradeNode(w http.ResponseWriter, r *http.Request) {
|
|
var req UpgradeRequest
|
|
if err := json.NewDecoder(io.LimitReader(r.Body, 1<<12)).Decode(&req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "the upgrade request is not valid JSON: "+err.Error())
|
|
return
|
|
}
|
|
if !release.Tag.MatchString(req.Version) {
|
|
writeError(w, http.StatusBadRequest, req.Version+" is not a release version")
|
|
return
|
|
}
|
|
u := a.ctl.Upgrader()
|
|
if u == nil {
|
|
writeError(w, http.StatusNotImplemented, "this control plane cannot upgrade nodes")
|
|
return
|
|
}
|
|
base := a.releases
|
|
if base == "" {
|
|
base = release.DefaultBase
|
|
}
|
|
if err := u.Upgrade(r.Context(), r.PathValue("id"), req.Version, base); err != nil {
|
|
writeError(w, http.StatusBadGateway, err.Error())
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]string{"node": r.PathValue("id"), "version": req.Version})
|
|
}
|
|
|
|
// SetServe supplies what serves a model from the registry.
|
|
func (a *API) SetServe(s *modelserve.Server) { a.serve = s }
|
|
|
|
func (a *API) serveModel(w http.ResponseWriter, r *http.Request) {
|
|
if a.serve == nil {
|
|
writeError(w, http.StatusNotImplemented,
|
|
"serving from the registry needs a mirror and a store, and this control plane lacks one")
|
|
return
|
|
}
|
|
var req modelserve.Request
|
|
if err := json.NewDecoder(io.LimitReader(r.Body, 1<<16)).Decode(&req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "the request is not valid JSON: "+err.Error())
|
|
return
|
|
}
|
|
req.By = actorOf(r)
|
|
job, err := a.serve.Start(req)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, err.Error())
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusAccepted, job)
|
|
}
|
|
|
|
func (a *API) job(w http.ResponseWriter, r *http.Request) {
|
|
if a.serve == nil {
|
|
writeError(w, http.StatusNotFound, "no such job")
|
|
return
|
|
}
|
|
job, ok := a.serve.Get(r.PathValue("id"))
|
|
if !ok {
|
|
writeError(w, http.StatusNotFound, "no such job; jobs are kept only until the control plane restarts")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, job)
|
|
}
|
|
|
|
func (a *API) listUpdates(w http.ResponseWriter, _ *http.Request) {
|
|
out := make([]modelserve.UpdateInfo, 0)
|
|
if a.serve != nil {
|
|
for _, u := range a.serve.Updates() {
|
|
out = append(out, u)
|
|
}
|
|
}
|
|
slices.SortFunc(out, func(x, y modelserve.UpdateInfo) int { return strings.Compare(x.Model, y.Model) })
|
|
writeJSON(w, http.StatusOK, out)
|
|
}
|
|
|
|
// CheckRequest asks the registry about one model now, or every one.
|
|
type CheckRequest struct {
|
|
Model string `json:"model,omitempty"`
|
|
}
|
|
|
|
func (a *API) checkUpdates(w http.ResponseWriter, r *http.Request) {
|
|
if a.serve == nil {
|
|
writeError(w, http.StatusNotImplemented, "updates need a mirror and a store, and this control plane lacks one")
|
|
return
|
|
}
|
|
var req CheckRequest
|
|
_ = json.NewDecoder(io.LimitReader(r.Body, 1<<12)).Decode(&req)
|
|
ctx, cancel := context.WithTimeout(r.Context(), 3*time.Minute)
|
|
defer cancel()
|
|
if req.Model != "" {
|
|
u, err := a.serve.CheckModel(ctx, req.Model)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, err.Error())
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, []modelserve.UpdateInfo{u})
|
|
return
|
|
}
|
|
a.serve.CheckUpdates(ctx)
|
|
a.listUpdates(w, r)
|
|
}
|
|
|
|
func (a *API) updateModel(w http.ResponseWriter, r *http.Request) {
|
|
if a.serve == nil {
|
|
writeError(w, http.StatusNotImplemented, "updates need a mirror and a store, and this control plane lacks one")
|
|
return
|
|
}
|
|
job, err := a.serve.StartUpdate(r.PathValue("name"), actorOf(r))
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, err.Error())
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusAccepted, job)
|
|
}
|
|
|
|
func (a *API) rollbackModel(w http.ResponseWriter, r *http.Request) {
|
|
if a.serve == nil {
|
|
writeError(w, http.StatusNotImplemented, "updates need a mirror and a store, and this control plane lacks one")
|
|
return
|
|
}
|
|
msg, err := a.serve.Rollback(r.Context(), r.PathValue("name"), actorOf(r))
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, err.Error())
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]string{"result": msg})
|
|
}
|
|
|
|
// OriginRequest names the repository a model's weights came from.
|
|
type OriginRequest struct {
|
|
Repo string `json:"repo"`
|
|
}
|
|
|
|
func (a *API) setOrigin(w http.ResponseWriter, r *http.Request) {
|
|
if a.serve == nil {
|
|
writeError(w, http.StatusNotImplemented, "updates need a mirror and a store, and this control plane lacks one")
|
|
return
|
|
}
|
|
var req OriginRequest
|
|
if err := json.NewDecoder(io.LimitReader(r.Body, 1<<12)).Decode(&req); err != nil || req.Repo == "" {
|
|
writeError(w, http.StatusBadRequest, "name the repository, owner/name")
|
|
return
|
|
}
|
|
o, err := a.serve.SetOriginOf(r.Context(), r.PathValue("name"), req.Repo)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, err.Error())
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, OriginView{Repo: o.Repo, Revision: o.Revision})
|
|
}
|
|
|
|
// PoolNodes is the fleet's ready nodes as a plan sees them: their cards,
|
|
// which of them are free, the fabric they share, and their memory.
|
|
func (a *API) PoolNodes() []modelplan.PoolNode {
|
|
var out []modelplan.PoolNode
|
|
for _, n := range a.Nodes() {
|
|
if n.State != "ready" {
|
|
continue
|
|
}
|
|
p := modelplan.PoolNode{ID: n.ID, GPUs: n.GPUTotal, FreeGPUs: n.GPUFree, HostMemGB: n.MemoryGB,
|
|
HostFreeGB: n.MemoryGB}
|
|
if len(n.Classes) > 0 {
|
|
p.Class = n.Classes[0]
|
|
p.GPUMemGB = modelplan.MemFromClass(p.Class)
|
|
}
|
|
if len(n.Fabrics) > 0 {
|
|
p.Fabric = n.Fabrics[0]
|
|
}
|
|
if l := n.Load; l != nil && l.MemTotal > 0 {
|
|
p.HostFreeGB = int((l.MemTotal - l.MemUsed) >> 30)
|
|
}
|
|
out = append(out, p)
|
|
}
|
|
return out
|
|
}
|
|
|
|
func (a *API) planModel(w http.ResponseWriter, r *http.Request) {
|
|
if a.serve == nil {
|
|
writeError(w, http.StatusNotImplemented, "planning needs a mirror and its registry, and this control plane lacks one")
|
|
return
|
|
}
|
|
q := r.URL.Query()
|
|
req := modelserve.PlanRequest{Repo: q.Get("repo"), Format: q.Get("format")}
|
|
if c := q.Get("context"); c != "" {
|
|
n, err := strconv.Atoi(c)
|
|
if err != nil || n < 0 {
|
|
writeError(w, http.StatusBadRequest, "context is a whole number of tokens")
|
|
return
|
|
}
|
|
req.Context = n
|
|
}
|
|
ctx, cancel := context.WithTimeout(r.Context(), time.Minute)
|
|
defer cancel()
|
|
p, err := a.serve.Plan(ctx, req)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, err.Error())
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, p)
|
|
}
|
|
|
|
// actorOf is who a change is recorded as made by: what the caller says, since
|
|
// the admin API answers only this machine and trusts whoever reaches it. Kept
|
|
// to a short printable name, so a claim cannot pass for more than one — no
|
|
// line breaks to forge a second log line, no length to bury the rest.
|
|
// Actor is who an admin API request says it comes from: the person who ran
|
|
// the command holding the admin token, as the CLI reports it.
|
|
func Actor(r *http.Request) string { return actorOf(r) }
|
|
|
|
func actorOf(r *http.Request) string {
|
|
claimed := r.Header.Get("X-Llmgw-Actor")
|
|
var b strings.Builder
|
|
for _, c := range claimed {
|
|
if b.Len() >= 64 {
|
|
break
|
|
}
|
|
if unicode.IsLetter(c) || unicode.IsDigit(c) || strings.ContainsRune("._-@", c) {
|
|
b.WriteRune(c)
|
|
}
|
|
}
|
|
if b.Len() == 0 {
|
|
return "unknown"
|
|
}
|
|
return b.String()
|
|
}
|
|
|
|
// NodeCPUs is the CPU counts of ready nodes with at least this much memory:
|
|
// where a replica needing it could go.
|
|
func (a *API) NodeCPUs(minMemoryGB int) []int {
|
|
var out []int
|
|
for _, n := range a.ctl.Inventory().Nodes() {
|
|
if n.State.String() != "ready" || n.CPUCount == 0 {
|
|
continue
|
|
}
|
|
if n.MemoryTotalMB > 0 && n.MemoryTotalMB/1024 < minMemoryGB {
|
|
continue
|
|
}
|
|
out = append(out, n.CPUCount)
|
|
}
|
|
return out
|
|
}
|
|
|
|
// reservationWarnings says what is wrong with what a model reserves.
|
|
//
|
|
// Resident memory includes the weights mapped from disk, which the kernel can
|
|
// take back, so it overstates what a model needs by the part of the weights
|
|
// not in use. A tenth over is let pass for that; more is a figure too low for
|
|
// placement to be trusted with.
|
|
func reservationWarnings(m catalog.Model, measuredGB float64) []string {
|
|
var out []string
|
|
for _, p := range m.Profiles {
|
|
if p.CPUOnly() && p.MinMemoryGB <= 0 {
|
|
out = append(out, fmt.Sprintf("profile %s declares no memory: placement cannot tell when a node is full "+
|
|
"(llmgw ctl model resources %s --memory GB)", p.Name, m.Name))
|
|
}
|
|
}
|
|
if len(m.Profiles) == 1 && m.Profiles[0].MinMemoryGB > 0 && measuredGB > float64(m.Profiles[0].MinMemoryGB)*1.1 {
|
|
out = append(out, fmt.Sprintf("uses %.1f GB and reserves %d: raise it, or nodes will be filled past their memory",
|
|
measuredGB, m.Profiles[0].MinMemoryGB))
|
|
}
|
|
return out
|
|
}
|
|
|
|
// externalLabel is "api.openai.com · gpt-4o-mini" for a model served elsewhere.
|
|
func externalLabel(m catalog.Model) string {
|
|
if !m.IsExternal() {
|
|
return ""
|
|
}
|
|
host := m.External.URL
|
|
if u, err := url.Parse(host); err == nil && u.Host != "" {
|
|
host = u.Host
|
|
}
|
|
return host + " · " + m.External.Model
|
|
}
|
|
|
|
// AdaptersRequest replaces a model's LoRA adapters with these.
|
|
type AdaptersRequest struct {
|
|
Adapters []catalog.Adapter `json:"adapters"`
|
|
}
|
|
|
|
// setAdapters replaces a model's adapters, after reading each new file to
|
|
// check it is an adapter for the model's architecture: one that is not is a
|
|
// replica that fails to start, found minutes later on a node.
|
|
func (a *API) setAdapters(w http.ResponseWriter, r *http.Request) {
|
|
if a.models == nil {
|
|
writeError(w, http.StatusNotImplemented, "this control plane has no store")
|
|
return
|
|
}
|
|
var req AdaptersRequest
|
|
if err := json.NewDecoder(io.LimitReader(r.Body, 1<<16)).Decode(&req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "the request is not valid JSON: "+err.Error())
|
|
return
|
|
}
|
|
name := r.PathValue("name")
|
|
m, ok := a.ctl.Catalog().Resolve(name)
|
|
if !ok {
|
|
writeError(w, http.StatusBadRequest, "no model "+name)
|
|
return
|
|
}
|
|
base, baseErr := gguf.KindFromURL(r.Context(), safedial.Client(safedial.ControlPlane, 30*time.Second), m.Source)
|
|
have := map[string]bool{}
|
|
for _, ad := range m.Adapters {
|
|
have[ad.Source] = true
|
|
}
|
|
for _, ad := range req.Adapters {
|
|
if have[ad.Source] {
|
|
continue // already checked when it was attached
|
|
}
|
|
h, err := gguf.KindFromURL(r.Context(), safedial.Client(safedial.ControlPlane, 30*time.Second), ad.Source)
|
|
switch {
|
|
case err != nil:
|
|
writeError(w, http.StatusBadRequest, fmt.Sprintf("adapter %s: %v", ad.Name, err))
|
|
return
|
|
case !h.IsAdapter():
|
|
writeError(w, http.StatusBadRequest, fmt.Sprintf("adapter %s: %s is a model, not a LoRA adapter", ad.Name, ad.Source))
|
|
return
|
|
case baseErr == nil && h.Architecture != base.Architecture:
|
|
writeError(w, http.StatusBadRequest, fmt.Sprintf("adapter %s is for %s models and %s is %s",
|
|
ad.Name, h.Architecture, m.Name, base.Architecture))
|
|
return
|
|
}
|
|
}
|
|
out, err := a.models.SetAdapters(r.Context(), m.Name, req.Adapters, actorOf(r))
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, err.Error())
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, AdaptersRequest{Adapters: out.Adapters})
|
|
}
|
|
|
|
// NodeEngineRequest chooses one engine's build for a node.
|
|
type NodeEngineRequest struct {
|
|
Engine string `json:"engine"`
|
|
// Build is "auto", a build such as "cu129", or "none".
|
|
Build string `json:"build"`
|
|
}
|
|
|
|
// setNodeEngine chooses an engine's build for a node, and has the node
|
|
// install it.
|
|
func (a *API) setNodeEngine(w http.ResponseWriter, r *http.Request) {
|
|
ch := a.ctl.EngineChoices()
|
|
if ch == nil {
|
|
writeError(w, http.StatusNotImplemented, "this control plane has no store to keep the choice in")
|
|
return
|
|
}
|
|
id := r.PathValue("id")
|
|
if _, ok := a.ctl.Inventory().Get(id); !ok {
|
|
writeError(w, http.StatusNotFound, "no node "+id)
|
|
return
|
|
}
|
|
var req NodeEngineRequest
|
|
if err := json.NewDecoder(io.LimitReader(r.Body, 1<<12)).Decode(&req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "the request is not valid JSON: "+err.Error())
|
|
return
|
|
}
|
|
if err := ch.Set(r.Context(), id, req.Engine, req.Build, actorOf(r)); err != nil {
|
|
writeError(w, http.StatusBadRequest, err.Error())
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, req)
|
|
}
|
|
|
|
// SetNodeEngine chooses an engine's build for a node, for the cluster page.
|
|
func (a *API) SetNodeEngine(ctx context.Context, node, engine, build, by string) error {
|
|
ch := a.ctl.EngineChoices()
|
|
if ch == nil {
|
|
return errors.New("this control plane has no store to keep the choice in")
|
|
}
|
|
if _, ok := a.ctl.Inventory().Get(node); !ok {
|
|
return errors.New("no node " + node)
|
|
}
|
|
return ch.Set(ctx, node, engine, build, by)
|
|
}
|