1430 lines
53 KiB
Rust
1430 lines
53 KiB
Rust
//! Container lifecycle ops: `StartContainer`, `StopContainer`,
|
|
//! `KillContainer`, `CreateContainer`, `UpdateContainer`, `DestroyContainer`
|
|
//! and `ListContainers`. Every op takes a logical agent name and reaches only
|
|
//! the `h-`-prefixed container [`container_system_name`] derives from it.
|
|
//!
|
|
//! `CreateContainer` / `UpdateContainer` apply a closure already in the store
|
|
//! and stream progress to the caller as [`PrivEvent::Line`]s when asked to.
|
|
|
|
use anyhow::{Context as _, Result, bail};
|
|
use hive_priv_sock::{PrivEvent, PrivStream, PrivStreamLine};
|
|
use tokio::io::{AsyncWriteExt, BufReader};
|
|
use tokio::net::unix::OwnedWriteHalf;
|
|
use tokio::process::Command;
|
|
|
|
use crate::shared::{RealSpawn, Spawn, container_system_name, validate_agent_name};
|
|
|
|
/// What the lifecycle ops run and how long they wait for it: the host's
|
|
/// tools and timings in [`Host::REAL`], stand-ins in tests.
|
|
struct Host<S> {
|
|
/// Starts the tools the ops run.
|
|
spawn: S,
|
|
/// How long to wait for machined to drop a machine's registration after a
|
|
/// shutdown has been asked for. Generous: a container with slow-stopping
|
|
/// units legitimately takes a while, and escalating early would SIGKILL a
|
|
/// shutdown that was going to finish on its own.
|
|
release_timeout: std::time::Duration,
|
|
/// Poll cadence while waiting for the registration to go away.
|
|
release_poll: std::time::Duration,
|
|
/// Cap on a single registration probe. The probe is a D-Bus round trip to
|
|
/// machined; if machined itself is wedged the call would otherwise sit on
|
|
/// the D-Bus method timeout, which is far longer than the whole stop
|
|
/// sequence should take.
|
|
probe_timeout: std::time::Duration,
|
|
}
|
|
|
|
impl Host<RealSpawn> {
|
|
const REAL: Self = Host {
|
|
spawn: RealSpawn,
|
|
release_timeout: std::time::Duration::from_secs(20),
|
|
release_poll: std::time::Duration::from_millis(500),
|
|
probe_timeout: std::time::Duration::from_secs(5),
|
|
};
|
|
}
|
|
|
|
impl<S: Spawn> Host<S> {
|
|
fn command(&self, program: &str) -> Command {
|
|
self.spawn.command(program)
|
|
}
|
|
}
|
|
|
|
/// Write one `PrivEvent::Line` to the client. Best-effort: a write
|
|
/// failure is logged but doesn't abort the running subprocess.
|
|
async fn write_line_event(writer: &mut OwnedWriteHalf, stream: PrivStream, data: &str) {
|
|
let event = PrivEvent::Line(PrivStreamLine {
|
|
stream,
|
|
data: data.to_owned(),
|
|
});
|
|
if let Ok(mut json) = serde_json::to_string(&event) {
|
|
json.push('\n');
|
|
if let Err(e) = writer.write_all(json.as_bytes()).await {
|
|
tracing::warn!(error = %e, "write_line_event: write failed");
|
|
}
|
|
}
|
|
}
|
|
|
|
/// `StartContainer`: clear any start-limit lockout left by earlier failures
|
|
/// so a now-correct start isn't blocked (`nixos-container start` does not do
|
|
/// this itself). Best-effort — if the unit doesn't exist yet (first-time
|
|
/// create), `reset-failed` is a no-op and the start proceeds regardless.
|
|
pub(crate) async fn start_container(name: &str) -> Result<(String, String)> {
|
|
start_container_with(&Host::REAL, name).await
|
|
}
|
|
|
|
async fn start_container_with(host: &Host<impl Spawn>, name: &str) -> Result<(String, String)> {
|
|
validate_agent_name(name)?;
|
|
let machine = container_system_name(name);
|
|
// Silenced: a first-time create means the unit doesn't exist yet, and
|
|
// `systemctl`'s own "Unit ... not loaded" complaint for that case would
|
|
// otherwise land in the journal at error priority on every such start —
|
|
// routine, not a signal, since the result is already discarded below.
|
|
let _ = host
|
|
.command("systemctl")
|
|
.args(["reset-failed", &format!("container@{machine}.service")])
|
|
.stdout(std::process::Stdio::null())
|
|
.stderr(std::process::Stdio::null())
|
|
.status()
|
|
.await;
|
|
container_run_with(host, &["start", &machine]).await
|
|
}
|
|
|
|
/// Most bytes of journal text `StartFailureLog` returns.
|
|
const START_LOG_CAP: usize = 8 * 1024;
|
|
/// Longest `journalctl` run.
|
|
const START_LOG_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
|
|
/// How far back `StartFailureLog` may look.
|
|
const START_LOG_MAX_LOOKBACK_SECS: u64 = 3600;
|
|
|
|
/// The fixed `journalctl` argv for `machine`'s start log. The match is
|
|
/// `_SYSTEMD_UNIT`, so only lines written by the container unit's own
|
|
/// processes come back, not systemd's job messages about it.
|
|
fn start_log_argv(machine: &str, since: u64) -> Vec<String> {
|
|
[
|
|
&format!("_SYSTEMD_UNIT=container@{machine}.service"),
|
|
"--since",
|
|
&format!("@{since}"),
|
|
"-n",
|
|
"30",
|
|
"-o",
|
|
"cat",
|
|
"--no-pager",
|
|
]
|
|
.map(str::to_owned)
|
|
.to_vec()
|
|
}
|
|
|
|
/// The last `cap` bytes of `text`, cut at a char boundary.
|
|
fn tail_bytes(text: &str, cap: usize) -> &str {
|
|
let mut start = text.len().saturating_sub(cap);
|
|
while !text.is_char_boundary(start) {
|
|
start += 1;
|
|
}
|
|
&text[start..]
|
|
}
|
|
|
|
/// `StartFailureLog`: the container unit's recent journal lines, capped.
|
|
/// `run` executes `journalctl` with the argv it is given.
|
|
async fn start_failure_log_with<F, Fut>(
|
|
name: &str,
|
|
since_unix: u64,
|
|
now: u64,
|
|
run: F,
|
|
) -> Result<(String, String)>
|
|
where
|
|
F: FnOnce(Vec<String>) -> Fut,
|
|
Fut: std::future::Future<Output = Result<String>>,
|
|
{
|
|
validate_agent_name(name)?;
|
|
let since = since_unix.clamp(now.saturating_sub(START_LOG_MAX_LOOKBACK_SECS), now);
|
|
let out = run(start_log_argv(&container_system_name(name), since)).await?;
|
|
Ok((tail_bytes(&out, START_LOG_CAP).to_owned(), String::new()))
|
|
}
|
|
|
|
/// `StartFailureLog` against the real `journalctl`.
|
|
pub(crate) async fn start_failure_log(name: &str, since_unix: u64) -> Result<(String, String)> {
|
|
let now = std::time::SystemTime::now()
|
|
.duration_since(std::time::UNIX_EPOCH)
|
|
.map_or(0, |d| d.as_secs());
|
|
start_failure_log_with(name, since_unix, now, |argv| {
|
|
run_journalctl(&Host::REAL, START_LOG_TIMEOUT, argv)
|
|
})
|
|
.await
|
|
}
|
|
|
|
/// `journalctl` with `argv`: its stdout, if it exits 0 within `limit`.
|
|
async fn run_journalctl(
|
|
host: &Host<impl Spawn>,
|
|
limit: std::time::Duration,
|
|
argv: Vec<String>,
|
|
) -> Result<String> {
|
|
let child = host
|
|
.command("journalctl")
|
|
.args(&argv)
|
|
.kill_on_drop(true)
|
|
.output();
|
|
let out = tokio::time::timeout(limit, child)
|
|
.await
|
|
.context("journalctl timed out")?
|
|
.context("invoke journalctl")?;
|
|
if !out.status.success() {
|
|
bail!("journalctl failed ({})", out.status);
|
|
}
|
|
Ok(String::from_utf8_lossy(&out.stdout).into_owned())
|
|
}
|
|
|
|
/// `StopContainer`.
|
|
pub(crate) async fn stop_container(name: &str) -> Result<(String, String)> {
|
|
stop_container_with(&Host::REAL, name).await
|
|
}
|
|
|
|
async fn stop_container_with(host: &Host<impl Spawn>, name: &str) -> Result<(String, String)> {
|
|
validate_agent_name(name)?;
|
|
stop_and_release(host, &container_system_name(name)).await
|
|
}
|
|
|
|
/// `KillContainer`: `nixos-container` has no kill verb, so this uses
|
|
/// `machinectl` to send `SIGKILL` to every process in the container — the
|
|
/// right semantics for a forced shutdown after a graceful stop has already
|
|
/// been attempted.
|
|
pub(crate) async fn kill_container(name: &str) -> Result<(String, String)> {
|
|
kill_container_with(&Host::REAL, name).await
|
|
}
|
|
|
|
async fn kill_container_with(host: &Host<impl Spawn>, name: &str) -> Result<(String, String)> {
|
|
validate_agent_name(name)?;
|
|
let machine = container_system_name(name);
|
|
machinectl_run(host, &["kill", &machine, "--signal=SIGKILL"]).await
|
|
}
|
|
|
|
/// `DestroyContainer`.
|
|
pub(crate) async fn destroy_container(name: &str) -> Result<(String, String)> {
|
|
destroy_container_with(&Host::REAL, name).await
|
|
}
|
|
|
|
async fn destroy_container_with(host: &Host<impl Spawn>, name: &str) -> Result<(String, String)> {
|
|
validate_agent_name(name)?;
|
|
container_run_with(host, &["destroy", &container_system_name(name)]).await
|
|
}
|
|
|
|
/// Shared body for `CreateContainer` / `UpdateContainer`: validate the name
|
|
/// and that `toplevel` is the closure realised for the agent, and apply it (streaming line events to `writer` when
|
|
/// `stream` is set). Nothing is built here: `toplevel` is a closure already in
|
|
/// the store, which `RealiseClosure` put there.
|
|
///
|
|
/// `create` passes `--system-path` so `nixos-container`'s own `buildFlake()`
|
|
/// never runs; it builds to a relative `.tmp` out-link that concurrent calls
|
|
/// share, which handed an agent another agent's closure.
|
|
///
|
|
/// `update` does not call `nixos-container update` at all: that action's
|
|
/// version-compat probe runs before `--system-path` is honored and dies on every
|
|
/// agent. [`swap_container_profile`] does what the action does past that probe:
|
|
/// `nix-env --set` the per-container profile, then `systemctl reload` if the
|
|
/// container is running.
|
|
pub(crate) async fn container_toplevel_action(
|
|
verb: &str,
|
|
name: &str,
|
|
toplevel: &str,
|
|
stream: bool,
|
|
writer: &mut OwnedWriteHalf,
|
|
) -> Result<(String, String)> {
|
|
validate_agent_name(name)?;
|
|
let toplevel = crate::closure::applicable_for(name, toplevel)?;
|
|
apply_toplevel(&Host::REAL, verb, name, toplevel, stream, writer).await
|
|
}
|
|
|
|
/// [`container_toplevel_action`] past the check that `toplevel` is the
|
|
/// closure realised for `name`.
|
|
async fn apply_toplevel(
|
|
host: &Host<impl Spawn>,
|
|
verb: &str,
|
|
name: &str,
|
|
toplevel: &str,
|
|
stream: bool,
|
|
writer: &mut OwnedWriteHalf,
|
|
) -> Result<(String, String)> {
|
|
let system_name = container_system_name(name);
|
|
if verb == "update" {
|
|
return swap_container_profile(host, &system_name, toplevel, stream, writer).await;
|
|
}
|
|
let args = [verb, &system_name, "--system-path", toplevel];
|
|
if stream {
|
|
container_run_streaming(host, &args, writer).await
|
|
} else {
|
|
container_run_with(host, &args).await
|
|
}
|
|
}
|
|
|
|
/// Apply a prebuilt toplevel to an existing container directly, without
|
|
/// going through `nixos-container update` (see `container_toplevel_action`'s
|
|
/// doc comment for why). Mirrors `nixos-container.pl`'s own `update`
|
|
/// action verbatim, past its version probe: point the per-container `nix-
|
|
/// env` profile at the new toplevel, then reload the container unit *if*
|
|
/// it's currently running — the container's own next start already reads
|
|
/// from the profile, so a stopped container needs nothing further.
|
|
///
|
|
/// Both steps are near-instant in the happy case, but still stream-
|
|
/// forwarded like the rest of this operation: a failure here (e.g. a
|
|
/// wedged `systemctl reload`) is exactly when the caller most wants the
|
|
/// live line in the build log, not just a summary error afterward.
|
|
async fn swap_container_profile(
|
|
host: &Host<impl Spawn>,
|
|
system_name: &str,
|
|
toplevel: &str,
|
|
stream: bool,
|
|
writer: &mut OwnedWriteHalf,
|
|
) -> Result<(String, String)> {
|
|
let profile = format!("/nix/var/nix/profiles/per-container/{system_name}/system");
|
|
let set_out = host
|
|
.command("nix-env")
|
|
.args(["-p", &profile, "--set", toplevel])
|
|
.output()
|
|
.await
|
|
.context("invoke nix-env --set")?;
|
|
let mut stdout = String::from_utf8_lossy(&set_out.stdout).into_owned();
|
|
let mut stderr = String::from_utf8_lossy(&set_out.stderr).into_owned();
|
|
log_and_forward(&stdout, &stderr, stream, writer).await;
|
|
if !set_out.status.success() {
|
|
bail!(
|
|
"nix-env -p {profile} --set failed ({}): {}",
|
|
set_out.status,
|
|
stderr.trim()
|
|
);
|
|
}
|
|
|
|
let unit = format!("container@{system_name}");
|
|
let state_out = host
|
|
.command("systemctl")
|
|
.args(["show", "--property=ActiveState", "--value", &unit])
|
|
.output()
|
|
.await
|
|
.context("query container ActiveState")?;
|
|
let active = String::from_utf8_lossy(&state_out.stdout).trim() == "active";
|
|
if active {
|
|
let reload_out = host
|
|
.command("systemctl")
|
|
.args(["reload", &unit])
|
|
.output()
|
|
.await
|
|
.context("invoke systemctl reload")?;
|
|
let r_stdout = String::from_utf8_lossy(&reload_out.stdout).into_owned();
|
|
let r_stderr = String::from_utf8_lossy(&reload_out.stderr).into_owned();
|
|
log_and_forward(&r_stdout, &r_stderr, stream, writer).await;
|
|
if !reload_out.status.success() {
|
|
bail!(
|
|
"systemctl reload {unit} failed ({}): {}",
|
|
reload_out.status,
|
|
r_stderr.trim()
|
|
);
|
|
}
|
|
stdout.push_str(&r_stdout);
|
|
stderr.push_str(&r_stderr);
|
|
}
|
|
Ok((stdout, stderr))
|
|
}
|
|
|
|
/// Log a command's captured stdout/stderr the same way every other
|
|
/// `nixos-container`-adjacent shellout in this file does, and — when
|
|
/// `stream` is set — forward each line to the client as a live
|
|
/// [`PrivEvent::Line`] too, so a caller watching the build log sees these
|
|
/// lines exactly like any other step's, not just a summary on failure.
|
|
async fn log_and_forward(stdout: &str, stderr: &str, stream: bool, writer: &mut OwnedWriteHalf) {
|
|
for line in stdout.lines() {
|
|
tracing::info!(target: "nixos-container", "{line}");
|
|
if stream {
|
|
write_line_event(writer, PrivStream::Stdout, line).await;
|
|
}
|
|
}
|
|
for line in stderr.lines() {
|
|
tracing::warn!(target: "nixos-container", "{line}");
|
|
if stream {
|
|
write_line_event(writer, PrivStream::Stderr, line).await;
|
|
}
|
|
}
|
|
}
|
|
|
|
/// How a [`run_bounded`] child ended.
|
|
pub(crate) enum BoundedRun {
|
|
Exited {
|
|
status: std::process::ExitStatus,
|
|
stderr: String,
|
|
},
|
|
/// Still running at the limit; its whole process group was killed.
|
|
TimedOut,
|
|
}
|
|
|
|
/// Run `cmd` to completion or until `limit`, draining both pipes: stdout is
|
|
/// discarded, stderr is kept and also logged as it arrives.
|
|
///
|
|
/// The child leads its own process group and a timeout `SIGKILL`s the whole
|
|
/// group, so a grandchild still holding the pipes (or the daemon connection)
|
|
/// dies with it instead of outliving the error.
|
|
pub(crate) async fn run_bounded(
|
|
mut cmd: Command,
|
|
limit: std::time::Duration,
|
|
) -> std::io::Result<BoundedRun> {
|
|
use tokio::io::AsyncBufReadExt as _;
|
|
|
|
let mut child = cmd
|
|
.stdout(std::process::Stdio::piped())
|
|
.stderr(std::process::Stdio::piped())
|
|
.process_group(0)
|
|
.spawn()?;
|
|
// Group leader, so its pid is the pgid.
|
|
let pgid = child.id().and_then(|pid| i32::try_from(pid).ok());
|
|
|
|
let stdout = child.stdout.take().expect("stdout piped");
|
|
let stderr = child.stderr.take().expect("stderr piped");
|
|
let run = async {
|
|
let mut stdout_lines = BufReader::new(stdout).lines();
|
|
let mut stderr_lines = BufReader::new(stderr).lines();
|
|
let mut stderr_buf = String::new();
|
|
|
|
// ⚠️ Both pipes are drained concurrently: reading one alone would let
|
|
// the other fill its pipe buffer and deadlock the child.
|
|
//
|
|
// ⚠️ Each stream's EOF is tracked separately rather than breaking on
|
|
// the first `None`: `next_line()` on a closed stream returns
|
|
// `Ok(None)` immediately and forever, so a loop that keeps polling a
|
|
// finished stream spins hot until the other one ends too.
|
|
let mut stdout_done = false;
|
|
let mut stderr_done = false;
|
|
while !(stdout_done && stderr_done) {
|
|
tokio::select! {
|
|
line = stdout_lines.next_line(), if !stdout_done => {
|
|
match line {
|
|
Ok(Some(_)) => {}
|
|
Ok(None) => stdout_done = true,
|
|
Err(e) => {
|
|
tracing::warn!(error = %e, "child stdout read error");
|
|
stdout_done = true;
|
|
}
|
|
}
|
|
}
|
|
line = stderr_lines.next_line(), if !stderr_done => {
|
|
match line {
|
|
Ok(Some(l)) => {
|
|
tracing::info!(target: "nix-store", "{l}");
|
|
if !stderr_buf.is_empty() {
|
|
stderr_buf.push('\n');
|
|
}
|
|
stderr_buf.push_str(&l);
|
|
}
|
|
Ok(None) => stderr_done = true,
|
|
Err(e) => {
|
|
tracing::warn!(error = %e, "child stderr read error");
|
|
stderr_done = true;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
let status = child.wait().await?;
|
|
Ok::<_, std::io::Error>(BoundedRun::Exited {
|
|
status,
|
|
stderr: stderr_buf,
|
|
})
|
|
};
|
|
if let Ok(exited) = tokio::time::timeout(limit, run).await {
|
|
return exited;
|
|
}
|
|
if let Some(pgid) = pgid {
|
|
// SAFETY: plain `kill(2)`; `-pgid` targets the child's own process
|
|
// group, which `process_group(0)` made it lead.
|
|
unsafe {
|
|
libc::kill(-pgid, libc::SIGKILL);
|
|
}
|
|
}
|
|
child.wait().await?;
|
|
Ok(BoundedRun::TimedOut)
|
|
}
|
|
|
|
/// Invoke `nixos-container` with the given args, log output to journald.
|
|
pub(crate) async fn container_run(args: &[&str]) -> Result<(String, String)> {
|
|
container_run_with(&Host::REAL, args).await
|
|
}
|
|
|
|
async fn container_run_with(host: &Host<impl Spawn>, args: &[&str]) -> Result<(String, String)> {
|
|
let out = host
|
|
.command("nixos-container")
|
|
.args(args)
|
|
.output()
|
|
.await
|
|
.context("invoke nixos-container")?;
|
|
let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
|
|
let stderr = String::from_utf8_lossy(&out.stderr).into_owned();
|
|
// `list` is a read-only enumeration called on the hot path (dashboard
|
|
// rescan, forge + boot sweeps) — its stdout is the return value, not
|
|
// progress, so logging every container name on every call floods the
|
|
// journal. Log stdout only for the mutating ops, where each line is
|
|
// genuine progress. stderr is always logged (errors matter regardless).
|
|
if args.first() != Some(&"list") {
|
|
for line in stdout.lines() {
|
|
tracing::info!(target: "nixos-container", "{line}");
|
|
}
|
|
}
|
|
for line in stderr.lines() {
|
|
tracing::warn!(target: "nixos-container", "{line}");
|
|
}
|
|
if !out.status.success() {
|
|
bail!(
|
|
"nixos-container {} failed ({}): {}",
|
|
args.join(" "),
|
|
out.status,
|
|
stderr.trim()
|
|
);
|
|
}
|
|
Ok((stdout, stderr))
|
|
}
|
|
|
|
/// Invoke `machinectl` with the given args, log output to journald.
|
|
/// Used for operations that nixos-container doesn't expose (e.g. sending
|
|
/// signals to running containers).
|
|
async fn machinectl_run(host: &Host<impl Spawn>, args: &[&str]) -> Result<(String, String)> {
|
|
let out = host
|
|
.command("machinectl")
|
|
.args(args)
|
|
.output()
|
|
.await
|
|
.context("invoke machinectl")?;
|
|
let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
|
|
let stderr = String::from_utf8_lossy(&out.stderr).into_owned();
|
|
for line in stdout.lines() {
|
|
tracing::info!(target: "machinectl", "{line}");
|
|
}
|
|
for line in stderr.lines() {
|
|
tracing::warn!(target: "machinectl", "{line}");
|
|
}
|
|
if !out.status.success() {
|
|
bail!(
|
|
"machinectl {} failed ({}): {}",
|
|
args.join(" "),
|
|
out.status,
|
|
stderr.trim()
|
|
);
|
|
}
|
|
Ok((stdout, stderr))
|
|
}
|
|
|
|
/// True when machined still holds a registration for `machine`.
|
|
///
|
|
/// This asks machined the same question the registration itself answers —
|
|
/// `machinectl show` resolves the name through `GetMachine`, the very lookup
|
|
/// that makes a later boot fail with `Failed to register machine: already
|
|
/// exists`. So a positive answer here is exactly the condition that breaks
|
|
/// the next start, not a proxy for it.
|
|
///
|
|
/// Deliberately NOT the container's systemd unit state: the unit can be
|
|
/// `inactive` while the registration is still held, and that gap is the
|
|
/// whole bug this probe exists to catch.
|
|
///
|
|
/// Errors and timeouts answer "still registered". Being wrong that way costs
|
|
/// a redundant SIGKILL to something already gone; being wrong the other way
|
|
/// hands back a stop that silently leaked the name.
|
|
async fn machine_registered(host: &Host<impl Spawn>, machine: &str) -> bool {
|
|
let probe = host
|
|
.command("machinectl")
|
|
.args(["show", machine, "--property=Name"])
|
|
// Don't leave a probe behind when the timeout below fires.
|
|
.kill_on_drop(true)
|
|
.output();
|
|
match tokio::time::timeout(host.probe_timeout, probe).await {
|
|
Ok(Ok(out)) => out.status.success(),
|
|
Ok(Err(e)) => {
|
|
tracing::warn!(%machine, error = %e, "machinectl show failed to run; assuming still registered");
|
|
true
|
|
}
|
|
Err(_) => {
|
|
tracing::warn!(%machine, "machinectl show timed out; assuming still registered");
|
|
true
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Poll [`machine_registered`] until the name is free or the timeout expires.
|
|
/// Returns true once the registration is gone.
|
|
async fn wait_name_released(host: &Host<impl Spawn>, machine: &str) -> bool {
|
|
let deadline = tokio::time::Instant::now() + host.release_timeout;
|
|
loop {
|
|
if !machine_registered(host, machine).await {
|
|
return true;
|
|
}
|
|
if tokio::time::Instant::now() >= deadline {
|
|
return false;
|
|
}
|
|
tokio::time::sleep(host.release_poll).await;
|
|
}
|
|
}
|
|
|
|
/// Stop `machine` and only report success once machined has actually released
|
|
/// the name.
|
|
///
|
|
/// `nixos-container stop` exiting 0 does not mean the machine is gone. A
|
|
/// process that sits in the machine's cgroup without being a child of the
|
|
/// container's init — a shell exec'd in from outside, say — never receives
|
|
/// the shutdown's SIGTERM if it has been stopped with SIGSTOP, so the
|
|
/// registration outlives the "successful" stop. Every later start of that
|
|
/// container then fails with `Failed to register machine: already exists`,
|
|
/// and machined re-persists the stale record across its own restart, so
|
|
/// there is no cleaning it up after the fact. The only place to catch it is
|
|
/// here, in the stop.
|
|
///
|
|
/// So: ask for the stop, wait for the name, and fail loudly if it is still
|
|
/// held — a caller that is told the stop worked will go on to start the
|
|
/// container and hit the confusing registration error instead of this one.
|
|
///
|
|
/// This used to escalate to `machinectl kill --signal=SIGKILL` here. Dropped
|
|
/// per real incident data: the case this guards is a container genuinely
|
|
/// wedged (e.g. a root-login process survived the stop), and SIGKILL
|
|
/// doesn't recover that in practice — only a host-level reboot has.
|
|
/// Pretending a kill attempt handled it hides a condition that needs a human
|
|
/// to look at the host, so this now just reports the failure instead of
|
|
/// quietly (and ineffectually) trying to force it.
|
|
///
|
|
/// The verify-then-fail lives in the helper rather than at a call site so
|
|
/// that every stop gets it: dashboard, reconcile, destroy, cold-start
|
|
/// fallback. The start path already distrusts its own exit code the same way;
|
|
/// this is the missing half of that pair.
|
|
async fn stop_and_release(host: &Host<impl Spawn>, machine: &str) -> Result<(String, String)> {
|
|
let stop = container_run_with(host, &["stop", machine]).await;
|
|
if wait_name_released(host, machine).await {
|
|
// Happy path, and also the path where a stop that reported failure
|
|
// nonetheless brought the machine down. Either way the caller gets
|
|
// the original result untouched.
|
|
return stop;
|
|
}
|
|
|
|
tracing::error!(
|
|
%machine,
|
|
"stop finished but machined still holds the registration — the \
|
|
container is likely wedged (a process outside its init tree \
|
|
survived the stop); this needs a host-level look, not another \
|
|
stop attempt"
|
|
);
|
|
bail!(
|
|
"stop {machine}: machined still holds the machine name after {}s. \
|
|
This container is likely wedged and starting it again will fail to \
|
|
register — needs host-level intervention (a reboot has been the \
|
|
only reliable fix in practice).",
|
|
host.release_timeout.as_secs()
|
|
)
|
|
}
|
|
|
|
/// Invoke `nixos-container` with the given args and forward output lines
|
|
/// to the caller as `PrivEvent::Line` messages in real time, logging each
|
|
/// line to journald as it arrives. Returns `(String::new(), String::new())`
|
|
/// on success (all output was streamed); the error string includes stderr
|
|
/// tail on failure.
|
|
async fn container_run_streaming(
|
|
host: &Host<impl Spawn>,
|
|
args: &[&str],
|
|
writer: &mut OwnedWriteHalf,
|
|
) -> Result<(String, String)> {
|
|
use tokio::io::AsyncBufReadExt as _;
|
|
|
|
let mut child = host
|
|
.command("nixos-container")
|
|
.args(args)
|
|
.stdout(std::process::Stdio::piped())
|
|
.stderr(std::process::Stdio::piped())
|
|
.spawn()
|
|
.context("invoke nixos-container (streaming)")?;
|
|
|
|
let stdout = child.stdout.take().expect("stdout piped");
|
|
let stderr = child.stderr.take().expect("stderr piped");
|
|
|
|
let mut stdout_lines = BufReader::new(stdout).lines();
|
|
let mut stderr_lines = BufReader::new(stderr).lines();
|
|
|
|
// Collect stderr for the error message; stream both to the client.
|
|
let mut stderr_buf = String::new();
|
|
|
|
// Drive stdout and stderr concurrently. Stderr is disabled once it
|
|
// hits EOF or an error: `next_line()` would otherwise resolve at once on
|
|
// every turn and spin the loop until stdout closes.
|
|
let mut stderr_done = false;
|
|
loop {
|
|
tokio::select! {
|
|
line = stdout_lines.next_line() => {
|
|
match line {
|
|
Ok(Some(l)) => {
|
|
tracing::info!(target: "nixos-container", "{l}");
|
|
write_line_event(writer, PrivStream::Stdout, &l).await;
|
|
}
|
|
Ok(None) => break,
|
|
Err(e) => {
|
|
tracing::warn!(error = %e, "nixos-container stdout read error");
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
line = stderr_lines.next_line(), if !stderr_done => {
|
|
match line {
|
|
Ok(Some(l)) => {
|
|
tracing::warn!(target: "nixos-container", "{l}");
|
|
write_line_event(writer, PrivStream::Stderr, &l).await;
|
|
if !stderr_buf.is_empty() {
|
|
stderr_buf.push('\n');
|
|
}
|
|
stderr_buf.push_str(&l);
|
|
}
|
|
Ok(None) => stderr_done = true,
|
|
Err(e) => {
|
|
tracing::warn!(error = %e, "nixos-container stderr read error");
|
|
stderr_done = true;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Drain any remaining stderr after stdout closed.
|
|
while let Ok(Some(l)) = stderr_lines.next_line().await {
|
|
tracing::warn!(target: "nixos-container", "{l}");
|
|
write_line_event(writer, PrivStream::Stderr, &l).await;
|
|
if !stderr_buf.is_empty() {
|
|
stderr_buf.push('\n');
|
|
}
|
|
stderr_buf.push_str(&l);
|
|
}
|
|
|
|
let status = child.wait().await.context("wait nixos-container")?;
|
|
if !status.success() {
|
|
// Only the last stderr line is embedded — the full stderr was
|
|
// already forwarded line-by-line as PrivEvent::Line messages and
|
|
// is captured in build_logs.sqlite by the caller. Keeping the
|
|
// error message short avoids bloating the anyhow chain.
|
|
bail!(
|
|
"nixos-container {} failed ({}): {}",
|
|
args.join(" "),
|
|
status,
|
|
stderr_buf.lines().last().unwrap_or("").trim()
|
|
);
|
|
}
|
|
Ok((String::new(), String::new()))
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::{BoundedRun, run_bounded};
|
|
use crate::shared::test_support::scratch;
|
|
|
|
/// Whether `pid` is gone, a zombie (`Z`), or being reaped (`X`): none of
|
|
/// these runs again, and a process passes through them in that order,
|
|
/// so once this is true it stays true.
|
|
fn is_dead(pid: &str) -> bool {
|
|
match std::fs::read_to_string(format!("/proc/{pid}/stat")) {
|
|
Err(_) => true,
|
|
Ok(stat) => stat_is_dead(&stat),
|
|
}
|
|
}
|
|
|
|
/// The state letter of a `/proc/<pid>/stat` line, which follows the
|
|
/// parenthesised command name, is `Z` or `X`.
|
|
fn stat_is_dead(stat: &str) -> bool {
|
|
stat.rsplit_once(") ")
|
|
.is_some_and(|(_, rest)| rest.starts_with(['Z', 'X']))
|
|
}
|
|
|
|
#[test]
|
|
fn zombie_and_reaped_states_count_as_dead() {
|
|
assert!(stat_is_dead("42 (sleep) Z 1 42 42"));
|
|
assert!(stat_is_dead("42 (sleep) X 1 42 42"));
|
|
assert!(!stat_is_dead("42 (sleep) S 1 42 42"));
|
|
assert!(!stat_is_dead("42 (sleep) R 1 42 42"));
|
|
}
|
|
|
|
/// A child still running at the limit comes back `TimedOut`, and the kill
|
|
/// reaches its whole process group: the backgrounded grandchild dies too,
|
|
/// not just the shell that started it.
|
|
#[tokio::test]
|
|
async fn a_child_past_its_limit_is_killed_with_its_group() {
|
|
use std::time::{Duration, Instant};
|
|
let dir = scratch();
|
|
let pidfile = dir.join("grandchild");
|
|
let mut cmd = tokio::process::Command::new("sh");
|
|
cmd.arg("-c")
|
|
.arg(format!("sleep 600 & echo $! > {}; wait", pidfile.display()));
|
|
|
|
let started = Instant::now();
|
|
let run = run_bounded(cmd, Duration::from_secs(2)).await.unwrap();
|
|
assert!(matches!(run, BoundedRun::TimedOut), "must time out");
|
|
assert!(
|
|
started.elapsed() < Duration::from_mins(1),
|
|
"the limit did not bound the run"
|
|
);
|
|
|
|
let pid = std::fs::read_to_string(&pidfile).unwrap();
|
|
let pid = pid.trim();
|
|
let deadline = Instant::now() + Duration::from_secs(5);
|
|
while !is_dead(pid) && Instant::now() < deadline {
|
|
tokio::time::sleep(Duration::from_millis(50)).await;
|
|
}
|
|
assert!(is_dead(pid), "grandchild {pid} outlived the timeout");
|
|
std::fs::remove_dir_all(&dir).ok();
|
|
}
|
|
|
|
/// The control: a child that exits inside its limit is `Exited`, with its
|
|
/// status and its stderr.
|
|
#[tokio::test]
|
|
async fn a_child_inside_its_limit_returns_its_status_and_stderr() {
|
|
let mut cmd = tokio::process::Command::new("sh");
|
|
cmd.arg("-c")
|
|
.arg("echo progress >&2; echo /nix/store/abc-toplevel");
|
|
match run_bounded(cmd, std::time::Duration::from_mins(1))
|
|
.await
|
|
.unwrap()
|
|
{
|
|
BoundedRun::Exited { status, stderr } => {
|
|
assert!(status.success());
|
|
assert_eq!(stderr, "progress");
|
|
}
|
|
BoundedRun::TimedOut => panic!("a fast child must not time out"),
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod start_log_tests {
|
|
use super::*;
|
|
|
|
const NOW: u64 = 1_000_000;
|
|
|
|
async fn argv_for(name: &str, since: u64) -> Vec<String> {
|
|
let seen = std::sync::Mutex::new(Vec::new());
|
|
start_failure_log_with(name, since, NOW, |argv| {
|
|
*seen.lock().unwrap() = argv;
|
|
async { Ok(String::new()) }
|
|
})
|
|
.await
|
|
.unwrap();
|
|
seen.into_inner().unwrap()
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn odd_names_are_refused_before_journalctl_runs() {
|
|
for bad in ["", "../x", "a b", "A", "x.service", "x\n--vacuum-size=1"] {
|
|
let r =
|
|
start_failure_log_with(bad, NOW, NOW, |_| async { panic!("runner must not run") })
|
|
.await;
|
|
assert!(r.is_err(), "{bad:?} accepted");
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn the_argv_is_fixed_and_the_unit_is_built_here() {
|
|
assert_eq!(
|
|
argv_for("bob", NOW - 5).await,
|
|
[
|
|
"_SYSTEMD_UNIT=container@h-bob.service",
|
|
"--since",
|
|
"@999995",
|
|
"-n",
|
|
"30",
|
|
"-o",
|
|
"cat",
|
|
"--no-pager"
|
|
]
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn since_is_clamped_to_the_last_hour_and_not_the_future() {
|
|
assert_eq!(argv_for("bob", 0).await[2], format!("@{}", NOW - 3600));
|
|
assert_eq!(argv_for("bob", NOW + 99).await[2], format!("@{NOW}"));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn the_output_is_capped_keeping_the_newest_bytes() {
|
|
let big = format!("{}end\n", "é".repeat(START_LOG_CAP));
|
|
let (out, _) = start_failure_log_with("bob", NOW, NOW, |_| async { Ok(big) })
|
|
.await
|
|
.unwrap();
|
|
assert!(out.len() <= START_LOG_CAP, "{}", out.len());
|
|
assert!(out.ends_with("end\n"));
|
|
let (small, _) = start_failure_log_with("bob", NOW, NOW, |_| async { Ok("a\n".into()) })
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(small, "a\n");
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod host_tests {
|
|
use std::path::PathBuf;
|
|
use std::time::Duration;
|
|
|
|
use tokio::io::AsyncReadExt as _;
|
|
use tokio::net::UnixStream;
|
|
|
|
use super::*;
|
|
use crate::shared::test_support::{MissingTool, ScriptSpawn, scratch};
|
|
|
|
const TOPLEVEL: &str = "/nix/store/0123456789abcdfghijklmnpqrsvwxyz-nixos-system-bob";
|
|
|
|
/// Programs that are `sh -c <script> <program> <args…>`: every call is
|
|
/// appended to a log as `program args…`, then `behave` decides the exit
|
|
/// status and output from `$0` (the program) and `$@` (its arguments).
|
|
/// `$D` is the stub's scratch directory.
|
|
struct Stub {
|
|
dir: PathBuf,
|
|
script: String,
|
|
}
|
|
|
|
impl Stub {
|
|
fn new(behave: &str) -> Self {
|
|
let dir = scratch();
|
|
let script = format!(
|
|
"D={d}\necho \"$0 $*\" >> \"$D/calls\"\n{behave}",
|
|
d = dir.display()
|
|
);
|
|
Self { dir, script }
|
|
}
|
|
|
|
/// Timings short enough for a test that waits them out.
|
|
fn host(&self) -> Host<ScriptSpawn> {
|
|
Self::host_over(ScriptSpawn(self.script.clone()))
|
|
}
|
|
|
|
/// [`Stub::host`]'s timings over another `spawn`.
|
|
fn host_over<S>(spawn: S) -> Host<S> {
|
|
Host {
|
|
spawn,
|
|
release_timeout: Duration::from_millis(300),
|
|
release_poll: Duration::from_millis(20),
|
|
probe_timeout: Duration::from_secs(10),
|
|
}
|
|
}
|
|
|
|
/// Every command run so far, in order.
|
|
fn calls(&self) -> Vec<String> {
|
|
std::fs::read_to_string(self.dir.join("calls"))
|
|
.map(|log| log.lines().map(str::to_owned).collect())
|
|
.unwrap_or_default()
|
|
}
|
|
}
|
|
|
|
impl Drop for Stub {
|
|
fn drop(&mut self) {
|
|
std::fs::remove_dir_all(&self.dir).ok();
|
|
}
|
|
}
|
|
|
|
/// A write half and the read end of its socket.
|
|
fn pipe() -> (OwnedWriteHalf, UnixStream) {
|
|
let (ours, theirs) = UnixStream::pair().unwrap();
|
|
(ours.into_split().1, theirs)
|
|
}
|
|
|
|
/// The `(stream, data)` of every `Line` event the writer sent. Takes the
|
|
/// writer so its end closes and the read finishes.
|
|
async fn lines_sent(writer: OwnedWriteHalf, mut peer: UnixStream) -> Vec<(PrivStream, String)> {
|
|
drop(writer);
|
|
let mut raw = String::new();
|
|
peer.read_to_string(&mut raw).await.unwrap();
|
|
raw.lines()
|
|
.map(|l| match serde_json::from_str::<PrivEvent>(l).unwrap() {
|
|
PrivEvent::Line(line) => (line.stream, line.data),
|
|
PrivEvent::Done(_) => panic!("unexpected terminal event"),
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
const BAD_NAMES: [&str; 8] = [
|
|
"",
|
|
"../x",
|
|
"bob/../x",
|
|
"A",
|
|
"a b",
|
|
"x.service",
|
|
"h_bob",
|
|
"x\n--vacuum-size=1",
|
|
];
|
|
|
|
fn err_text(r: Result<(String, String)>) -> String {
|
|
format!("{:#}", r.unwrap_err())
|
|
}
|
|
|
|
// ---- name validation: nothing runs for a refused name ----
|
|
|
|
#[tokio::test]
|
|
async fn every_lifecycle_verb_refuses_a_bad_name_before_running_anything() {
|
|
let stub = Stub::new("exit 0");
|
|
let host = stub.host();
|
|
for bad in BAD_NAMES {
|
|
let results = [
|
|
start_container_with(&host, bad).await,
|
|
stop_container_with(&host, bad).await,
|
|
kill_container_with(&host, bad).await,
|
|
destroy_container_with(&host, bad).await,
|
|
];
|
|
for r in results {
|
|
assert!(
|
|
err_text(r).contains("invalid name"),
|
|
"{bad:?} not refused as a name"
|
|
);
|
|
}
|
|
}
|
|
assert_eq!(stub.calls(), Vec::<String>::new(), "a command ran");
|
|
|
|
// Control: a good name runs, so an empty call log means "refused".
|
|
kill_container_with(&host, "bob").await.unwrap();
|
|
assert_eq!(stub.calls().len(), 1);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn the_apply_verbs_check_the_name_before_the_closure() {
|
|
for bad in BAD_NAMES {
|
|
let (writer, _peer) = &mut pipe();
|
|
let err = container_toplevel_action("create", bad, TOPLEVEL, false, writer)
|
|
.await
|
|
.unwrap_err();
|
|
assert!(
|
|
format!("{err:#}").contains("invalid name"),
|
|
"{bad:?}: {err:#}"
|
|
);
|
|
}
|
|
}
|
|
|
|
// ---- the apply gate ----
|
|
|
|
/// Only the closure realised for the agent is applied. Nothing is
|
|
/// realised for these names, so each request is refused at the gate: the
|
|
/// message is the gate's, not a `nixos-container` failure.
|
|
#[tokio::test]
|
|
async fn an_unrealised_or_foreign_closure_is_refused_at_the_gate() {
|
|
for verb in ["create", "update"] {
|
|
let (writer, _peer) = &mut pipe();
|
|
let err = container_toplevel_action(verb, "never-realised", TOPLEVEL, false, writer)
|
|
.await
|
|
.unwrap_err();
|
|
let msg = format!("{err:#}");
|
|
assert!(
|
|
msg.contains("is not the closure realised for this agent (none)"),
|
|
"{verb}: {msg}"
|
|
);
|
|
}
|
|
for not_a_store_path in ["/bin/sh", "", "/nix/store/../etc/passwd"] {
|
|
let (writer, _peer) = &mut pipe();
|
|
let err = container_toplevel_action(
|
|
"create",
|
|
"never-realised",
|
|
not_a_store_path,
|
|
false,
|
|
writer,
|
|
)
|
|
.await
|
|
.unwrap_err();
|
|
assert!(
|
|
format!("{err:#}").contains("is not a store path"),
|
|
"{not_a_store_path:?}: {err:#}"
|
|
);
|
|
}
|
|
}
|
|
|
|
// ---- start ----
|
|
|
|
#[tokio::test]
|
|
async fn start_clears_the_start_limit_then_starts_the_prefixed_machine() {
|
|
let stub = Stub::new("echo started");
|
|
let (stdout, _) = start_container_with(&stub.host(), "bob").await.unwrap();
|
|
assert_eq!(stdout, "started\n");
|
|
assert_eq!(
|
|
stub.calls(),
|
|
[
|
|
"systemctl reset-failed container@h-bob.service",
|
|
"nixos-container start h-bob"
|
|
]
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_failed_reset_does_not_stop_the_start_but_a_failed_start_is_an_error() {
|
|
let stub = Stub::new("[ \"$0\" = systemctl ] && exit 1; echo ok");
|
|
start_container_with(&stub.host(), "bob").await.unwrap();
|
|
|
|
let stub =
|
|
Stub::new("[ \"$0\" = nixos-container ] && { echo no unit >&2; exit 3; }; exit 0");
|
|
let msg = err_text(start_container_with(&stub.host(), "bob").await);
|
|
assert!(msg.contains("nixos-container start h-bob failed"), "{msg}");
|
|
assert!(
|
|
msg.contains("no unit"),
|
|
"stderr must be in the error: {msg}"
|
|
);
|
|
}
|
|
|
|
// ---- kill / destroy ----
|
|
|
|
#[tokio::test]
|
|
async fn kill_sends_sigkill_to_the_prefixed_machine_through_machinectl() {
|
|
let stub = Stub::new("echo killed");
|
|
let (stdout, _) = kill_container_with(&stub.host(), "bob").await.unwrap();
|
|
assert_eq!(stdout, "killed\n");
|
|
assert_eq!(stub.calls(), ["machinectl kill h-bob --signal=SIGKILL"]);
|
|
|
|
let stub = Stub::new("echo not running >&2; exit 1");
|
|
let msg = err_text(kill_container_with(&stub.host(), "bob").await);
|
|
assert!(
|
|
msg.contains("machinectl kill h-bob --signal=SIGKILL failed"),
|
|
"{msg}"
|
|
);
|
|
assert!(msg.contains("not running"), "{msg}");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn destroy_names_only_the_prefixed_machine() {
|
|
let stub = Stub::new("exit 0");
|
|
destroy_container_with(&stub.host(), "bob").await.unwrap();
|
|
assert_eq!(stub.calls(), ["nixos-container destroy h-bob"]);
|
|
}
|
|
|
|
// ---- container_run / machinectl_run ----
|
|
|
|
#[tokio::test]
|
|
async fn container_run_returns_both_streams_and_fails_with_the_trimmed_stderr() {
|
|
let stub = Stub::new("echo out; echo err >&2");
|
|
let (out, err) = container_run_with(&stub.host(), &["list"]).await.unwrap();
|
|
assert_eq!((out.as_str(), err.as_str()), ("out\n", "err\n"));
|
|
|
|
let stub = Stub::new("echo bad thing >&2; exit 2");
|
|
let msg = err_text(container_run_with(&stub.host(), &["stop", "h-bob"]).await);
|
|
assert!(
|
|
msg.starts_with("nixos-container stop h-bob failed ("),
|
|
"{msg}"
|
|
);
|
|
assert!(msg.ends_with("): bad thing"), "{msg}");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_missing_tool_is_reported_by_name() {
|
|
let host = Stub::host_over(MissingTool);
|
|
let msg = err_text(container_run_with(&host, &["list"]).await);
|
|
assert!(msg.contains("invoke nixos-container"), "{msg}");
|
|
let msg = err_text(machinectl_run(&host, &["list"]).await);
|
|
assert!(msg.contains("invoke machinectl"), "{msg}");
|
|
}
|
|
|
|
// ---- journalctl ----
|
|
|
|
#[tokio::test]
|
|
async fn journalctl_runs_with_exactly_the_given_argv() {
|
|
let stub = Stub::new("echo line");
|
|
let argv = start_log_argv("h-bob", 5);
|
|
let out = run_journalctl(&stub.host(), Duration::from_secs(10), argv)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(out, "line\n");
|
|
assert_eq!(
|
|
stub.calls(),
|
|
["journalctl _SYSTEMD_UNIT=container@h-bob.service --since @5 -n 30 -o cat --no-pager"]
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn journalctl_failure_and_timeout_are_errors() {
|
|
let stub = Stub::new("exit 1");
|
|
let r = run_journalctl(&stub.host(), Duration::from_secs(10), vec![]).await;
|
|
assert!(format!("{:#}", r.unwrap_err()).contains("journalctl failed"));
|
|
|
|
let stub = Stub::new("exec sleep 30");
|
|
let r = run_journalctl(&stub.host(), Duration::from_millis(100), vec![]).await;
|
|
assert!(format!("{:#}", r.unwrap_err()).contains("journalctl timed out"));
|
|
}
|
|
|
|
// ---- stop: success means the name is free ----
|
|
|
|
#[tokio::test]
|
|
async fn a_stop_that_releases_the_name_returns_the_stop_output() {
|
|
let stub = Stub::new("[ \"$0\" = machinectl ] && exit 1; echo stopped");
|
|
let (stdout, _) = stop_container_with(&stub.host(), "bob").await.unwrap();
|
|
assert_eq!(stdout, "stopped\n");
|
|
assert_eq!(
|
|
stub.calls(),
|
|
[
|
|
"nixos-container stop h-bob",
|
|
"machinectl show h-bob --property=Name"
|
|
]
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_stop_that_leaves_the_name_registered_is_an_error_even_if_it_exited_zero() {
|
|
let stub = Stub::new("exit 0");
|
|
let msg = err_text(stop_container_with(&stub.host(), "bob").await);
|
|
assert!(
|
|
msg.contains("machined still holds the machine name"),
|
|
"{msg}"
|
|
);
|
|
assert!(msg.contains("stop h-bob"), "{msg}");
|
|
let probes = stub
|
|
.calls()
|
|
.iter()
|
|
.filter(|c| c.starts_with("machinectl show"))
|
|
.count();
|
|
assert!(probes > 1, "the name was polled, not probed once: {probes}");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn the_name_may_free_up_after_a_few_polls() {
|
|
let stub = Stub::new(
|
|
"if [ \"$0\" = machinectl ]; then\n\
|
|
n=$(wc -l < \"$D/probes\" 2>/dev/null || echo 0)\n\
|
|
echo x >> \"$D/probes\"\n\
|
|
[ \"$n\" -ge 2 ] && exit 1\n\
|
|
exit 0\n\
|
|
fi\n\
|
|
echo stopped",
|
|
);
|
|
let host = Host {
|
|
release_timeout: Duration::from_secs(10),
|
|
..stub.host()
|
|
};
|
|
let (stdout, _) = stop_container_with(&host, "bob").await.unwrap();
|
|
assert_eq!(stdout, "stopped\n");
|
|
let probes = stub
|
|
.calls()
|
|
.iter()
|
|
.filter(|c| c.starts_with("machinectl show"))
|
|
.count();
|
|
assert_eq!(probes, 3);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_failed_stop_that_freed_the_name_keeps_its_own_error() {
|
|
let stub = Stub::new("[ \"$0\" = machinectl ] && exit 1; echo refused >&2; exit 1");
|
|
let msg = err_text(stop_container_with(&stub.host(), "bob").await);
|
|
assert!(msg.contains("nixos-container stop h-bob failed"), "{msg}");
|
|
assert!(msg.contains("refused"), "{msg}");
|
|
assert!(!msg.contains("machined still holds"), "{msg}");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn the_registration_probe_reads_exit_status_and_fails_safe() {
|
|
let registered = Stub::new("exit 0");
|
|
assert!(machine_registered(®istered.host(), "h-bob").await);
|
|
let free = Stub::new("exit 1");
|
|
assert!(!machine_registered(&free.host(), "h-bob").await);
|
|
|
|
let host = Stub::host_over(MissingTool);
|
|
assert!(
|
|
machine_registered(&host, "h-bob").await,
|
|
"a probe that cannot run must answer registered"
|
|
);
|
|
|
|
let wedged = Stub::new("exec sleep 30");
|
|
let host = Host {
|
|
probe_timeout: Duration::from_millis(100),
|
|
..wedged.host()
|
|
};
|
|
assert!(
|
|
machine_registered(&host, "h-bob").await,
|
|
"a probe that times out must answer registered"
|
|
);
|
|
}
|
|
|
|
// ---- create / update ----
|
|
|
|
#[tokio::test]
|
|
async fn create_passes_the_closure_as_system_path_not_a_flake() {
|
|
let stub = Stub::new("echo created");
|
|
let (writer, _peer) = &mut pipe();
|
|
let (stdout, _) = apply_toplevel(&stub.host(), "create", "bob", TOPLEVEL, false, writer)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(stdout, "created\n");
|
|
assert_eq!(
|
|
stub.calls(),
|
|
[format!(
|
|
"nixos-container create h-bob --system-path {TOPLEVEL}"
|
|
)]
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_streamed_create_forwards_each_line_and_returns_nothing() {
|
|
let stub = Stub::new("echo building; echo warning >&2");
|
|
let (writer, peer) = pipe();
|
|
let mut writer = writer;
|
|
let (stdout, stderr) =
|
|
apply_toplevel(&stub.host(), "create", "bob", TOPLEVEL, true, &mut writer)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!((stdout.as_str(), stderr.as_str()), ("", ""));
|
|
let mut seen = lines_sent(writer, peer).await;
|
|
seen.sort_by_key(|(s, _)| *s == PrivStream::Stderr);
|
|
assert_eq!(
|
|
seen,
|
|
[
|
|
(PrivStream::Stdout, "building".to_owned()),
|
|
(PrivStream::Stderr, "warning".to_owned())
|
|
]
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_failed_streamed_create_carries_only_the_last_stderr_line() {
|
|
let stub = Stub::new("echo first >&2; echo last >&2; exit 4");
|
|
let (writer, peer) = pipe();
|
|
let mut writer = writer;
|
|
let err = apply_toplevel(&stub.host(), "create", "bob", TOPLEVEL, true, &mut writer)
|
|
.await
|
|
.unwrap_err();
|
|
let msg = format!("{err:#}");
|
|
assert!(
|
|
msg.contains("nixos-container create h-bob --system-path"),
|
|
"{msg}"
|
|
);
|
|
assert!(msg.ends_with(": last"), "{msg}");
|
|
let seen = lines_sent(writer, peer).await;
|
|
assert_eq!(
|
|
seen.len(),
|
|
2,
|
|
"every line is forwarded even though the error keeps one"
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn update_sets_the_per_container_profile_and_never_calls_nixos_container() {
|
|
let stub = Stub::new(
|
|
"[ \"$0\" = systemctl ] && [ \"$2\" = --property=ActiveState ] && echo inactive; exit 0",
|
|
);
|
|
let (writer, _peer) = &mut pipe();
|
|
apply_toplevel(&stub.host(), "update", "bob", TOPLEVEL, false, writer)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(
|
|
stub.calls(),
|
|
[
|
|
format!(
|
|
"nix-env -p /nix/var/nix/profiles/per-container/h-bob/system --set {TOPLEVEL}"
|
|
),
|
|
"systemctl show --property=ActiveState --value container@h-bob".to_owned(),
|
|
],
|
|
"a stopped container is not reloaded"
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn update_reloads_a_running_container_and_streams_both_steps() {
|
|
let stub = Stub::new(
|
|
"case \"$0 $1\" in\n\
|
|
'systemctl show') echo active;;\n\
|
|
'nix-env -p') echo set-done;;\n\
|
|
'systemctl reload') echo reloaded; echo reload-warn >&2;;\n\
|
|
esac",
|
|
);
|
|
let (writer, peer) = pipe();
|
|
let mut writer = writer;
|
|
let (stdout, stderr) =
|
|
apply_toplevel(&stub.host(), "update", "bob", TOPLEVEL, true, &mut writer)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(stdout, "set-done\nreloaded\n");
|
|
assert_eq!(stderr, "reload-warn\n");
|
|
assert_eq!(
|
|
stub.calls().last().unwrap(),
|
|
"systemctl reload container@h-bob"
|
|
);
|
|
assert_eq!(
|
|
lines_sent(writer, peer).await,
|
|
[
|
|
(PrivStream::Stdout, "set-done".to_owned()),
|
|
(PrivStream::Stdout, "reloaded".to_owned()),
|
|
(PrivStream::Stderr, "reload-warn".to_owned()),
|
|
]
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn update_stops_at_a_failed_profile_swap_and_reports_a_failed_reload() {
|
|
let stub = Stub::new("[ \"$0\" = nix-env ] && { echo no space >&2; exit 1; }; echo active");
|
|
let (writer, _peer) = &mut pipe();
|
|
let msg =
|
|
err_text(apply_toplevel(&stub.host(), "update", "bob", TOPLEVEL, false, writer).await);
|
|
assert!(
|
|
msg.contains("/per-container/h-bob/system --set failed"),
|
|
"{msg}"
|
|
);
|
|
assert!(msg.contains("no space"), "{msg}");
|
|
assert_eq!(stub.calls().len(), 1, "nothing runs after a failed swap");
|
|
|
|
let stub =
|
|
Stub::new("case \"$1\" in reload) echo wedged >&2; exit 1;; show) echo active;; esac");
|
|
let (writer, _peer) = &mut pipe();
|
|
let msg =
|
|
err_text(apply_toplevel(&stub.host(), "update", "bob", TOPLEVEL, false, writer).await);
|
|
assert!(
|
|
msg.contains("systemctl reload container@h-bob failed"),
|
|
"{msg}"
|
|
);
|
|
assert!(msg.contains("wedged"), "{msg}");
|
|
}
|
|
|
|
// ---- streaming: a closed stream must not spin the reader ----
|
|
|
|
/// CPU time of the calling thread. The `#[tokio::test]` runtime runs on
|
|
/// this thread, so tests running in parallel do not count.
|
|
fn thread_cpu() -> Duration {
|
|
let mut ts = libc::timespec {
|
|
tv_sec: 0,
|
|
tv_nsec: 0,
|
|
};
|
|
let rc = unsafe { libc::clock_gettime(libc::CLOCK_THREAD_CPUTIME_ID, &raw mut ts) };
|
|
assert_eq!(rc, 0);
|
|
Duration::new(
|
|
ts.tv_sec.try_into().unwrap(),
|
|
ts.tv_nsec.try_into().unwrap(),
|
|
)
|
|
}
|
|
|
|
/// Runs `create` against a `nixos-container` that executes `script`,
|
|
/// returning the client's lines and the CPU the wait cost.
|
|
async fn streamed_cpu(script: &str) -> (Vec<(PrivStream, String)>, Duration) {
|
|
let stub = Stub::new(&format!("case \"$0\" in nixos-container) {script};; esac"));
|
|
let (mut writer, peer) = pipe();
|
|
let before = thread_cpu();
|
|
apply_toplevel(&stub.host(), "create", "bob", TOPLEVEL, true, &mut writer)
|
|
.await
|
|
.unwrap();
|
|
let used = thread_cpu().saturating_sub(before);
|
|
(lines_sent(writer, peer).await, used)
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_closed_stderr_does_not_spin_while_stdout_stays_open() {
|
|
let (lines, used) = streamed_cpu("exec 2>&-; sleep 1; echo done").await;
|
|
assert_eq!(lines, [(PrivStream::Stdout, "done".to_owned())]);
|
|
// A spinning loop burns about 1s of CPU across the 1s sleep.
|
|
assert!(
|
|
used < Duration::from_millis(500),
|
|
"burned {used:?} of cpu in a 1s wait"
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_closed_stdout_does_not_spin_while_stderr_stays_open() {
|
|
let (lines, used) = streamed_cpu("exec 1>&-; sleep 1; echo done >&2").await;
|
|
assert_eq!(lines, [(PrivStream::Stderr, "done".to_owned())]);
|
|
assert!(
|
|
used < Duration::from_millis(500),
|
|
"burned {used:?} of cpu in a 1s wait"
|
|
);
|
|
}
|
|
|
|
// ---- client events ----
|
|
|
|
#[tokio::test]
|
|
async fn a_line_event_is_one_json_line_and_a_dead_client_is_not_fatal() {
|
|
let (mut writer, peer) = pipe();
|
|
write_line_event(&mut writer, PrivStream::Stderr, "boom").await;
|
|
assert_eq!(
|
|
lines_sent(writer, peer).await,
|
|
[(PrivStream::Stderr, "boom".to_owned())]
|
|
);
|
|
|
|
let (mut writer, peer) = pipe();
|
|
drop(peer);
|
|
// Best-effort: the failed write is logged, not propagated.
|
|
write_line_event(&mut writer, PrivStream::Stdout, "x").await;
|
|
write_line_event(&mut writer, PrivStream::Stdout, "y").await;
|
|
}
|
|
}
|