From 54de18f754c93ba4365f27e288f4304665f45f23 Mon Sep 17 00:00:00 2001 From: Maksym Pavlenko Date: Fri, 4 Sep 2026 15:24:11 -0700 Subject: [PATCH] runc-shim: add unit tests Adds test coverage for the shim, driven through the ttrpc Task service over a fake OCI runtime so the tests describe observable behaviour rather than the types underneath. Covers the container and exec lifecycle, exit and wait handling, published events and error mapping, plus the leaf helpers in common.rs, io.rs and runc.rs. RuncFactory gains an injectable spawner, completing a seam the rest of the chain already had. No behaviour change. --- crates/runc-shim/Cargo.toml | 3 + crates/runc-shim/src/common.rs | 198 +++++- crates/runc-shim/src/io.rs | 14 + crates/runc-shim/src/runc.rs | 131 ++-- crates/runc-shim/src/service.rs | 7 +- crates/runc-shim/src/task.rs | 1151 +++++++++++++++++++++++++++++++ 6 files changed, 1456 insertions(+), 48 deletions(-) diff --git a/crates/runc-shim/Cargo.toml b/crates/runc-shim/Cargo.toml index fa12b6c9..9444b048 100644 --- a/crates/runc-shim/Cargo.toml +++ b/crates/runc-shim/Cargo.toml @@ -40,6 +40,9 @@ async-trait.workspace = true tokio = { workspace = true, features = ["macros", "rt-multi-thread", "process", "sync", "fs", "io-util", "net", "time", "signal"] } rustix = { version = "1.1", default-features = false, features = ["std", "termios"] } +[dev-dependencies] +tempfile.workspace = true + [package.metadata.cargo-machete] ignored = ["libc"] diff --git a/crates/runc-shim/src/common.rs b/crates/runc-shim/src/common.rs index 6421ce6e..7100950d 100644 --- a/crates/runc-shim/src/common.rs +++ b/crates/runc-shim/src/common.rs @@ -155,7 +155,7 @@ pub fn create_runc( namespace: &str, bundle: impl AsRef, opts: &Options, - spawner: Option>, + spawner: Arc, ) -> containerd_shim::Result { let runtime = if runtime.is_empty() { DEFAULT_COMMAND @@ -177,9 +177,7 @@ pub fn create_runc( .log(log) .log_json() .systemd_cgroup(opts.systemd_cgroup); - if let Some(s) = spawner { - gopts.custom_spawner(s); - } + gopts.custom_spawner(spawner); gopts .build() .map_err(other_error!("unable to create runc instance")) @@ -263,3 +261,195 @@ where )), } } + +#[cfg(test)] +mod tests { + use std::path::PathBuf; + + use containerd_shim::{ + api::ExecProcessRequest, + protos::protobuf::{well_known_types::any::Any, MessageField}, + Error, + }; + use oci_spec::runtime::{ + LinuxNamespace, LinuxNamespaceBuilder, LinuxNamespaceType, Process, Spec, + }; + + use super::{check_kill_error, create_io, get_spec_from_request, has_shared_pid_namespace}; + use crate::io::Stdio; + + // ----------------------------------------------------------------------- + // check_kill_error: turning runc stderr into a shim error + // ----------------------------------------------------------------------- + + #[test] + fn kill_error_finished() { + for msg in [ + "process already finished", + "container not running", + "no such process", + // Casing comes straight off runc stderr, so matching must be insensitive. + "Container Not Running", + ] { + assert!( + matches!(check_kill_error(msg.to_string()), Error::NotFoundError(_)), + "{:?} should be reported as not found", + msg + ); + } + } + + #[test] + fn kill_error_no_container() { + match check_kill_error("container does not exist".to_string()) { + Error::NotFoundError(msg) => assert_eq!(msg, "no such container"), + other => panic!("expected NotFoundError, got {:?}", other), + } + } + + #[test] + fn kill_error_unknown() { + let err = check_kill_error("disk on fire".to_string()); + assert!( + !matches!(err, Error::NotFoundError(_)), + "an unknown failure must not be flattened into not-found" + ); + assert!(err.to_string().contains("disk on fire")); + } + + // ----------------------------------------------------------------------- + // has_shared_pid_namespace: whether the shim must reap the container's children + // ----------------------------------------------------------------------- + + fn spec_with_namespaces(namespaces: Option>) -> Spec { + let mut spec = Spec::default(); + let mut linux = spec + .linux() + .clone() + .expect("the default spec has a linux section"); + linux.set_namespaces(namespaces); + spec.set_linux(Some(linux)); + spec + } + + fn pid_namespace(path: Option<&str>) -> LinuxNamespace { + let mut builder = LinuxNamespaceBuilder::default().typ(LinuxNamespaceType::Pid); + if let Some(path) = path { + builder = builder.path(PathBuf::from(path)); + } + builder.build().expect("build pid namespace") + } + + #[test] + fn pid_ns() { + // No path means the container gets a namespace of its own. + let private = spec_with_namespaces(Some(vec![pid_namespace(None)])); + assert!(!has_shared_pid_namespace(&private)); + + // A path means it joins a namespace that already exists. + let joined = spec_with_namespaces(Some(vec![pid_namespace(Some("/proc/1/ns/pid"))])); + assert!(has_shared_pid_namespace(&joined)); + + // Nothing isolates the container, so it shares whatever it was given. + assert!(has_shared_pid_namespace(&spec_with_namespaces(Some( + vec![] + )))); + assert!(has_shared_pid_namespace(&spec_with_namespaces(None))); + + let mut no_linux = Spec::default(); + no_linux.set_linux(None); + assert!(has_shared_pid_namespace(&no_linux)); + } + + // ----------------------------------------------------------------------- + // get_spec_from_request + // ----------------------------------------------------------------------- + + fn exec_request_with_spec(spec: Option<&Process>, terminal: bool) -> ExecProcessRequest { + let mut req = ExecProcessRequest { + terminal, + ..Default::default() + }; + if let Some(spec) = spec { + let mut any = Any::new(); + // Mirrors what containerd sends, even though the shim decodes the + // payload as JSON without consulting the type url. + any.type_url = "types.containerd.io/opencontainers/runtime-spec/1/Process".to_string(); + any.value = serde_json::to_vec(spec).expect("encode process spec"); + req.spec = MessageField::some(any); + } + req + } + + #[test] + fn spec_terminal_override() { + let mut spec = Process::default(); + spec.set_terminal(Some(false)); + + let parsed = get_spec_from_request(&exec_request_with_spec(Some(&spec), true)) + .expect("parse process spec"); + assert_eq!( + parsed.terminal(), + Some(true), + "the exec request decides whether the process gets a tty" + ); + + let parsed = get_spec_from_request(&exec_request_with_spec(Some(&spec), false)) + .expect("parse process spec"); + assert_eq!(parsed.terminal(), Some(false)); + } + + #[test] + fn spec_missing() { + let err = get_spec_from_request(&exec_request_with_spec(None, false)) + .expect_err("a spec is required to exec"); + assert!( + matches!(err, Error::InvalidArgument(_)), + "expected InvalidArgument, got {:?}", + err + ); + } + + // ----------------------------------------------------------------------- + // create_io: choosing the stdio driver from the containerd-supplied paths + // ----------------------------------------------------------------------- + + #[test] + fn io_null() { + let stdio = Stdio::default(); + assert!(stdio.is_null()); + + let pio = create_io("id", 0, 0, &stdio).expect("create io"); + assert!(pio.io.is_some(), "null stdio still needs a driver"); + assert!(pio.uri.is_none()); + assert!(!pio.copy); + } + + #[test] + fn io_fifo() { + let stdio = Stdio::new("/run/in", "/run/out", "/run/err", false); + assert!(!stdio.is_null()); + + let pio = create_io("id", 0, 0, &stdio).expect("create io"); + assert_eq!(pio.uri.as_deref(), Some("fifo:///run/out")); + assert!(pio.io.is_some()); + assert!( + !pio.copy, + "runc writes to the fifos directly, so the shim does not copy" + ); + } + + /// Documents current behaviour, which is not obviously the intended one: a + /// non-fifo scheme yields no io driver and leaves `copy` false, so nothing + /// is wired up and the container's output goes nowhere. Recorded so a fix + /// shows up here as a deliberate change. + #[test] + fn io_scheme() { + let stdio = Stdio::new("", "binary:///usr/bin/log", "", false); + + let pio = create_io("id", 0, 0, &stdio).expect("create io"); + assert_eq!(pio.uri.as_deref(), Some("binary:///usr/bin/log")); + assert!(pio.io.is_none()); + assert!(!pio.copy); + } +} diff --git a/crates/runc-shim/src/io.rs b/crates/runc-shim/src/io.rs index e7a43679..0677d32c 100644 --- a/crates/runc-shim/src/io.rs +++ b/crates/runc-shim/src/io.rs @@ -36,3 +36,17 @@ impl Stdio { self.stdin.is_empty() && self.stdout.is_empty() && self.stderr.is_empty() } } + +#[cfg(test)] +mod tests { + use super::Stdio; + + #[test] + fn is_null() { + assert!(Stdio::new("", "", "", false).is_null()); + assert!(Stdio::new("", "", "", true).is_null(), "a tty is not stdio"); + assert!(!Stdio::new("/run/in", "", "", false).is_null()); + assert!(!Stdio::new("", "/run/out", "", false).is_null()); + assert!(!Stdio::new("", "", "/run/err", false).is_null()); + } +} diff --git a/crates/runc-shim/src/runc.rs b/crates/runc-shim/src/runc.rs index 031abfd9..5c1aca75 100644 --- a/crates/runc-shim/src/runc.rs +++ b/crates/runc-shim/src/runc.rs @@ -75,8 +75,20 @@ pub type InitProcess = ProcessTemplate; pub type RuncContainer = ContainerTemplate; -#[derive(Clone, Default)] -pub(crate) struct RuncFactory {} +pub(crate) struct RuncFactory { + /// How the OCI runtime binary is launched. Defaults to [`ShimExecutor`], + /// which reaps through the shim's exit monitor; substitutable so that a + /// caller can supply its own execution strategy. + pub(crate) spawner: Arc, +} + +impl Default for RuncFactory { + fn default() -> Self { + Self { + spawner: Arc::new(ShimExecutor::default()), + } + } +} #[async_trait] impl ContainerFactory for RuncFactory { @@ -111,13 +123,7 @@ impl ContainerFactory for RuncFactory { mount_rootfs(&m, rootfs.as_path()).await? } - let runc = create_runc( - runtime, - ns, - bundle, - &opts, - Some(Arc::new(ShimExecutor::default())), - )?; + let runc = create_runc(runtime, ns, bundle, &opts, self.spawner.clone())?; let id = req.id(); let stdio = Stdio::new(req.stdin(), req.stdout(), req.stderr(), req.terminal()); @@ -268,7 +274,7 @@ pub struct RuncInitLifecycle { opts: Options, bundle: String, exit_signal: Arc, - /// Cache for cgroup paths to avoid repeated /proc//cgroup parsing + /// Cache for cgroup paths to avoid repeated `/proc//cgroup` parsing #[cfg(target_os = "linux")] cgroup_cache: RwLock>, } @@ -841,42 +847,85 @@ async fn wait_pid(pid: i32, s: Subscription) -> i32 { #[cfg(test)] mod tests { - use std::{os::unix::process::ExitStatusExt, path::Path, process::ExitStatus}; - - use containerd_shim::util::{mkdir, write_str_to_file}; - use runc::error::Error::CommandFailed; - use tokio::fs::remove_dir_all; + use std::{os::unix::process::ExitStatusExt, process::ExitStatus}; + + use super::runtime_error; + use crate::common::LOG_JSON_FILE; + + fn command_failed() -> ::runc::error::Error { + ::runc::error::Error::CommandFailed { + // Raw wait(2) status: the exit code lives in bits 8-15, so a bare + // `1` would render as "killed by SIGHUP" rather than "exit code 1". + status: ExitStatus::from_raw(1 << 8), + stdout: String::new(), + stderr: String::new(), + } + } - use crate::{common::LOG_JSON_FILE, runc::runtime_error}; + #[tokio::test] + async fn runtime_error_from_log() { + let bundle = tempfile::tempdir().expect("create bundle"); + std::fs::write( + bundle.path().join(LOG_JSON_FILE), + "{\"level\":\"info\",\"msg\":\"hello world\",\"time\":\"2022-11-25\"}\n\ + {\"level\":\"error\",\"msg\":\"failed error\",\"time\":\"2022-11-26\"}\n\ + {\"level\":\"error\",\"msg\":\"panic\",\"time\":\"2022-11-27\"}\n", + ) + .expect("write runtime log"); + + let err = runtime_error( + bundle.path().to_str().unwrap(), + command_failed(), + "OCI runtime create failed", + ) + .await; + let msg = err.to_string(); + assert!( + msg.contains("OCI runtime create failed"), + "the caller context is kept: {:?}", + msg + ); + assert!( + msg.contains("panic"), + "the last error line wins, got {:?}", + msg + ); + } #[tokio::test] - async fn test_runtime_error() { - let empty_err = CommandFailed { - status: ExitStatus::from_raw(1), - stdout: "".to_string(), - stderr: "".to_string(), - }; - let log_json = "\ - {\"level\":\"info\",\"msg\":\"hello world\",\"time\":\"2022-11-25\"}\n\ - {\"level\":\"error\",\"msg\":\"failed error\",\"time\":\"2022-11-26\"}\n\ - {\"level\":\"error\",\"msg\":\"panic\",\"time\":\"2022-11-27\"}\n\ - "; - let test_dir = "/tmp/shim-test"; - let _ = mkdir(test_dir, 0o744).await; - write_str_to_file(Path::new(test_dir).join(LOG_JSON_FILE).as_path(), log_json) - .await - .expect("write log json should not be error"); + async fn runtime_error_no_error_line() { + let bundle = tempfile::tempdir().expect("create bundle"); + std::fs::write( + bundle.path().join(LOG_JSON_FILE), + "{\"level\":\"info\",\"msg\":\"nothing went wrong\",\"time\":\"2022-11-25\"}\n", + ) + .expect("write runtime log"); + + let err = runtime_error( + bundle.path().to_str().unwrap(), + command_failed(), + "OCI runtime start failed", + ) + .await; + assert!(err.to_string().contains("no OCI runtime error in logfile")); + } - let expectd_msg = "panic"; - let actual_err = runtime_error(test_dir, empty_err, "").await; - remove_dir_all(test_dir) - .await - .expect("remove test dir should not be error"); + #[tokio::test] + async fn runtime_error_no_log() { + let bundle = tempfile::tempdir().expect("create bundle"); + + let err = runtime_error( + bundle.path().to_str().unwrap(), + command_failed(), + "OCI runtime delete failed", + ) + .await; + let msg = err.to_string(); + assert!(msg.contains("OCI runtime delete failed"), "got {:?}", msg); assert!( - actual_err.to_string().contains(expectd_msg), - "actual error \"{}\" should contains \"{}\"", - actual_err, - expectd_msg + msg.contains("unable to open OCI runtime log file"), + "got {:?}", + msg ); } } diff --git a/crates/runc-shim/src/service.rs b/crates/runc-shim/src/service.rs index f78f33b0..2706f011 100644 --- a/crates/runc-shim/src/service.rs +++ b/crates/runc-shim/src/service.rs @@ -113,7 +113,7 @@ impl Shim for Service { namespace, &bundle, &opts, - Some(Arc::new(ShimExecutor::default())), + Arc::new(ShimExecutor::default()), )?; let pid = read_pid_from_file(&bundle.join(INIT_PID_FILE)) .await @@ -149,7 +149,7 @@ impl Shim for Service { } } -async fn process_exits( +pub(crate) async fn process_exits( s: Subscription, task: &TaskService, tx: Sender<(String, Box)>, @@ -167,7 +167,8 @@ async fn process_exits( let mut change_process: Vec<&mut (dyn Process + Send + Sync)> = Vec::new(); // pid belongs to container init process if cont.init.pid == pid { - // kill all children process if the container has a private PID namespace + // Reap the container's children ourselves unless it has a + // private PID namespace, in which case the kernel does it. if should_kill_all_on_exit(&bundle).await { cont.kill(None, 9, true).await.unwrap_or_else(|e| { error!("failed to kill init's children: {}", e) diff --git a/crates/runc-shim/src/task.rs b/crates/runc-shim/src/task.rs index d5f74e83..67c2b3df 100644 --- a/crates/runc-shim/src/task.rs +++ b/crates/runc-shim/src/task.rs @@ -481,3 +481,1154 @@ where Ok(Empty::default()) } } + +/// Behaviour of the ttrpc `Task` service, driven over a fake OCI runtime. +/// +/// Everything here goes through the `Task` trait, which is the API containerd +/// itself calls. That keeps the tests about observable behaviour rather than the +/// types underneath, so they stay valid as those types change. +#[cfg(test)] +mod tests { + use std::{ + collections::HashMap, + os::unix::process::ExitStatusExt, + process::ExitStatus, + sync::{ + atomic::{AtomicI32, Ordering}, + Arc, Mutex, + }, + time::Duration, + }; + + use async_trait::async_trait; + use containerd_shim::{ + api::{ + CloseIORequest, ConnectRequest, CreateTaskRequest, DeleteRequest, ExecProcessRequest, + KillRequest, Options, PidsRequest, ResizePtyRequest, ShutdownRequest, StartRequest, + StateRequest, StatsRequest, Status, WaitRequest, + }, + asynchronous::{monitor::Subscription, ExitSignal}, + monitor::{ExitEvent, Subject}, + protos::{ + events::task::{TaskDelete, TaskExit}, + protobuf::{well_known_types::any::Any, Message, MessageDyn, MessageField}, + shim::oci::ProcessDetails, + shim_async::Task, + ttrpc, + ttrpc::{r#async::TtrpcContext, Code}, + }, + }; + use oci_spec::runtime::{Process, Spec}; + use runc::{Command, Spawner}; + use tempfile::TempDir; + use tokio::sync::mpsc::{ + channel, error::TryRecvError, unbounded_channel, Receiver, UnboundedSender, + }; + + use crate::{ + runc::{RuncContainer, RuncFactory}, + service::process_exits, + task::TaskService, + }; + + // =========================================================================== + // Harness + // =========================================================================== + + /// The concrete `TaskService` under test, aliased so that collapsing + /// `TaskService` touches this line rather than the fixture. + type Shim = TaskService; + + /// Hands out pids that no real process can own. + /// + /// Starting above `pid_max` means that if a code path ever issues a real + /// `kill(2)` against one of these it fails with `ESRCH` rather than signalling + /// something real. Pids stay unique per test so failures name one container. + fn next_fake_pid() -> i32 { + static NEXT: AtomicI32 = AtomicI32::new(0x4000_0000); + NEXT.fetch_add(1, Ordering::Relaxed) + } + + #[derive(Debug, Default)] + struct FakeState { + /// Full argv of every invocation, in order. + calls: Vec>, + /// JSON payload for `ps --format=json`. + ps_pids: Vec, + /// stderr to fail with, keyed by subcommand. + failures: HashMap, + } + + /// A [`Spawner`] that records invocations and simulates the side effects the + /// shim depends on, so the Task API can be driven without a real OCI runtime. + /// + /// Every `Runc` method funnels through a single `Spawner::execute` call, so this + /// intercepts all of them. + #[derive(Debug, Default)] + struct FakeRunc { + state: Mutex, + } + + impl FakeRunc { + /// Makes `subcommand` exit non-zero, writing `stderr` on the runtime stderr. + fn fail(&self, subcommand: &str, stderr: &str) { + self.state + .lock() + .unwrap() + .failures + .insert(subcommand.to_string(), stderr.to_string()); + } + + /// Sets what `runc ps` reports. + fn set_ps_pids(&self, pids: Vec) { + self.state.lock().unwrap().ps_pids = pids; + } + + /// Argv of every invocation of `subcommand`, from the subcommand onwards. + fn calls_for(&self, subcommand: &str) -> Vec> { + self.state + .lock() + .unwrap() + .calls + .iter() + .map(|argv| tail_from_subcommand(argv)) + .filter(|tail| tail[0] == subcommand) + .map(<[String]>::to_vec) + .collect() + } + } + + /// Splits an argv at the runc subcommand, returning the subcommand and + /// everything after it. + /// + /// `Runc::launch_io` builds `[global_args, command_args].concat()`, and the + /// global set is closed: `--root`, `--log` and `--log-format` each take a + /// separate value, `--debug` and `--systemd-cgroup` take none, and + /// `--rootless=` is a single token. Decoding that prefix is exact, where + /// scanning for a known verb would silently mis-parse an argv whose shape + /// changed. + fn tail_from_subcommand(argv: &[String]) -> &[String] { + let mut i = 0; + while let Some(arg) = argv.get(i) { + match arg.as_str() { + "--root" | "--log" | "--log-format" => i += 2, + a if a.starts_with('-') => i += 1, + _ => return &argv[i..], + } + } + panic!("fake runc: no subcommand in argv {:?}", argv); + } + + /// Value following `flag` in an argv, if present. + fn flag_value(argv: &[String], flag: &str) -> Option { + let i = argv.iter().position(|a| a == flag)?; + argv.get(i + 1).cloned() + } + + #[async_trait] + impl Spawner for FakeRunc { + async fn execute(&self, cmd: Command) -> runc::Result<(ExitStatus, u32, String, String)> { + let argv: Vec = cmd + .as_std() + .get_args() + .map(|a| a.to_string_lossy().into_owned()) + .collect(); + let subcommand = tail_from_subcommand(&argv)[0].clone(); + let pid_file = flag_value(&argv, "--pid-file"); + + let failure = { + let mut state = self.state.lock().unwrap(); + let failure = state.failures.get(&subcommand).cloned(); + state.calls.push(argv); + failure + }; + let pid = next_fake_pid(); + + if let Some(stderr) = failure { + // Exit code 1. `launch_io` turns any non-success status into + // `Error::CommandFailed { status, stdout, stderr }`, which is what + // `runtime_error` and `check_kill_error` consume. + return Ok(( + ExitStatus::from_raw(1 << 8), + pid as u32, + String::new(), + stderr, + )); + } + + // Real runc writes the container pid here; the shim reads it back + // immediately after create and exec. + if let Some(path) = pid_file { + std::fs::write(&path, pid.to_string()) + .unwrap_or_else(|e| panic!("fake runc: write pid file {}: {}", path, e)); + } + + let stdout = if subcommand == "ps" { + serde_json::to_string(&self.state.lock().unwrap().ps_pids).unwrap() + } else { + String::new() + }; + + Ok((ExitStatus::from_raw(0), pid as u32, stdout, String::new())) + } + } + + /// A `TaskService` backed by a fake runtime and a throwaway bundle directory. + struct TestShim { + task: Arc, + runc: Arc, + exit: Arc, + /// Feeds process-exit events to this fixture's `process_exits` pump. + /// + /// The shim's own exit monitor is a process-global singleton. Handing the + /// pump a private channel instead keeps fixtures fully isolated from each + /// other, and lets the pump task end on its own when the fixture drops and + /// this sender goes with it. + exits: UnboundedSender, + events: Receiver<(String, Box)>, + bundle: TempDir, + } + + impl TestShim { + /// Builds the service and starts the exit pump, mirroring what + /// `Service::create_task_service` wires up in production. + async fn new() -> Self { + let bundle = tempfile::tempdir().expect("create bundle dir"); + + // `should_kill_all_on_exit` reads this when the init process exits; + // without it that read fails and the code takes a log-and-continue path. + std::fs::write( + bundle.path().join("config.json"), + serde_json::to_string(&Spec::default()).unwrap(), + ) + .expect("write config.json"); + + let runc = Arc::new(FakeRunc::default()); + let (tx, events) = channel(128); + let exit = Arc::new(ExitSignal::default()); + + let mut task = Shim::new("runc-shim-test", exit.clone(), tx.clone()); + task.factory.spawner = runc.clone(); + let task = Arc::new(task); + + // A private stand-in for the global pid monitor. The id is never + // registered anywhere; `process_exits` only passes it to an unsubscribe + // call on its way out, and ids issued by the real monitor start at 0. + let (exits, rx) = unbounded_channel(); + process_exits(Subscription { id: -1, rx }, &task, tx).await; + + Self { + task, + runc, + exit, + exits, + events, + bundle, + } + } + + fn bundle(&self) -> String { + self.bundle.path().to_string_lossy().into_owned() + } + + /// A create request for `id` against this shim's bundle. + /// + /// stdio is left empty on purpose: that makes `Stdio::is_null()` true, so + /// the shim uses `NullIo` and skips all fifo and console plumbing. + fn create_request(&self, id: &str) -> CreateTaskRequest { + let mut opts = Options::new(); + // `GlobalOpts::build` resolves the runtime through `binary_path`, + // which needs a real file on disk even though FakeRunc never + // executes it — and which returns None when PATH is unset, absolute + // path or not. So these tests require PATH to be present. + opts.binary_name = std::env::current_exe() + .expect("current_exe") + .to_string_lossy() + .into_owned(); + // Keep the runc state root inside the bundle so nothing reaches for + // /run/containerd. + opts.root = self + .bundle + .path() + .join("runc-root") + .to_string_lossy() + .into_owned(); + + let mut any = Any::new(); + any.type_url = "containerd.runc.v1.Options".to_string(); + any.value = opts.write_to_bytes().expect("encode options"); + + CreateTaskRequest { + id: id.to_string(), + bundle: self.bundle(), + options: MessageField::some(any), + ..Default::default() + } + } + + /// Reports that `pid` exited, and waits for the shim to act on it. + /// + /// The exit pump runs as a spawned task, so acting and asserting in the + /// same breath can outrun it. The published `TaskExit` is the sync point. + async fn exit_process(&mut self, pid: i32, code: i32) -> TaskExit { + self.exits + .send(ExitEvent { + subject: Subject::Pid(pid), + exit_code: code, + }) + .expect("exit pump should still be running"); + self.await_event::("/tasks/exit").await + } + + /// Creates and starts a container, returning its init pid. + async fn start_container(&self, id: &str) -> i32 { + self.task + .create(&ctx(), self.create_request(id)) + .await + .expect("create container"); + let resp = self + .task + .start(&ctx(), start_request(id, "")) + .await + .expect("start container"); + resp.pid as i32 + } + + /// Creates and starts an exec process, returning its pid. + async fn start_exec(&self, id: &str, exec_id: &str) -> i32 { + self.task + .exec(&ctx(), exec_request(id, exec_id)) + .await + .expect("exec"); + let resp = self + .task + .start(&ctx(), start_request(id, exec_id)) + .await + .expect("start exec"); + resp.pid as i32 + } + + /// Topics of every event published so far, in order. + fn event_topics(&mut self) -> Vec { + let mut out = Vec::new(); + loop { + match self.events.try_recv() { + Ok((topic, _)) => out.push(topic), + Err(TryRecvError::Empty) | Err(TryRecvError::Disconnected) => return out, + } + } + } + + /// Waits for an event on `topic`, decoded as `T`. + /// + /// Exit events are published from a spawned task, so a test that acts and + /// then asserts immediately can outrun the publisher. + async fn await_event(&mut self, topic: &str) -> T { + loop { + let (got, msg) = tokio::time::timeout(Duration::from_secs(5), self.events.recv()) + .await + .unwrap_or_else(|_| panic!("timed out waiting for {}", topic)) + .expect("event channel closed"); + if got == topic { + return T::parse_from_bytes(&msg.write_to_bytes_dyn().expect("encode event")) + .expect("decode event"); + } + } + } + } + + /// A request context. The shim ignores it, but the generated trait requires one. + fn ctx() -> TtrpcContext { + TtrpcContext { + mh: Default::default(), + metadata: Default::default(), + timeout_nano: 0, + } + } + + fn start_request(id: &str, exec_id: &str) -> StartRequest { + StartRequest { + id: id.to_string(), + exec_id: exec_id.to_string(), + ..Default::default() + } + } + + fn state_request(id: &str, exec_id: &str) -> StateRequest { + StateRequest { + id: id.to_string(), + exec_id: exec_id.to_string(), + ..Default::default() + } + } + + fn delete_request(id: &str, exec_id: &str) -> DeleteRequest { + DeleteRequest { + id: id.to_string(), + exec_id: exec_id.to_string(), + ..Default::default() + } + } + + fn wait_request(id: &str, exec_id: &str) -> WaitRequest { + WaitRequest { + id: id.to_string(), + exec_id: exec_id.to_string(), + ..Default::default() + } + } + + fn kill_request(id: &str, signal: u32, all: bool) -> KillRequest { + KillRequest { + id: id.to_string(), + signal, + all, + ..Default::default() + } + } + + fn pids_request(id: &str) -> PidsRequest { + PidsRequest { + id: id.to_string(), + ..Default::default() + } + } + + fn stats_request(id: &str) -> StatsRequest { + StatsRequest { + id: id.to_string(), + ..Default::default() + } + } + + fn connect_request(id: &str) -> ConnectRequest { + ConnectRequest { + id: id.to_string(), + ..Default::default() + } + } + + /// An exec request carrying a default process spec. + fn exec_request(id: &str, exec_id: &str) -> ExecProcessRequest { + let mut any = Any::new(); + any.type_url = "types.containerd.io/opencontainers/runtime-spec/1/Process".to_string(); + any.value = serde_json::to_vec(&Process::default()).expect("encode process spec"); + + ExecProcessRequest { + id: id.to_string(), + exec_id: exec_id.to_string(), + spec: MessageField::some(any), + ..Default::default() + } + } + + /// The ttrpc status code carried by an error, if it has one. + fn code_of(err: &ttrpc::Error) -> Option { + match err { + ttrpc::Error::RpcStatus(s) => Some(s.code()), + _ => None, + } + } + + // =========================================================================== + // Dispatch between the init process and execs + // =========================================================================== + + #[tokio::test] + async fn state_dispatch() { + let shim = TestShim::new().await; + let init_pid = shim.start_container("dispatch").await; + let exec_pid = shim.start_exec("dispatch", "e1").await; + + // Fixture precondition, not a claim about the shim: the pid assertions + // below can only tell the two processes apart if the fake runtime handed + // out different pids for them. + assert_ne!(init_pid, exec_pid, "fake runtime reused a pid"); + + // Empty exec_id addresses the init process. + let init = shim + .task + .state(&ctx(), state_request("dispatch", "")) + .await + .expect("state of init"); + assert_eq!(init.id, "dispatch"); + assert_eq!(init.pid, init_pid as u32); + assert_eq!(init.status(), Status::RUNNING); + assert_eq!(init.bundle, shim.bundle()); + + // A known exec_id addresses that exec. + let exec = shim + .task + .state(&ctx(), state_request("dispatch", "e1")) + .await + .expect("state of exec"); + assert_eq!(exec.id, "e1"); + assert_eq!(exec.pid, exec_pid as u32); + assert_eq!(exec.status(), Status::RUNNING); + + // An unknown exec_id is not found. + let err = shim + .task + .state(&ctx(), state_request("dispatch", "nope")) + .await + .expect_err("unknown exec should not resolve"); + assert_eq!(code_of(&err), Some(Code::NOT_FOUND)); + } + + #[tokio::test] + async fn unknown_container() { + let shim = TestShim::new().await; + + // Every method that takes a container id rejects one it does not know. + let errs = [ + ( + "state", + shim.task + .state(&ctx(), state_request("ghost", "")) + .await + .err(), + ), + ( + "kill", + shim.task + .kill(&ctx(), kill_request("ghost", 9, false)) + .await + .err(), + ), + ( + "delete", + shim.task + .delete(&ctx(), delete_request("ghost", "")) + .await + .err(), + ), + ( + "pids", + shim.task.pids(&ctx(), pids_request("ghost")).await.err(), + ), + ( + "exec", + shim.task + .exec(&ctx(), exec_request("ghost", "e1")) + .await + .err(), + ), + ( + "stats", + shim.task.stats(&ctx(), stats_request("ghost")).await.err(), + ), + ]; + + for (method, err) in errs { + let err = + err.unwrap_or_else(|| panic!("{} on an unknown container should fail", method)); + assert_eq!(code_of(&err), Some(Code::NOT_FOUND), "{}: {}", method, err); + } + } + + // =========================================================================== + // Exit: waiters, retained metadata, published events + // =========================================================================== + + /// Concurrent waiters are all served the task's exit status. + /// + /// The pending check below rules out a waiter that returned early, but it + /// cannot prove one registered: a spawned task that has not been polled yet + /// is pending too, and would take the same short-circuit `wait_after_exit` + /// covers. Proving registration means reading `wait_chan_tx`, which is the + /// implementation detail these tests deliberately stay off. What is pinned + /// here is that no waiter is dropped or served a different result. + #[tokio::test] + async fn concurrent_waiters() { + let mut shim = TestShim::new().await; + let pid = shim.start_container("waiters").await; + + let mut waiters: Vec<_> = (0..3) + .map(|_| { + let task = shim.task.clone(); + tokio::spawn(async move { task.wait(&ctx(), wait_request("waiters", "")).await }) + }) + .collect(); + + for waiter in &mut waiters { + assert!( + tokio::time::timeout(Duration::from_millis(50), waiter) + .await + .is_err(), + "a waiter returned before the task exited" + ); + } + + shim.exit_process(pid, 7).await; + + for waiter in waiters { + // Bounded so a regression that drops a waiter fails here instead of + // hanging CI until the job timeout. + let resp = tokio::time::timeout(Duration::from_secs(5), waiter) + .await + .expect("a waiter was never woken after the task exited") + .expect("waiter task panicked") + .expect("wait should succeed"); + assert_eq!(resp.exit_status, 7); + assert!( + resp.exited_at.is_some(), + "an exited process must carry a timestamp" + ); + } + } + + #[tokio::test] + async fn wait_after_exit() { + let mut shim = TestShim::new().await; + let pid = shim.start_container("late-wait").await; + + shim.exit_process(pid, 3).await; + + // Exit metadata is retained, not consumed: a wait issued afterwards still + // reports it, and reports it repeatedly. + for _ in 0..2 { + let resp = shim + .task + .wait(&ctx(), wait_request("late-wait", "")) + .await + .expect("wait after exit"); + assert_eq!(resp.exit_status, 3); + assert!(resp.exited_at.is_some()); + } + } + + #[tokio::test] + async fn state_after_exit() { + let mut shim = TestShim::new().await; + let pid = shim.start_container("stopped").await; + + shim.exit_process(pid, 42).await; + + let resp = shim + .task + .state(&ctx(), state_request("stopped", "")) + .await + .expect("state after exit"); + assert_eq!(resp.status(), Status::STOPPED); + assert_eq!(resp.exit_status, 42); + assert!(resp.exited_at.is_some()); + } + + #[tokio::test] + async fn exit_event() { + let mut shim = TestShim::new().await; + let pid = shim.start_container("exit-event").await; + + let event = shim.exit_process(pid, 9).await; + assert_eq!(event.container_id, "exit-event"); + assert_eq!(event.id, "exit-event"); + assert_eq!(event.pid, pid as u32); + assert_eq!(event.exit_status, 9); + assert!(event.exited_at.is_some()); + } + + #[tokio::test] + async fn exec_exit_event() { + let mut shim = TestShim::new().await; + shim.start_container("exec-exit").await; + let exec_pid = shim.start_exec("exec-exit", "e1").await; + + let event = shim.exit_process(exec_pid, 5).await; + assert_eq!(event.container_id, "exec-exit"); + assert_eq!( + event.id, "e1", + "the event identifies the exec, not the task" + ); + assert_eq!(event.pid, exec_pid as u32); + assert_eq!(event.exit_status, 5); + } + + #[tokio::test] + async fn start_after_exit() { + let mut shim = TestShim::new().await; + let pid = shim.start_container("dead").await; + shim.exit_process(pid, 0).await; + + let err = shim + .task + .start(&ctx(), start_request("dead", "")) + .await + .expect_err("starting into a dead container should fail"); + assert_eq!(code_of(&err), Some(Code::FAILED_PRECONDITION)); + } + + // =========================================================================== + // Delete + // =========================================================================== + + #[tokio::test] + async fn delete_exec() { + let mut shim = TestShim::new().await; + shim.start_container("del-exec").await; + let exec_pid = shim.start_exec("del-exec", "e1").await; + shim.exit_process(exec_pid, 4).await; + + let resp = shim + .task + .delete(&ctx(), delete_request("del-exec", "e1")) + .await + .expect("delete exec"); + assert_eq!(resp.pid, exec_pid as u32); + assert_eq!(resp.exit_status, 4); + + // The exec is gone... + let err = shim + .task + .state(&ctx(), state_request("del-exec", "e1")) + .await + .expect_err("deleted exec should not resolve"); + assert_eq!(code_of(&err), Some(Code::NOT_FOUND)); + + // ...but the container is not. + shim.task + .state(&ctx(), state_request("del-exec", "")) + .await + .expect("container should survive deleting one of its execs"); + } + + #[tokio::test] + async fn delete_container() { + let mut shim = TestShim::new().await; + let pid = shim.start_container("del-task").await; + shim.exit_process(pid, 0).await; + + let resp = shim + .task + .delete(&ctx(), delete_request("del-task", "")) + .await + .expect("delete container"); + assert_eq!(resp.pid, pid as u32); + + let err = shim + .task + .state(&ctx(), state_request("del-task", "")) + .await + .expect_err("deleted container should not resolve"); + assert_eq!(code_of(&err), Some(Code::NOT_FOUND)); + } + + #[tokio::test] + async fn delete_event() { + let mut shim = TestShim::new().await; + let pid = shim.start_container("del-event").await; + shim.exit_process(pid, 6).await; + + shim.task + .delete(&ctx(), delete_request("del-event", "")) + .await + .expect("delete"); + + let event = shim.await_event::("/tasks/delete").await; + assert_eq!(event.container_id, "del-event"); + assert_eq!(event.pid, pid as u32); + assert_eq!(event.exit_status, 6); + } + + // =========================================================================== + // Published events for the happy path + // =========================================================================== + + #[tokio::test] + async fn lifecycle_events() { + let mut shim = TestShim::new().await; + shim.start_container("lifecycle").await; + shim.start_exec("lifecycle", "e1").await; + + assert_eq!( + shim.event_topics(), + [ + "/tasks/create", + "/tasks/start", + "/tasks/exec-added", + "/tasks/exec-started", + ] + ); + } + + // =========================================================================== + // Errors coming back from the runtime + // =========================================================================== + + #[tokio::test] + async fn create_failure() { + let shim = TestShim::new().await; + + // The shim reads the last error line out of the runtime log to explain a + // failed create. + std::fs::write( + std::path::Path::new(&shim.bundle()).join(crate::common::LOG_JSON_FILE), + "{\"level\":\"info\",\"msg\":\"hello\",\"time\":\"2024-01-01\"}\n\ + {\"level\":\"error\",\"msg\":\"rootfs is not a directory\",\"time\":\"2024-01-01\"}\n", + ) + .expect("seed runtime log"); + shim.runc.fail("create", "exit status 1"); + + let err = shim + .task + .create(&ctx(), shim.create_request("bad-create")) + .await + .expect_err("create should fail"); + assert!( + err.to_string().contains("rootfs is not a directory"), + "expected the runtime log error in {:?}", + err.to_string() + ); + } + + #[tokio::test] + async fn kill_error_mapping() { + let shim = TestShim::new().await; + shim.start_container("kill-err").await; + + // A failed kill leaves the container untouched, so one fixture covers + // every phrasing runc might report. + for stderr in [ + "process already finished", + "container not running", + "no such process", + "container does not exist", + ] { + shim.runc.fail("kill", stderr); + + let err = shim + .task + .kill(&ctx(), kill_request("kill-err", 9, false)) + .await + .err() + .unwrap_or_else(|| panic!("kill should fail for {:?}", stderr)); + assert_eq!( + code_of(&err), + Some(Code::NOT_FOUND), + "runtime said {:?}, shim reported {}", + stderr, + err + ); + } + } + + #[tokio::test] + async fn kill_args() { + let shim = TestShim::new().await; + shim.start_container("kill-args").await; + + shim.task + .kill(&ctx(), kill_request("kill-args", 15, true)) + .await + .expect("kill"); + + let calls = shim.runc.calls_for("kill"); + assert_eq!(calls.len(), 1, "expected exactly one runc kill"); + assert_eq!(calls[0], ["kill", "--all", "kill-args", "15"]); + } + + // =========================================================================== + // Pids, connect, shutdown, stdio + // =========================================================================== + + #[tokio::test] + async fn pids_exec_details() { + let shim = TestShim::new().await; + let init_pid = shim.start_container("pids").await; + let exec_pid = shim.start_exec("pids", "e1").await; + shim.runc + .set_ps_pids(vec![init_pid as usize, exec_pid as usize]); + + let resp = shim + .task + .pids(&ctx(), pids_request("pids")) + .await + .expect("pids"); + assert_eq!(resp.processes.len(), 2); + + let init = resp + .processes + .iter() + .find(|p| p.pid == init_pid as u32) + .expect("init pid reported"); + assert!( + init.info.is_none(), + "the init process carries no exec details" + ); + + let exec = resp + .processes + .iter() + .find(|p| p.pid == exec_pid as u32) + .expect("exec pid reported"); + let info = exec.info.as_ref().expect("exec details attached"); + let details = + ProcessDetails::parse_from_bytes(&info.value).expect("decode process details"); + assert_eq!(details.exec_id, "e1"); + } + + #[tokio::test] + async fn connect() { + let shim = TestShim::new().await; + let pid = shim.start_container("connect").await; + + let resp = shim + .task + .connect(&ctx(), connect_request("connect")) + .await + .expect("connect"); + assert_eq!(resp.shim_pid, std::process::id()); + assert_eq!(resp.task_pid, pid as u32); + + // An unknown container reports no task pid rather than failing. + let resp = shim + .task + .connect(&ctx(), connect_request("ghost")) + .await + .expect("connect to unknown container"); + assert_eq!(resp.task_pid, 0); + } + + #[tokio::test] + async fn shutdown() { + let mut shim = TestShim::new().await; + let pid = shim.start_container("shutdown").await; + + shim.task + .shutdown(&ctx(), ShutdownRequest::default()) + .await + .expect("shutdown with a live container"); + assert!( + tokio::time::timeout(Duration::from_millis(50), shim.exit.wait()) + .await + .is_err(), + "shutdown must not signal exit while a container is still held" + ); + + shim.exit_process(pid, 0).await; + shim.task + .delete(&ctx(), delete_request("shutdown", "")) + .await + .expect("delete"); + + shim.task + .shutdown(&ctx(), ShutdownRequest::default()) + .await + .expect("shutdown when empty"); + tokio::time::timeout(Duration::from_secs(5), shim.exit.wait()) + .await + .expect("shutdown should signal exit once empty"); + } + + #[tokio::test] + async fn resize_and_close_io() { + let shim = TestShim::new().await; + shim.start_container("stdio").await; + shim.start_exec("stdio", "e1").await; + + for exec_id in ["", "e1"] { + shim.task + .resize_pty( + &ctx(), + ResizePtyRequest { + id: "stdio".to_string(), + exec_id: exec_id.to_string(), + width: 80, + height: 24, + ..Default::default() + }, + ) + .await + .unwrap_or_else(|e| panic!("resize_pty({:?}): {}", exec_id, e)); + + // `stdin` is left unset on purpose: `TaskService::close_io` never + // reads the flag and always closes stdin, so setting it here would + // imply coverage that does not exist. + shim.task + .close_io( + &ctx(), + CloseIORequest { + id: "stdio".to_string(), + exec_id: exec_id.to_string(), + ..Default::default() + }, + ) + .await + .unwrap_or_else(|e| panic!("close_io({:?}): {}", exec_id, e)); + } + + let err = shim + .task + .resize_pty( + &ctx(), + ResizePtyRequest { + id: "stdio".to_string(), + exec_id: "nope".to_string(), + width: 80, + height: 24, + ..Default::default() + }, + ) + .await + .expect_err("resizing an unknown exec should fail"); + assert_eq!(code_of(&err), Some(Code::NOT_FOUND)); + } + + /// The fake runtime hands out pids that own no cgroup, so metrics collection + /// has nothing to read. It must say so rather than panic or invent numbers. + #[tokio::test] + async fn stats_no_cgroup() { + let shim = TestShim::new().await; + shim.start_container("stats").await; + + let err = shim + .task + .stats(&ctx(), stats_request("stats")) + .await + .expect_err("a container with no cgroup has no metrics to report"); + + #[cfg(target_os = "linux")] + assert!( + err.to_string().contains("cgroup"), + "expected a cgroup failure, got {}", + err + ); + #[cfg(not(target_os = "linux"))] + assert!( + err.to_string().contains("Unimplemented method"), + "stats is not supported off Linux, got {}", + err + ); + } + + // =========================================================================== + // Pause and resume are Linux-only + // =========================================================================== + + #[cfg(target_os = "linux")] + #[tokio::test] + async fn pause_resume() { + use containerd_shim::api::{PauseRequest, ResumeRequest}; + + let shim = TestShim::new().await; + shim.start_container("paused").await; + shim.start_exec("paused", "e1").await; + + shim.task + .pause( + &ctx(), + PauseRequest { + id: "paused".to_string(), + ..Default::default() + }, + ) + .await + .expect("pause"); + + // A paused task projects its status onto every exec inside it, even though + // the exec process itself was never touched. + for exec_id in ["", "e1"] { + let resp = shim + .task + .state(&ctx(), state_request("paused", exec_id)) + .await + .unwrap_or_else(|e| panic!("state({:?}): {}", exec_id, e)); + assert_eq!( + resp.status(), + Status::PAUSED, + "exec_id {:?} should report PAUSED", + exec_id + ); + } + + shim.task + .resume( + &ctx(), + ResumeRequest { + id: "paused".to_string(), + ..Default::default() + }, + ) + .await + .expect("resume"); + + for exec_id in ["", "e1"] { + let resp = shim + .task + .state(&ctx(), state_request("paused", exec_id)) + .await + .unwrap_or_else(|e| panic!("state({:?}): {}", exec_id, e)); + assert_eq!(resp.status(), Status::RUNNING); + } + } + + #[cfg(not(target_os = "linux"))] + #[tokio::test] + async fn pause_unsupported() { + use containerd_shim::api::PauseRequest; + + let shim = TestShim::new().await; + shim.start_container("no-pause").await; + + let err = shim + .task + .pause( + &ctx(), + PauseRequest { + id: "no-pause".to_string(), + ..Default::default() + }, + ) + .await + .expect_err("pause is not supported off Linux"); + + // Not `Code::UNIMPLEMENTED`: `Error::Unimplemented` has no ttrpc mapping + // and reaches containerd as an opaque error. `unimplemented_no_code` + // below pins that gap on every platform. + assert_eq!(code_of(&err), None, "got {}", err); + assert!( + err.to_string().contains("Unimplemented method"), + "got {}", + err + ); + } + + // ======================================================================= + // How shim errors reach containerd + // ======================================================================= + + fn ttrpc_code(err: containerd_shim::Error) -> Option { + code_of(&err.into()) + } + + #[test] + fn error_codes() { + use containerd_shim::Error; + + assert_eq!( + ttrpc_code(Error::InvalidArgument("x".to_string())), + Some(Code::INVALID_ARGUMENT) + ); + assert_eq!( + ttrpc_code(Error::NotFoundError("x".to_string())), + Some(Code::NOT_FOUND) + ); + assert_eq!( + ttrpc_code(Error::FailedPreconditionError("x".to_string())), + Some(Code::FAILED_PRECONDITION) + ); + } + + /// `Error::Unimplemented` is not mapped to a ttrpc status, so every method + /// the shim does not support on the current platform (`pause`, `resume`, + /// `update` and `stats` off Linux; `update`/`stats`/`ps`/`pause`/`resume` on + /// an exec) reaches containerd as an opaque error rather than + /// `Code::UNIMPLEMENTED`. + /// + /// Pinned so that closing the gap is a deliberate change, not an accident. + #[test] + fn unimplemented_no_code() { + let err = containerd_shim::Error::Unimplemented("pause".to_string()); + assert!(err.to_string().contains("Unimplemented method")); + assert_eq!(ttrpc_code(err), None); + } +}