hyperhive/hive-c0re/src/stats/container_stats.rs

//! Live per-agent-container resource load (CPU + memory, plus the slow
//! disk sample from hive-priv in [`disk_sampler_loop`]) for the OTEL
//! exporter. 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@<machine>.service` — not a machined
//! `machine-<name>.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/networking/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 utoipa::ToSchema;

use crate::coordinator::Coordinator;

/// Sampling interval for the two CPU reads. Short enough to keep a sample
/// quick, 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, ToSchema)]
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,
    /// Cumulative CPU time (`cpu.stat` `usage_usec`, microseconds,
    /// monotonic). `None` if the read failed. Feeds the OTEL semconv
    /// `container.cpu.time` counter (which wants absolute cumulative time,
    /// converted to seconds), distinct from the sampled `cpu_pct`.
    pub cpu_time_usec: Option<u64>,
    /// 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<u64>,
    /// Memory ceiling (`memory.max`), bytes. `None` when unlimited
    /// (the file reads `max`).
    pub mem_max_bytes: Option<u64>,
    /// Last-sampled total disk usage — the agent's `state/` and `work/` dirs
    /// plus the container's writable rootfs, excluding the shared read-only
    /// nix store — in bytes. `None` when it is unknown: the slow background
    /// sampler ([`disk_sampler_loop`]) has not measured this agent yet, or
    /// its last measurement failed. Never 0 for a failed measurement.
    pub disk_bytes: Option<u64>,
}

/// Cadence for the slow disk sampler. Each pass has hive-priv walk every
/// agent's full tree as root, which is real I/O on the disk the agents
/// work on, and the sizes move over hours, not minutes.
const DISK_SAMPLE_INTERVAL: Duration = Duration::from_mins(30);

/// Last per-agent disk measurement, keyed by agent name; `None` is a failed
/// one. Written by [`disk_sampler_loop`], read by [`gather`], so a sample
/// never measures.
fn disk_cache() -> &'static RwLock<HashMap<String, Option<u64>>> {
    static CACHE: OnceLock<RwLock<HashMap<String, Option<u64>>>> = OnceLock::new();
    CACHE.get_or_init(|| RwLock::new(HashMap::new()))
}

/// Total disk usage of `name`'s `state/` + `work/` + container rootfs,
/// measured by hive-priv as root (`hive-core` can read none of the rootfs
/// and not all of the agent-owned dirs). `None` on any failure.
async fn measure_agent_disk(name: &hive_types::Ident) -> Option<u64> {
    disk_reading(
        name,
        crate::priv_client::measure_agent_disk(name.as_str()).await,
    )
}

/// A failed measurement is unknown, never a 0 contribution.
fn disk_reading(name: &hive_types::Ident, measured: anyhow::Result<u64>) -> Option<u64> {
    measured
        .inspect_err(|e| {
            tracing::warn!(agent = %name, error = %format!("{e:#}"), "disk usage measurement failed");
        })
        .ok()
}

/// Background loop: every [`DISK_SAMPLE_INTERVAL`], remeasure disk usage
/// for every kept-state agent and refresh [`disk_cache`]. The first pass
/// runs immediately so the exporter has 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.into_string(), 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-<name>.scope` — the
/// container's cgroup *is* the launching service unit,
/// `container@<machine>.service`, placed under `machine.slice`.
/// systemd-machined still logs "New machine `<name>`" (registration), but the
/// cgroup stays on the service unit. So the path is
/// `machine.slice/container@<machine>.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-<agent>.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<u64> {
    std::fs::read_to_string(path)
        .ok()?
        .trim()
        .parse::<u64>()
        .ok()
}

/// `memory.max` reads `max` when there's no limit — map that to `None`.
fn read_mem_max(path: &std::path::Path) -> Option<u64> {
    let s = std::fs::read_to_string(path).ok()?;
    let s = s.trim();
    if s == "max" {
        None
    } else {
        s.parse::<u64>().ok()
    }
}

/// Parse `usage_usec` (cumulative CPU time, microseconds) from a
/// scope's `cpu.stat`.
fn read_usage_usec(dir: &std::path::Path) -> Option<u64> {
    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::<u64>().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<usize> {
        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::<usize>().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<ContainerResource> {
    // Machines that currently have a cgroup scope (= running). Agent
    // machine name is `h-<state-dir name>`.
    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.into_string(), 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<Option<u64>> = candidates.iter().map(|(_, d)| read_usage_usec(d)).collect();
    sleep(CPU_SAMPLE).await;

    let nproc = f64::from(u32::try_from(host_nproc()).unwrap_or(u32::MAX));
    let interval_secs = CPU_SAMPLE.as_secs_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 measured carry `disk_bytes = None` too.
    let disk = disk_cache().read().map(|c| c.clone()).unwrap_or_default();

    let mut out: Vec<ContainerResource> = Vec::with_capacity(candidates.len());
    for (i, (name, dir)) in candidates.iter().enumerate() {
        // The second cumulative read — also exposed raw as `cpu_time_usec`
        // for the OTEL `container.cpu.time` counter (which wants absolute
        // cumulative CPU time, not the sampled percent).
        let usage_now = read_usage_usec(dir);
        let cpu_pct = match (t0[i], usage_now) {
            (Some(a), Some(b)) => {
                let delta_secs = Duration::from_micros(b.saturating_sub(a)).as_secs_f64();
                (delta_secs / (interval_secs * nproc)) * 100.0
            }
            _ => 0.0,
        };
        out.push(ContainerResource {
            name: name.clone(),
            cpu_pct,
            cpu_time_usec: usage_now,
            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().flatten(),
        });
    }
    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@<machine>.service` (literal name, no `\x2d`),
        // under machine.slice — NOT a machined `machine-<name>.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")
        );
    }

    #[test]
    fn failed_disk_measurement_is_unknown_not_zero() {
        let name = hive_types::Ident::parse("iris").expect("valid ident");
        assert_eq!(
            disk_reading(&name, Err(anyhow::anyhow!("du for iris timed out"))),
            None
        );
        assert_eq!(disk_reading(&name, Ok(0)), Some(0));
        assert_eq!(
            disk_reading(&name, Ok(13_000_000_000)),
            Some(13_000_000_000)
        );
    }
}