diff --git a/hive-c0re/src/stats/container_stats.rs b/hive-c0re/src/stats/container_stats.rs index 0cd39c967..747b26edd 100644 --- a/hive-c0re/src/stats/container_stats.rs +++ b/hive-c0re/src/stats/container_stats.rs @@ -64,6 +64,13 @@ pub struct ContainerResource { /// High-water memory usage since container start (`memory.peak`), /// bytes. `None` if the kernel doesn't expose `memory.peak`. pub mem_peak_bytes: Option, + /// Processes the kernel OOM-killed inside the container since it started + /// (`memory.events` `oom_kill`, cumulative, counting every descendant + /// cgroup). `None` if the file or the key is missing, never 0. + pub oom_kills: Option, + /// Memory currently swapped out (`memory.swap.current`), bytes. `None` + /// if the kernel doesn't expose it. + pub swap_current_bytes: Option, /// Memory ceiling (`memory.max`), bytes. `None` when unlimited /// (the file reads `max`). pub mem_max_bytes: Option, @@ -168,6 +175,19 @@ fn read_mem_max(path: &std::path::Path) -> Option { } } +/// `oom_kill` from a cgroup's `memory.events`, the count of processes the +/// kernel killed for memory in the cgroup and everything below it. +fn read_oom_kills(dir: &std::path::Path) -> Option { + oom_kills_in(&std::fs::read_to_string(dir.join("memory.events")).ok()?) +} + +/// The `oom_kill` value in `memory.events` text (`key value` lines). +fn oom_kills_in(events: &str) -> Option { + events + .lines() + .find_map(|line| line.strip_prefix("oom_kill ")?.trim().parse::().ok()) +} + /// Parse `usage_usec` (cumulative CPU time, microseconds) from a /// scope's `cpu.stat`. fn read_usage_usec(dir: &std::path::Path) -> Option { @@ -279,6 +299,8 @@ pub async fn gather() -> Vec { 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")), + oom_kills: read_oom_kills(dir), + swap_current_bytes: read_u64(&dir.join("memory.swap.current")), mem_max_bytes: read_mem_max(&dir.join("memory.max")), disk_bytes: agent.as_ref().and_then(|a| disk.get(a).copied().flatten()), }); @@ -338,6 +360,25 @@ mod tests { assert_eq!(running_candidates(slice.path(), &[]).len(), 1); } + const EVENTS: &str = "low 0\nhigh 0\nmax 12\noom 5\noom_kill 3\noom_group_kill 1\n"; + + #[test] + fn oom_kills_reads_the_oom_kill_key_not_its_neighbours() { + assert_eq!(oom_kills_in(EVENTS), Some(3)); + assert_eq!(oom_kills_in("oom_kill 0\n"), Some(0)); + // Neighbouring keys with the same prefix or value are not it. + assert_eq!(oom_kills_in("oom 5\noom_group_kill 1\n"), None); + assert_eq!(oom_kills_in("oom_kill x\n"), None); + } + + #[test] + fn a_missing_memory_events_is_no_sample_not_zero() { + let dir = tempfile::tempdir().expect("tempdir"); + assert_eq!(read_oom_kills(dir.path()), None); + // Control: the same reader finds the count once the file exists. + std::fs::write(dir.path().join("memory.events"), EVENTS).expect("write"); + assert_eq!(read_oom_kills(dir.path()), Some(3)); + } #[test] fn failed_disk_measurement_is_unknown_not_zero() { let name = hive_types::Ident::parse("iris").expect("valid ident"); diff --git a/hive-c0re/src/stats/otel_metrics.rs b/hive-c0re/src/stats/otel_metrics.rs index e19865248..d466a1e8a 100644 --- a/hive-c0re/src/stats/otel_metrics.rs +++ b/hive-c0re/src/stats/otel_metrics.rs @@ -138,6 +138,59 @@ fn build_provider( Ok(provider) } +/// The custom (`hyperhive.`) memory series: peak, OOM kills and swap. +fn register_memory_series( + meter: &opentelemetry::metrics::Meter, + snapshot: &Arc>>, +) { + let snap = snapshot.clone(); + meter + .u64_observable_gauge("hyperhive.container.memory.peak") + .with_unit("By") + .with_callback(move |obs| { + if let Ok(g) = snap.lock() { + for c in g.iter() { + if let Some(peak) = c.mem_peak_bytes { + obs.observe(peak, &attrs(c)); + } + } + } + }) + .build(); + + // Processes the kernel OOM-killed in the container (cumulative). An agent + // with a RAM budget has no swap, so reaching the budget ends in a kill. + let snap = snapshot.clone(); + meter + .u64_observable_counter("hyperhive.container.memory.oom_kills") + .with_unit("{process}") + .with_callback(move |obs| { + if let Ok(g) = snap.lock() { + for c in g.iter() { + if let Some(kills) = c.oom_kills { + obs.observe(kills, &attrs(c)); + } + } + } + }) + .build(); + + let snap = snapshot.clone(); + meter + .u64_observable_gauge("hyperhive.container.memory.swap.usage") + .with_unit("By") + .with_callback(move |obs| { + if let Ok(g) = snap.lock() { + for c in g.iter() { + if let Some(swap) = c.swap_current_bytes { + obs.observe(swap, &attrs(c)); + } + } + } + }) + .build(); +} + /// Register the observable instruments. Each callback reads the shared /// snapshot and reports one data point per agent. Instruments are held by the /// meter/provider (kept alive in `PROVIDER`). @@ -192,23 +245,10 @@ fn register_instruments(provider: &SdkMeterProvider, snapshot: Arc