chadtmiller.com/knotless
git that lives in your Atmosphere account
7
//! Where a run's steps run: in a shell on the runner's host, or in one
//! container the runner's docker daemon starts for the run with the
//! checkout mounted, so each step finds what the steps before it left,
//! installed tools included, as on one machine.
use anyhow::{Context, Result, bail};
use rustix::process::{Pid, Signal};
use std::collections::{BTreeSet, HashSet};
use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::sync::Mutex;
use std::time::{Duration, Instant};
use tokio::process::Command;
/// The runner's own variables a step sees, so it finds its tools and the
/// docker client its daemon. Nothing else of the runner's environment, its
/// credentials least of all, reaches a step.
const PASSED: [&str; 7] = [
"PATH",
"HOME",
"LANG",
"TMPDIR",
"USER",
"DOCKER_HOST",
"DOCKER_CONFIG",
];
#[derive(Clone, Debug, PartialEq)]
pub enum Executor {
/// `sh -c` on the host, as the runner's user. For trusted repositories.
Shell,
/// `docker run --rm`, in the workflow's image or this one, within
/// `limits`.
Docker { image: String, limits: Limits },
}
/// What one run's container may take of the host, under docker: each
/// unset, as much as it likes, as before there were limits. A runner
/// running code it does not trust sets them all, so that no run takes
/// the machine from the rest, or from what else runs there.
#[derive(Clone, Debug, Default, PartialEq)]
pub struct Limits {
/// CPUs, as docker's `--cpus` takes them: `2`, `1.5`.
pub cpus: Option<String>,
/// Bytes of memory, swap none beyond it: a run past it is killed by
/// the kernel, its step failing.
pub memory: Option<u64>,
/// Processes and threads at once, so a fork bomb ends at it.
pub pids: Option<u64>,
/// The docker network runs join, made if missing as a bridge whose
/// containers cannot reach each other: so a run reaches neither
/// another run nor what the host's other containers serve on theirs.
pub network: Option<String>,
/// The container runtime docker runs it with, as its `--runtime`
/// takes one: `runsc`, gVisor's, puts a kernel of its own between a
/// run and the host's.
pub runtime: Option<String>,
/// The runtime, in place of `runtime`, of a run [`Setup::vouched`]
/// for: `runc`, docker's own, spares it gVisor's cost, as the code
/// runs on the merge queue's runner with none once it is queued.
pub trusted_runtime: Option<String>,
/// A file on the docker host that runs see as their
/// `/etc/resolv.conf`. Docker's own resolver on a network of its
/// making is reached through NAT rules in the container's network
/// namespace, which gVisor's network stack does not apply: under
/// `runsc` a run resolves nothing without one naming resolvers of
/// its own.
pub resolv: Option<String>,
}
/// The option a network made for [`Limits::network`] is made with: no
/// traffic between its containers.
const ISOLATED: &str = "com.docker.network.bridge.enable_icc=false";
/// What a run's steps run in, from its first step to its last.
#[derive(Debug, PartialEq)]
pub enum Session {
/// A shell in the checkout, with the run's variables.
Shell {
dir: PathBuf,
vars: Vec<(String, String)>,
},
/// A running container, by ID, through this docker client, and the
/// overlay volumes it mounts.
Docker {
docker: PathBuf,
id: String,
volumes: Vec<String>,
},
}
/// What a step came to.
#[derive(Debug, PartialEq)]
pub struct Exit {
/// None when it was killed: out of time, or cancelled.
pub code: Option<i32>,
/// Whether it was killed for being cancelled.
pub cancelled: bool,
pub elapsed: Duration,
}
impl Exit {
pub fn success(&self) -> bool {
self.code == Some(0)
}
}
/// What a run's steps start in.
pub struct Setup<'a> {
/// The checkout, which steps run in.
pub dir: &'a Path,
pub vars: &'a [(String, String)],
/// The image, where not the executor's own.
pub image: Option<&'a str>,
/// Overlays of what is kept from run to run, under docker only.
pub caches: &'a [Cache],
/// Where steps leave what the run made to be looked at, named to them
/// as `KNOTLESS_RUN_OUTPUTS`.
pub outputs: &'a Path,
/// The checkouts of the shared steps' repositories, if any: mounted
/// read-only at [`STEPS`] under docker.
pub steps: Option<&'a Path>,
/// Whose the run is, as its container is labeled under docker.
pub owner: Owner<'a>,
/// Whether it gives way to everything else on the host for CPU, as a
/// pull's run does: under docker, a weight of about 10 against a
/// container's 100, so a burst of pulls leaves the site, deploys and
/// main's runs what they need.
pub behind: bool,
/// Whether its code is vouched for by an owner the runner trusts, to
/// run under [`Limits::trusted_runtime`].
pub vouched: bool,
}
/// The `--cpu-shares` a run behind the rest is given: docker makes it a
/// cgroup v2 `cpu.weight` of 10.
const BEHIND: &str = "256";
/// The runner a run is of, by its account's DID, and the run, by its
/// check's key.
#[derive(Clone, Copy)]
pub struct Owner<'a> {
pub runner: &'a str,
pub run: &'a str,
}
/// The variable naming [`Setup::outputs`] to the steps.
pub const OUTPUTS: &str = "KNOTLESS_RUN_OUTPUTS";
/// Where the shared steps' repositories are, in the container: each at
/// `<STEPS>/<n>`, named to its steps as `KNOTLESS_STEP_DIR`.
pub const STEPS: &str = "/steps";
/// An overlay of a cache, mounted at `path` in a run's container: a
/// volume of docker's `local` driver over a base, `lower`, read only, with
/// the run's own `upper` layer taking what it writes.
#[derive(Debug, Clone, PartialEq)]
pub struct Cache {
pub volume: String,
pub path: String,
/// The run it is of, which the volume is labelled with.
pub run: String,
pub lower: PathBuf,
pub upper: PathBuf,
pub work: PathBuf,
}
/// The label a run's overlay volumes carry, naming the run.
pub const OVERLAY_LABEL: &str = "dev.knotless.overlay";
/// The labels a run's container carries: the runner's DID, and the run's
/// check's key.
pub const RUNNER_LABEL: &str = "dev.knotless.runner";
pub const RUN_LABEL: &str = "dev.knotless.run";
impl Executor {
pub fn parse(name: &str, image: &str) -> Option<Executor> {
match name {
"shell" => Some(Executor::Shell),
"docker" => Some(Executor::Docker {
image: image.to_string(),
limits: Limits::default(),
}),
_ => None,
}
}
/// Where the steps find the `i`th shared step's repository, checked
/// out at `at` on the host: under docker, where it is mounted.
pub fn step_dir(&self, i: usize, at: &Path) -> String {
match self {
Executor::Shell => at.to_string_lossy().into_owned(),
Executor::Docker { .. } => format!("{STEPS}/{i}"),
}
}
/// Starts what a run's steps run in, as `setup` says: under docker, a
/// container whose start, an image pulled among it, is written to
/// `log` and may take `timeout`.
pub async fn start(
&self,
setup: &Setup<'_>,
log: &std::fs::File,
timeout: Duration,
) -> Result<Session> {
self.start_with(Path::new("docker"), setup, log, timeout)
.await
}
/// [`Executor::start`], with `docker` as the docker client.
pub async fn start_with(
&self,
docker: &Path,
setup: &Setup<'_>,
log: &std::fs::File,
timeout: Duration,
) -> Result<Session> {
let Setup {
dir,
vars,
image,
caches,
outputs,
steps: _,
owner: _,
behind: _,
vouched: _,
} = *setup;
let mut vars = vars.to_vec();
let Executor::Docker {
image: default,
limits,
} = self
else {
vars.push((OUTPUTS.into(), outputs.display().to_string()));
return Ok(Session::Shell {
dir: dir.to_path_buf(),
vars,
});
};
vars.push((OUTPUTS.into(), "/outputs".into()));
let image = image.unwrap_or(default);
let mut volumes = Vec::new();
let mut made = Ok(());
for cache in caches {
made = overlay(docker, cache, log).await;
if made.is_err() {
break;
}
volumes.push(cache.volume.clone());
}
let started = match made {
Ok(()) => {
let start = Start {
image,
limits,
vars: &vars,
timeout,
};
container(docker, setup, start, log).await
}
Err(err) => Err(err),
};
// A container that did not start leaves no overlay mounted, for
// its layers to be removed.
if started.is_err() {
remove_volumes(docker, &volumes).await;
}
Ok(Session::Docker {
docker: docker.to_path_buf(),
id: started?,
volumes,
})
}
}
/// Makes the overlay volume `cache`, writing what goes wrong to `log`.
async fn overlay(docker: &Path, cache: &Cache, log: &std::fs::File) -> Result<()> {
let layers = format!(
"o=lowerdir={},upperdir={},workdir={},userxattr",
cache.lower.display(),
cache.upper.display(),
cache.work.display()
);
let mut cmd = client(docker, &[]);
cmd.args(["volume", "create", "--driver", "local", "--label"])
.arg(format!("{OVERLAY_LABEL}={}", cache.run))
.args(["--opt", "type=overlay", "--opt", "device=overlay", "--opt"])
.arg(layers)
.arg(&cache.volume)
.stdout(Stdio::null())
.stderr(log.try_clone()?);
let made = cmd.status().await.context("making a cache's overlay")?;
if !made.success() {
bail!("the overlay at {} was not made: see the log", cache.path);
}
Ok(())
}
/// What a run's container starts from: its image, within its limits, with
/// its variables, in at most its timeout.
struct Start<'a> {
image: &'a str,
limits: &'a Limits,
vars: &'a [(String, String)],
timeout: Duration,
}
/// Starts the container a run's steps run in, as `start` says, mounting
/// the checkout, the outputs and the overlays `setup` names: its ID.
async fn container(
docker: &Path,
setup: &Setup<'_>,
start: Start<'_>,
log: &std::fs::File,
) -> Result<String> {
let Start {
image,
limits,
vars,
timeout,
} = start;
if let Some(network) = &limits.network {
isolated(docker, network, log).await?;
}
let mut cmd = client(docker, vars);
cmd.args(["run", "-d", "--init", "--label"])
.arg(format!("{RUNNER_LABEL}={}", setup.owner.runner))
.arg("--label")
.arg(format!("{RUN_LABEL}={}", setup.owner.run))
.arg("-v")
.arg(format!("{}:/work", setup.dir.display()))
.arg("-v")
.arg(format!("{}:/outputs", setup.outputs.display()));
if let Some(steps) = setup.steps {
cmd.arg("-v").arg(format!("{}:{STEPS}:ro", steps.display()));
}
cmd.args(["-w", "/work"]);
if setup.behind {
cmd.args(["--cpu-shares", BEHIND]);
}
cmd.args(limited(limits, setup.vouched));
for cache in setup.caches {
let mount = format!("type=volume,src={},dst={}", cache.volume, cache.path);
cmd.arg("--mount").arg(mount);
}
// Named only: the values reach the container from the client's
// environment, never its arguments.
for (k, _) in vars {
cmd.arg("-e").arg(k);
}
// Kept running until the run ends, for each step to be run in.
cmd.args([image, "tail", "-f", "/dev/null"])
.stdin(Stdio::null())
.stderr(log.try_clone()?)
.kill_on_drop(true);
let started = tokio::time::timeout(timeout, cmd.output()).await;
let out = match started {
Ok(out) => out.context("starting a container")?,
Err(_) => bail!("{image} did not start in {}s", timeout.as_secs()),
};
if !out.status.success() {
bail!("{image} did not start: see the log");
}
let id = String::from_utf8_lossy(&out.stdout).trim().to_string();
crate::telemetry::container("container started", &id, Some(image));
Ok(id)
}
/// The arguments to `docker run` that hold a container to `limits`, a
/// `vouched` one's under its trusted runtime, if any.
fn limited(limits: &Limits, vouched: bool) -> Vec<String> {
let mut args = Vec::new();
if let Some(cpus) = &limits.cpus {
args.extend(["--cpus".into(), cpus.clone()]);
}
if let Some(bytes) = limits.memory {
// Swap as large as the memory is none beyond it.
let bytes = bytes.to_string();
args.extend([
"--memory".into(),
bytes.clone(),
"--memory-swap".into(),
bytes,
]);
}
if let Some(pids) = limits.pids {
args.extend(["--pids-limit".into(), pids.to_string()]);
}
if let Some(network) = &limits.network {
args.extend(["--network".into(), network.clone()]);
}
let trusted = limits.trusted_runtime.as_ref().filter(|_| vouched);
if let Some(runtime) = trusted.or(limits.runtime.as_ref()) {
args.extend(["--runtime".into(), runtime.clone()]);
}
if let Some(resolv) = &limits.resolv {
args.extend(["-v".into(), format!("{resolv}:/etc/resolv.conf:ro")]);
}
args
}
/// Makes the network `name` if docker has none of that name, as one whose
/// containers cannot reach each other, writing what goes wrong to `log`.
/// One there already is used as it is, and so is one another run's start
/// made meanwhile.
async fn isolated(docker: &Path, name: &str, log: &std::fs::File) -> Result<()> {
if network_there(docker, name).await? {
return Ok(());
}
let mut create = client(docker, &[]);
create
.args([
"network", "create", "--driver", "bridge", "--opt", ISOLATED, name,
])
.stdout(Stdio::null())
.stderr(log.try_clone()?);
let made = create.status().await.context("making a network")?;
if !made.success() && !network_there(docker, name).await? {
bail!("the network {name} was not made: see the log");
}
Ok(())
}
/// Whether docker has a network named `name`.
async fn network_there(docker: &Path, name: &str) -> Result<bool> {
let mut inspect = client(docker, &[]);
inspect
.args(["network", "inspect", name])
.stdout(Stdio::null())
.stderr(Stdio::null());
let asked = inspect.status().await.context("asking for a network")?;
Ok(asked.success())
}
impl Executor {
/// Removes what runs of an earlier runner left in docker, under
/// docker: their overlay volumes, and the volumes caches were kept in
/// before they were overlays. What it returns is the runs whose
/// overlays are still mounted, and so stay.
pub async fn clean_caches(&self) -> Vec<String> {
self.clean_caches_with(Path::new("docker")).await
}
/// [`Executor::clean_caches`], with `docker` as the docker client.
pub async fn clean_caches_with(&self, docker: &Path) -> Vec<String> {
if !matches!(self, Executor::Docker { .. }) {
return Vec::new();
}
let overlays = format!("label={OVERLAY_LABEL}");
for filter in [overlays.as_str(), "name=knotless-cache-"] {
let names = listed(docker, &["-q", "--filter", filter]).await;
remove_volumes(docker, &names).await;
}
let runs = format!("{{{{.Label \"{OVERLAY_LABEL}\"}}}}");
let mut left = listed(docker, &["--filter", &overlays, "--format", &runs]).await;
left.sort();
left.dedup();
left
}
/// Removes the containers the runner `runner` left, under docker:
/// those labeled as its own, and with `work`, those that mount a path
/// under it, as a run's checkout and outputs are, unless of a run in
/// `going`, by its check's key, read once the containers are listed, so
/// one started meanwhile is never taken for left. Called as it starts, with none going,
/// each is left by a runner that was killed or crashed, one from before
/// runs were labeled among them; and from time to time after. Any other
/// container is never touched.
pub async fn remove_left(
&self,
runner: &str,
going: &Mutex<HashSet<String>>,
work: Option<&Path>,
) {
let docker = Path::new("docker");
self.remove_left_with(docker, runner, going, work).await
}
/// [`Executor::remove_left`], with `docker` as the docker client.
pub async fn remove_left_with(
&self,
docker: &Path,
runner: &str,
going: &Mutex<HashSet<String>>,
work: Option<&Path>,
) {
if !matches!(self, Executor::Docker { .. }) {
return;
}
let all = words(client(docker, &[]).args(["ps", "-aq"])).await;
if all.is_empty() {
return;
}
// Each as `<id>|<runner>|<run>|<source>;<source>;...`.
let format = format!(
"{{{{.Id}}}}|{{{{index .Config.Labels \"{RUNNER_LABEL}\"}}}}|{{{{index .Config.Labels \"{RUN_LABEL}\"}}}}|{{{{range .Mounts}}}}{{{{.Source}}}};{{{{end}}}}"
);
let mut inspect = client(docker, &[]);
inspect.args(["inspect", "--format", &format]).args(&all);
let described = words(&mut inspect).await;
let left: Vec<&str> = described
.iter()
.filter_map(|line| {
let mut fields = line.splitn(4, '|');
let (id, mine, run) = (fields.next()?, fields.next()?, fields.next()?);
let mounts = fields.next().unwrap_or_default().split(';');
let under = |m: &str| work.is_some_and(|w| Path::new(m).starts_with(w));
let ours = mine == runner || mounts.into_iter().any(under);
(ours && !going.lock().unwrap().contains(run)).then_some(id)
})
.collect();
if left.is_empty() {
return;
}
let _ = client(docker, &[])
.args(["rm", "-f"])
.args(left)
.stdout(Stdio::null())
.stderr(Stdio::null())
.status()
.await;
}
}
/// The lines `cmd` prints, or none if it cannot be run.
async fn words(cmd: &mut Command) -> Vec<String> {
let listed = cmd.stderr(Stdio::null()).output().await;
let text = listed.map(|o| String::from_utf8_lossy(&o.stdout).into_owned());
let text = text.unwrap_or_default();
text.lines()
.filter(|l| !l.is_empty())
.map(str::to_string)
.collect()
}
/// The words `docker volume ls` prints, given `args`.
async fn listed(docker: &Path, args: &[&str]) -> Vec<String> {
let listed = client(docker, &[])
.args(["volume", "ls"])
.args(args)
.stderr(Stdio::null())
.output()
.await;
let text = listed.map(|o| String::from_utf8_lossy(&o.stdout).into_owned());
let text = text.unwrap_or_default();
text.split_whitespace().map(str::to_string).collect()
}
/// Removes the volumes `names`, as far as nothing still uses them.
async fn remove_volumes(docker: &Path, names: &[String]) {
if names.is_empty() {
return;
}
let _ = client(docker, &[])
.args(["volume", "rm", "-f"])
.args(names)
.stdout(Stdio::null())
.stderr(Stdio::null())
.status()
.await;
}
/// Lists the processes in a container by ID, but for itself, from `/proc`,
/// which every image has where `ps` may not be.
const RUNNING: &str =
r#"for p in /proc/[0-9]*; do p=${p#/proc/}; [ "$p" = "$$" ] || echo "$p"; done"#;
/// The docker client, seeing `vars` and the variables in [`PASSED`].
fn client(docker: &Path, vars: &[(String, String)]) -> Command {
let mut cmd = Command::new(docker);
passed(&mut cmd, vars);
cmd
}
/// Clears `cmd`'s environment to the variables in [`PASSED`] and `vars`.
fn passed(cmd: &mut Command, vars: &[(String, String)]) {
cmd.env_clear();
for name in PASSED {
if let Some(v) = std::env::var_os(name) {
cmd.env(name, v);
}
}
cmd.envs(vars.iter().map(|(k, v)| (k, v)));
}
impl Session {
/// The process that runs `script` as a step: `sh -e`, so a command
/// that fails ends the step, failed, rather than only its last one
/// counting. It sees the run's variables and `own`, this step's alone,
/// such as the secrets it is given, which the container never holds.
pub fn command(&self, script: &str, own: &[(String, String)]) -> Command {
match self {
Session::Shell { dir, vars } => {
let mut cmd = Command::new("sh");
cmd.arg("-ec").arg(script).current_dir(dir);
passed(&mut cmd, vars);
cmd.envs(own.iter().map(|(k, v)| (k, v)));
cmd
}
Session::Docker { docker, id, .. } => {
// The container has the run's variables already. Named
// only: the values reach the step from the client's
// environment, never its arguments.
let mut cmd = client(docker, own);
cmd.arg("exec");
for (k, _) in own {
cmd.arg("-e").arg(k);
}
cmd.args([id, "sh", "-ec", script]);
cmd
}
}
}
/// The processes running in the container, by ID, for
/// [`Session::end_left`] to tell which a step left: none under the
/// shell executor, whose steps share the runner's host.
pub async fn running(&self) -> Option<BTreeSet<u32>> {
let Session::Docker { docker, id, .. } = self else {
return None;
};
let mut list = client(docker, &[]);
list.args(["exec", id, "sh", "-c", RUNNING]);
let ids = words(&mut list).await;
Some(ids.iter().filter_map(|p| p.parse().ok()).collect())
}
/// Ends what is running in the container that was not in `before`, as
/// [`Session::running`] had it before `step`, and says so in `log`: the
/// processes a step given secrets left, which would hold them for the
/// steps after it to read. Nothing, with no `before`.
pub async fn end_left(
&self,
before: Option<&BTreeSet<u32>>,
step: &str,
mut log: &std::fs::File,
) -> Result<()> {
let (Some(before), Session::Docker { docker, id, .. }) = (before, self) else {
return Ok(());
};
let now = self.running().await.unwrap_or_default();
let left: Vec<String> = now.difference(before).map(u32::to_string).collect();
if left.is_empty() {
return Ok(());
}
let mut kill = client(docker, &[]);
kill.args(["exec", id, "kill", "-9"]).args(&left);
let _ = kill
.stdout(Stdio::null())
.stderr(Stdio::null())
.status()
.await;
use std::io::Write;
let n = left.len();
let said = format!("==> ended {n} of what {step} left running, as it was given secrets\n");
log.write_all(said.as_bytes())?;
Ok(())
}
/// Ends the run's container, and with it anything a step left
/// running, one killed for its time among them; then removes its
/// overlay volumes, which unmounts them, for its layers to be kept or
/// removed.
pub async fn stop(self) {
if let Session::Docker {
docker,
id,
volumes,
} = self
{
let _ = client(&docker, &[])
.args(["rm", "-f", &id])
.stdout(Stdio::null())
.stderr(Stdio::null())
.status()
.await;
crate::telemetry::container("container stopped", &id, None);
remove_volumes(&docker, &volumes).await;
}
}
}
/// Runs `cmd` to its end, `timeout` or `cancelled`, writing what it prints
/// to `log`. One that runs out of time or is cancelled is killed with
/// everything it started.
pub async fn run(
mut cmd: Command,
log: &std::fs::File,
timeout: Duration,
cancelled: impl Future<Output = ()>,
) -> Result<Exit> {
let started = Instant::now();
cmd.stdin(Stdio::null())
.stdout(log.try_clone()?)
.stderr(log.try_clone()?)
.kill_on_drop(true)
.process_group(0);
let mut child = cmd.spawn().context("starting a step")?;
// Its exit, or, killed, whether for being cancelled.
let ended = tokio::select! {
status = tokio::time::timeout(timeout, child.wait()) => status.map_err(|_| false),
_ = cancelled => Err(true),
};
let (code, cancelled) = match ended {
Ok(status) => (Some(status?.code().unwrap_or(-1)), false),
Err(cancelled) => {
// The group the step leads: a shell's children with it. Killed
// by the call itself, not a `kill` command, which the runner's
// image does not have: without one, a step cancelled or out of
// time ran on to its end, and one that hung held its job, and
// a stop, for good.
let group = child.id().and_then(|pid| i32::try_from(pid).ok());
if let Some(group) = group.and_then(Pid::from_raw) {
let _ = rustix::process::kill_process_group(group, Signal::KILL);
}
let _ = child.wait().await;
(None, cancelled)
}
};
Ok(Exit {
code,
cancelled,
elapsed: started.elapsed(),
})
}
#[cfg(test)]
mod tests {
use super::*;
fn args(cmd: &Command) -> Vec<String> {
let std = cmd.as_std();
std::iter::once(std.get_program())
.chain(std.get_args())
.map(|a| a.to_string_lossy().into_owned())
.collect()
}
#[test]
fn executors() {
assert_eq!(Executor::parse("shell", "x"), Some(Executor::Shell));
assert_eq!(Executor::parse("vm", "x"), None);
let at = Path::new("/tmp/run/steps/2");
assert_eq!(Executor::Shell.step_dir(2, at), "/tmp/run/steps/2");
let docker = Executor::parse("docker", "x").unwrap();
assert_eq!(docker.step_dir(2, at), format!("{STEPS}/2"));
let vars = [("SECRET".to_string(), "v".to_string())];
let shell = Session::Shell {
dir: "/w".into(),
vars: vars.to_vec(),
};
let own = [("KEY".to_string(), "k".to_string())];
let cmd = shell.command("make", &own);
assert_eq!(args(&cmd), ["sh", "-ec", "make"]);
let env: Vec<_> = cmd.as_std().get_envs().collect();
assert!(env.contains(&("SECRET".as_ref(), Some("v".as_ref()))));
assert!(env.contains(&("KEY".as_ref(), Some("k".as_ref()))));
let docker = Session::Docker {
docker: "docker".into(),
id: "c1".into(),
volumes: Vec::new(),
};
assert_eq!(
args(&docker.command("make", &[])).join(" "),
"docker exec c1 sh -ec make"
);
let cmd = docker.command("make", &own);
assert_eq!(args(&cmd).join(" "), "docker exec -e KEY c1 sh -ec make");
let env: Vec<_> = cmd.as_std().get_envs().collect();
assert!(env.contains(&("KEY".as_ref(), Some("k".as_ref()))));
// A secret passed to a shared step under the name its step file
// asks for reaches the client by that name alone.
let step = knotless::workflow::Step {
secrets: Some([("SSH_KEY".to_string(), "DEPLOY_KEY".to_string())].into()),
..Default::default()
};
let granted = [("DEPLOY_KEY".to_string(), "k".to_string())];
let cmd = docker.command("make", &step.seen(&granted));
assert_eq!(
args(&cmd).join(" "),
"docker exec -e SSH_KEY c1 sh -ec make"
);
let env: Vec<_> = cmd.as_std().get_envs().collect();
assert!(env.contains(&("SSH_KEY".as_ref(), Some("k".as_ref()))));
assert!(!env.iter().any(|(k, _)| *k == "DEPLOY_KEY"));
}
/// A stand-in docker client, in `dir`, that writes what it is asked
/// to `dir/calls`: it starts container `c1` unless the image is
/// `bad` or `slow`, makes every volume but `knotless-overlay-bad`,
/// lists overlay volumes `o1` and `o2`, of runs `r1` and `r0`, and an
/// old cache volume `old`, has network `have`, makes any other but
/// `gone` and `raced` (made meanwhile by another, as its making
/// fails), and runs what else it is asked to exec here.
fn fake_docker(dir: &Path) -> PathBuf {
let path = dir.join("docker");
let calls = dir.join("calls");
let script = format!(
"#!/bin/sh\necho \"$*\" >> {calls}\ncase \"$*\" in\n 'network inspect have') exit 0;;\n 'network inspect raced') test -e {raced}; exit $?;;\n 'network inspect '*) exit 1;;\n 'network create '*' raced') touch {raced}; exit 1;;\n 'network create '*' gone') echo 'no network' >&2; exit 1;;\n 'network create '*) exit 0;;\n *' bad tail '*) echo 'no such image' >&2; exit 1;;\n *' slow tail '*) sleep 5;;\n run\\ *) echo c1;;\n *' knotless-overlay-bad') echo 'no overlay' >&2; exit 1;;\n *' for p in /proc/'*) echo 1; echo 7; echo 9; echo x;;\n 'exec c1 kill -9 '*) ;;\n exec\\ *) shift 2; exec \"$@\";;\n 'volume ls -q --filter label='*) echo o1; echo o2;;\n 'volume ls -q --filter name='*) echo old;;\n 'volume ls --filter '*) echo r1; echo r1; echo r0;;\n 'ps -aq') echo c7; echo c8; echo c9; echo c10;;\n inspect\\ *) echo 'c7|did:left|r7|/x;'; echo 'c8|did:left|r8|/w/work/a/src;'; echo 'c9|||/w/work/b/src;/w/work/b/outputs;'; echo 'c10|did:other||/w/workshop;';;\nesac\n",
calls = calls.display(),
raced = dir.join("raced").display()
);
// Copied into place by another process: a file this one wrote can
// be held open for writing by a test's fork meanwhile, and running
// it then fails, "Text file busy".
let written = dir.join("docker.sh");
std::fs::write(&written, script).unwrap();
let copied = std::process::Command::new("cp")
.arg(&written)
.arg(&path)
.status()
.unwrap();
assert!(copied.success());
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).unwrap();
path
}
#[tokio::test]
async fn what_a_step_leaves_running_is_ended() {
let dir = tempfile::tempdir().unwrap();
let docker = fake_docker(dir.path());
let log_path = dir.path().join("log");
let log = std::fs::File::create(&log_path).unwrap();
let session = Session::Docker {
docker,
id: "c1".into(),
volumes: Vec::new(),
};
let running = session.running().await.unwrap();
assert_eq!(running, BTreeSet::from([1, 7, 9]));
let before = BTreeSet::from([1, 7]);
session
.end_left(Some(&before), "deploy", &log)
.await
.unwrap();
session
.end_left(Some(&running), "quiet", &log)
.await
.unwrap();
session.end_left(None, "unwatched", &log).await.unwrap();
let calls = std::fs::read_to_string(dir.path().join("calls")).unwrap();
assert!(calls.contains("exec c1 kill -9 9\n"), "{calls}");
assert_eq!(calls.matches("kill").count(), 1, "{calls}");
let said = std::fs::read_to_string(&log_path).unwrap();
assert_eq!(
said,
"==> ended 1 of what deploy left running, as it was given secrets\n"
);
let shell = Session::Shell {
dir: "/w".into(),
vars: Vec::new(),
};
assert_eq!(shell.running().await, None);
shell.end_left(Some(&before), "s", &log).await.unwrap();
}
fn overlay_of(volume: &str, path: &str) -> Cache {
Cache {
volume: volume.into(),
path: path.into(),
run: "r1".into(),
lower: "/c/bases/t/0".into(),
upper: format!("/c/runs/r1/{volume}/upper").into(),
work: format!("/c/runs/r1/{volume}/work").into(),
}
}
#[tokio::test]
async fn one_container_a_run() {
let dir = tempfile::tempdir().unwrap();
let docker = fake_docker(dir.path());
let log = std::fs::File::create(dir.path().join("log")).unwrap();
let vars = [("SECRET".to_string(), "v".to_string())];
let executor = Executor::parse("docker", "debian:stable-slim").unwrap();
let time = Duration::from_secs(10);
let src = Path::new("/w/src");
let caches = [overlay_of("x", "/work/target"), overlay_of("y", "/root")];
let setup = |image, vars, caches| Setup {
dir: src,
vars,
image,
caches,
outputs: Path::new("/w/outputs"),
steps: None,
owner: Owner {
runner: "did:r",
run: "k1",
},
behind: false,
vouched: false,
};
// A pull's, behind the rest for CPU.
let behind = Setup {
behind: true,
steps: Some(Path::new("/w/steps")),
..setup(Some("rust:1"), &vars, &caches)
};
let session = executor
.start_with(&docker, &behind, &log, time)
.await
.unwrap();
let ran = run(
session.command("echo ran", &[]),
&log,
time,
std::future::pending(),
);
assert!(ran.await.unwrap().success());
// Its overlays go with its container.
session.stop().await;
// Containers a runner left are removed: its own, but for a run
// going, and any mounting a path under its work directory, as one
// from before runs were labeled does. One that left none removes
// none, and one with no docker to ask removes nothing.
let work = Some(Path::new("/w/work"));
let going = Mutex::new(HashSet::from(["r8".to_string()]));
executor
.remove_left_with(&docker, "did:left", &going, work)
.await;
let none = Mutex::default();
executor
.remove_left_with(&docker, "did:none", &none, None)
.await;
let gone = Path::new("/nonexistent/docker");
executor
.remove_left_with(gone, "did:left", &none, work)
.await;
Executor::Shell.remove_left("did:left", &none, None).await;
let calls = std::fs::read_to_string(dir.path().join("calls")).unwrap();
let inspect = "inspect --format {{.Id}}|{{index .Config.Labels \"dev.knotless.runner\"}}|\
{{index .Config.Labels \"dev.knotless.run\"}}|{{range .Mounts}}{{.Source}};{{end}} \
c7 c8 c9 c10";
assert_eq!(
calls,
format!(
"volume create --driver local --label dev.knotless.overlay=r1 --opt type=overlay \
--opt device=overlay --opt o=lowerdir=/c/bases/t/0,upperdir=/c/runs/r1/x/upper,\
workdir=/c/runs/r1/x/work,userxattr x\n\
volume create --driver local --label dev.knotless.overlay=r1 --opt type=overlay \
--opt device=overlay --opt o=lowerdir=/c/bases/t/0,upperdir=/c/runs/r1/y/upper,\
workdir=/c/runs/r1/y/work,userxattr y\n\
run -d --init --label dev.knotless.runner=did:r --label dev.knotless.run=k1 -v /w/src:/work -v /w/outputs:/outputs -v /w/steps:/steps:ro -w /work \
--cpu-shares 256 --mount type=volume,src=x,dst=/work/target \
--mount type=volume,src=y,dst=/root \
-e SECRET -e KNOTLESS_RUN_OUTPUTS rust:1 tail -f /dev/null\n\
exec c1 sh -ec echo ran\n\
rm -f c1\n\
volume rm -f x y\n\
ps -aq\n{inspect}\n\
rm -f c7 c9\n\
ps -aq\n{inspect}\n"
)
);
assert_eq!(
std::fs::read_to_string(dir.path().join("log")).unwrap(),
"ran\n"
);
// An image that will not start, or not in time, is an error, and
// the overlays made for it go.
std::fs::remove_file(dir.path().join("calls")).unwrap();
let err = executor
.start_with(&docker, &setup(Some("bad"), &[], &caches[..1]), &log, time)
.await
.unwrap_err();
assert_eq!(err.to_string(), "bad did not start: see the log");
let calls = std::fs::read_to_string(dir.path().join("calls")).unwrap();
assert!(
calls.ends_with("tail -f /dev/null\nvolume rm -f x\n"),
"{calls}"
);
// One whose overlay is not made does not start.
std::fs::remove_file(dir.path().join("calls")).unwrap();
let bad = [
caches[0].clone(),
overlay_of("knotless-overlay-bad", "/root"),
];
let err = executor
.start_with(&docker, &setup(Some("rust:1"), &[], &bad), &log, time)
.await
.unwrap_err();
assert_eq!(
err.to_string(),
"the overlay at /root was not made: see the log"
);
let calls = std::fs::read_to_string(dir.path().join("calls")).unwrap();
assert!(!calls.contains("run -d"), "{calls}");
assert!(
calls.ends_with(" knotless-overlay-bad\nvolume rm -f x\n"),
"{calls}"
);
let err = executor
.start_with(
&docker,
&setup(Some("slow"), &[], &[]),
&log,
Duration::from_millis(100),
)
.await
.unwrap_err();
assert_eq!(err.to_string(), "slow did not start in 0s");
let gone = Path::new("/nonexistent/docker");
let err = executor
.start_with(gone, &setup(None, &[], &[]), &log, time)
.await
.unwrap_err();
assert_eq!(err.to_string(), "starting a container");
let err = executor
.start_with(gone, &setup(None, &[], &caches), &log, time)
.await
.unwrap_err();
assert_eq!(err.to_string(), "making a cache's overlay");
// The shell's is the checkout itself.
let shell = Executor::Shell
.start(&setup(None, &vars, &caches), &log, time)
.await
.unwrap();
assert_eq!(
shell,
Session::Shell {
dir: src.into(),
vars: vec![vars[0].clone(), (OUTPUTS.into(), "/w/outputs".into())]
}
);
shell.stop().await;
}
#[tokio::test]
async fn an_earlier_runners_caches_go() {
let dir = tempfile::tempdir().unwrap();
let docker = fake_docker(dir.path());
let executor = Executor::parse("docker", "debian:stable-slim").unwrap();
// The runs whose overlays are still mounted stay, each once.
assert_eq!(executor.clean_caches_with(&docker).await, ["r0", "r1"]);
assert!(Executor::Shell.clean_caches().await.is_empty());
let calls = std::fs::read_to_string(dir.path().join("calls")).unwrap();
assert_eq!(
calls,
"volume ls -q --filter label=dev.knotless.overlay\n\
volume rm -f o1 o2\n\
volume ls -q --filter name=knotless-cache-\n\
volume rm -f old\n\
volume ls --filter label=dev.knotless.overlay --format {{.Label \"dev.knotless.overlay\"}}\n"
);
}
#[tokio::test]
async fn a_run_within_its_limits() {
let dir = tempfile::tempdir().unwrap();
let docker = fake_docker(dir.path());
let log = std::fs::File::create(dir.path().join("log")).unwrap();
let time = Duration::from_secs(10);
let caches = [overlay_of("x", "/work/target")];
let setup = Setup {
dir: Path::new("/w/src"),
vars: &[],
image: None,
caches: &caches,
outputs: Path::new("/w/outputs"),
steps: None,
owner: Owner {
runner: "did:r",
run: "k1",
},
behind: false,
vouched: false,
};
let within = |network: &str| Executor::Docker {
image: "rust:1".into(),
limits: Limits {
cpus: Some("1.5".into()),
memory: Some(4 << 30),
pids: Some(512),
network: Some(network.into()),
runtime: Some("runsc".into()),
trusted_runtime: Some("runc".into()),
resolv: Some("/r".into()),
},
};
let calls = || {
let path = dir.path().join("calls");
let text = std::fs::read_to_string(&path).unwrap();
std::fs::remove_file(&path).unwrap();
text
};
let create = "network create --driver bridge --opt \
com.docker.network.bridge.enable_icc=false";
let run = "run -d --init --label dev.knotless.runner=did:r --label dev.knotless.run=k1 \
-v /w/src:/work -v /w/outputs:/outputs -w /work \
--cpus 1.5 --memory 4294967296 --memory-swap 4294967296 --pids-limit 512";
let mount = "--mount type=volume,src=x,dst=/work/target -e KNOTLESS_RUN_OUTPUTS \
rust:1 tail -f /dev/null";
// A network there already is joined as it is; one missing is made,
// its containers kept apart; one made meanwhile by another run's
// start is joined too.
for (network, asked) in [
("have", "network inspect have\n".to_string()),
("new", format!("network inspect new\n{create} new\n")),
(
"raced",
format!("network inspect raced\n{create} raced\nnetwork inspect raced\n"),
),
] {
let session = within(network)
.start_with(&docker, &setup, &log, time)
.await
.unwrap();
session.stop().await;
let calls = calls();
let started = format!(
"{asked}{run} --network {network} --runtime runsc \
-v /r:/etc/resolv.conf:ro {mount}\n"
);
assert!(calls.contains(&started), "{calls}");
}
// Code vouched for runs under the trusted runtime instead.
let vouched = Setup {
vouched: true,
..setup
};
let session = within("have")
.start_with(&docker, &vouched, &log, time)
.await
.unwrap();
session.stop().await;
let ran = calls();
assert!(ran.contains(" --runtime runc -v /r:"), "{ran}");
// One that cannot be made starts nothing, and its overlays go.
let err = within("gone")
.start_with(&docker, &setup, &log, time)
.await
.unwrap_err();
assert_eq!(
err.to_string(),
"the network gone was not made: see the log"
);
let calls = calls();
assert!(!calls.contains("run -d"), "{calls}");
assert!(
calls.ends_with("network inspect gone\nvolume rm -f x\n"),
"{calls}"
);
// With no docker to ask, it is not known to be there.
let gone = Path::new("/nonexistent/docker");
let err = within("have")
.start_with(
gone,
&Setup {
caches: &[],
..setup
},
&log,
time,
)
.await
.unwrap_err();
assert_eq!(err.to_string(), "asking for a network");
let text = std::fs::read_to_string(dir.path().join("log")).unwrap();
assert_eq!(text, "no network\n");
}
#[tokio::test]
async fn runs_and_times_out() {
let dir = tempfile::tempdir().unwrap();
let log = std::fs::File::create(dir.path().join("log")).unwrap();
let vars = [("WHO".to_string(), "step".to_string())];
let shell = Session::Shell {
dir: dir.path().into(),
vars: vars.to_vec(),
};
let cmd = shell.command("echo hi $WHO; echo err >&2; exit 3", &[]);
let never = std::future::pending::<()>;
let exit = run(cmd, &log, Duration::from_secs(10), never())
.await
.unwrap();
assert_eq!(exit.code, Some(3));
assert!(!exit.success());
// A child of the shell is killed with it.
let cmd = shell.command("sleep 30 & sleep 30", &[]);
let exit = run(cmd, &log, Duration::from_millis(200), never())
.await
.unwrap();
assert_eq!((exit.code, exit.cancelled), (None, false));
assert!(exit.elapsed < Duration::from_secs(5));
// Cancelled, likewise.
let cmd = shell.command("sleep 30 & sleep 30", &[]);
let soon = tokio::time::sleep(Duration::from_millis(200));
let exit = run(cmd, &log, Duration::from_secs(10), soon).await.unwrap();
assert_eq!((exit.code, exit.cancelled), (None, true));
assert!(exit.elapsed < Duration::from_secs(5));
let text = std::fs::read_to_string(dir.path().join("log")).unwrap();
assert_eq!(text, "hi step\nerr\n");
let mut missing = shell.command("true", &[]);
missing.current_dir(dir.path().join("gone"));
assert!(
run(missing, &log, Duration::from_secs(1), never())
.await
.is_err()
);
}
}