Add test for signals

Signed-off-by: Jorge Prendes <[email protected]>

crates/containerd-shim-wasm/Cargo.toml

@@ -78,6 +78,7 @@ windows-sys = { workspace = true, features = [
 ] }
 
 [dev-dependencies]
+tokio = { workspace = true, features = ["signal"] }
 containerd-shim-wasm-test-modules = { workspace = true }
 env_logger = { workspace = true }
 tempfile = { workspace = true }

crates/containerd-shim-wasm/src/lib.rs

@@ -11,6 +11,12 @@ pub mod sandbox;
 pub(crate) mod sys;
 
 #[cfg(any(test, feature = "testing"))]
+/// Utilities for writing shims tests.
+/// You can use this to test your runwasi based shim.
 pub mod testing;
 
+#[cfg(test)]
+/// Tests for runwasi's containerd-shim-wasm.
+mod test;
+
 pub use containerd_shim::Config;

crates/containerd-shim-wasm/src/test/mod.rs

@@ -0,0 +1 @@
+mod signals;

crates/containerd-shim-wasm/src/test/signals.rs

@@ -0,0 +1,157 @@
+//! This module includes a test for sending signals to containers when
+//! the shim is managing two or more containers.
+//! 
+//! See https://github.com/containerd/runwasi/issues/755 for context.
+//!
+//! This test is currently broken for the reasons explained in #755.
+//! Running the test will result in a failure:
+//! ```
+//! cargo test -p containerd-shim-wasm -- test::signals::test_handling_signals --exact --show-output --nocapture --ignored
+//! ```
+//!
+//! This is because the current implementation breaks `tokio::signal`.
+//! You can verify this by using `libc::signal` instead, and the test will succeed
+//! ```
+//! USE_LIBC=1 cargo test -p containerd-shim-wasm -- test::signals::test_handling_signals --exact --show-output --nocapture --ignored
+//! ```
+//!
+//! Once #755 is fixed we can remove the libc based implementation and
+//! remove the ignore attribute from the test.
+
+use std::future::pending;
+use std::sync::mpsc::channel;
+use std::sync::{Arc, LazyLock};
+use std::thread::sleep;
+use std::time::Duration;
+
+use anyhow::{bail, Result};
+use containerd_shim_wasm_test_modules::HELLO_WORLD;
+use tokio::sync::Notify;
+
+use crate::container::{Engine, Instance, RuntimeContext};
+use crate::sandbox::Stdio;
+use crate::testing::WasiTest;
+
+#[derive(Clone, Default)]
+pub struct SomeEngine;
+
+async fn ctrl_c(use_libc: bool) {
+    static CANCELLATION: LazyLock<Notify> = LazyLock::new(|| Notify::new());
+
+    fn on_ctr_c(_: libc::c_int) {
+        CANCELLATION.notify_waiters();
+    }
+
+    if use_libc {
+        unsafe { libc::signal(libc::SIGINT, on_ctr_c as _) };
+        CANCELLATION.notified().await;
+    } else {
+        let _ = tokio::signal::ctrl_c().await;
+    }
+}
+
+impl Engine for SomeEngine {
+    fn name() -> &'static str {
+        "some-engine"
+    }
+
+    fn run_wasi(&self, ctx: &impl RuntimeContext, stdio: Stdio) -> Result<i32> {
+        stdio.redirect()?;
+        let name = ctx.entrypoint().func;
+        tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()?
+            .block_on(async move {
+                use tokio::time::sleep;
+                let use_libc = std::env::var("USE_LIBC").unwrap_or_default();
+                let use_libc = !use_libc.is_empty() && use_libc != "0";
+                let signal = async {
+                    println!("{name}> waiting for signal!");
+                    ctrl_c(use_libc).await;
+                    println!("{name}> received signal, bye!");
+                };
+                let task = async {
+                    sleep(Duration::from_millis(10)).await;
+                    eprintln!("{name}> ready");
+                    pending().await
+                };
+                tokio::select! {
+                    _ = signal => {}
+                    _ = task => {}
+                };
+                Ok(0)
+            })
+    }
+}
+
+type SomeInstance = Instance<SomeEngine>;
+
+struct KillGuard(Arc<WasiTest<SomeInstance>>);
+impl Drop for KillGuard {
+    fn drop(&mut self) {
+        let _ = self.0.kill();
+    }
+}
+
+#[test]
+#[ignore = "this currently fails due to tokio's global state"]
+fn test_handling_signals() -> Result<()> {
+    // use a thread scope to ensure we join all threads at the end
+    std::thread::scope(|s| -> Result<()> {
+        let mut containers = vec![];
+
+        for i in 0..20 {
+            let container = WasiTest::<SomeInstance>::builder()?
+                .with_name(format!("test-{i}"))
+                .with_start_fn(format!("test-{i}"))
+                .with_stdout("/proc/self/fd/1")?
+                .with_wasm(HELLO_WORLD)?
+                .build()?;
+            containers.push(Arc::new(container));
+        }
+
+        let guard: Vec<_> = containers.iter().cloned().map(KillGuard).collect();
+
+        for container in containers.iter() {
+            container.start()?;
+        }
+
+        let (tx, rx) = channel();
+
+        for (i, container) in containers.iter().cloned().enumerate() {
+            let tx = tx.clone();
+            s.spawn(move || -> anyhow::Result<()> {
+                println!("shim> waiting for container {i}");
+                let (code, ..) = container.wait(Duration::from_secs(10000))?;
+                println!("shim> container test-{i} exited with code {code}");
+                tx.send(i)?;
+                Ok(())
+            });
+        }
+
+        'outer: for (i, container) in containers.iter().enumerate() {
+            for _ in 0..100 {
+                let stderr = container.read_stderr()?.unwrap_or_default();
+                if stderr.contains("ready") {
+                    continue 'outer;
+                }
+                sleep(Duration::from_millis(1));
+            }
+            bail!("timeout waiting for container test-{i}");
+        }
+
+        println!("shim> all containers ready");
+
+        for (i, container) in containers.iter().enumerate() {
+            println!("shim> sending ctrl-c to container test-{i}");
+            let _ = container.ctrl_c()?;
+            let id = rx.recv_timeout(Duration::from_secs(5))?;
+            println!("shim> received exit from container test-{id} (expected test-{i})");
+            assert_eq!(id, i);
+        }
+
+        drop(guard);
+
+        Ok(())
+    })
+}

crates/containerd-shim-wasm/src/testing.rs

@@ -4,6 +4,11 @@ use std::collections::HashMap;
 use std::fs::{self, create_dir, read, read_to_string, write, File};
 use std::marker::PhantomData;
 use std::ops::Add;
+#[cfg(unix)]
+use std::os::unix::fs::symlink;
+#[cfg(windows)]
+use std::os::windows::fs::symlink_file as symlink;
+use std::path::Path;
 use std::time::Duration;
 
 use anyhow::{bail, Result};
@@ -76,6 +81,11 @@ where
         Ok(builder)
     }
 
+    pub fn with_name(mut self, name: impl Into<String>) -> Self {
+        self.container_name = name.into();
+        self
+    }
+
     pub fn with_host_network(mut self) -> Self {
         // Removing the `network` namespace results in the binding to the host's socket.
         // This allows for direct communication with the host's networking interface.
@@ -113,6 +123,32 @@ where
         Ok(self)
     }
 
+    pub fn with_stdout(self, stdout: impl AsRef<Path>) -> Result<Self> {
+        let stdout = fs::canonicalize(stdout.as_ref())?;
+
+        let dir = self.tempdir.path();
+
+        log::info!("setting wasi test stdout to {:?}", stdout);
+
+        std::fs::remove_file(dir.join("stdout"))?;
+        symlink(stdout, dir.join("stdout"))?;
+
+        Ok(self)
+    }
+
+    pub fn with_stderr(self, stderr: impl AsRef<Path>) -> Result<Self> {
+        let stderr = fs::canonicalize(stderr.as_ref())?;
+
+        let dir = self.tempdir.path();
+
+        log::info!("setting wasi test stderr to {:?}", stderr);
+
+        std::fs::remove_file(dir.join("stderr"))?;
+        symlink(stderr, dir.join("stderr"))?;
+
+        Ok(self)
+    }
+
     pub fn as_oci_image(
         mut self,
         image_name: Option<String>,
@@ -229,9 +265,13 @@ where
         Ok(self)
     }
 
-    pub fn wait(&self, timeout: Duration) -> Result<(u32, String, String)> {
-        let dir = self.tempdir.path();
+    pub fn kill(&self) -> Result<&Self> {
+        log::info!("sending SIGKILL");
+        self.instance.kill(SIGKILL as u32)?;
+        Ok(self)
+    }
 
+    pub fn wait(&self, timeout: Duration) -> Result<(u32, String, String)> {
         log::info!("waiting wasi test");
         let (status, _) = match self.instance.wait_timeout(timeout) {
             Some(res) => res,
@@ -241,15 +281,35 @@ where
             }
         };
 
-        let stdout = read_to_string(dir.join("stdout"))?;
-        let stderr = read_to_string(dir.join("stderr"))?;
+        let stdout = self.read_stdout()?.unwrap_or_default();
+        let stderr = self.read_stderr()?.unwrap_or_default();
 
         self.instance.delete()?;
 
         log::info!("wasi test status is {status}");
 
         Ok((status, stdout, stderr))
     }
+
+    pub fn root(&self) -> &Path {
+        self.tempdir.path()
+    }
+
+    pub fn read_stdout(&self) -> Result<Option<String>> {
+        let path = self.tempdir.path().join("stdout");
+        if path.is_symlink() {
+            return Ok(None);
+        }
+        Ok(Some(read_to_string(path)?))
+    }
+
+    pub fn read_stderr(&self) -> Result<Option<String>> {
+        let path = self.tempdir.path().join("stderr");
+        if path.is_symlink() {
+            return Ok(None);
+        }
+        Ok(Some(read_to_string(path)?))
+    }
 }
 
 pub mod oci_helpers {