llmgw/internal/control/fleetmetrics/fleetmetrics.go

// 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"
	"strings"
	"time"

	"github.com/prometheus/client_golang/prometheus"

	"gitea.mynodes.xyz/ilker/llmgw/internal/control/fabrics"
	"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.
// HealthSource adds what the reconciler knows about each model and where
// engine installs stand, and whether this control plane is the one placing
// work. Asked for by type assertion, so a source without it — a test's —
// publishes what it did before.
type HealthSource interface {
	ModelHealth() []reconciler.ModelHealth
	IsLeader() bool
}

type Collector struct {
	src Source

	modelDesired      *prometheus.Desc
	modelFailures     *prometheus.Desc
	modelParked       *prometheus.Desc
	modelUnplaceable  *prometheus.Desc
	modelPreempted    *prometheus.Desc
	capacityHolds     *prometheus.Desc
	capacityGuarantee *prometheus.Desc
	capacityLimit     *prometheus.Desc
	engineInstall     *prometheus.Desc
	leader            *prometheus.Desc

	nodeUp       *prometheus.Desc
	nodeGPUs     *prometheus.Desc
	nodeGPUsFree *prometheus.Desc
	nodeMemory   *prometheus.Desc
	nodeSeen     *prometheus.Desc
	nodeDiskFree *prometheus.Desc
	nodeDiskSize *prometheus.Desc

	gpuMemoryTotal *prometheus.Desc
	gpuMemoryUsed  *prometheus.Desc
	gpuUtil        *prometheus.Desc
	gpuAssigned    *prometheus.Desc
	gpuPCIeReplay  *prometheus.Desc
	gpuRetired     *prometheus.Desc
	gpuRetirePend  *prometheus.Desc
	gpuRemapped    *prometheus.Desc
	gpuRemapPend   *prometheus.Desc
	gpuRemapFail   *prometheus.Desc
	gpuNVLinkErr   *prometheus.Desc
	gpuXID         *prometheus.Desc
	gpuLastXID     *prometheus.Desc
	gpuMemUtil     *prometheus.Desc
	gpuSMActive    *prometheus.Desc
	gpuSMOccupancy *prometheus.Desc
	gpuTensor      *prometheus.Desc
	gpuDRAMActive  *prometheus.Desc
	gpuSMClock     *prometheus.Desc
	gpuTemp        *prometheus.Desc
	gpuMemTemp     *prometheus.Desc
	gpuPower       *prometheus.Desc
	gpuPowerLimit  *prometheus.Desc
	gpuThrottled   *prometheus.Desc
	nodeCPUTemp    *prometheus.Desc
	nodeCPUPower   *prometheus.Desc
	nodeInletTemp  *prometheus.Desc
	gpuHealthy     *prometheus.Desc
	gpuNVLink      *prometheus.Desc
	gpuECC         *prometheus.Desc

	replicaState *prometheus.Desc
	replicaIdle  *prometheus.Desc
	replicaReady *prometheus.Desc

	nodeCPU      *prometheus.Desc
	nodeCPUs     *prometheus.Desc
	nodeLoad     *prometheus.Desc
	nodeMemUsed  *prometheus.Desc
	nodeMemAvail *prometheus.Desc
	shardCPU     *prometheus.Desc
	shardMemory  *prometheus.Desc

	ibActive  *prometheus.Desc
	ibRate    *prometheus.Desc
	ibCounter *prometheus.Desc

	gpuNVLJoined *prometheus.Desc
	imexUp       *prometheus.Desc

	fabricSMs      *prometheus.Desc
	fabricMaster   *prometheus.Desc
	fabricProblems *prometheus.Desc
}

// loadFresh is how old a node's load may be and still be reported. A node
// that stopped sending it has an unknown load, and repeating its last figure
// would draw a busy machine long after it went away.
const loadFresh = time.Minute

// 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),
		// Only the disk a node fills itself: space on a shared filer is not
		// this node's to spend, and summing it per node would have a fleet of
		// twenty each report the same free terabyte. Absent rather than zero
		// when nothing was reported, so an alert on "free is low" does not
		// fire for every node that was never asked about its disk.
		nodeDiskFree: prometheus.NewDesc("llmgw_node_disk_free_bytes",
			"Free space on the local filesystems holding model weights. Absent when the node reports no storage.", node, nil),
		nodeDiskSize: prometheus.NewDesc("llmgw_node_disk_total_bytes",
			"Size of those filesystems, so free space can be read as a fraction.", 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",
			"The share of the last sample period in which any kernel ran on a GPU (nvidia-smi's "+
				"utilization.gpu). Not how much of the chip: a kernel on one SM reads 100%. 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, and "+
				"llmgw_gpu_sm_active_percent how much.", gpu, nil),
		gpuSMActive: prometheus.NewDesc("llmgw_gpu_sm_active_percent",
			"The share of time a GPU's SMs had work, averaged over all of them: how much of the chip "+
				"the busy figure is using. From NVML's GPM (Hopper and later) or a dcgm-exporter on the node; "+
				"absent elsewhere.", gpu, nil),
		gpuSMOccupancy: prometheus.NewDesc("llmgw_gpu_sm_occupancy_percent",
			"How full a GPU's active SMs were with warps, of the most they can hold.", gpu, nil),
		gpuTensor: prometheus.NewDesc("llmgw_gpu_tensor_active_percent",
			"The share of time a GPU's tensor cores were working. Low in decode at batch one is normal; "+
				"prefill and batches are where it should rise.", gpu, nil),
		gpuDRAMActive: prometheus.NewDesc("llmgw_gpu_dram_active_percent",
			"The share of time a GPU's memory interface was moving data, from its profiling counters.", gpu, nil),
		gpuPCIeReplay: prometheus.NewDesc("llmgw_gpu_pcie_replays",
			"PCIe link retries on a GPU's link since the driver loaded; one that keeps climbing is a riser, slot or cable.", gpu, nil),
		gpuRetired: prometheus.NewDesc("llmgw_gpu_retired_pages",
			"Memory pages retired, by cause (sbe, dbe), on GPUs before Ampere.", append(gpu, "cause"), nil),
		gpuRetirePend: prometheus.NewDesc("llmgw_gpu_retirement_pending",
			"1 when retired pages wait for a GPU reset to take effect.", gpu, nil),
		gpuRemapped: prometheus.NewDesc("llmgw_gpu_remapped_rows",
			"Memory rows remapped, by kind (correctable, uncorrectable), on Ampere and later.", append(gpu, "kind"), nil),
		gpuRemapPend: prometheus.NewDesc("llmgw_gpu_remap_pending",
			"1 when remapped rows wait for a GPU reset to take effect.", gpu, nil),
		gpuRemapFail: prometheus.NewDesc("llmgw_gpu_remap_failure",
			"1 when a GPU has run out of spare memory rows: it needs replacing.", gpu, nil),
		gpuNVLinkErr: prometheus.NewDesc("llmgw_gpu_nvlink_errors",
			"NVLink errors summed over a GPU's links, by kind (crc, replay, recovery).", append(gpu, "kind"), nil),
		gpuXID: prometheus.NewDesc("llmgw_gpu_xid_errors",
			"The driver's XID error events for a GPU since its node's agent started.", gpu, nil),
		gpuLastXID: prometheus.NewDesc("llmgw_gpu_last_xid",
			"The code of a GPU's last XID error since its agent started; absent when there was none.", gpu, nil),
		gpuAssigned: prometheus.NewDesc("llmgw_gpu_assigned",
			"1 for each GPU a replica holds: which model, which replica, and whose quota it counts against. "+
				"Joined on node and gpu with the other llmgw_gpu_ series, it says what each model's cards are doing.",
			[]string{"node", "gpu", "model", "replica", "owner"}, nil),
		gpuMemUtil: prometheus.NewDesc("llmgw_gpu_memory_utilization_percent",
			"How much of the sample period a GPU's memory was being read or written: "+
				"decode is bound by it, so a card busy here and idle above is waiting on memory.", gpu, nil),
		gpuSMClock: prometheus.NewDesc("llmgw_gpu_sm_clock_mhz",
			"A GPU's streaming multiprocessor clock now; one held low under load is the card being throttled.", gpu, nil),
		gpuMemTemp: prometheus.NewDesc("llmgw_gpu_memory_temperature_celsius",
			"A GPU's memory (HBM) temperature, where the driver reports it.", gpu, nil),
		gpuPower: prometheus.NewDesc("llmgw_gpu_power_watts",
			"A GPU's power draw now.", gpu, nil),
		gpuPowerLimit: prometheus.NewDesc("llmgw_gpu_power_limit_watts",
			"The power limit a GPU is held to.", gpu, nil),
		gpuThrottled: prometheus.NewDesc("llmgw_gpu_throttled",
			"1 while something holds a GPU's clocks down: sw_power_cap, hw_slowdown, sw_thermal, hw_thermal, power_brake.",
			append(gpu, "reason"), nil),
		nodeCPUTemp: prometheus.NewDesc("llmgw_node_cpu_temperature_celsius",
			"A node's hottest CPU sensor.", node, nil),
		nodeCPUPower: prometheus.NewDesc("llmgw_node_cpu_power_watts",
			"A node's CPU packages' power draw, from RAPL.", node, nil),
		nodeInletTemp: prometheus.NewDesc("llmgw_node_inlet_temperature_celsius",
			"A node's inlet or ambient air sensor, where its board has one.", node, 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),
		gpuNVLink: prometheus.NewDesc("llmgw_gpu_nvlink_peers",
			"GPUs reachable over NVLink from this one. Always 0 on an agent built without "+
				"the nvml tag, which is every portable binary — so a flat zero across a fleet "+
				"means the topology is unknown rather than absent, and tensor groups are being "+
				"chosen by index.", 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. Published, as 0 when none are, for every "+
				"model that should be running; a stopped model, or an on-demand one nobody is "+
				"asking for, publishes nothing.", []string{"model"}, nil),
		modelDesired: prometheus.NewDesc("llmgw_model_replicas_desired",
			"Replicas a model should have now: its minimum, or for an on-demand model 1 while "+
				"it is being asked for.", []string{"model"}, nil),
		modelFailures: prometheus.NewDesc("llmgw_model_start_failures",
			"Replicas of a model that failed to start in a row. Resets when one comes up.",
			[]string{"model"}, nil),
		modelParked: prometheus.NewDesc("llmgw_model_parked",
			"1 when a model failed so often in a row that it is no longer retried.",
			[]string{"model"}, nil),
		modelUnplaceable: prometheus.NewDesc("llmgw_model_unplaceable",
			"1 when a model is short of replicas and the last attempt found no node to place one on.",
			[]string{"model"}, nil),
		capacityHolds: prometheus.NewDesc("llmgw_capacity_holds",
			"What a part of the tenancy tree holds of a resource, its own models' replicas and those beneath it.",
			[]string{"scope", "resource"}, nil),
		capacityGuarantee: prometheus.NewDesc("llmgw_capacity_guarantee",
			"What a part of the tenancy tree is guaranteed of a resource.",
			[]string{"scope", "resource"}, nil),
		capacityLimit: prometheus.NewDesc("llmgw_capacity_limit",
			"The most of a resource a part of the tenancy tree may hold.",
			[]string{"scope", "resource"}, nil),
		modelPreempted: prometheus.NewDesc("llmgw_model_preempted_total",
			"Replicas of a model stopped to make room for a model of a higher capacity class.",
			[]string{"model", "class"}, nil),
		engineInstall: prometheus.NewDesc("llmgw_node_engine_install",
			"Where installing an engine stands on a node: 1 for the current state.",
			[]string{"node", "engine", "version", "state"}, nil),
		leader: prometheus.NewDesc("llmgw_control_plane_leader",
			"1 when this control plane is the one placing work, 0 when it is a standby.", nil, 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),

		// A node's own use, from its heartbeat. Absent when the agent has
		// not measured it lately, which is not the same as idle.
		nodeCPU: prometheus.NewDesc("llmgw_node_cpu_utilization_percent",
			"How busy a node's CPUs were between its last two heartbeats, 0 to 100 across all cores.", node, nil),
		nodeCPUs: prometheus.NewDesc("llmgw_node_cpus",
			"CPUs a node has, so a load average can be read against it.", node, nil),
		nodeLoad: prometheus.NewDesc("llmgw_node_load",
			"The kernel's load average over 1, 5 and 15 minutes.", append(node, "window"), nil),
		nodeMemUsed: prometheus.NewDesc("llmgw_node_memory_used_bytes",
			"RAM in use: total less what the kernel could hand to a new process.", node, nil),
		nodeMemAvail: prometheus.NewDesc("llmgw_node_memory_available_bytes",
			"RAM the kernel could hand to a new process without swapping.", node, nil),
		shardCPU: prometheus.NewDesc("llmgw_shard_cpu_cores",
			"Cores an engine process group kept busy, the way top counts: 2.5 is two and a half.",
			[]string{"node", "model", "replica", "shard"}, nil),
		shardMemory: prometheus.NewDesc("llmgw_shard_memory_bytes",
			"Resident memory of an engine process group.",
			[]string{"node", "model", "replica", "shard"}, nil),

		// Every InfiniBand or RoCE port, re-read on each heartbeat.
		ibActive: prometheus.NewDesc("llmgw_ib_port_active",
			"1 when an InfiniBand or RoCE port can carry traffic. phys_state tells a cable "+
				"unplugged (Polling, Disabled) from a link with no subnet manager (LinkUp).",
			[]string{"node", "hca", "port", "link_layer", "state", "phys_state"}, nil),
		ibRate: prometheus.NewDesc("llmgw_ib_port_rate_gbps",
			"A port's negotiated rate. A link that trained below its usual speed is a "+
				"cable or transceiver to look at.", []string{"node", "hca", "port"}, nil),
		gpuNVLJoined: prometheus.NewDesc("llmgw_gpu_nvlink_joined",
			"1 when a GPU is joined to its multi-node NVLink domain, 0 when it reports a fabric it has not "+
				"joined. Absent on cards with no multi-node fabric.", []string{"node", "gpu", "domain", "state"}, nil),
		imexUp: prometheus.NewDesc("llmgw_imex_up",
			"1 when nvidia-imex is active with a channel, on nodes whose GPUs report a fabric.", []string{"node"}, nil),
		fabricSMs: prometheus.NewDesc("llmgw_fabric_subnet_managers",
			"Subnet managers a physical fabric lists, by state. No standby means the fabric "+
				"cannot recover from a change if its master stops.", []string{"fabric", "physical", "state"}, nil),
		fabricMaster: prometheus.NewDesc("llmgw_fabric_master_sm",
			"1 for a physical fabric's master subnet manager, as most of its ports see it; a "+
				"change of sm_guid is a failover.", []string{"fabric", "physical", "sm_guid"}, nil),
		fabricProblems: prometheus.NewDesc("llmgw_fabric_problems",
			"Disagreements about the fabrics, by level: fail means multi-node work will not "+
				"place or will not run where it should.", []string{"level"}, nil),
		ibCounter: prometheus.NewDesc("llmgw_ib_port_counter_total",
			"A port's cumulative counters, by sysfs name: symbol_error, link_downed and the "+
				"other errors say a link is going bad before a job fails on it; port_xmit_data "+
				"and port_rcv_data are bytes.", []string{"node", "hca", "port", "counter"}, 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.nodeDiskFree, c.nodeDiskSize,
		c.nodeCPU, c.nodeCPUs, c.nodeLoad, c.nodeMemUsed, c.nodeMemAvail, c.shardCPU, c.shardMemory,
		c.ibActive, c.ibRate, c.ibCounter, c.fabricSMs, c.fabricMaster, c.fabricProblems,
		c.gpuNVLJoined, c.imexUp,
		c.gpuMemoryTotal, c.gpuMemoryUsed, c.gpuUtil, c.gpuMemUtil, c.gpuSMClock,
		c.gpuSMActive, c.gpuSMOccupancy, c.gpuTensor, c.gpuDRAMActive, c.gpuAssigned, c.gpuTemp, c.gpuHealthy,
		c.gpuPCIeReplay, c.gpuRetired, c.gpuRetirePend, c.gpuRemapped, c.gpuRemapPend, c.gpuRemapFail,
		c.gpuNVLinkErr, c.gpuXID, c.gpuLastXID,
		c.gpuMemTemp, c.gpuPower, c.gpuPowerLimit, c.gpuThrottled, c.nodeCPUTemp, c.nodeCPUPower, c.nodeInletTemp,
		c.gpuNVLink, c.gpuECC,
		c.replicaState, c.replicaReady, c.replicaIdle,
		c.modelDesired, c.modelFailures, c.modelParked, c.modelUnplaceable, c.modelPreempted,
		c.capacityHolds, c.capacityGuarantee, c.capacityLimit,
		c.engineInstall, c.leader,
	} {
		ch <- d
	}
}

// Collect implements prometheus.Collector.
func (c *Collector) Collect(ch chan<- prometheus.Metric) {
	now := time.Now()
	c.collectQuotas(ch)
	c.collectAssignments(ch)

	loads := map[string]*inventory.Load{}
	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)
		// Nothing reported means nothing is known, so no sample at all. A zero
		// here would read as a full disk and page someone about a node that
		// was simply never asked.
		if free, total := n.LocalDisk(); total > 0 {
			ch <- prometheus.MustNewConstMetric(c.nodeDiskFree, prometheus.GaugeValue,
				float64(free), n.ID)
			ch <- prometheus.MustNewConstMetric(c.nodeDiskSize, prometheus.GaugeValue,
				float64(total), n.ID)
		}

		if l := n.Load; l != nil && now.Sub(l.At) < loadFresh {
			// Each only when measured: a VM has none of these, and a zero
			// would read as a machine at 0 °C drawing nothing.
			for _, v := range []struct {
				d   *prometheus.Desc
				val float64
			}{{c.nodeCPUTemp, l.CPUTempC}, {c.nodeCPUPower, l.CPUPowerW}, {c.nodeInletTemp, l.InletTempC}} {
				if v.val > 0 {
					ch <- prometheus.MustNewConstMetric(v.d, prometheus.GaugeValue, v.val, n.ID)
				}
			}
			ch <- prometheus.MustNewConstMetric(c.nodeCPU, prometheus.GaugeValue, l.CPUPercent, n.ID)
			if l.CPUCount > 0 {
				ch <- prometheus.MustNewConstMetric(c.nodeCPUs, prometheus.GaugeValue, float64(l.CPUCount), n.ID)
			}
			for w, v := range map[string]float64{"1m": l.Load1, "5m": l.Load5, "15m": l.Load15} {
				ch <- prometheus.MustNewConstMetric(c.nodeLoad, prometheus.GaugeValue, v, n.ID, w)
			}
			if l.MemUsedBytes > 0 || l.MemAvailBytes > 0 {
				ch <- prometheus.MustNewConstMetric(c.nodeMemUsed, prometheus.GaugeValue, float64(l.MemUsedBytes), n.ID)
				ch <- prometheus.MustNewConstMetric(c.nodeMemAvail, prometheus.GaugeValue, float64(l.MemAvailBytes), n.ID)
			}
			loads[n.ID] = l
		}

		for _, p := range n.IBPorts {
			port := strconv.Itoa(p.Port)
			active := 0.0
			if p.Active() {
				active = 1
			}
			ch <- prometheus.MustNewConstMetric(c.ibActive, prometheus.GaugeValue, active,
				n.ID, p.HCA, port, p.LinkLayer, p.State, p.PhysState)
			ch <- prometheus.MustNewConstMetric(c.ibRate, prometheus.GaugeValue, float64(p.RateGbps),
				n.ID, p.HCA, port)
			for name, v := range p.Counters {
				ch <- prometheus.MustNewConstMetric(c.ibCounter, prometheus.CounterValue, float64(v),
					n.ID, p.HCA, port, name)
			}
		}

		fabricGPUs := 0
		for _, g := range n.GPUs {
			fb := g.Fabric
			if fb == nil || strings.EqualFold(fb.State, "Not Supported") {
				continue
			}
			fabricGPUs++
			joined := 0.0
			if fb.Domain() != "" {
				joined = 1
			}
			ch <- prometheus.MustNewConstMetric(c.gpuNVLJoined, prometheus.GaugeValue, joined,
				n.ID, strconv.Itoa(g.Index), fb.Domain(), fb.State)
		}
		if fabricGPUs > 0 {
			up := 0.0
			if n.IMEX.State == "active" && n.IMEX.Channels > 0 {
				up = 1
			}
			ch <- prometheus.MustNewConstMetric(c.imexUp, prometheus.GaugeValue, up, 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}

			// The sensors are the card's. A MIG instance shares its card's
			// heat and power, but not its memory, and has no utilisation of
			// its own that any driver reports: its memory is the instance's
			// own reading, and its utilisation is left out rather than
			// published as the card's or as zero.
			s, fresh := sensorFor(n.Load, g.CardIndex(), now)
			if g.IsMIG() {
				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...)
			} else if fresh && s.Busy() {
				// What the card is doing now, sampled every heartbeat.
				c.gpuBusy(ch, s, labels)
			} else {
				// An agent too old to sample it: what it read at
				// registration, which is all there is.
				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...)
				}
			}
			// The sensors' fresh reading when there is one; the temperature
			// read at registration otherwise, which is all an older agent sends.
			if fresh {
				c.gpuSensor(ch, s, labels)
			} else 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.gpuNVLink, prometheus.GaugeValue,
				float64(len(g.NVLinkPeers)), labels...)
			ecc := float64(g.Health.ECCErrors)
			if fresh && s.Busy() && s.ECCUncorrected >= 0 {
				ecc = s.ECCUncorrected
			}
			ch <- prometheus.MustNewConstMetric(c.gpuECC, prometheus.GaugeValue, ecc, labels...)
		}
		for class, n2 := range byClass {
			ch <- prometheus.MustNewConstMetric(c.nodeGPUs, prometheus.GaugeValue,
				float64(n2), n.ID, class)
		}
	}

	// The fabrics as the adapters find them: their subnet managers, and how
	// many things disagree.
	rep := fabrics.Analyze(c.src.Nodes(), nil, now)
	for _, ph := range rep.Physical {
		name := strings.Join(ph.Declared, ",")
		if ph.SM != nil && ph.SM.GUID != "" {
			ch <- prometheus.MustNewConstMetric(c.fabricMaster, prometheus.GaugeValue, 1, name, ph.Key, ph.SM.GUID)
		}
		byState := map[string]int{}
		for _, sm := range ph.SMs {
			byState[sm.State]++
		}
		for st, n := range byState {
			ch <- prometheus.MustNewConstMetric(c.fabricSMs, prometheus.GaugeValue, float64(n), name, ph.Key, st)
		}
	}
	levels := map[fabrics.Level]int{fabrics.Fail: 0, fabrics.Warn: 0, fabrics.Info: 0}
	for _, p := range rep.Problems {
		levels[p.Level]++
	}
	for l, n := range levels {
		ch <- prometheus.MustNewConstMetric(c.fabricProblems, prometheus.GaugeValue, float64(n), string(l))
	}

	// 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() {
		// Each engine's share of its node, labelled with what it serves, so
		// "which model is eating this machine" is one query.
		for _, sh := range rep.Shards {
			if l := loads[sh.NodeID]; l != nil {
				if use, ok := l.Shards[sh.ID]; ok {
					labels := []string{sh.NodeID, rep.Model, rep.ID, sh.ID}
					ch <- prometheus.MustNewConstMetric(c.shardCPU, prometheus.GaugeValue, use.CPUCores, labels...)
					ch <- prometheus.MustNewConstMetric(c.shardMemory, prometheus.GaugeValue, float64(use.RSSBytes), labels...)
				}
			}
		}
		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])
	}
	hs, ok := c.src.(HealthSource)
	if !ok {
		for model, v := range ready {
			ch <- prometheus.MustNewConstMetric(c.replicaReady, prometheus.GaugeValue,
				float64(v), model)
		}
		return
	}
	for _, n := range c.src.Nodes() {
		for _, p := range n.Provision {
			ch <- prometheus.MustNewConstMetric(c.engineInstall, prometheus.GaugeValue, 1,
				n.ID, p.Engine, p.Version, p.State)
		}
	}
	lead := 0.0
	if hs.IsLeader() {
		lead = 1
	}
	ch <- prometheus.MustNewConstMetric(c.leader, prometheus.GaugeValue, lead)
	for _, h := range hs.ModelHealth() {
		// Ready is published at zero for what should be running — the one
		// value ModelHasNothingServing exists to catch, and which, emitted
		// only for replicas that were Ready, it never saw.
		if h.Desired > 0 || ready[h.Model] > 0 {
			ch <- prometheus.MustNewConstMetric(c.replicaReady, prometheus.GaugeValue,
				float64(ready[h.Model]), h.Model)
		}
		ch <- prometheus.MustNewConstMetric(c.modelDesired, prometheus.GaugeValue, float64(h.Desired), h.Model)
		ch <- prometheus.MustNewConstMetric(c.modelFailures, prometheus.GaugeValue, float64(h.Failures), h.Model)
		ch <- prometheus.MustNewConstMetric(c.modelParked, prometheus.GaugeValue, b2f(h.Parked), h.Model)
		ch <- prometheus.MustNewConstMetric(c.modelUnplaceable, prometheus.GaugeValue, b2f(h.Unplaceable), h.Model)
		ch <- prometheus.MustNewConstMetric(c.modelPreempted, prometheus.CounterValue, float64(h.Preempted), h.Model, h.Class)
	}
}

func b2f(b bool) float64 {
	if b {
		return 1
	}
	return 0
}

// OwnerSource says whose quota each model counts against, for labelling
// what its replicas hold. Asked for by type assertion.
type OwnerSource interface {
	ModelOwners() map[string]string
}

// QuotaSource adds the tenancy tree's capacity quotas. Asked for by type
// assertion, like HealthSource.
type QuotaSource interface {
	QuotaUse() []reconciler.QuotaUse
}

func (c *Collector) collectQuotas(ch chan<- prometheus.Metric) {
	qs, ok := c.src.(QuotaSource)
	if !ok {
		return
	}
	for _, q := range qs.QuotaUse() {
		for _, r := range []struct {
			name     string
			holds    int
			gua, lim *int
		}{
			{"gpus", q.Holds.GPUs, q.Guarantee.GPUs, q.Limit.GPUs},
			{"cpu_cores", q.Holds.CPUCores, q.Guarantee.CPUCores, q.Limit.CPUCores},
			{"memory_gb", q.Holds.MemoryGB, q.Guarantee.MemoryGB, q.Limit.MemoryGB},
		} {
			if r.gua == nil && r.lim == nil {
				continue
			}
			ch <- prometheus.MustNewConstMetric(c.capacityHolds, prometheus.GaugeValue, float64(r.holds), q.Scope, r.name)
			if r.gua != nil {
				ch <- prometheus.MustNewConstMetric(c.capacityGuarantee, prometheus.GaugeValue, float64(*r.gua), q.Scope, r.name)
			}
			if r.lim != nil {
				ch <- prometheus.MustNewConstMetric(c.capacityLimit, prometheus.GaugeValue, float64(*r.lim), q.Scope, r.name)
			}
		}
	}
}

// sensorFor is a GPU's fresh sensor reading, if the node sent one.
func sensorFor(l *inventory.Load, index int, now time.Time) (inventory.GPUSensor, bool) {
	if l == nil || now.Sub(l.At) >= loadFresh {
		return inventory.GPUSensor{}, false
	}
	for _, g := range l.GPUs {
		if g.Index == index {
			return g, true
		}
	}
	return inventory.GPUSensor{}, false
}

// gpuBusy is how busy a card is, from a sample that carries it.
func (c *Collector) gpuBusy(ch chan<- prometheus.Metric, s inventory.GPUSensor, labels []string) {
	ch <- prometheus.MustNewConstMetric(c.gpuMemoryTotal, prometheus.GaugeValue, s.MemoryTotalMB*1024*1024, labels...)
	ch <- prometheus.MustNewConstMetric(c.gpuMemoryUsed, prometheus.GaugeValue, s.MemoryUsedMB*1024*1024, labels...)
	// Zero is a real reading here — an idle card — and only -1, the driver
	// not saying, leaves the series out.
	if s.UtilPercent >= 0 {
		ch <- prometheus.MustNewConstMetric(c.gpuUtil, prometheus.GaugeValue, s.UtilPercent, labels...)
	}
	if s.MemUtilPercent >= 0 {
		ch <- prometheus.MustNewConstMetric(c.gpuMemUtil, prometheus.GaugeValue, s.MemUtilPercent, labels...)
	}
	if s.SMClockMHz > 0 {
		ch <- prometheus.MustNewConstMetric(c.gpuSMClock, prometheus.GaugeValue, s.SMClockMHz, labels...)
	}
	// The card's profile, here rather than with the sensors because it is
	// the whole card's: a MIG instance must not be published as having it.
	if p := s.Profile; p != nil {
		for _, v := range []struct {
			d   *prometheus.Desc
			val float64
		}{{c.gpuSMActive, p.SMActive}, {c.gpuSMOccupancy, p.SMOccupancy}, {c.gpuTensor, p.TensorActive}, {c.gpuDRAMActive, p.DRAMActive}} {
			// Zero is an idle chip; only -1, not reported, is left out.
			if v.val >= 0 {
				ch <- prometheus.MustNewConstMetric(v.d, prometheus.GaugeValue, v.val, labels...)
			}
		}
	}
}

func (c *Collector) gpuSensor(ch chan<- prometheus.Metric, s inventory.GPUSensor, labels []string) {
	for _, v := range []struct {
		d   *prometheus.Desc
		val float64
	}{{c.gpuTemp, s.TempC}, {c.gpuMemTemp, s.MemoryTempC}, {c.gpuPower, s.PowerW}, {c.gpuPowerLimit, s.PowerLimitW}} {
		if v.val > 0 {
			ch <- prometheus.MustNewConstMetric(v.d, prometheus.GaugeValue, v.val, labels...)
		}
	}
	for _, reason := range s.Throttle {
		ch <- prometheus.MustNewConstMetric(c.gpuThrottled, prometheus.GaugeValue, 1, append(labels, reason)...)
	}
	if w := s.Wear; w != nil {
		c.gpuWear(ch, w, labels)
	}
}

// gpuWear is what wears a card out. A count the card does not keep (-1) is
// left out, not published as zero.
func (c *Collector) gpuWear(ch chan<- prometheus.Metric, w *inventory.GPUWear, labels []string) {
	gauge := func(d *prometheus.Desc, v float64, extra ...string) {
		ch <- prometheus.MustNewConstMetric(d, prometheus.GaugeValue, v, append(append([]string{}, labels...), extra...)...)
	}
	flag := func(b bool) float64 {
		if b {
			return 1
		}
		return 0
	}
	if w.PCIeReplays >= 0 {
		gauge(c.gpuPCIeReplay, float64(w.PCIeReplays))
	}
	if w.RetiredSBE >= 0 || w.RetiredDBE >= 0 {
		gauge(c.gpuRetired, float64(max(w.RetiredSBE, 0)), "sbe")
		gauge(c.gpuRetired, float64(max(w.RetiredDBE, 0)), "dbe")
		gauge(c.gpuRetirePend, flag(w.RetirePending))
	}
	if w.RemappedCorrectable >= 0 || w.RemappedUncorrectable >= 0 {
		gauge(c.gpuRemapped, float64(max(w.RemappedCorrectable, 0)), "correctable")
		gauge(c.gpuRemapped, float64(max(w.RemappedUncorrectable, 0)), "uncorrectable")
		gauge(c.gpuRemapPend, flag(w.RemapPending))
		gauge(c.gpuRemapFail, flag(w.RemapFailure))
	}
	gauge(c.gpuNVLinkErr, float64(w.NVLinkCRC), "crc")
	gauge(c.gpuNVLinkErr, float64(w.NVLinkReplay), "replay")
	gauge(c.gpuNVLinkErr, float64(w.NVLinkRecovery), "recovery")
	gauge(c.gpuXID, float64(w.XIDs))
	if w.XIDs > 0 {
		gauge(c.gpuLastXID, float64(w.LastXID))
	}
}

// collectAssignments says which replica holds each GPU.
func (c *Collector) collectAssignments(ch chan<- prometheus.Metric) {
	var owners map[string]string
	if o, ok := c.src.(OwnerSource); ok {
		owners = o.ModelOwners()
	}
	for _, r := range c.src.Replicas() {
		if !r.State.Counts() {
			continue
		}
		for _, sh := range r.Shards {
			for _, g := range sh.GPUIndices {
				ch <- prometheus.MustNewConstMetric(c.gpuAssigned, prometheus.GaugeValue, 1,
					sh.NodeID, strconv.Itoa(g), r.Model, r.ID, owners[r.Model])
			}
		}
	}
}