brokerd: the Podman runtime
Implemented-By: OpenCode session (model recorded in docs/implementer-log.md)
This commit is contained in:
@@ -0,0 +1,276 @@
|
||||
//! The Podman runtime: one fresh container per call, with the limits of `[runner]`. Whatever the
|
||||
//! tool prints is the result's content; every failure is a fixed sentence, and what Podman itself
|
||||
//! said goes only to `brokerd`'s log. M3b spec, section 6.
|
||||
|
||||
use std::ffi::OsString;
|
||||
use std::io::{Read, Write};
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::process::{Child, Command, ExitStatus, Stdio};
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::thread::JoinHandle;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use crate::config::Runner;
|
||||
use crate::podman;
|
||||
use crate::runner::{RunError, RunOutput, RunSpec, Runtime};
|
||||
|
||||
pub const RUNBOOK: &str = "see docs/runbook.md#runner-unavailable";
|
||||
pub const COULD_NOT_RUN: &str = "the tool could not run";
|
||||
pub const CANNOT_START: &str = "the tool runner could not start the container";
|
||||
pub const KILLED: &str = "the tool was stopped: it ran out of memory or was killed";
|
||||
pub const TIMED_OUT: &str = "the tool ran past its time limit";
|
||||
pub const UNEXPECTED: &str = "the tool failed with an unexpected status";
|
||||
/// How often a running container is checked.
|
||||
pub const POLL: Duration = Duration::from_millis(50);
|
||||
/// How much of Podman's standard error is kept for the log.
|
||||
pub const STDERR_KEPT: usize = 4096;
|
||||
|
||||
/// A log sink. Call it as `(self.log)("a line")`.
|
||||
pub type Log = Arc<dyn Fn(&str) + Send + Sync>;
|
||||
|
||||
pub struct Podman {
|
||||
runner: Runner,
|
||||
egress_dir: PathBuf,
|
||||
log: Log,
|
||||
next: AtomicU64,
|
||||
}
|
||||
|
||||
impl Podman {
|
||||
/// `egress_dir` is `Config::egress_dir()`; task 12 uses it.
|
||||
pub fn new(runner: Runner, egress_dir: PathBuf, log: Log) -> Podman {
|
||||
Podman {
|
||||
runner,
|
||||
egress_dir,
|
||||
log,
|
||||
next: AtomicU64::new(0),
|
||||
}
|
||||
}
|
||||
|
||||
/// Where `http_fetch` calls get their directories. Public, so the field counts as read.
|
||||
pub fn egress_dir(&self) -> &Path {
|
||||
&self.egress_dir
|
||||
}
|
||||
|
||||
/// Run one short `podman` command (`kill`, `rm -f`) whose result only matters for the log.
|
||||
pub(crate) fn podman(&self, args: &[&str]) {
|
||||
let what = args.join(" ");
|
||||
let podman = &self.runner.podman;
|
||||
match Command::new(podman)
|
||||
.args(args)
|
||||
.stdin(Stdio::null())
|
||||
.stdout(Stdio::null())
|
||||
.stderr(Stdio::null())
|
||||
.status()
|
||||
{
|
||||
Ok(status) if status.success() => {}
|
||||
Ok(status) => (self.log)(&format!("brokerd: podman {what} exited {status}")),
|
||||
Err(e) => (self.log)(&format!("brokerd: podman {what} failed: {e}")),
|
||||
}
|
||||
}
|
||||
|
||||
/// Wait for `child` until `limit` has passed since `started`. `Some(status)` if it ended;
|
||||
/// `None` if it ran too long, after `podman kill <name>`, `podman rm -f <name>`,
|
||||
/// `child.kill()` and `child.wait()` (step 5).
|
||||
fn wait(
|
||||
&self,
|
||||
child: &mut Child,
|
||||
name: &str,
|
||||
started: Instant,
|
||||
limit: Duration,
|
||||
) -> Option<ExitStatus> {
|
||||
loop {
|
||||
if let Ok(Some(status)) = child.try_wait() {
|
||||
return Some(status);
|
||||
}
|
||||
if started.elapsed() >= limit {
|
||||
self.podman(&["kill", name]);
|
||||
self.podman(&["rm", "-f", name]);
|
||||
let _ = child.kill();
|
||||
let _ = child.wait();
|
||||
return None;
|
||||
}
|
||||
std::thread::sleep(POLL);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Podman {
|
||||
/// Steps 3 to 7 for one container: spawn, feed and read it, wait, answer. Task 12 calls it too.
|
||||
/// Already written: it only joins the helpers below.
|
||||
fn run_container(
|
||||
&self,
|
||||
name: &str,
|
||||
args: Vec<OsString>,
|
||||
input: String,
|
||||
limit: Duration,
|
||||
) -> Result<RunOutput, RunError> {
|
||||
let Some(mut child) = self.spawn(args) else {
|
||||
return Err(RunError::Unavailable(CANNOT_START.to_string()));
|
||||
};
|
||||
let started = Instant::now();
|
||||
let cap = usize::try_from(self.runner.output_cap).unwrap_or(usize::MAX);
|
||||
let io = Io::start(&mut child, input, cap);
|
||||
let status = self.wait(&mut child, name, started, limit);
|
||||
let (out, truncated, err) = io.finish();
|
||||
self.answer(name, status, out, truncated, &err)
|
||||
}
|
||||
|
||||
/// Step 3: spawn `podman` with `args` and all three standard streams piped. On failure, log
|
||||
/// `brokerd: cannot start {podman path}: {e}` + "\n" + RUNBOOK and return `None`.
|
||||
fn spawn(&self, args: Vec<OsString>) -> Option<Child> {
|
||||
let podman = &self.runner.podman;
|
||||
match Command::new(podman)
|
||||
.args(args)
|
||||
.stdin(Stdio::piped())
|
||||
.stdout(Stdio::piped())
|
||||
.stderr(Stdio::piped())
|
||||
.spawn()
|
||||
{
|
||||
Ok(child) => Some(child),
|
||||
Err(e) => {
|
||||
(self.log)(&format!(
|
||||
"brokerd: cannot start {}: {}\n{}",
|
||||
podman.display(),
|
||||
e,
|
||||
RUNBOOK
|
||||
));
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Step 7: the answer, by the table in the task. `status` is `None` when the time limit was
|
||||
/// passed. `out` is the kept standard output, `err` the kept standard error (for the log only:
|
||||
/// it never goes into a `RunError`).
|
||||
fn answer(
|
||||
&self,
|
||||
name: &str,
|
||||
status: Option<ExitStatus>,
|
||||
out: Vec<u8>,
|
||||
truncated: bool,
|
||||
err: &str,
|
||||
) -> Result<RunOutput, RunError> {
|
||||
match status {
|
||||
None => {
|
||||
(self.log)(&format!(
|
||||
"brokerd: stopped container {}: it ran past its time limit",
|
||||
name
|
||||
));
|
||||
Err(RunError::Failed(TIMED_OUT.to_string()))
|
||||
}
|
||||
Some(status) => match status.code() {
|
||||
Some(0) | Some(1) => Ok(RunOutput {
|
||||
content: String::from_utf8_lossy(&out).into_owned(),
|
||||
truncated,
|
||||
}),
|
||||
Some(2) => {
|
||||
(self.log)(err);
|
||||
Err(RunError::Failed(COULD_NOT_RUN.to_string()))
|
||||
}
|
||||
Some(125..=127) => {
|
||||
(self.log)(&format!("{err}\n{RUNBOOK}"));
|
||||
Err(RunError::Unavailable(CANNOT_START.to_string()))
|
||||
}
|
||||
Some(137) => Err(RunError::Failed(KILLED.to_string())),
|
||||
_ => {
|
||||
(self.log)(&format!("brokerd: container {name} exited {status}\n{err}"));
|
||||
Err(RunError::Failed(UNEXPECTED.to_string()))
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Runtime for Podman {
|
||||
/// Steps 1 and 2, then `run_container`. Already written.
|
||||
fn run(&self, spec: &RunSpec) -> Result<RunOutput, RunError> {
|
||||
let n = self.next.fetch_add(1, Ordering::SeqCst);
|
||||
let name = podman::container_name(spec.session(), spec.call(), n);
|
||||
let args = podman::tool_args(spec, &self.runner, &name, None);
|
||||
let input = spec.arguments().canonical_json();
|
||||
let limit = self.runner.time_limit(spec.tool());
|
||||
self.run_container(&name, args, input, limit)
|
||||
}
|
||||
}
|
||||
|
||||
/// Step 4: the three threads that feed and read one container, so no pipe can block another.
|
||||
struct Io {
|
||||
writer: Option<JoinHandle<()>>,
|
||||
stdout: Option<JoinHandle<(Vec<u8>, bool)>>,
|
||||
stderr: Option<JoinHandle<(Vec<u8>, bool)>>,
|
||||
}
|
||||
|
||||
impl Io {
|
||||
/// Take `child`'s three pipes (`child.stdin.take()` and so on) and start one thread for each:
|
||||
/// write `input` to standard input and then drop it; `read_capped` standard output with `cap`;
|
||||
/// `read_capped` standard error with `STDERR_KEPT`. A pipe that is `None` gets no thread.
|
||||
fn start(child: &mut Child, input: String, cap: usize) -> Io {
|
||||
let writer = child.stdin.take().map(|mut stdin| {
|
||||
std::thread::spawn(move || {
|
||||
let _ = stdin.write_all(input.as_bytes());
|
||||
})
|
||||
});
|
||||
let stdout = child
|
||||
.stdout
|
||||
.take()
|
||||
.map(|stdout| std::thread::spawn(move || read_capped(stdout, cap)));
|
||||
let stderr = child
|
||||
.stderr
|
||||
.take()
|
||||
.map(|stderr| std::thread::spawn(move || read_capped(stderr, STDERR_KEPT)));
|
||||
Io {
|
||||
writer,
|
||||
stdout,
|
||||
stderr,
|
||||
}
|
||||
}
|
||||
|
||||
/// Step 6: join the three threads. A thread that is missing or panicked counts as empty
|
||||
/// (`join().ok()`, `unwrap_or_default()`). Returns the kept standard output, whether there was
|
||||
/// more, and the kept standard error decoded with `from_utf8_lossy`.
|
||||
fn finish(self) -> (Vec<u8>, bool, String) {
|
||||
if let Some(handle) = self.writer {
|
||||
let _ = handle.join();
|
||||
}
|
||||
let (out, truncated) = match self.stdout {
|
||||
Some(handle) => handle.join().ok().unwrap_or_default(),
|
||||
None => (Vec::new(), false),
|
||||
};
|
||||
let err = match self.stderr {
|
||||
Some(handle) => {
|
||||
let (bytes, _) = handle.join().ok().unwrap_or_default();
|
||||
String::from_utf8_lossy(&bytes).into_owned()
|
||||
}
|
||||
None => String::new(),
|
||||
};
|
||||
(out, truncated, err)
|
||||
}
|
||||
}
|
||||
|
||||
/// Everything `from` gives, keeping the first `cap` bytes; `true` if there was more. Reads on past
|
||||
/// the cap, so the writer is never blocked or broken by a closed pipe. No indexing: `buf.get(..n)`.
|
||||
fn read_capped(from: impl Read, cap: usize) -> (Vec<u8>, bool) {
|
||||
let mut from = from;
|
||||
let mut kept: Vec<u8> = Vec::new();
|
||||
let mut truncated = false;
|
||||
let mut chunk = [0u8; 8192];
|
||||
loop {
|
||||
let n = match from.read(&mut chunk) {
|
||||
Ok(0) => break,
|
||||
Ok(n) => n,
|
||||
Err(_) => break,
|
||||
};
|
||||
let remaining = cap - kept.len();
|
||||
if remaining == 0 {
|
||||
truncated = true;
|
||||
} else {
|
||||
let take = remaining.min(n);
|
||||
kept.extend_from_slice(&chunk[..take]);
|
||||
if n > take {
|
||||
truncated = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
(kept, truncated)
|
||||
}
|
||||
Reference in New Issue
Block a user