//! Live per-agent-container resource load (CPU + memory) for the //! dashboard. Reads cgroup v2 stat files for each running agent //! machine directly — same privsep-clean posture as the turn-stats //! sqlite reads (`cpu.stat` / `memory.*` are world-readable; no //! `hive-priv` needed). Runs host-side in hive-c0re, where //! `/sys/fs/cgroup/machine.slice/` holds the per-container cgroups. Because //! nixos-container runs `systemd-nspawn --keep-unit` (with //! `Slice = "machine.slice"`), each container's cgroup is its launching //! service unit `container@.service` — not a machined //! `machine-.scope`. See [`scope_dir`]. //! //! No network: agents share the host network namespace //! (`privateNetwork = false`), so there is no per-container net //! counter to read. Per-agent network only becomes meaningful with the //! netns-isolation roadmap (`docs/network.md`). //! //! CPU is cumulative (`usage_usec` is monotonic), so a single read is //! meaningless — we sample every machine's counter, sleep one short //! interval, sample again, and divide the delta by `interval × nproc` //! to get a host-normalised percentage (0..100 across all cores). use std::collections::HashMap; use std::path::PathBuf; use std::sync::{OnceLock, RwLock}; use std::time::Duration; use serde::Serialize; use tokio::time::sleep; use crate::coordinator::Coordinator; /// Sampling interval for the two CPU reads. Short enough to keep the /// endpoint snappy, long enough that the delta isn't dominated by read /// jitter. const CPU_SAMPLE: Duration = Duration::from_millis(200); const MACHINE_SLICE: &str = "/sys/fs/cgroup/machine.slice"; #[derive(Debug, Serialize)] pub struct ContainerResource { /// Agent name (without the `h-` machine prefix). pub name: String, /// Host-normalised CPU usage over the sample interval, as a /// percentage of total host CPU (0..100 across all cores). pub cpu_pct: f64, /// Current memory usage (`memory.current`), bytes. pub mem_current_bytes: u64, /// High-water memory usage since container start (`memory.peak`), /// bytes. `None` if the kernel doesn't expose `memory.peak`. pub mem_peak_bytes: Option, /// Memory ceiling (`memory.max`), bytes. `None` when unlimited /// (the file reads `max`). pub mem_max_bytes: Option, /// Last-sampled total disk usage — the agent's state dir plus the /// container's writable rootfs, excluding the shared read-only nix /// store — in bytes. `None` until the slow background sampler /// ([`disk_sampler_loop`]) has populated the cache at least once. /// Decoupled from this hot per-poll path because a `du` tree-walk is /// far too expensive at the 5s `/api/container-load` cadence. pub disk_bytes: Option, } /// Cadence for the slow disk sampler. Disk size changes slowly relative /// to CPU/mem, and a `du` walk is expensive, so this runs far less often /// than the per-poll cgroup reads — the row just carries the last value. const DISK_SAMPLE_INTERVAL: Duration = Duration::from_mins(5); /// Root of the per-container writable rootfs trees that nixos-containers /// manages (`/var/lib/nixos-containers/`). const NIXOS_CONTAINERS_ROOT: &str = "/var/lib/nixos-containers"; /// Last-sampled per-agent disk usage (bytes), keyed by agent name. /// Written by [`disk_sampler_loop`] (~every 5 min), read by [`gather`] /// so the hot poll never runs `du`. fn disk_cache() -> &'static RwLock> { static CACHE: OnceLock>> = OnceLock::new(); CACHE.get_or_init(|| RwLock::new(HashMap::new())) } /// `du -sxb ` → apparent total bytes. `-s` summary, `-b` apparent /// size, `-x` stay on one filesystem so the walk never descends into the /// bind-mounted read-only shared nix store (or any other bind mount) — /// exactly the per-container-only measure we want. Best-effort: a /// missing/unreadable path (e.g. a stopped container with no rootfs) /// yields `None`, which the caller treats as 0. async fn du_bytes(path: &std::path::Path) -> Option { let out = tokio::process::Command::new("du") .arg("-sxb") .arg(path) .output() .await .ok()?; if !out.status.success() { return None; } let stdout = String::from_utf8_lossy(&out.stdout); stdout.split_whitespace().next()?.parse::().ok() } /// Total per-agent disk: the agent's host-side state dir /// (`/var/lib/hyperhive/agents//state`) plus the container's writable /// rootfs (`/var/lib/nixos-containers/h-`). Each measured with `du -sxb` /// so the shared nix store and other bind mounts are excluded. A path that /// doesn't exist contributes 0. /// /// The state dir is resolved via [`Coordinator::agent_notes_dir`] — the same /// host path the dashboard reads agent state from. Hardcoding the /// container-internal bind-mount path (`/agents//state`) instead made /// `du` fail host-side (that path only exists inside the container), so every /// agent's state-dir contribution was 0; with the writable rootfs nearly empty /// (almost everything is bind-mounted), that surfaced as all agents reporting /// 0 disk. async fn measure_agent_disk(name: &str) -> u64 { let state_dir = Coordinator::agent_notes_dir(name); let rootfs = PathBuf::from(format!("{NIXOS_CONTAINERS_ROOT}/h-{name}")); let mut total = 0u64; for path in [state_dir, rootfs] { if let Some(bytes) = du_bytes(&path).await { total = total.saturating_add(bytes); } } total } /// Background loop: every [`DISK_SAMPLE_INTERVAL`], recompute disk usage /// for every kept-state agent and refresh [`disk_cache`]. The first pass /// runs immediately so the dashboard shows disk shortly after boot. /// Runs forever; spawn once at hive-c0re startup via [`spawn_disk_sampler`]. pub async fn disk_sampler_loop() { loop { for name in Coordinator::kept_state_names() { let bytes = measure_agent_disk(&name).await; if let Ok(mut cache) = disk_cache().write() { cache.insert(name, bytes); } } sleep(DISK_SAMPLE_INTERVAL).await; } } /// Spawn the slow disk sampler as a detached background task. Called once /// from `main` alongside the other host-side loops. pub fn spawn_disk_sampler() { tokio::spawn(disk_sampler_loop()); } /// Cgroup directory for a container's machine. /// /// nixos-container runs `systemd-nspawn --keep-unit` with /// `Slice = "machine.slice"` (see nixpkgs /// `virtualisation/nixos-containers.nix`). `--keep-unit` means nspawn does /// **not** create a separate machined `machine-.scope` — the /// container's cgroup *is* the launching service unit, /// `container@.service`, placed under `machine.slice`. /// systemd-machined still logs "New machine " (registration), but the /// cgroup stays on the service unit. So the path is /// `machine.slice/container@.service`, and the service unit name is /// used verbatim — no `\x2d` escaping (that only applies when a string is /// converted *into* a scope/slice unit name, not to an already-formed /// instance unit; the journal shows the literal `container@h-.service`). fn scope_dir(machine: &str) -> PathBuf { PathBuf::from(MACHINE_SLICE).join(format!("container@{machine}.service")) } /// Read a single unsigned integer from a one-line cgroup file. fn read_u64(path: &std::path::Path) -> Option { std::fs::read_to_string(path) .ok()? .trim() .parse::() .ok() } /// `memory.max` reads `max` when there's no limit — map that to `None`. fn read_mem_max(path: &std::path::Path) -> Option { let s = std::fs::read_to_string(path).ok()?; let s = s.trim(); if s == "max" { None } else { s.parse::().ok() } } /// Parse `usage_usec` (cumulative CPU time, microseconds) from a /// scope's `cpu.stat`. fn read_usage_usec(dir: &std::path::Path) -> Option { let text = std::fs::read_to_string(dir.join("cpu.stat")).ok()?; for line in text.lines() { if let Some(rest) = line.strip_prefix("usage_usec ") { return rest.trim().parse::().ok(); } } None } /// Total online host CPUs, for normalising CPU% to "% of total host". /// Reads `/sys/devices/system/cpu/online` (a comma-separated list of /// ranges, e.g. `0-19`) rather than `available_parallelism()`, which /// reflects the hive-core process's CPU affinity — if hive-core is ever /// affinity-pinned to a subset, that would under-count the denominator /// and inflate the percentage. Falls back to the process count, then 1. fn host_nproc() -> usize { fn from_sysfs() -> Option { let s = std::fs::read_to_string("/sys/devices/system/cpu/online").ok()?; let mut count = 0usize; for part in s.trim().split(',') { let part = part.trim(); if part.is_empty() { continue; } if let Some((a, b)) = part.split_once('-') { let a: usize = a.trim().parse().ok()?; let b: usize = b.trim().parse().ok()?; count += b.saturating_sub(a) + 1; } else { part.parse::().ok()?; count += 1; } } (count > 0).then_some(count) } from_sysfs() .or_else(|| std::thread::available_parallelism().ok().map(usize::from)) .unwrap_or(1) } /// Sample CPU + memory for every running agent container. Best-effort: /// agents whose scope dir is absent (not running) or unreadable are /// skipped, never fatal. Sorted by name for a stable display order. pub async fn gather() -> Vec { // Machines that currently have a cgroup scope (= running). Agent // machine name is `h-`. let candidates: Vec<(String, PathBuf)> = Coordinator::kept_state_names() .into_iter() .filter_map(|name| { let dir = scope_dir(&format!("h-{name}")); dir.join("cpu.stat").exists().then_some((name, dir)) }) .collect(); // First CPU sample for all, then one shared sleep, then the second // — so N agents cost one interval, not N. let t0: Vec> = candidates.iter().map(|(_, d)| read_usage_usec(d)).collect(); sleep(CPU_SAMPLE).await; #[allow( clippy::cast_precision_loss, reason = "CPU-utilisation math; the operands (core count, microsecond sample interval, cgroup time delta) stay well under f64's 2^53 exact-integer range, so no precision is lost" )] let nproc = host_nproc() as f64; #[allow( clippy::cast_precision_loss, reason = "CPU-utilisation math; the operands (core count, microsecond sample interval, cgroup time delta) stay well under f64's 2^53 exact-integer range, so no precision is lost" )] let interval_usec = CPU_SAMPLE.as_micros() as f64; // Snapshot the slow disk cache once (cheap clone of a small map) so // the per-agent loop below is a plain map lookup — no `du` on this // hot path. Agents not yet sampled simply carry `disk_bytes = None`. let disk = disk_cache().read().map(|c| c.clone()).unwrap_or_default(); let mut out: Vec = Vec::with_capacity(candidates.len()); for (i, (name, dir)) in candidates.iter().enumerate() { let cpu_pct = match (t0[i], read_usage_usec(dir)) { (Some(a), Some(b)) => { #[allow( clippy::cast_precision_loss, reason = "CPU-utilisation math; the operands (core count, microsecond sample interval, cgroup time delta) stay well under f64's 2^53 exact-integer range, so no precision is lost" )] let delta = b.saturating_sub(a) as f64; (delta / (interval_usec * nproc)) * 100.0 } _ => 0.0, }; out.push(ContainerResource { name: name.clone(), cpu_pct, mem_current_bytes: read_u64(&dir.join("memory.current")).unwrap_or(0), mem_peak_bytes: read_u64(&dir.join("memory.peak")), mem_max_bytes: read_mem_max(&dir.join("memory.max")), disk_bytes: disk.get(name).copied(), }); } out } #[cfg(test)] mod tests { use super::*; #[test] fn scope_dir_is_the_keep_unit_service_under_machine_slice() { // nixos-container `--keep-unit` keeps the cgroup on the service // unit `container@.service` (literal name, no `\x2d`), // under machine.slice — NOT a machined `machine-.scope`. assert_eq!( scope_dir("h-atlas"), PathBuf::from("/sys/fs/cgroup/machine.slice/container@h-atlas.service") ); // Underscores in agent names are likewise verbatim. assert_eq!( scope_dir("h-foo_bar"), PathBuf::from("/sys/fs/cgroup/machine.slice/container@h-foo_bar.service") ); } }