container stats: bound hive-c0re's wait for the disk measurement, no total for an agent with no paths
hive-c0re waited on hive-priv's answer with no timeout, so a hung hive-priv would stall the disk sampler indefinitely. The wait is now hive-priv's own 10 min bound plus a minute (the bound is shared as MEASURE_AGENT_DISK_TIMEOUT); a timeout is an error, so the agent shows unknown. An agent with none of state/, work/ or the rootfs has no measurable size. hive-priv now answers with an error for it instead of a total of 0, which read as a real measurement.
This commit is contained in:
parent
98ccc31e2b
commit
960cf9ce30
3 changed files with 131 additions and 36 deletions
|
|
@ -8,8 +8,9 @@
|
|||
|
||||
use anyhow::{Context as _, Result, bail};
|
||||
use hive_priv_sock::{
|
||||
BindMount, CredentialMount, InfraAction, InfraContainer, JournalQuery, NetworkIsolation,
|
||||
PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse, PrivStream,
|
||||
BindMount, CredentialMount, InfraAction, InfraContainer, JournalQuery,
|
||||
MEASURE_AGENT_DISK_TIMEOUT, NetworkIsolation, PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse,
|
||||
PrivStream,
|
||||
};
|
||||
use std::os::fd::{AsRawFd as _, OwnedFd, RawFd};
|
||||
|
||||
|
|
@ -429,25 +430,44 @@ pub async fn read_subvolume_usage(agent_name: &str) -> Result<(u64, u64)> {
|
|||
parse_qgroup_usage(&stdout).with_context(|| format!("parse qgroup usage for {agent_name}"))
|
||||
}
|
||||
|
||||
/// How long hive-c0re waits for hive-priv's answer to a disk measurement:
|
||||
/// hive-priv's own bound plus a margin, so a hung hive-priv can't stall the
|
||||
/// caller while a slow walk still arrives as hive-priv's timeout error.
|
||||
const MEASURE_AGENT_DISK_WAIT: std::time::Duration =
|
||||
MEASURE_AGENT_DISK_TIMEOUT.saturating_add(std::time::Duration::from_mins(1));
|
||||
|
||||
/// Measure an agent's disk usage in bytes (`state/` + `work/` + container
|
||||
/// rootfs, one bounded `du` run by hive-priv as root).
|
||||
///
|
||||
/// # Errors
|
||||
/// Returns an error if the hive-priv call fails, the `du` walk times out or
|
||||
/// exits non-zero, or the returned total isn't a number.
|
||||
/// Returns an error if the hive-priv call fails or doesn't answer within
|
||||
/// [`MEASURE_AGENT_DISK_WAIT`], the `du` walk times out or exits non-zero,
|
||||
/// the agent has none of the paths, or the returned total isn't a number.
|
||||
pub async fn measure_agent_disk(agent_name: &str) -> Result<u64> {
|
||||
let (stdout, _) = check(
|
||||
let resp = within(
|
||||
MEASURE_AGENT_DISK_WAIT,
|
||||
call(&PrivRequest::MeasureAgentDisk {
|
||||
agent_name: agent_name.to_owned(),
|
||||
})
|
||||
.await?,
|
||||
)?;
|
||||
}),
|
||||
)
|
||||
.await?;
|
||||
let (stdout, _) = check(resp)?;
|
||||
stdout
|
||||
.trim()
|
||||
.parse()
|
||||
.with_context(|| format!("parse disk usage for {agent_name}: {stdout:?}"))
|
||||
}
|
||||
|
||||
/// Await `fut`, failing if it takes longer than `limit`.
|
||||
async fn within<T>(
|
||||
limit: std::time::Duration,
|
||||
fut: impl std::future::Future<Output = Result<T>>,
|
||||
) -> Result<T> {
|
||||
tokio::time::timeout(limit, fut)
|
||||
.await
|
||||
.map_err(|_| anyhow::anyhow!("hive-priv did not answer within {}s", limit.as_secs()))?
|
||||
}
|
||||
|
||||
/// Set or clear a btrfs qgroup size limit on an agent's state subvolume.
|
||||
/// `limit_bytes = None` clears it. Requires quota enabled. Via hive-priv.
|
||||
///
|
||||
|
|
@ -577,3 +597,26 @@ fn ok(resp: PrivResponse) -> Result<()> {
|
|||
check(resp)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::within;
|
||||
use std::time::Duration;
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_silent_hive_priv_times_out_as_an_error() {
|
||||
let err = within(
|
||||
Duration::from_millis(20),
|
||||
std::future::pending::<Result<u64, _>>(),
|
||||
)
|
||||
.await
|
||||
.expect_err("a call that never answers must fail");
|
||||
assert!(err.to_string().contains("did not answer"), "{err}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn an_answer_inside_the_limit_passes_through() {
|
||||
let got = within(Duration::from_secs(5), async { Ok(7_u64) }).await;
|
||||
assert_eq!(got.expect("answered in time"), 7);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -291,6 +291,11 @@ pub struct NetworkIsolation {
|
|||
pub gateway_ip: String,
|
||||
}
|
||||
|
||||
/// Wall-clock bound hive-priv puts on one [`PrivRequest::MeasureAgentDisk`]
|
||||
/// walk. A caller's own wait has to be longer than this, so a slow walk
|
||||
/// reaches it as hive-priv's timeout error rather than as its own.
|
||||
pub const MEASURE_AGENT_DISK_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(10);
|
||||
|
||||
/// A request to the privileged helper.
|
||||
///
|
||||
/// Wire format: one JSON object per line over `/run/hive/priv.sock`.
|
||||
|
|
@ -565,7 +570,8 @@ pub enum PrivRequest {
|
|||
/// bytes as a decimal string in `stdout`.
|
||||
///
|
||||
/// hive-priv derives all three paths from `agent_name`; the request
|
||||
/// carries no path. A path that doesn't exist is skipped. A timeout, a
|
||||
/// carries no path. A path that doesn't exist is skipped; an agent with
|
||||
/// none of the three has no measurable size and gets an error. A timeout, a
|
||||
/// non-zero `du` exit or unparseable output is an error, never a partial
|
||||
/// or zero total. Requires root: `hive-core` cannot read the rootfs or
|
||||
/// the agent-owned files under `state/` and `work/`.
|
||||
|
|
|
|||
|
|
@ -23,9 +23,9 @@ use std::path::{Path, PathBuf};
|
|||
use anyhow::{Context as _, Result, anyhow, bail};
|
||||
use hive_priv_sock::{
|
||||
AGENT_PREFIX, AGENT_RUNTIME_ROOT, AGENT_STATE_ROOT, BindMount, CredentialMount, InfraAction,
|
||||
InfraContainer, JournalQuery, META_DIR, MIGRATE_STAGING_ROOT, NetworkIsolation,
|
||||
PAUSED_MARKER_FILE, PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse, PrivStream,
|
||||
PrivStreamLine, SIBLING_CONTAINERS,
|
||||
InfraContainer, JournalQuery, MEASURE_AGENT_DISK_TIMEOUT, META_DIR, MIGRATE_STAGING_ROOT,
|
||||
NetworkIsolation, PAUSED_MARKER_FILE, PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse,
|
||||
PrivStream, PrivStreamLine, SIBLING_CONTAINERS,
|
||||
};
|
||||
use tokio::io::{AsyncWriteExt, BufReader};
|
||||
use tokio::net::unix::OwnedWriteHalf;
|
||||
|
|
@ -1684,10 +1684,6 @@ 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.
|
||||
|
|
@ -1703,10 +1699,8 @@ fn agent_disk_paths(agent_name: &str) -> [PathBuf; 3] {
|
|||
}
|
||||
|
||||
/// `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.
|
||||
/// returning the grand total in bytes. An agent with none of them has no
|
||||
/// measurable size, which is an error rather than a total of 0.
|
||||
async fn measure_agent_disk(agent_name: &str) -> Result<(String, String)> {
|
||||
let mut paths = Vec::new();
|
||||
for path in agent_disk_paths(agent_name) {
|
||||
|
|
@ -1716,18 +1710,33 @@ async fn measure_agent_disk(agent_name: &str) -> Result<(String, String)> {
|
|||
Err(e) => return Err(e).with_context(|| format!("stat {}", path.display())),
|
||||
}
|
||||
}
|
||||
// `du` with no operand would measure its working directory instead.
|
||||
let total = du_total(agent_name, du_command(&paths)?, MEASURE_AGENT_DISK_TIMEOUT).await?;
|
||||
Ok((total.to_string(), String::new()))
|
||||
}
|
||||
|
||||
/// `du -s -x -b -c -P -- <paths>`. `-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. Errors on no paths: `du` with no operand would
|
||||
/// measure its working directory instead.
|
||||
fn du_command(paths: &[PathBuf]) -> Result<Command> {
|
||||
if paths.is_empty() {
|
||||
return Ok(("0".to_owned(), String::new()));
|
||||
bail!("none of the agent's state, work or rootfs paths exist");
|
||||
}
|
||||
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 {
|
||||
cmd.args(["-s", "-x", "-b", "-c", "-P", "--"]).args(paths);
|
||||
Ok(cmd)
|
||||
}
|
||||
|
||||
/// Run `cmd` (a [`du_command`]) for at most `limit` and parse its total.
|
||||
/// The child is killed on timeout. A timeout, a spawn failure, a non-zero
|
||||
/// exit or output without a total is an error, never a partial total.
|
||||
async fn du_total(agent_name: &str, mut cmd: Command, limit: std::time::Duration) -> Result<u64> {
|
||||
cmd.kill_on_drop(true);
|
||||
let Ok(out) = tokio::time::timeout(limit, cmd.output()).await else {
|
||||
bail!(
|
||||
"du for {agent_name} timed out after {}s; killed it",
|
||||
DISK_USAGE_TIMEOUT.as_secs()
|
||||
limit.as_secs()
|
||||
);
|
||||
};
|
||||
let out = out.with_context(|| format!("spawn du for {agent_name}"))?;
|
||||
|
|
@ -1743,9 +1752,8 @@ async fn measure_agent_disk(agent_name: &str) -> Result<(String, String)> {
|
|||
);
|
||||
}
|
||||
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()))
|
||||
parse_du_total(&stdout)
|
||||
.ok_or_else(|| anyhow!("no `total` line in du output for {agent_name}: {stdout:?}"))
|
||||
}
|
||||
|
||||
/// The byte count on `du -c`'s final `<bytes>\ttotal` line. `None` when the
|
||||
|
|
@ -3178,16 +3186,17 @@ fn ensure_socket_dir_in(root: &Path, name: &str) -> Result<()> {
|
|||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{
|
||||
BindMount, BoundedRun, OwnedFd, PAUSED_MARKER_FILE, PrivRequest, StagedFile,
|
||||
BindMount, BoundedRun, Command, OwnedFd, PAUSED_MARKER_FILE, PrivRequest, StagedFile,
|
||||
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,
|
||||
du_command, du_total, 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};
|
||||
use std::time::Duration;
|
||||
|
||||
fn bind(container_path: &str) -> BindMount {
|
||||
BindMount {
|
||||
|
|
@ -4240,4 +4249,41 @@ mod tests {
|
|||
assert_eq!(parse_du_total("-1\ttotal\n"), None);
|
||||
assert_eq!(parse_du_total("100 total\n"), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn no_existing_paths_is_an_error_not_zero() {
|
||||
let err = du_command(&[]).expect_err("no operands must not run du");
|
||||
assert!(err.to_string().contains("none of"), "{err}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn non_zero_du_exit_is_an_error() {
|
||||
let err = du_total("iris", Command::new("false"), Duration::from_secs(5))
|
||||
.await
|
||||
.expect_err("a failed du must not yield a total");
|
||||
assert!(err.to_string().contains("failed"), "{err}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_walk_past_the_limit_is_killed_and_errors() {
|
||||
let mut cmd = Command::new("sleep");
|
||||
cmd.arg("30");
|
||||
let started = std::time::Instant::now();
|
||||
let err = du_total("iris", cmd, Duration::from_millis(50))
|
||||
.await
|
||||
.expect_err("a walk past the limit must fail");
|
||||
assert!(err.to_string().contains("timed out"), "{err}");
|
||||
assert!(started.elapsed() < Duration::from_secs(10));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn du_total_sums_a_real_walk() {
|
||||
let dir = std::env::temp_dir().join(format!("hive-priv-du-{}", std::process::id()));
|
||||
std::fs::create_dir_all(&dir).unwrap();
|
||||
std::fs::write(dir.join("f"), b"hello").unwrap();
|
||||
let cmd = du_command(std::slice::from_ref(&dir)).unwrap();
|
||||
let total = du_total("iris", cmd, Duration::from_secs(30)).await;
|
||||
std::fs::remove_dir_all(&dir).unwrap();
|
||||
assert!(total.unwrap() >= 5);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue