host stats: read pressure stall totals for the host and each container

hive-c0re/src/stats/container_stats.rs

@@ -71,6 +71,9 @@ pub struct ContainerResource {
     /// Memory currently swapped out (`memory.swap.current`), bytes. `None`
     /// if the kernel doesn't expose it.
     pub swap_current_bytes: Option<u64>,
+    /// Memory pressure-stall totals (`memory.pressure`). `None` if the
+    /// kernel has no PSI or the cgroup file.
+    pub mem_pressure: Option<crate::host_stats::Pressure>,
     /// Memory ceiling (`memory.max`), bytes. `None` when unlimited
     /// (the file reads `max`).
     pub mem_max_bytes: Option<u64>,
@@ -301,6 +304,7 @@ pub async fn gather() -> Vec<ContainerResource> {
             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_pressure: crate::host_stats::read_pressure(&dir.join("memory.pressure")),
             mem_max_bytes: read_mem_max(&dir.join("memory.max")),
             disk_bytes: agent.as_ref().and_then(|a| disk.get(a).copied().flatten()),
         });

hive-c0re/src/stats/host_stats.rs

@@ -8,7 +8,9 @@
 //! after. [`refresh_disk_warning`] drives a held guard from a periodic task
 //! (see `spawn` in `main.rs`). The same probe backs the
 //! `hyperhive.hive.nix_store.utilization` gauge ([`disk_utilization`]), which is
-//! what the early-warning Grafana alerts read.
+//! what the early-warning Grafana alerts read. Pressure-stall totals from
+//! `/proc/pressure` back the `hyperhive.hive.pressure.stall` counter
+//! ([`host_pressure`]).
 //!
 //! The registry has no dashboard banner any more. A raised warning reaches
 //! the operator as the `hyperhive.hive.warning` gauge and a Grafana alert
@@ -91,6 +93,73 @@ pub fn disk_utilization() -> Option<f64> {
     nix_disk_usage().map(|d| d.used_pct / 100.0)
 }
 
+/// Cumulative stall time of one pressure file (`/proc/pressure/*` or a
+/// cgroup's `memory.pressure`), in microseconds. `None` for a line the file
+/// does not carry: `cpu` has no `full` line before kernel 5.13.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, ToSchema)]
+pub struct Pressure {
+    /// `total` of the `some` line: time at least one task was stalled.
+    pub some_us: Option<u64>,
+    /// `total` of the `full` line: time all non-idle tasks were stalled.
+    pub full_us: Option<u64>,
+}
+
+/// Parse PSI text (`some avg10=… avg60=… avg300=… total=N` plus an optional
+/// `full` line). `None` when no `some`/`full` line is present or one of them
+/// has no numeric `total=`; unknown line kinds are ignored.
+#[must_use]
+pub fn parse_pressure(text: &str) -> Option<Pressure> {
+    let mut out = Pressure {
+        some_us: None,
+        full_us: None,
+    };
+    for line in text.lines() {
+        let slot = match line.split_whitespace().next() {
+            Some("some") => &mut out.some_us,
+            Some("full") => &mut out.full_us,
+            _ => continue,
+        };
+        let total = line
+            .split_whitespace()
+            .find_map(|f| f.strip_prefix("total="))?;
+        *slot = Some(total.parse::<u64>().ok()?);
+    }
+    (out.some_us.is_some() || out.full_us.is_some()).then_some(out)
+}
+
+/// Read and parse a pressure file; `None` when it is missing, unreadable or
+/// malformed.
+#[must_use]
+pub fn read_pressure(path: &std::path::Path) -> Option<Pressure> {
+    parse_pressure(&std::fs::read_to_string(path).ok()?)
+}
+
+/// The resources the kernel reports host-wide pressure for.
+const PRESSURE_RESOURCES: [&str; 3] = ["cpu", "memory", "io"];
+
+/// Host-wide pressure per resource (`cpu`, `memory`, `io`) from
+/// `/proc/pressure`. A resource whose file cannot be read is left out; a
+/// kernel without PSI (missing file, or `EOPNOTSUPP` under `psi=0`) logs one
+/// debug line for the process lifetime and yields an empty list.
+#[must_use]
+pub fn host_pressure() -> Vec<(&'static str, Pressure)> {
+    host_pressure_in(std::path::Path::new("/proc/pressure"))
+}
+
+fn host_pressure_in(dir: &std::path::Path) -> Vec<(&'static str, Pressure)> {
+    static LOGGED: std::sync::Once = std::sync::Once::new();
+    let out: Vec<_> = PRESSURE_RESOURCES
+        .into_iter()
+        .filter_map(|r| Some((r, read_pressure(&dir.join(r))?)))
+        .collect();
+    if out.is_empty() {
+        LOGGED.call_once(|| {
+            tracing::debug!(dir = %dir.display(), "no pressure stall information; skipping PSI series");
+        });
+    }
+    out
+}
+
 /// The held `pending_logins` warning; see [`refresh_pending_logins`].
 static PENDING_LOGINS: std::sync::Mutex<Option<crate::warnings::WarningGuard>> =
     std::sync::Mutex::new(None);
@@ -355,6 +424,51 @@ mod tests {
         );
     }
 
+    const MEMORY_PSI: &str = "some avg10=0.12 avg60=0.34 avg300=0.56 total=123456\n\
+                              full avg10=0.00 avg60=0.01 avg300=0.02 total=7890\n";
+
+    #[test]
+    fn pressure_parses_some_and_full_totals() {
+        assert_eq!(
+            parse_pressure(MEMORY_PSI),
+            Some(Pressure {
+                some_us: Some(123_456),
+                full_us: Some(7890)
+            })
+        );
+    }
+
+    #[test]
+    fn pressure_without_a_full_line_has_no_full_total() {
+        let cpu = "some avg10=1.00 avg60=2.00 avg300=3.00 total=42\n";
+        assert_eq!(
+            parse_pressure(cpu),
+            Some(Pressure {
+                some_us: Some(42),
+                full_us: None
+            })
+        );
+    }
+
+    #[test]
+    fn malformed_pressure_is_no_sample() {
+        assert_eq!(parse_pressure(""), None);
+        assert_eq!(parse_pressure("some avg10=0.00 total=abc\n"), None);
+        assert_eq!(parse_pressure("some avg10=0.00 avg60=0.00\n"), None);
+        assert_eq!(parse_pressure("garbage\n"), None);
+    }
+
+    #[test]
+    fn host_pressure_skips_missing_files_and_keeps_present_ones() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        assert!(host_pressure_in(dir.path()).is_empty(), "no PSI files");
+        std::fs::write(dir.path().join("memory"), MEMORY_PSI).expect("write");
+        let got = host_pressure_in(dir.path());
+        assert_eq!(got.len(), 1, "only the present file: {got:?}");
+        assert_eq!(got[0].0, "memory");
+        assert_eq!(got[0].1.some_us, Some(123_456));
+    }
+
     /// The gauge value is a fraction, so a percent leaking through would put
     /// every alert threshold 100x off.
     #[test]

hive-c0re/src/stats/otel_metrics.rs

@@ -175,6 +175,24 @@ fn register_memory_series(
         })
         .build();
 
+    // Cumulative time tasks in the container's cgroup stalled on memory
+    // (`memory.pressure`); the rate is the stall fraction, which rises when
+    // the container is throttled at `MemoryHigh`.
+    let snap = snapshot.clone();
+    meter
+        .u64_observable_counter("hyperhive.container.memory.pressure.stall")
+        .with_unit("us")
+        .with_callback(move |obs| {
+            if let Ok(g) = snap.lock() {
+                for c in g.iter() {
+                    if let Some(p) = c.mem_pressure {
+                        observe_pressure(obs, p, &attrs(c));
+                    }
+                }
+            }
+        })
+        .build();
+
     let snap = snapshot.clone();
     meter
         .u64_observable_gauge("hyperhive.container.memory.swap.usage")
@@ -191,6 +209,21 @@ fn register_memory_series(
         .build();
 }
 
+/// Report `some`/`full` stall totals as `kind`-labelled points.
+fn observe_pressure(
+    obs: &dyn opentelemetry::metrics::AsyncInstrument<u64>,
+    p: crate::host_stats::Pressure,
+    base: &[KeyValue],
+) {
+    for (kind, total) in [("some", p.some_us), ("full", p.full_us)] {
+        if let Some(us) = total {
+            let mut kvs = base.to_vec();
+            kvs.push(KeyValue::new("kind", kind));
+            obs.observe(us, &kvs);
+        }
+    }
+}
+
 /// 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`).
@@ -463,6 +496,20 @@ fn register_hive_instruments(meter: &opentelemetry::metrics::Meter) {
             }
         })
         .build();
+
+    // Host-wide stall totals from `/proc/pressure`, by `resource` (`cpu`,
+    // `memory`, `io`) and `kind` (`some`, `full`). Counters in microseconds,
+    // so a rate over them is the stalled fraction of wall time. No sample on a
+    // kernel without PSI.
+    meter
+        .u64_observable_counter("hyperhive.hive.pressure.stall")
+        .with_unit("us")
+        .with_callback(|obs| {
+            for (resource, p) in crate::host_stats::host_pressure() {
+                observe_pressure(obs, p, &[KeyValue::new("resource", resource)]);
+            }
+        })
+        .build();
 }
 
 /// Attributes for the hive-scoped family: **none**.
@@ -629,6 +676,7 @@ mod tests {
             mem_peak_bytes: None,
             oom_kills: None,
             swap_current_bytes: None,
+            mem_pressure: None,
             mem_max_bytes: None,
             disk_bytes: None,
         };
@@ -675,6 +723,7 @@ mod tests {
             mem_peak_bytes: None,
             oom_kills: None,
             swap_current_bytes: None,
+            mem_pressure: None,
             mem_max_bytes: None,
             disk_bytes: None,
         };