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); + } +}