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

remove the hivectl-only quota verbs and the trollshell status feed

Browse source
This commit is contained in:
atlas 2026-11-13 16:00:30 +01:00
commit cf360b74b9
3 changed files with 51 additions and 521 deletions
Download patch file Download diff file Expand all files Collapse all files

15
hive-c0re/src/container_view.rs
View file

@ -55,9 +55,7 @@ pub struct ContainerView {
impl From<ContainerView> for hive_sh4re::container::AgentStatusRow {
/// The one place a `ContainerView` becomes the wire row `host.sock`
/// serves it as — `handle_agent_status`'s poll path and the
/// `SubscribeAgentStatus` push path both go through this, so the two
/// can never quietly diverge on which fields make the cut.
/// serves it as (`handle_agent_status`).
///
/// `pending_reminders` has no source here: reminders are agent-local
/// now, and c0re has no cross-agent visibility into pending counts
@ -97,8 +95,8 @@ fn agent_url(name: &str) -> Option<String> {
/// is testable without an env var in the loop. `/agent/<name>/` is the
/// gateway's real path scheme
/// (`nix/host-modules/hive-c0re/vhosts.nix`'s per-agent `agentLocations`
/// block) — computed once here so a client (trollshell's sidebar, the
/// dashboard) never has to derive it and silently drift if that scheme
/// block) — computed once here so a client (e.g. the dashboard) never has
/// to derive it and silently drift if that scheme
/// ever changes.
fn agent_url_for_domain(domain: &str, name: &str) -> String {
format!("https://{domain}/agent/{name}/")
@ -362,10 +360,9 @@ mod tests {
}
}
/// The `From` impl both `handle_agent_status`'s poll path and
/// `stream_agent_status`'s push path go through — pins the field
/// mapping so the two can't quietly diverge, and that `port` (no
/// equivalent on the row) is dropped on purpose rather than by omission.
/// The `From` impl `handle_agent_status` goes through — pins the field
/// mapping, and that `port` (no equivalent on the row) is dropped on
/// purpose rather than by omission.
#[test]
fn agent_status_row_carries_every_field_the_row_has_and_drops_the_rest() {
use super::ContainerView;

86
hive-c0re/src/priv_client.rs
View file

@ -384,34 +384,6 @@ pub async fn delete_agent_subvolume(agent_name: &str) -> Result<()> {
.await?)
}
/// Enable btrfs qgroup accounting on the agent-state filesystem (operator
/// opt-in; prerequisite for usage reads + quotas). Idempotent; no-op off
/// btrfs. Via hive-priv (root).
///
/// # Errors
/// Returns an error if the hive-priv call fails or `btrfs quota enable`
/// reports a non-zero exit.
pub async fn ensure_btrfs_quota() -> Result<()> {
ok(call(&PrivRequest::EnsureBtrfsQuota).await?)
}
/// Read an agent state subvolume's btrfs qgroup usage, returning
/// `(referenced_bytes, exclusive_bytes)`. Via hive-priv (root).
///
/// # Errors
/// Returns an error if the hive-priv call fails, `btrfs qgroup show` exits
/// non-zero (e.g. quota not enabled — the message propagates so the caller
/// can surface it), or the output has no level-0 qgroup row to parse.
pub async fn read_subvolume_usage(agent_name: &str) -> Result<(u64, u64)> {
let (stdout, _) = check(
call(&PrivRequest::ReadSubvolumeUsage {
agent_name: agent_name.to_owned(),
})
.await?,
)?;
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.
@ -450,20 +422,6 @@ async fn within<T>(
.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.
///
/// # Errors
/// Returns an error if the hive-priv call fails or `btrfs qgroup limit`
/// reports a non-zero exit (e.g. quota not enabled).
pub async fn set_subvolume_quota(agent_name: &str, limit_bytes: Option<u64>) -> Result<()> {
ok(call(&PrivRequest::SetSubvolumeQuota {
agent_name: agent_name.to_owned(),
limit_bytes,
})
.await?)
}
/// Convert an existing plain-dir agent state root into a btrfs subvolume in
/// place (operator opt-in; via hive-priv as root). The caller must stop the
/// agent first (so its state bind-mount is gone) and restart it after.
@ -553,29 +511,6 @@ fn send_snapshot_request(
}
}
/// Parse `(referenced, exclusive)` bytes from `btrfs qgroup show -f --raw`
/// output (a qgroup row is `<id-with-slash> <rfer> <excl> …`).
///
/// `-f <path>` already restricts the listing to qgroups impacting that path
/// (excluding ancestral qgroups — see btrfs-qgroup-show(8)), so it never
/// mixes in other agents' subvolumes. Among the rows it returns we select
/// the **level-0** qgroup (`0/<subvolid>`) — the subvolume's own automatic
/// usage qgroup — rather than blindly taking the last line. Picking the
/// `0/` leaf is unambiguous even if an operator has assigned the subvolume
/// to a higher-level aggregate qgroup (`1/<id>`, …) that `-F` would surface.
fn parse_qgroup_usage(out: &str) -> Result<(u64, u64)> {
for line in out.lines() {
let cols: Vec<&str> = line.split_whitespace().collect();
if cols.len() >= 3
&& cols[0].starts_with("0/")
&& let (Ok(rfer), Ok(excl)) = (cols[1].parse::<u64>(), cols[2].parse::<u64>())
{
return Ok((rfer, excl));
}
}
bail!("no level-0 qgroup data row in `btrfs qgroup show` output: {out:?}")
}
fn check(resp: PrivResponse) -> Result<(String, String)> {
if resp.ok {
Ok((resp.stdout, resp.stderr))
@ -597,8 +532,8 @@ mod tests {
use super::{
BindMount, CredentialMount, HiveFacts, NetworkIsolation, NixBuilds, PrivRequest,
PrivResponse, PrivStream, call_at, call_streaming_at, check, create_container_request,
older_hive_priv_refused, parse_qgroup_usage, resource_limits_request,
send_snapshot_request, update_container_request, within, write_nspawn_flags_request,
older_hive_priv_refused, resource_limits_request, send_snapshot_request,
update_container_request, within, write_nspawn_flags_request,
};
use anyhow::Context as _;
use serde_json::{Value, json};
@ -748,23 +683,6 @@ mod tests {
assert_eq!(incremental["snapshot_name"], "snap-2");
}
#[test]
fn the_level_zero_qgroup_row_gives_the_usage() {
let out = "qgroupid rfer excl\n\
-------- ---- ----\n\
1/100 999 888\n\
0/257 4096 2048\n";
assert_eq!(parse_qgroup_usage(out).expect("a 0/ row"), (4096, 2048));
}
#[test]
fn output_with_no_level_zero_row_is_an_error() {
let err = parse_qgroup_usage("qgroupid rfer excl\n1/100 1 2\n").expect_err("no 0/ row");
assert!(err.to_string().contains("no level-0 qgroup"), "{err}");
// A 0/ row whose columns are not numbers is skipped, not misread.
parse_qgroup_usage("0/257 x y\n").expect_err("non-numeric columns");
}
#[test]
fn a_refused_response_surfaces_the_error_text() {
let refused = PrivResponse {

163
hive-c0re/src/server.rs
View file

@ -93,21 +93,12 @@ async fn handle_with(
continue;
}
};
// The one verb that answers with more than one response — see its
// doc comment. Hands `write` over and stops reading further
// requests on this connection; a client wanting anything else
// opens a fresh one.
if matches!(req, HostRequest::SubscribeAgentStatus) {
return stream_agent_status(write, coord).await;
}
let resp = dispatch(&req, coord.clone()).await;
write_response(&mut write, &resp).await?;
}
}
/// Serialize one `HostResponse` as a JSON line and flush it. Shared by the
/// ordinary one-response-per-request path and `stream_agent_status`'s
/// multi-response one, so the two can't quietly drift on framing.
/// Serialize one `HostResponse` as a JSON line and flush it.
async fn write_response(
write: &mut tokio::net::unix::OwnedWriteHalf,
resp: &HostResponse,
@ -167,25 +158,11 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
// see the request's doc comment for why the client asks.
HostRequest::AgentExists { name } => HostResponse::agent_exists(agent_exists(name)?),
HostRequest::AgentStatus => handle_agent_status().await,
// Intercepted in `handle()` before a request ever reaches
// `dispatch` — see `stream_agent_status`'s doc comment. This
// arm exists only so the match stays exhaustive; reaching it
// would mean a future caller invoked `dispatch` directly.
HostRequest::SubscribeAgentStatus => {
HostResponse::error("SubscribeAgentStatus is handled by the connection loop")
}
// The hive domain + per-surface public URLs are injected into
// c0re's service env by the hyperhive module; surface them so the
// operator CLI can fill in this hive's own identity (the
// federation peer-config block).
HostRequest::Urls => HostResponse::urls(hive_urls()),
HostRequest::QuotaEnable => handle_quota_enable().await?,
HostRequest::QuotaLimit { name, limit } => {
handle_quota_limit(name.as_str(), *limit).await?
}
HostRequest::QuotaShow { name } => {
handle_quota_show(name.as_ref().map(hive_types::Ident::as_str)).await?
}
HostRequest::UpgradeSubvolume { name } => {
handle_upgrade_subvolume(name.as_str()).await?
}
@ -264,49 +241,6 @@ async fn handle_agent_status() -> HostResponse {
}
}
/// `SubscribeAgentStatus` — ack once, then push one single-row
/// [`HostResponse::agent_statuses`] per agent-status change for as long as
/// the client stays connected. Consumes `write` (rather than borrowing it,
/// like every other handler here) because this is the one verb that keeps
/// writing after its own response, so `handle()` hands over ownership and
/// stops reading further requests on this connection once it calls this.
///
/// Rides the same `Coordinator.dashboard_events` broadcast channel the
/// dashboard's own SSE route reads (`dashboard/state_snapshot.rs`) —
/// `SetStatus` already triggers a `rescan_containers_and_emit` on every
/// status change (`socket_server/mod.rs::handle_set_status`), so that
/// channel already carries every event this needs; no separate plumbing.
async fn stream_agent_status(
mut write: tokio::net::unix::OwnedWriteHalf,
coord: Arc<Coordinator>,
) -> Result<()> {
let mut events = coord.dashboard_subscribe();
write_response(&mut write, &HostResponse::success()).await?;
loop {
match events.recv().await {
Ok(crate::dashboard_events::DashboardEvent::ContainerStateChanged {
container,
..
}) => {
let row = hive_sh4re::container::AgentStatusRow::from(container);
write_response(&mut write, &HostResponse::agent_statuses(vec![row])).await?;
}
// Every other event kind on this channel is dashboard-only
// (broker traffic, the queue, …) — nothing this
// subscriber asked for.
Ok(_) => {}
// Best-effort, same contract the dashboard's own live channel
// already has (see `HostRequest::SubscribeAgentStatus`'s doc
// comment) — a slow reader drops updates rather than stalling
// the broadcaster for everyone else on the channel.
Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
tracing::warn!(skipped, "agent-status subscribe: receiver lagged");
}
Err(tokio::sync::broadcast::error::RecvError::Closed) => return Ok(()),
}
}
}
/// True when `name` has a state dir under the agents root, i.e. it's a
/// managed agent rather than a bare (operator/human) account.
///
@ -320,67 +254,6 @@ fn agent_exists(name: &hive_types::Ident) -> Result<bool> {
.with_context(|| format!("check agent state dir for {name}"))
}
async fn handle_quota_enable() -> Result<HostResponse> {
crate::priv_client::ensure_btrfs_quota()
.await
.context("enable btrfs qgroup accounting")?;
Ok(HostResponse::messages(vec![
"btrfs qgroup accounting enabled on the agent-state filesystem.".to_owned(),
"(usage may read 0 until btrfs finishes its background rescan)".to_owned(),
]))
}
async fn handle_quota_limit(name: &str, limit: Option<u64>) -> Result<HostResponse> {
crate::priv_client::set_subvolume_quota(name, limit)
.await
.with_context(|| format!("set quota for {name}"))?;
// Bare success: the client prints the human-readable confirmation from
// the value it sent (it holds the `human_bytes` formatter).
Ok(HostResponse::success())
}
async fn handle_quota_show(name: Option<&str>) -> Result<HostResponse> {
let agents: Vec<String> = match name {
Some(n) => vec![n.to_owned()],
None => Coordinator::kept_state_names()
.into_iter()
.map(hive_types::Ident::into_string)
.collect(),
};
let mut rows = Vec::with_capacity(agents.len());
for agent in &agents {
match crate::priv_client::read_subvolume_usage(agent).await {
Ok((referenced, exclusive)) => rows.push(hive_host_sock::QuotaRow {
agent: agent.clone(),
referenced: Some(referenced),
exclusive: Some(exclusive),
note: None,
}),
Err(e) => {
let msg = format!("{e:#}");
// btrfs-progs prints "ERROR: ... quota not enabled" to stderr
// when qgroups are off; short-circuit the whole sweep with the
// enable hint (case-insensitive fragment match — the wording
// varies across btrfs-progs versions).
if msg.to_ascii_lowercase().contains("quota not enabled") {
return Ok(HostResponse::error(
"btrfs quota not enabled — run `hivectl quota-enable` first",
));
}
// A plain-dir agent (no subvolume) has no qgroup; note it
// inline and keep going rather than aborting the whole sweep.
rows.push(hive_host_sock::QuotaRow {
agent: agent.clone(),
referenced: None,
exclusive: None,
note: Some(format!("no qgroup data — plain dir or: {msg}")),
});
}
}
}
Ok(HostResponse::quota(rows))
}
async fn handle_upgrade_subvolume(name: &str) -> Result<HostResponse> {
crate::priv_client::upgrade_agent_subvolume(name)
.await
@ -748,12 +621,34 @@ mod tests {
/// `parse error` by the connection loop) while a served verb parses.
#[test]
fn a_removed_verb_is_a_parse_error_and_a_served_one_parses() {
let err =
serde_json::from_slice::<HostRequest>(br#"{"cmd":"list"}"#).expect_err("list is gone");
assert!(err.to_string().contains("unknown variant `list`"), "{err}");
assert!(
serde_json::from_slice::<HostRequest>(br#"{"cmd":"subscribe_agent_status"}"#).is_ok()
);
for gone in [
"list",
"subscribe_agent_status",
"quota_enable",
"quota_limit",
"quota_show",
] {
let line = format!(r#"{{"cmd":"{gone}"}}"#);
let err = serde_json::from_slice::<HostRequest>(line.as_bytes()).expect_err(gone);
assert!(
err.to_string()
.contains(&format!("unknown variant `{gone}`")),
"{gone}: {err}"
);
}
for served in [
br#"{"cmd":"agent_status"}"#.as_slice(),
br#"{"cmd":"upgrade_subvolume","name":"atlas"}"#,
br#"{"cmd":"snapshot_subvolume","name":"atlas","label":"hive-x"}"#,
br#"{"cmd":"delete_snapshot","name":"atlas","label":"hive-x"}"#,
br#"{"cmd":"send_snapshot","name":"atlas","label":"hive-x","dest":"out"}"#,
] {
assert!(
serde_json::from_slice::<HostRequest>(served).is_ok(),
"{}",
String::from_utf8_lossy(served)
);
}
}
/// A real `Coordinator` over a throwaway sqlite dir; `handle_with` takes