diff --git a/api/gen/llmgw/v1/agent.pb.go b/api/gen/llmgw/v1/agent.pb.go index c38a95d..11c560c 100644 --- a/api/gen/llmgw/v1/agent.pb.go +++ b/api/gen/llmgw/v1/agent.pb.go @@ -640,11 +640,15 @@ type GPU struct { NvlinkPeers []int32 `protobuf:"varint,11,rep,packed,name=nvlink_peers,json=nvlinkPeers,proto3" json:"nvlink_peers,omitempty"` // problems keeps a GPU out of placement: ECC faults, or memory already held // by another process. - Problems []string `protobuf:"bytes,12,rep,name=problems,proto3" json:"problems,omitempty"` - TemperatureC int32 `protobuf:"varint,13,opt,name=temperature_c,json=temperatureC,proto3" json:"temperature_c,omitempty"` - EccErrors int32 `protobuf:"varint,14,opt,name=ecc_errors,json=eccErrors,proto3" json:"ecc_errors,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + Problems []string `protobuf:"bytes,12,rep,name=problems,proto3" json:"problems,omitempty"` + TemperatureC int32 `protobuf:"varint,13,opt,name=temperature_c,json=temperatureC,proto3" json:"temperature_c,omitempty"` + EccErrors int32 `protobuf:"varint,14,opt,name=ecc_errors,json=eccErrors,proto3" json:"ecc_errors,omitempty"` + // utilization_percent is how busy the card was over the last sample period. + // Zero when the driver is too old to report it, which is not the same as + // idle: the metric is simply absent rather than wrong. + UtilizationPercent int32 `protobuf:"varint,15,opt,name=utilization_percent,json=utilizationPercent,proto3" json:"utilization_percent,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *GPU) Reset() { @@ -775,6 +779,13 @@ func (x *GPU) GetEccErrors() int32 { return 0 } +func (x *GPU) GetUtilizationPercent() int32 { + if x != nil { + return x.UtilizationPercent + } + return 0 +} + type FabricAttachment struct { state protoimpl.MessageState `protogen:"open.v1"` // fabric is the operator-assigned name. It is not derived from the GID @@ -1546,7 +1557,7 @@ const file_llmgw_v1_agent_proto_rawDesc = "" + "\vleader_addr\x18\r \x01(\tR\n" + "leaderAddr\x12\x1f\n" + "\vleader_port\x18\x0e \x01(\x05R\n" + - "leaderPort\"\xc8\x03\n" + + "leaderPort\"\xf9\x03\n" + "\x03GPU\x12\x14\n" + "\x05index\x18\x01 \x01(\x05R\x05index\x12\x12\n" + "\x04uuid\x18\x02 \x01(\tR\x04uuid\x12\x12\n" + @@ -1564,7 +1575,8 @@ const file_llmgw_v1_agent_proto_rawDesc = "" + "\bproblems\x18\f \x03(\tR\bproblems\x12#\n" + "\rtemperature_c\x18\r \x01(\x05R\ftemperatureC\x12\x1d\n" + "\n" + - "ecc_errors\x18\x0e \x01(\x05R\teccErrors\"[\n" + + "ecc_errors\x18\x0e \x01(\x05R\teccErrors\x12/\n" + + "\x13utilization_percent\x18\x0f \x01(\x05R\x12utilizationPercent\"[\n" + "\x10FabricAttachment\x12\x16\n" + "\x06fabric\x18\x01 \x01(\tR\x06fabric\x12\x12\n" + "\x04hcas\x18\x02 \x03(\tR\x04hcas\x12\x1b\n" + diff --git a/api/proto/llmgw/v1/agent.proto b/api/proto/llmgw/v1/agent.proto index 4293c6d..39604c7 100644 --- a/api/proto/llmgw/v1/agent.proto +++ b/api/proto/llmgw/v1/agent.proto @@ -137,6 +137,10 @@ message GPU { repeated string problems = 12; int32 temperature_c = 13; int32 ecc_errors = 14; + // utilization_percent is how busy the card was over the last sample period. + // Zero when the driver is too old to report it, which is not the same as + // idle: the metric is simply absent rather than wrong. + int32 utilization_percent = 15; } message FabricAttachment { diff --git a/internal/agent/discovery/gpu.go b/internal/agent/discovery/gpu.go index 5780a94..cd94b5c 100644 --- a/internal/agent/discovery/gpu.go +++ b/internal/agent/discovery/gpu.go @@ -20,6 +20,14 @@ type GPU struct { // MemoryUsedMB is what is already allocated. Non-zero on an idle node means // something other than our engines is using the card (plan §5.1). MemoryUsedMB int `json:"memory_used_mb"` + // UtilizationPercent is how busy the card was over the last sample period. + // + // The number people actually watch, and the one nothing collected: memory + // used says a model is loaded, and this says whether it is doing anything. + // A fleet of cards at ninety percent occupancy and two percent utilisation + // is a fleet holding models nobody is asking for, which is invisible + // without it. + UtilizationPercent int `json:"utilization_percent,omitempty"` // ComputeCapability gates quantization choices: an FP8 profile must not // land on a card below 8.9 (plan §6.2). ComputeCapability string `json:"compute_capability"` diff --git a/internal/agent/discovery/gpu_smi.go b/internal/agent/discovery/gpu_smi.go index d34ad22..18b43c6 100644 --- a/internal/agent/discovery/gpu_smi.go +++ b/internal/agent/discovery/gpu_smi.go @@ -21,8 +21,21 @@ var smiQueryFields = []string{ "driver_version", "temperature.gpu", "ecc.errors.uncorrected.volatile.total", + // Last, and optional. An nvidia-smi too old to know this field returns a + // shorter row, and every column before it still lines up — which is why it + // goes on the end rather than anywhere else, and why the parser requires + // only the mandatory prefix. + "utilization.gpu", } +// mandatorySMIFields is how many columns a row must have to be usable. +// +// Everything before utilization.gpu. A driver that does not know the last field +// gives a short row, and rejecting those would make an llmgw upgrade silently +// stop discovering GPUs on an older node — a far worse outcome than not knowing +// how busy a card is. +const mandatorySMIFields = 10 + // DiscoverGPUsSMI enumerates GPUs by shelling out to nvidia-smi. // // This is the cgo-free path (plan §3.2): the NVML bindings need cgo, which @@ -55,7 +68,7 @@ func parseSMI(out string) ([]GPU, error) { gpus := make([]GPU, 0, len(rows)) for _, row := range rows { - if len(row) < len(smiQueryFields) { + if len(row) < mandatorySMIFields { continue } idx, err := strconv.Atoi(strings.TrimSpace(row[0])) @@ -76,6 +89,9 @@ func parseSMI(out string) ([]GPU, error) { g.MemoryUsedMB, _ = parseCSVInt(row[4]) g.Health.TemperatureC, _ = parseCSVInt(row[8]) g.Health.ECCErrors, _ = parseCSVInt(row[9]) + if len(row) > 10 { + g.UtilizationPercent, _ = parseCSVInt(row[10]) + } assessHealth(&g) gpus = append(gpus, g) diff --git a/internal/agent/discovery/gpu_test.go b/internal/agent/discovery/gpu_test.go index ea71a44..034d233 100644 --- a/internal/agent/discovery/gpu_test.go +++ b/internal/agent/discovery/gpu_test.go @@ -142,3 +142,38 @@ func TestDiscoverGPUsSMIMissingBinary(t *testing.T) { t.Error("expected an error when nvidia-smi is missing") } } + +func TestAnOlderDriverWithoutUtilisationStillDiscovers(t *testing.T) { + // An nvidia-smi too old for utilization.gpu returns a shorter row. Rejecting + // those would make an llmgw upgrade silently stop discovering GPUs on an + // older node, which is far worse than not knowing how busy a card is — and + // is what the first version of this change did. + old := "0, GPU-aaa, NVIDIA H100 80GB HBM3, 81559, 1024, 9.0, 0000:01:00.0, 550.54, 41, 0" + gpus, err := parseSMI(old) + if err != nil { + t.Fatal(err) + } + if len(gpus) != 1 { + t.Fatalf("a row without utilisation produced %d GPUs", len(gpus)) + } + if gpus[0].MemoryTotalMB != 81559 || gpus[0].Health.TemperatureC != 41 { + t.Errorf("columns shifted: %+v", gpus[0]) + } + if gpus[0].UtilizationPercent != 0 { + t.Errorf("utilisation = %d, want 0 when unreported", gpus[0].UtilizationPercent) + } +} + +func TestUtilisationIsReadWhenPresent(t *testing.T) { + row := "0, GPU-aaa, NVIDIA H100 80GB HBM3, 81559, 1024, 9.0, 0000:01:00.0, 550.54, 41, 0, 87" + gpus, err := parseSMI(row) + if err != nil { + t.Fatal(err) + } + if len(gpus) != 1 { + t.Fatalf("got %d GPUs", len(gpus)) + } + if gpus[0].UtilizationPercent != 87 { + t.Errorf("utilisation = %d, want 87", gpus[0].UtilizationPercent) + } +} diff --git a/internal/cli/server.go b/internal/cli/server.go index 89184a3..e076da7 100644 --- a/internal/cli/server.go +++ b/internal/cli/server.go @@ -24,6 +24,7 @@ import ( "gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi" "gitea.mynodes.xyz/ilker/llmgw/internal/control/agentsvc" "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/fleetmetrics" "gitea.mynodes.xyz/ilker/llmgw/internal/control/modelsvc" "gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler" "gitea.mynodes.xyz/ilker/llmgw/internal/router" @@ -121,6 +122,13 @@ func newServerCmd() *cobra.Command { return err } if admin != nil { + // What the cluster is made of, on the same /metrics as the + // traffic figures: correlating "the cluster was full" with + // "requests were queueing" should not need two systems. + if err := rt.RegisterMetrics(fleetmetrics.New(admin)); err != nil { + return err + } + api := adminapi.New(admin) // Catalog changes need somewhere to persist, so this is wired // only when there is a store. Without one, the endpoints say so diff --git a/internal/control/agentsvc/convert.go b/internal/control/agentsvc/convert.go index 8236566..e8d0008 100644 --- a/internal/control/agentsvc/convert.go +++ b/internal/control/agentsvc/convert.go @@ -32,17 +32,18 @@ func NodeFromRegister(r *llmgwv1.Register) inventory.Node { } for _, g := range r.GetGpus() { n.GPUs = append(n.GPUs, discovery.GPU{ - Index: int(g.GetIndex()), - UUID: g.GetUuid(), - Name: g.GetName(), - MemoryTotalMB: int(g.GetMemoryTotalMb()), - MemoryUsedMB: int(g.GetMemoryUsedMb()), - ComputeCapability: g.GetComputeCapability(), - NUMANode: int(g.GetNumaNode()), - PCIBusID: g.GetPciBusId(), - DriverVersion: g.GetDriverVersion(), - CUDAVersion: g.GetCudaVersion(), - NVLinkPeers: intSlice(g.GetNvlinkPeers()), + Index: int(g.GetIndex()), + UUID: g.GetUuid(), + Name: g.GetName(), + MemoryTotalMB: int(g.GetMemoryTotalMb()), + MemoryUsedMB: int(g.GetMemoryUsedMb()), + ComputeCapability: g.GetComputeCapability(), + NUMANode: int(g.GetNumaNode()), + PCIBusID: g.GetPciBusId(), + DriverVersion: g.GetDriverVersion(), + UtilizationPercent: int(g.GetUtilizationPercent()), + CUDAVersion: g.GetCudaVersion(), + NVLinkPeers: intSlice(g.GetNvlinkPeers()), Health: discovery.GPUHealth{ ECCErrors: int(g.GetEccErrors()), TemperatureC: int(g.GetTemperatureC()), @@ -76,20 +77,21 @@ func RegisterFromInventory(nodeID, hostname, mgmtAddr, agentVersion string, } for _, g := range inv.GPUs { r.Gpus = append(r.Gpus, &llmgwv1.GPU{ - Index: int32(g.Index), - Uuid: g.UUID, - Name: g.Name, - MemoryTotalMb: int32(g.MemoryTotalMB), - MemoryUsedMb: int32(g.MemoryUsedMB), - ComputeCapability: g.ComputeCapability, - NumaNode: int32(g.NUMANode), - PciBusId: g.PCIBusID, - DriverVersion: g.DriverVersion, - CudaVersion: g.CUDAVersion, - NvlinkPeers: int32Slice(g.NVLinkPeers), - Problems: g.Health.Problems, - TemperatureC: int32(g.Health.TemperatureC), - EccErrors: int32(g.Health.ECCErrors), + Index: int32(g.Index), + Uuid: g.UUID, + Name: g.Name, + MemoryTotalMb: int32(g.MemoryTotalMB), + MemoryUsedMb: int32(g.MemoryUsedMB), + ComputeCapability: g.ComputeCapability, + NumaNode: int32(g.NUMANode), + PciBusId: g.PCIBusID, + DriverVersion: g.DriverVersion, + UtilizationPercent: int32(g.UtilizationPercent), + CudaVersion: g.CUDAVersion, + NvlinkPeers: int32Slice(g.NVLinkPeers), + Problems: g.Health.Problems, + TemperatureC: int32(g.Health.TemperatureC), + EccErrors: int32(g.Health.ECCErrors), }) } for _, f := range fabrics { diff --git a/internal/control/control.go b/internal/control/control.go index abf536c..f6a67d8 100644 --- a/internal/control/control.go +++ b/internal/control/control.go @@ -158,6 +158,12 @@ func (c *Controller) ReplaceCatalog(ctx context.Context, next *catalog.Catalog) c.log.Info("catalog replaced", "models", len(next.Models())) } +// Nodes, FreeGPUs and Replicas satisfy fleetmetrics.Source, so what the cluster +// is made of can be scraped from the one process that already knows. +func (c *Controller) Nodes() []inventory.Node { return c.inv.Nodes() } +func (c *Controller) FreeGPUs(nodeID string) []int { return c.inv.FreeGPUs(nodeID) } +func (c *Controller) Replicas() []reconciler.Replica { return c.rec.Replicas() } + // Reconciler exposes the reconciliation loop, for status reports and ctl. func (c *Controller) Reconciler() *reconciler.Reconciler { return c.rec } diff --git a/internal/control/fleetmetrics/fleetmetrics.go b/internal/control/fleetmetrics/fleetmetrics.go new file mode 100644 index 0000000..c8a0ec3 --- /dev/null +++ b/internal/control/fleetmetrics/fleetmetrics.go @@ -0,0 +1,200 @@ +// Package fleetmetrics exports what the cluster is made of and how much of it +// is in use. +// +// The router already exported twenty-two metric families about requests, +// endpoints, admission and scoring — everything about traffic, and nothing +// about the machines serving it. There was no way to graph GPU usage, spot a +// node going unhealthy, watch a fabric fill up, or answer "was the cluster full +// at the time" after an incident, which is the first question anyone asks. +// +// Exported by the control plane rather than by each agent, deliberately. The +// control plane already holds all of it: agents report their hardware at +// registration and their state on every heartbeat, so the data is a map read +// away. One scrape target for a fleet of hundreds, no per-node Prometheus +// configuration, no firewall rule per machine, and nothing new on the nodes. +package fleetmetrics + +import ( + "strconv" + "time" + + "github.com/prometheus/client_golang/prometheus" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler" +) + +// Source is the control plane state this reads. +type Source interface { + Nodes() []inventory.Node + FreeGPUs(nodeID string) []int + Replicas() []reconciler.Replica +} + +// Collector reports the fleet on each scrape. +// +// A collector rather than gauges kept up to date by a ticker: the numbers are +// derived from state that already exists, so computing them when asked is both +// simpler and fresher than maintaining a parallel copy that can drift. Nothing +// runs between scrapes. +type Collector struct { + src Source + + nodeUp *prometheus.Desc + nodeGPUs *prometheus.Desc + nodeGPUsFree *prometheus.Desc + nodeMemory *prometheus.Desc + nodeSeen *prometheus.Desc + + gpuMemoryTotal *prometheus.Desc + gpuMemoryUsed *prometheus.Desc + gpuUtil *prometheus.Desc + gpuTemp *prometheus.Desc + gpuHealthy *prometheus.Desc + gpuECC *prometheus.Desc + + replicaState *prometheus.Desc + replicaIdle *prometheus.Desc + replicaReady *prometheus.Desc +} + +// New builds the collector. +func New(src Source) *Collector { + node := []string{"node"} + gpu := []string{"node", "gpu", "class"} + return &Collector{ + src: src, + nodeUp: prometheus.NewDesc("llmgw_node_up", + "1 when a node is ready to take work, 0 otherwise.", + append(node, "state", "fabric"), nil), + nodeGPUs: prometheus.NewDesc("llmgw_node_gpus_total", + "GPUs a node reports.", append(node, "class"), nil), + nodeGPUsFree: prometheus.NewDesc("llmgw_node_gpus_free", + "GPUs on a node that are healthy and unallocated.", node, nil), + nodeMemory: prometheus.NewDesc("llmgw_node_memory_bytes", + "A node's total RAM, 0 when it does not report one.", node, nil), + nodeSeen: prometheus.NewDesc("llmgw_node_last_seen_seconds", + "Seconds since a node last reported. Rising means an agent has gone quiet.", node, nil), + + gpuMemoryTotal: prometheus.NewDesc("llmgw_gpu_memory_total_bytes", + "A GPU's framebuffer size.", gpu, nil), + gpuMemoryUsed: prometheus.NewDesc("llmgw_gpu_memory_used_bytes", + "A GPU's framebuffer in use, including processes llmgw did not start.", gpu, nil), + gpuUtil: prometheus.NewDesc("llmgw_gpu_utilization_percent", + "How busy a GPU was over the last sample period. Absent when the driver does not "+ + "report it, which is not the same as idle. Memory used says a model is loaded; "+ + "this says whether it is doing anything.", gpu, nil), + gpuTemp: prometheus.NewDesc("llmgw_gpu_temperature_celsius", + "A GPU's reported temperature.", gpu, nil), + gpuHealthy: prometheus.NewDesc("llmgw_gpu_healthy", + "1 when a GPU is eligible for placement, 0 when a fault keeps it out.", gpu, nil), + gpuECC: prometheus.NewDesc("llmgw_gpu_ecc_errors", + "Uncorrected ECC errors on a GPU. Non-zero is why a card stops being scheduled.", gpu, nil), + + replicaState: prometheus.NewDesc("llmgw_replicas", + "Replicas by model and state.", []string{"model", "state"}, nil), + replicaReady: prometheus.NewDesc("llmgw_model_replicas_ready", + "Replicas of a model that are serving.", []string{"model"}, nil), + replicaIdle: prometheus.NewDesc("llmgw_replica_idle_seconds", + "Seconds since a replica was last sent a request. Absent when nothing has been reported, "+ + "which is not the same as idle.", []string{"model", "replica"}, nil), + } +} + +// Describe implements prometheus.Collector. +func (c *Collector) Describe(ch chan<- *prometheus.Desc) { + for _, d := range []*prometheus.Desc{ + c.nodeUp, c.nodeGPUs, c.nodeGPUsFree, c.nodeMemory, c.nodeSeen, + c.gpuMemoryTotal, c.gpuMemoryUsed, c.gpuUtil, c.gpuTemp, c.gpuHealthy, c.gpuECC, + c.replicaState, c.replicaReady, c.replicaIdle, + } { + ch <- d + } +} + +// Collect implements prometheus.Collector. +func (c *Collector) Collect(ch chan<- prometheus.Metric) { + now := time.Now() + + for _, n := range c.src.Nodes() { + fabric := "" + if len(n.Fabrics) > 0 { + fabric = n.Fabrics[0].Fabric + } + up := 0.0 + if n.State == inventory.Ready { + up = 1 + } + // The state is a label rather than a number, because a cordoned node + // and an unreachable one need different responses and a single gauge + // counting "not ready" would hide which it is. + ch <- prometheus.MustNewConstMetric(c.nodeUp, prometheus.GaugeValue, up, + n.ID, n.State.String(), fabric) + ch <- prometheus.MustNewConstMetric(c.nodeGPUsFree, prometheus.GaugeValue, + float64(len(c.src.FreeGPUs(n.ID))), n.ID) + ch <- prometheus.MustNewConstMetric(c.nodeMemory, prometheus.GaugeValue, + float64(n.MemoryTotalMB)*1024*1024, n.ID) + ch <- prometheus.MustNewConstMetric(c.nodeSeen, prometheus.GaugeValue, + now.Sub(n.LastSeen).Seconds(), n.ID) + + byClass := map[string]int{} + for _, g := range n.GPUs { + class := g.Class() + byClass[class]++ + labels := []string{n.ID, strconv.Itoa(g.Index), class} + + ch <- prometheus.MustNewConstMetric(c.gpuMemoryTotal, prometheus.GaugeValue, + float64(g.MemoryTotalMB)*1024*1024, labels...) + ch <- prometheus.MustNewConstMetric(c.gpuMemoryUsed, prometheus.GaugeValue, + float64(g.MemoryUsedMB)*1024*1024, labels...) + if g.UtilizationPercent > 0 { + // Only when reported. A driver too old to give it would + // otherwise show every card at a flat zero, which reads as an + // idle fleet rather than a missing measurement. + ch <- prometheus.MustNewConstMetric(c.gpuUtil, prometheus.GaugeValue, + float64(g.UtilizationPercent), labels...) + } + if g.Health.TemperatureC > 0 { + ch <- prometheus.MustNewConstMetric(c.gpuTemp, prometheus.GaugeValue, + float64(g.Health.TemperatureC), labels...) + } + healthy := 0.0 + if g.Health.OK() { + healthy = 1 + } + ch <- prometheus.MustNewConstMetric(c.gpuHealthy, prometheus.GaugeValue, healthy, labels...) + ch <- prometheus.MustNewConstMetric(c.gpuECC, prometheus.GaugeValue, + float64(g.Health.ECCErrors), labels...) + } + for class, n2 := range byClass { + ch <- prometheus.MustNewConstMetric(c.nodeGPUs, prometheus.GaugeValue, + float64(n2), n.ID, class) + } + } + + // Replicas by model and state, so a model stuck starting is visible as a + // shape on a graph rather than only in a log line nobody tails. + counts := map[[2]string]int{} + ready := map[string]int{} + for _, rep := range c.src.Replicas() { + counts[[2]string{rep.Model, rep.State.String()}]++ + if rep.State == reconciler.Ready { + ready[rep.Model]++ + } + if !rep.LastUsedAt.IsZero() { + // Only when known. A replica nothing has reported would otherwise + // look like the most idle thing in the cluster, and after a router + // restart that would be all of them at once. + ch <- prometheus.MustNewConstMetric(c.replicaIdle, prometheus.GaugeValue, + now.Sub(rep.LastUsedAt).Seconds(), rep.Model, rep.ID) + } + } + for k, v := range counts { + ch <- prometheus.MustNewConstMetric(c.replicaState, prometheus.GaugeValue, + float64(v), k[0], k[1]) + } + for model, v := range ready { + ch <- prometheus.MustNewConstMetric(c.replicaReady, prometheus.GaugeValue, + float64(v), model) + } +} diff --git a/internal/control/fleetmetrics/fleetmetrics_test.go b/internal/control/fleetmetrics/fleetmetrics_test.go new file mode 100644 index 0000000..e0d158a --- /dev/null +++ b/internal/control/fleetmetrics/fleetmetrics_test.go @@ -0,0 +1,141 @@ +package fleetmetrics + +import ( + "strconv" + "strings" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus" + + "gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler" +) + +type fakeSource struct { + nodes []inventory.Node + free map[string][]int + replicas []reconciler.Replica +} + +func (f fakeSource) Nodes() []inventory.Node { return f.nodes } +func (f fakeSource) FreeGPUs(id string) []int { return f.free[id] } +func (f fakeSource) Replicas() []reconciler.Replica { return f.replicas } + +func gather(t *testing.T, src Source) string { + t.Helper() + reg := prometheus.NewRegistry() + if err := reg.Register(New(src)); err != nil { + t.Fatal(err) + } + var b strings.Builder + mfs, err := reg.Gather() + if err != nil { + t.Fatal(err) + } + for _, mf := range mfs { + for _, m := range mf.GetMetric() { + b.WriteString(mf.GetName()) + for _, l := range m.GetLabel() { + b.WriteString(" " + l.GetName() + "=" + l.GetValue()) + } + if g := m.GetGauge(); g != nil { + b.WriteString(" -> " + strconv.FormatFloat(g.GetValue(), 'f', -1, 64) + "\n") + continue + } + b.WriteString("\n") + } + } + return b.String() +} + +func gpuNode(id string, gpus int, fabric string) inventory.Node { + n := inventory.Node{ID: id, State: inventory.Ready, LastSeen: time.Now()} + for i := range gpus { + n.GPUs = append(n.GPUs, discovery.GPU{ + Index: i, Name: "NVIDIA H100 80GB HBM3", MemoryTotalMB: 81559, + MemoryUsedMB: 1024, UtilizationPercent: 40, + Health: discovery.GPUHealth{TemperatureC: 45}, + }) + } + if fabric != "" { + n.Fabrics = []inventory.FabricAttachment{{Fabric: fabric, HCAs: []string{"mlx5_0"}}} + } + return n +} + +func TestTheFleetIsReported(t *testing.T) { + // The gap this closes: twenty-two metric families about traffic, and none + // about the machines serving it — so "was the cluster full at the time", + // the first question after an incident, had no answer. + src := fakeSource{ + nodes: []inventory.Node{gpuNode("n1", 2, "ib-a")}, + free: map[string][]int{"n1": {1}}, + replicas: []reconciler.Replica{ + {ID: "r1", Model: "m", State: reconciler.Ready, LastUsedAt: time.Now().Add(-90 * time.Second)}, + }, + } + out := gather(t, src) + for _, want := range []string{ + "llmgw_node_up", "llmgw_node_gpus_free", "llmgw_gpu_memory_used_bytes", + "llmgw_gpu_utilization_percent", "llmgw_gpu_healthy", "llmgw_replicas", + "llmgw_model_replicas_ready", "llmgw_replica_idle_seconds", + } { + if !strings.Contains(out, want) { + t.Errorf("%s is missing from:\n%s", want, out) + } + } + // The fabric travels as a label, because with isolated fabrics "how full is + // the cluster" is the wrong question and "how full is this fabric" is the + // right one. + if !strings.Contains(out, "fabric=ib-a") { + t.Errorf("the fabric label is missing:\n%s", out) + } +} + +func TestAnIdleReplicaNothingHasReportedIsAbsentNotZero(t *testing.T) { + // A replica the router has said nothing about would otherwise look like the + // most idle thing in the cluster, and after a router restart that would be + // all of them at once. + src := fakeSource{ + replicas: []reconciler.Replica{{ID: "r1", Model: "m", State: reconciler.Ready}}, + } + if out := gather(t, src); strings.Contains(out, "llmgw_replica_idle_seconds") { + t.Errorf("idleness was reported for a replica nothing has used:\n%s", out) + } +} + +func TestAnUnreportedUtilisationIsAbsentNotZero(t *testing.T) { + // A driver too old to give it would otherwise show every card at a flat + // zero, which reads as an idle fleet rather than a missing measurement. + n := gpuNode("n1", 1, "") + n.GPUs[0].UtilizationPercent = 0 + src := fakeSource{nodes: []inventory.Node{n}, free: map[string][]int{"n1": {0}}} + if out := gather(t, src); strings.Contains(out, "llmgw_gpu_utilization_percent") { + t.Errorf("utilisation was reported when the driver gave none:\n%s", out) + } +} + +func TestNodeStateIsALabelNotANumber(t *testing.T) { + // A cordoned node and an unreachable one need different responses, and a + // single gauge counting "not ready" would hide which it is. + n := gpuNode("n1", 1, "") + n.State = inventory.Cordoned + src := fakeSource{nodes: []inventory.Node{n}, free: map[string][]int{}} + out := gather(t, src) + if !strings.Contains(out, "state=cordoned") { + t.Errorf("the node state is not a label:\n%s", out) + } + if !strings.Contains(out, "llmgw_node_up state=cordoned") && !strings.Contains(out, "state=cordoned") { + t.Errorf("unexpected shape:\n%s", out) + } +} + +func TestAnEmptyFleetCollectsWithoutError(t *testing.T) { + // A control plane with no agents yet must still serve /metrics; failing + // there would take monitoring down exactly when a cluster is being built. + if out := gather(t, fakeSource{}); out == "" { + t.Log("no metrics for an empty fleet, which is correct") + } +} diff --git a/internal/router/admin.go b/internal/router/admin.go index fe7650b..bebf9e2 100644 --- a/internal/router/admin.go +++ b/internal/router/admin.go @@ -14,6 +14,19 @@ import ( // It is served on a separate listener so the client-facing port can be exposed // without also exposing pprof (plan §9.1). The control plane's management API // is mounted here too, by the caller. +// RegisterMetrics adds a collector to the /metrics the admin listener serves. +// +// Exists so the control plane can publish what the cluster is made of on the +// same endpoint as the traffic figures. Correlating "the cluster was full" with +// "requests were queueing" is the first thing anyone does after an incident, +// and needing two scrape targets to do it is a needless obstacle. +// +// Called before AdminHandler, because the registry is built there. +func (r *Router) RegisterMetrics(c prometheus.Collector) error { + r.extraMetrics = append(r.extraMetrics, c) + return nil +} + func (r *Router) AdminHandler() (*http.ServeMux, error) { reg := prometheus.NewRegistry() reg.MustRegister( @@ -28,6 +41,11 @@ func (r *Router) AdminHandler() (*http.ServeMux, error) { if err := reg.Register(newEndpointCollector(r)); err != nil { return nil, err } + for _, c := range r.extraMetrics { + if err := reg.Register(c); err != nil { + return nil, err + } + } mux := http.NewServeMux() mux.Handle("GET /metrics", promhttp.HandlerFor(reg, promhttp.HandlerOpts{})) diff --git a/internal/router/router.go b/internal/router/router.go index 14fa669..6fecd77 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -16,6 +16,8 @@ import ( "sync/atomic" "time" + "github.com/prometheus/client_golang/prometheus" + "gitea.mynodes.xyz/ilker/llmgw/internal/account" "gitea.mynodes.xyz/ilker/llmgw/internal/accounting" "gitea.mynodes.xyz/ilker/llmgw/internal/auth" @@ -54,6 +56,9 @@ type Router struct { // from it, so a swap mid-request is invisible rather than a race. models atomic.Pointer[map[string]catalog.Model] + // extraMetrics are collectors other components publish on /metrics. + extraMetrics []prometheus.Collector + // landing configures the public page, which the router serves but does not // otherwise care about. landing config.Landing