vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

container stats: measure agent disk usage through hive-priv, unknown on failure

Browse source 
The disk sampler ran `du -sxb` as the unprivileged hive-core user over
each agent's state/, work/ and container rootfs. hive-core cannot read
the rootfs or all of the agent-owned files, so `du` failed and each
failed path silently counted as 0: the dashboard showed a few MB for
agents using over 13 GB.

Add a hive-priv `MeasureAgentDisk { agent_name }` op. hive-priv
validates the name, derives the three paths itself (the request carries
no path), skips the ones that don't exist, and runs one
`du -s -x -b -c -P` as root with a 10 min timeout. A timeout, a non-zero
exit or output without a `total` line is an error.

hive-c0re calls it every 30 min instead of running `du` every 5 min:
each pass walks every agent's full tree as root on the disk the agents
work on, and the sizes move slowly. A failed measurement is cached as
None, so `disk_bytes` is null, the OTel storage.usage sample is omitted
and the dashboard shows "unknown", never 0.

`Coordinator::agent_work_dir` had no other caller and is removed.

Closes #4956
This commit is contained in:
atlas 2026-10-05 16:12:33 +02:00 • committed by mara
commit 98ccc31e2b
10 changed files with 244 additions and 179 deletions
Download patch file Download diff file Expand all files Collapse all files

146
hive-priv/src/main.rs
View file

@ -430,6 +430,10 @@ async fn exec(
limit_bytes,
} => exec_set_subvolume_quota(agent_name, limit_bytes).await,
PrivRequest::MeasureAgentDisk { ref agent_name } => {
exec_measure_agent_disk(agent_name).await
}
PrivRequest::UpgradeAgentSubvolume { ref agent_name } => {
exec_upgrade_agent_subvolume(agent_name).await
}
@ -558,6 +562,12 @@ async fn exec_read_subvolume_usage(agent_name: &str) -> Result<(String, String)>
read_subvolume_usage(agent_name).await
}
/// `MeasureAgentDisk`.
async fn exec_measure_agent_disk(agent_name: &str) -> Result<(String, String)> {
validate_agent_name(agent_name)?;
measure_agent_disk(agent_name).await
}
/// `SetSubvolumeQuota`.
async fn exec_set_subvolume_quota(
agent_name: &str,
@ -1674,6 +1684,82 @@ async fn read_subvolume_usage(agent_name: &str) -> Result<(String, String)> {
))
}
/// Wall-clock bound on one [`measure_agent_disk`] walk. It only stops a walk
/// wedged on I/O from holding the caller's sampler forever.
const DISK_USAGE_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(10);
/// The host paths `MeasureAgentDisk` sums for `agent_name`: its `state/`
/// and `work/` dirs under `AGENT_STATE_ROOT` and its container rootfs.
/// These are the only paths that op ever walks.
// `state` / `work` must match hive-c0re's `agent_notes_dir` and
// `paths::WORK_SUBDIR`.
fn agent_disk_paths(agent_name: &str) -> [PathBuf; 3] {
let agent_root = PathBuf::from(AGENT_STATE_ROOT).join(agent_name);
[
agent_root.join("state"),
agent_root.join("work"),
PathBuf::from("/var/lib/nixos-containers").join(container_system_name(agent_name)),
]
}
/// `MeasureAgentDisk` — one `du` over the [`agent_disk_paths`] that exist,
/// returning the grand total in bytes. `-x` keeps each walk on its
/// operand's filesystem (so the bind-mounted nix store is never counted),
/// `-P` never follows a symlink, `-b` counts apparent size, `-c` adds the
/// `total` line [`parse_du_total`] reads.
async fn measure_agent_disk(agent_name: &str) -> Result<(String, String)> {
let mut paths = Vec::new();
for path in agent_disk_paths(agent_name) {
match std::fs::symlink_metadata(&path) {
Ok(_) => paths.push(path),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
Err(e) => return Err(e).with_context(|| format!("stat {}", path.display())),
}
}
// `du` with no operand would measure its working directory instead.
if paths.is_empty() {
return Ok(("0".to_owned(), String::new()));
}
let mut cmd = Command::new("du");
cmd.args(["-s", "-x", "-b", "-c", "-P", "--"])
.args(&paths)
.kill_on_drop(true);
let Ok(out) = tokio::time::timeout(DISK_USAGE_TIMEOUT, cmd.output()).await else {
bail!(
"du for {agent_name} timed out after {}s; killed it",
DISK_USAGE_TIMEOUT.as_secs()
);
};
let out = out.with_context(|| format!("spawn du for {agent_name}"))?;
if !out.status.success() {
bail!(
"du for {agent_name} failed ({}): {}",
out.status,
String::from_utf8_lossy(&out.stderr)
.lines()
.last()
.unwrap_or("")
.trim()
);
}
let stdout = String::from_utf8_lossy(&out.stdout);
let total = parse_du_total(&stdout)
.ok_or_else(|| anyhow!("no `total` line in du output for {agent_name}: {stdout:?}"))?;
Ok((total.to_string(), String::new()))
}
/// The byte count on `du -c`'s final `<bytes>\ttotal` line. `None` when the
/// last line isn't that shape, so truncated or unexpected output is never
/// read as a smaller total.
fn parse_du_total(stdout: &str) -> Option<u64> {
let last = stdout.lines().rev().find(|l| !l.trim().is_empty())?;
let (bytes, label) = last.split_once('\t')?;
if label != "total" {
return None;
}
bytes.parse().ok()
}
/// Best-effort removal of a leftover migration path from a prior aborted
/// upgrade: try `btrfs subvolume delete` (in case it's a half-created
/// subvolume) then a plain recursive remove. Both failures are ignored —
@ -3093,11 +3179,12 @@ fn ensure_socket_dir_in(root: &Path, name: &str) -> Result<()> {
mod tests {
use super::{
BindMount, BoundedRun, OwnedFd, PAUSED_MARKER_FILE, PrivRequest, StagedFile,
check_fd_agreement, contains_secret_shaped_run, describe_forge_admin,
ensure_plain_filename, ensure_socket_dir_in, git_overlay_flags, limits_dropin_body,
open_dir, open_export_dest, partial_name, publish_file, redact_secret_line,
remove_marker_in, run_bounded, single_output_path, toplevel_attr, validate_credential_name,
validate_snapshot_name, write_agent_dir_file, write_bridge_dns_marker_in,
agent_disk_paths, check_fd_agreement, contains_secret_shaped_run, describe_forge_admin,
ensure_plain_filename, ensure_socket_dir_in, exec_measure_agent_disk, git_overlay_flags,
limits_dropin_body, open_dir, open_export_dest, parse_du_total, partial_name, publish_file,
redact_secret_line, remove_marker_in, run_bounded, single_output_path, toplevel_attr,
validate_credential_name, validate_snapshot_name, write_agent_dir_file,
write_bridge_dns_marker_in,
};
use std::path::PathBuf;
use std::sync::atomic::{AtomicU32, Ordering};
@ -4104,4 +4191,53 @@ mod tests {
"must not have touched the existing file"
);
}
#[test]
fn disk_paths_are_derived_from_the_name_only() {
assert_eq!(
agent_disk_paths("iris"),
[
PathBuf::from("/var/lib/hyperhive/agents/iris/state"),
PathBuf::from("/var/lib/hyperhive/agents/iris/work"),
PathBuf::from("/var/lib/nixos-containers/h-iris"),
]
);
}
/// Every name that could steer the walk out of the agent's own paths is
/// refused before any path is built, let alone `du` spawned.
#[tokio::test]
async fn measure_rejects_names_that_are_not_agent_names() {
for bad in [
"", "..", "../etc", "a/b", "/etc", "Iris", "a b", "a\0b", "foo_bar",
] {
let err = exec_measure_agent_disk(bad)
.await
.expect_err(&format!("{bad:?} must be rejected"));
assert!(
err.to_string().contains("invalid name"),
"{bad:?}: wrong error: {err}"
);
}
}
#[test]
fn du_total_reads_the_total_line() {
let out = "100\t/var/lib/hyperhive/agents/iris/state\n\
2000\t/var/lib/nixos-containers/h-iris\n\
2100\ttotal\n";
assert_eq!(parse_du_total(out), Some(2100));
assert_eq!(parse_du_total("0\ttotal"), Some(0));
}
#[test]
fn du_total_rejects_output_without_a_total_line() {
assert_eq!(parse_du_total(""), None);
assert_eq!(parse_du_total("\n"), None);
// A per-path line is not the total, even when it is the last one.
assert_eq!(parse_du_total("100\t/var/lib/x\n"), None);
assert_eq!(parse_du_total("abc\ttotal\n"), None);
assert_eq!(parse_du_total("-1\ttotal\n"), None);
assert_eq!(parse_du_total("100 total\n"), None);
}
}