use std::sync::Arc;
use gunnar_core::{NewRepo, RepoRef};
use gunnar_grpc::{ControlError, ControlPlane as _};
use gunnar_proto::{CapacityRequest, CapacityVerdict, SetGrowth};
use gunnar_server::control_plane::CoreControlPlane;
use gunnar_server::stores::ServerStores;
use gunnar_ssh::{DiskFloor, DiskHeadroomGate};
use prost::Message as _;
const MIB: u64 = 1024 * 1024;
const WRITTEN: u64 = 64 * MIB;
struct Rig {
_dir: tempfile::TempDir,
stores: Arc<ServerStores>,
}
impl Rig {
fn new() -> Rig {
let dir = tempfile::tempdir().expect("tempdir");
let stores = Arc::new(ServerStores::open(dir.path()).expect("stores"));
stores
.catalog()
.create_namespace("team")
.expect("namespace");
Rig { _dir: dir, stores }
}
fn gate(&self, floor: DiskFloor) -> Arc<DiskHeadroomGate> {
Arc::new(DiskHeadroomGate::new(self.stores.root(), floor))
}
fn plane(&self, floor: DiskFloor) -> CoreControlPlane {
CoreControlPlane::new(Arc::clone(&self.stores)).with_headroom(self.gate(floor))
}
fn burn(&self, name: &str, bytes: u64) {
use std::io::Write as _;
let path = self.stores.root().join(name);
let mut f = std::fs::File::create(&path).expect("create");
let chunk = vec![0x5au8; MIB as usize];
let mut left = bytes;
while left > 0 {
let n = left.min(MIB) as usize;
f.write_all(&chunk[..n]).expect("write");
left -= n as u64;
}
f.flush().expect("flush");
f.sync_all()
.expect("fsync: the bytes must be on the filesystem, not in a page cache");
}
}
fn ask(
plane: &CoreControlPlane,
requested: u64,
namespace: &str,
) -> gunnar_proto::CapacityResponse {
plane
.capacity(&CapacityRequest {
requested_bytes: requested,
namespace: namespace.to_string(),
disks: false,
})
.expect("a plane with a disk gate answers")
}
#[test]
fn the_free_bytes_are_a_live_statvfs_and_move_when_the_filesystem_moves() {
let rig = Rig::new();
let plane = rig.plane(DiskFloor {
percent: 0,
bytes: 1,
});
let before = ask(&plane, 1, "");
assert!(before.total_bytes > 0, "{before:?}");
assert!(
before.free_bytes > 0 && before.free_bytes < before.total_bytes,
"{before:?}"
);
assert_eq!(
before.servable_bytes,
before.free_bytes - before.floor_bytes,
"servable is free minus the floor; quoting `free` would sell the reserve"
);
let gate = rig.gate(DiskFloor {
percent: 0,
bytes: 1,
});
let room = gate.headroom().expect("statvfs on the store root");
assert_eq!(
before.total_bytes, room.total,
"the RPC and the disk gate must measure the SAME filesystem"
);
assert!(
before.free_bytes.abs_diff(room.free) < 4 * MIB,
"the RPC's free bytes and the gate's differ by more than a moment's noise: {} vs {}",
before.free_bytes,
room.free
);
rig.burn("ballast", WRITTEN);
let after = ask(&plane, 1, "");
let dropped = before.free_bytes.saturating_sub(after.free_bytes);
assert!(
dropped >= WRITTEN * 3 / 4,
"writing {WRITTEN} bytes must be visible in the answer; free went {} -> {} (a drop of \
{dropped}). A hardcoded constant cannot move, and neither can a statvfs of the wrong \
path.",
before.free_bytes,
after.free_bytes
);
assert_eq!(
after.total_bytes, before.total_bytes,
"the filesystem did not change size, only its free space"
);
assert_eq!(
before.servable_bytes.saturating_sub(after.servable_bytes),
dropped,
"servable must move with free, byte for byte"
);
assert_eq!(after.root, rig.stores.root().display().to_string());
}
#[test]
fn a_raise_beyond_the_servable_bytes_is_answered_no_with_the_shortfall_named() {
let rig = Rig::new();
let plane = rig.plane(DiskFloor {
percent: 0,
bytes: 1,
});
let yes = ask(&plane, 1, "");
assert_eq!(
yes.verdict,
CapacityVerdict::CapacityAvailable as i32,
"{}",
yes.reason
);
assert_eq!(yes.shortfall_bytes, 0);
assert!(yes.reason.contains("YES"), "{}", yes.reason);
let huge = 1024 * 1024 * 1024 * 1024 * 1024u64;
let no = ask(&plane, huge, "");
assert_eq!(
no.verdict,
CapacityVerdict::CapacityExhausted as i32,
"{}",
no.reason
);
assert_eq!(
no.additional_bytes, huge,
"no namespace was named, so the whole cap is new"
);
assert_eq!(
no.shortfall_bytes,
huge - no.servable_bytes,
"the shortfall is what is MISSING, computed from the measured servable bytes"
);
assert!(no.shortfall_bytes > 0);
assert!(
no.reason.contains("NO."),
"the sentence must say no: {}",
no.reason
);
assert!(
no.reason
.contains(&gunnar_ssh::headroom::human(no.servable_bytes)),
"and must name what CAN be served: {}",
no.reason
);
let free = no.free_bytes;
let half = rig.plane(DiskFloor {
percent: 0,
bytes: free / 2,
});
let over = ask(&half, free / 4 * 3, "");
assert_eq!(
over.verdict,
CapacityVerdict::CapacityExhausted as i32,
"{} of {} free was accepted with a {} floor in force: the reserve was sold",
gunnar_ssh::headroom::human(free / 4 * 3),
gunnar_ssh::headroom::human(free),
gunnar_ssh::headroom::human(free / 2),
);
let under = ask(&half, free / 4, "");
assert_eq!(
under.verdict,
CapacityVerdict::CapacityAvailable as i32,
"and the same disk, asked for a quarter, must still say yes — or this test proves nothing but that the gate refuses everything: {}",
under.reason
);
assert_eq!(
under.floor_bytes,
free / 2,
"the floor reported is the one this plane enforces, not a default"
);
}
#[test]
fn a_named_namespace_is_only_charged_for_the_bytes_it_does_not_already_hold() {
let rig = Rig::new();
let path = RepoRef::new("team", "proj").expect("legal");
rig.stores
.create_repo(NewRepo::new(path.clone()), false)
.expect("repo");
let git_dir = rig.stores.repo_dir(&path).expect("a path");
std::fs::write(git_dir.join("ballast.pack"), vec![0x5au8; 8 * MIB as usize]).expect("write");
let plane = rig.plane(DiskFloor {
percent: 0,
bytes: 1,
});
let held = plane
.usage(Some("team"))
.expect("usage")
.first()
.expect("one row")
.pack_bytes;
assert!(
held >= 8 * MIB,
"this test needs the namespace to really hold bytes, not zero; it holds {held}"
);
let a = ask(&plane, held + 1, "team");
assert_eq!(
a.held_bytes, held,
"the RPC must report the SAME meter Usage.Read does (one implementation)"
);
assert_eq!(
a.additional_bytes,
1,
"raising a tenant from {held} to {} needs ONE new byte, not {}",
held + 1,
held + 1
);
assert_eq!(
a.verdict,
CapacityVerdict::CapacityAvailable as i32,
"{}",
a.reason
);
assert!(
a.reason.contains("team"),
"the sentence names the tenant: {}",
a.reason
);
let anonymous = ask(&plane, held + 1, "");
assert_eq!(anonymous.held_bytes, 0);
assert_eq!(anonymous.additional_bytes, held + 1);
assert_ne!(
anonymous.additional_bytes, a.additional_bytes,
"naming the namespace must change the arithmetic, or `held_bytes` is decoration"
);
let err = plane
.capacity(&CapacityRequest {
requested_bytes: 1,
namespace: "no-such-tenant".into(),
disks: false,
})
.expect_err("an unknown namespace must not be treated as holding nothing");
assert!(matches!(err, ControlError::NotFound(_)), "{err:?}");
}
#[test]
fn the_growth_answer_on_this_developer_box_is_never_sealed() {
let rig = Rig::new();
let plane = rig.plane(DiskFloor {
percent: 0,
bytes: 1,
});
let a = ask(&plane, 1, "");
assert_ne!(
a.growth,
SetGrowth::GrowthSealed as i32,
"gunnar is not PID 1 here ({}), so nothing may report a sealed appliance: {}",
std::process::id(),
a.reason
);
assert_ne!(
a.growth,
SetGrowth::Unspecified as i32,
"every answer carries a growth verdict; UNSPECIFIED is a field nobody filled"
);
assert!(
!a.reason.contains("UNUSED AND INVISIBLE"),
"and must not tell an operator on a shelled box that a disk is wasted: {}",
a.reason
);
}
#[test]
fn a_plane_with_no_disk_gate_declines_instead_of_quoting_a_floor_it_does_not_enforce() {
let rig = Rig::new();
let plane = CoreControlPlane::new(Arc::clone(&rig.stores));
let err = plane
.capacity(&CapacityRequest {
requested_bytes: 1,
namespace: String::new(),
disks: false,
})
.expect_err("no gate wired, no quote");
let ControlError::Unimplemented(line) = err else {
panic!("Unimplemented, not {err:?}");
};
assert!(line.contains("--disk-floor"), "{line}");
}
fn meminfo_here() -> (u64, u64) {
let text = std::fs::read_to_string("/proc/meminfo").expect("/proc/meminfo on a Linux host");
let kib = |key: &str| -> u64 {
text.lines()
.find(|l| l.starts_with(key))
.and_then(|l| l.split_whitespace().nth(1))
.and_then(|n| n.parse::<u64>().ok())
.unwrap_or_else(|| panic!("{key} in /proc/meminfo"))
};
(kib("MemTotal:") * 1024, kib("MemAvailable:") * 1024)
}
fn cgroup_memory_max_here() -> Option<u64> {
let own = std::fs::read_to_string("/proc/self/cgroup").ok()?;
let rel = own
.lines()
.find_map(|l| l.strip_prefix("0::"))?
.trim()
.trim_start_matches('/');
let max = std::fs::read_to_string(format!("/sys/fs/cgroup/{rel}/memory.max")).ok()?;
max.trim().parse().ok()
}
fn cpus_here() -> (usize, Option<usize>) {
let status = std::fs::read_to_string("/proc/self/status").expect("/proc/self/status");
let list = status
.lines()
.find_map(|l| l.strip_prefix("Cpus_allowed_list:"))
.expect("Cpus_allowed_list")
.trim();
let count = |list: &str| {
let mut n = 0usize;
for range in list.split(',') {
n += match range.split_once('-') {
Some((lo, hi)) => hi.parse::<usize>().unwrap() - lo.parse::<usize>().unwrap() + 1,
None => 1,
};
}
n
};
let online = std::fs::read_to_string("/sys/devices/system/cpu/online")
.expect("/sys/devices/system/cpu/online");
let allowed = count(list).min(count(online.trim()));
let quota = (|| {
let own = std::fs::read_to_string("/proc/self/cgroup").ok()?;
let rel = own
.lines()
.find_map(|l| l.strip_prefix("0::"))?
.trim()
.trim_start_matches('/');
let raw = std::fs::read_to_string(format!("/sys/fs/cgroup/{rel}/cpu.max")).ok()?;
let mut parts = raw.split_whitespace();
let quota: f64 = parts.next()?.parse().ok()?;
let period: f64 = parts.next()?.parse().ok()?;
Some((quota / period).ceil().max(1.0) as usize)
})();
(allowed, quota)
}
#[test]
fn the_ram_fields_are_this_hosts_meminfo_read_by_a_second_reader() {
let rig = Rig::new();
let plane = rig.plane(DiskFloor {
percent: 0,
bytes: 1,
});
let a = ask(&plane, 1, "");
let (mem_total, mem_available) = meminfo_here();
let expected_total = match cgroup_memory_max_here() {
Some(cap) if cap < mem_total => cap,
_ => mem_total,
};
assert_eq!(
a.ram_bytes_total,
Some(expected_total),
"ram_bytes_total must be what /proc/meminfo says on this host (MemTotal {mem_total}, \
cgroup cap {:?}): {}",
cgroup_memory_max_here(),
a.machine_reason
);
let available = a
.ram_bytes_available
.expect("MemAvailable is readable on this host");
assert!(
available > 0 && available <= expected_total,
"{available} of {expected_total}"
);
let drift = available.abs_diff(mem_available.min(expected_total));
assert!(
drift < expected_total / 5,
"MemAvailable on the wire ({available}) is not within a fifth of an independent read \
({mem_available}) — that is not two readings of one file"
);
assert!(
a.machine_reason.contains("/proc/meminfo"),
"{}",
a.machine_reason
);
}
#[test]
fn the_cores_on_the_wire_are_the_affinity_masks_popcount_in_millicores() {
let rig = Rig::new();
let plane = rig.plane(DiskFloor {
percent: 0,
bytes: 1,
});
let a = ask(&plane, 1, "");
let (allowed, quota) = cpus_here();
let expected = quota.map_or(allowed, |q| q.min(allowed));
assert_eq!(
a.cpu_millicores_total,
Some(expected as u64 * 1000),
"Cpus_allowed_list counts {allowed} CPUs, cpu.max quota {quota:?}: {}",
a.machine_reason
);
assert!(expected >= 1, "a Linux box has a CPU");
let busy = a
.cpu_millicores_busy
.expect("/proc/loadavg is readable on this host");
let load1: f64 = std::fs::read_to_string("/proc/loadavg")
.expect("loadavg")
.split_whitespace()
.next()
.unwrap()
.parse()
.unwrap();
assert!(
(busy as f64 - load1 * 1000.0).abs() < 1000.0 + load1 * 250.0,
"busy {busy} vs an independent load read of {load1}"
);
assert!(
a.machine_reason.contains("APPROXIMATION"),
"{}",
a.machine_reason
);
}
#[derive(Clone, PartialEq, prost::Message)]
struct OlderCapacityResponse {
#[prost(enumeration = "CapacityVerdict", tag = "1")]
verdict: i32,
#[prost(enumeration = "SetGrowth", tag = "2")]
growth: i32,
#[prost(string, tag = "3")]
root: String,
#[prost(uint64, tag = "4")]
total_bytes: u64,
#[prost(uint64, tag = "5")]
free_bytes: u64,
#[prost(uint64, tag = "6")]
floor_bytes: u64,
#[prost(uint64, tag = "7")]
servable_bytes: u64,
#[prost(uint64, tag = "8")]
requested_bytes: u64,
#[prost(uint64, tag = "9")]
held_bytes: u64,
#[prost(uint64, tag = "10")]
additional_bytes: u64,
#[prost(uint64, tag = "11")]
shortfall_bytes: u64,
#[prost(string, tag = "12")]
data_set: String,
#[prost(string, tag = "13")]
reason: String,
}
#[test]
fn an_older_client_decodes_the_same_disk_answer_from_todays_bytes() {
let rig = Rig::new();
let plane = rig.plane(DiskFloor {
percent: 0,
bytes: 1,
});
let today = ask(&plane, 1, "");
assert!(
today.ram_bytes_total.is_some() && today.cpu_millicores_total.is_some(),
"this test needs the new fields PRESENT, or it proves nothing: {}",
today.machine_reason
);
let bytes = today.encode_to_vec();
let older = OlderCapacityResponse::decode(bytes.as_slice()).expect("an older client decodes");
assert_eq!(
(
older.verdict,
older.growth,
&older.root,
older.total_bytes,
older.free_bytes,
older.floor_bytes,
older.servable_bytes
),
(
today.verdict,
today.growth,
&today.root,
today.total_bytes,
today.free_bytes,
today.floor_bytes,
today.servable_bytes
)
);
assert_eq!(
(
older.requested_bytes,
older.held_bytes,
older.additional_bytes,
older.shortfall_bytes,
&older.data_set,
&older.reason
),
(
today.requested_bytes,
today.held_bytes,
today.additional_bytes,
today.shortfall_bytes,
&today.data_set,
&today.reason
)
);
let stripped = gunnar_proto::CapacityResponse {
cpu_millicores_total: None,
ram_bytes_total: None,
ram_bytes_available: None,
cpu_millicores_busy: None,
machine_reason: String::new(),
grow_state: None,
hotplug: false,
disk_test: 0,
disk_seq_write_bps: None,
disk_seq_read_bps: None,
disk_sync_milliops_per_s: None,
disk_read_how: String::new(),
disk_reason: String::new(),
..today.clone()
};
assert_eq!(stripped.encode_to_vec(), older.encode_to_vec());
assert!(
bytes.len() > older.encoded_len(),
"today's bytes carry more than the older message — else the new fields were absent"
);
}
#[test]
fn the_box_facts_ride_on_admin_capacity_and_add_no_rpc_to_the_census() {
let admin: Vec<&str> = gunnar_grpc::auth::METHODS
.iter()
.copied()
.filter(|m| m.starts_with("Admin."))
.collect();
assert_eq!(
admin,
[
"Admin.AuditTrail",
"Admin.Capacity",
"Admin.GrowResume",
"Admin.GrowStart",
"Admin.GrowStatus",
"Admin.Health",
"Admin.IssuerInfo",
"Admin.SetTunnelAddress",
"Admin.IssueTwinCertificate",
"Admin.Stats",
"Admin.StorageInfo",
"Admin.Topology",
"Admin.VerifyChain",
"Admin.GetSetting",
"Admin.SetSetting",
],
"the Admin family changed; the box facts were meant to ride on Admin.Capacity"
);
assert_eq!(
gunnar_grpc::auth::METHODS
.iter()
.filter(|m| **m == "Admin.Capacity")
.count(),
1
);
}
#[test]
fn every_disk_is_reported_only_when_asked_and_the_data_root_is_placed_on_its_mount() {
let rig = Rig::new();
let floor = DiskFloor {
percent: 5,
bytes: 2 * MIB,
};
let plane = rig.plane(floor);
assert!(
ask(&plane, 1, "").disks.is_none(),
"a quote paid for the disk report"
);
let answer = plane
.capacity(&CapacityRequest {
requested_bytes: 0,
namespace: String::new(),
disks: true,
})
.expect("a plane with a disk gate answers");
let disks = answer.disks.as_ref().expect("asked for, so reported");
assert!(disks.problems.is_empty(), "{:?}", disks.problems);
assert!(disks.measured_at_unix_ms > 0);
let data = disks
.mounts
.iter()
.find(|m| m.labels.iter().any(|l| l == "data"))
.unwrap_or_else(|| panic!("the store root is on no reported mount: {:?}", disks.mounts));
assert!(data.measured, "{data:?}");
assert_eq!(
data.total_bytes, answer.total_bytes,
"two statvfs of one filesystem disagree on its size"
);
let expected_floor = (data.total_bytes / 100 * 5)
.max(2 * MIB)
.min(data.total_bytes);
assert!(
data.floor_bytes.abs_diff(expected_floor) <= data.total_bytes / 1000 + 1,
"the floor on the data mount is not the --disk-floor in force: {} vs {expected_floor}",
data.floor_bytes
);
assert!(
disks
.mounts
.iter()
.any(|m| m.labels.iter().any(|l| l == "system")),
"`/` is not labelled system: {:?}",
disks.mounts
);
let bytes = answer.encode_to_vec();
let back = gunnar_proto::CapacityResponse::decode(bytes.as_slice()).expect("decodes");
assert_eq!(back.disks.as_ref(), Some(disks));
}