[stress-test] Add support for native containers

crates/stress-test/Cargo.toml

@@ -9,7 +9,7 @@ homepage.workspace = true
 
 [dependencies]
 anyhow = { workspace = true }
-trapeze = "0.7"
+trapeze = "0.7.6"
 prost = "0.13"
 prost-types = "0.13"
 tokio = { workspace = true, features = ["rt-multi-thread", "macros", "fs", "process", "signal"] }
@@ -27,6 +27,7 @@ tonic = "0.12"
 serde_json = { workspace = true }
 serde = { workspace = true }
 futures = "0.3"
+sha256 = { workspace = true }
 
 
 [package.metadata.cargo-machete]

crates/stress-test/protos/github.com/containerd/containerd/api/runtime/task/v3/shim.proto

@@ -0,0 +1,48 @@
+/*
+	Copyright The containerd Authors.
+
+	Licensed under the Apache License, Version 2.0 (the "License");
+	you may not use this file except in compliance with the License.
+	You may obtain a copy of the License at
+
+		http://www.apache.org/licenses/LICENSE-2.0
+
+	Unless required by applicable law or agreed to in writing, software
+	distributed under the License is distributed on an "AS IS" BASIS,
+	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+	See the License for the specific language governing permissions and
+	limitations under the License.
+*/
+
+syntax = "proto3";
+
+package containerd.task.v3;
+
+import "google/protobuf/empty.proto";
+import "github.com/containerd/containerd/api/runtime/task/v2/shim.proto";
+
+option go_package = "github.com/containerd/containerd/api/runtime/task/v3;task";
+
+// Shim service is launched for each container and is responsible for owning the IO
+// for the container and its additional processes.  The shim is also the parent of
+// each container and allows reattaching to the IO and receiving the exit status
+// for the container processes.
+service Task {
+	rpc State(containerd.task.v2.StateRequest) returns (containerd.task.v2.StateResponse);
+	rpc Create(containerd.task.v2.CreateTaskRequest) returns (containerd.task.v2.CreateTaskResponse);
+	rpc Start(containerd.task.v2.StartRequest) returns (containerd.task.v2.StartResponse);
+	rpc Delete(containerd.task.v2.DeleteRequest) returns (containerd.task.v2.DeleteResponse);
+	rpc Pids(containerd.task.v2.PidsRequest) returns (containerd.task.v2.PidsResponse);
+	rpc Pause(containerd.task.v2.PauseRequest) returns (google.protobuf.Empty);
+	rpc Resume(containerd.task.v2.ResumeRequest) returns (google.protobuf.Empty);
+	rpc Checkpoint(containerd.task.v2.CheckpointTaskRequest) returns (google.protobuf.Empty);
+	rpc Kill(containerd.task.v2.KillRequest) returns (google.protobuf.Empty);
+	rpc Exec(containerd.task.v2.ExecProcessRequest) returns (google.protobuf.Empty);
+	rpc ResizePty(containerd.task.v2.ResizePtyRequest) returns (google.protobuf.Empty);
+	rpc CloseIO(containerd.task.v2.CloseIORequest) returns (google.protobuf.Empty);
+	rpc Update(containerd.task.v2.UpdateTaskRequest) returns (google.protobuf.Empty);
+	rpc Wait(containerd.task.v2.WaitRequest) returns (containerd.task.v2.WaitResponse);
+	rpc Stats(containerd.task.v2.StatsRequest) returns (containerd.task.v2.StatsResponse);
+	rpc Connect(containerd.task.v2.ConnectRequest) returns (containerd.task.v2.ConnectResponse);
+	rpc Shutdown(containerd.task.v2.ShutdownRequest) returns (google.protobuf.Empty);
+}

crates/stress-test/src/containerd/client.rs

@@ -21,7 +21,7 @@ use containerd_client::services::v1::{
 };
 use containerd_client::types::Mount;
 use humantime::format_rfc3339;
-use oci_spec::image::{ImageConfiguration, ImageManifest};
+use oci_spec::image::{Arch, ImageConfiguration, ImageIndex, ImageManifest};
 use oci_spec::runtime::Spec;
 use prost_types::Any;
 use tokio_async_drop::tokio_async_drop as async_drop;
@@ -125,9 +125,29 @@ impl ClientInner {
         let request = self.with_metadata(request);
         let response = client.get(request).await?.into_inner();
         let image = response.image.context("Could not find image")?;
-
         let descriptor = image.target.context("Could not find image descriptor")?;
-        let manifest = self.read_content(&descriptor.digest).await?;
+        let mut manifest = self.read_content(&descriptor.digest).await?;
+
+        // If this is a multiplatform image, the manifest will be an index manifest
+        // rather than an image manifest.
+        if let Ok(index) = ImageIndex::from_reader(Cursor::new(&manifest)) {
+            let descriptor = index
+                .manifests()
+                .iter()
+                .find(|m| {
+                    let Some(platform) = m.platform() else {
+                        return false;
+                    };
+                    match platform.architecture() {
+                        Arch::Amd64 => cfg!(target_arch = "x86_64"),
+                        Arch::ARM64 => cfg!(target_arch = "aarch64"),
+                        _ => false,
+                    }
+                })
+                .context("host platform not supported")?;
+            manifest = self.read_content(descriptor.digest()).await?;
+        }
+
         let manifest = ImageManifest::from_reader(Cursor::new(manifest))?;
         let config = self.read_content(manifest.config().digest()).await?;
         let config = ImageConfiguration::from_reader(Cursor::new(config))?;
@@ -138,12 +158,12 @@ impl ClientInner {
     pub(crate) async fn snapshot(
         &self,
         id: String,
-        diffs: Vec<String>,
+        parent: String,
     ) -> Result<Vec<containerd_client::types::Mount>> {
         let mut client = SnapshotsClient::new(self.channel.clone());
         let request = PrepareSnapshotRequest {
             key: id,
-            parent: diffs.join(" "),
+            parent,
             snapshotter: String::from("overlayfs"),
             ..Default::default()
         };
@@ -191,13 +211,15 @@ impl ClientInner {
         container_id: impl Into<String>,
         mounts: impl Into<Vec<Mount>>,
         stdout: impl Into<String>,
+        stderr: impl Into<String>,
     ) -> Result<()> {
         let mut client = TasksClient::new(self.channel.clone());
 
         let request = CreateTaskRequest {
             container_id: container_id.into(),
             rootfs: mounts.into(),
             stdout: stdout.into(),
+            stderr: stderr.into(),
             ..Default::default()
         };
         let request = self.with_metadata(request);
@@ -221,7 +243,7 @@ impl ClientInner {
         Ok(())
     }
 
-    async fn wait_task(&self, container_id: impl Into<String>) -> Result<()> {
+    async fn wait_task(&self, container_id: impl Into<String>) -> Result<u32> {
         let mut client = TasksClient::new(self.channel.clone());
 
         let request = WaitRequest {
@@ -230,9 +252,9 @@ impl ClientInner {
         };
         let request = self.with_metadata(request);
 
-        client.wait(request).await?;
-
-        Ok(())
+        let response = client.wait(request).await?;
+        let status = response.into_inner().exit_status;
+        Ok(status)
     }
 
     async fn kill_task(&self, container_id: impl Into<String>) -> Result<()> {
@@ -305,7 +327,8 @@ impl Client {
     ) -> Result<Vec<Mount>> {
         let config = self.0.image_config(image.into()).await?;
         let diffs = config.rootfs().diff_ids().clone();
-        let mounts = self.0.snapshot(id.into(), diffs).await?;
+        let chain_id = chain_id(diffs);
+        let mounts = self.0.snapshot(id.into(), chain_id).await?;
         Ok(mounts)
     }
 
@@ -335,15 +358,18 @@ impl Client {
         container_id: impl Into<String>,
         mounts: impl Into<Vec<Mount>>,
         stdout: impl Into<String>,
+        stderr: impl Into<String>,
     ) -> Result<()> {
-        self.0.create_task(container_id, mounts, stdout).await
+        self.0
+            .create_task(container_id, mounts, stdout, stderr)
+            .await
     }
 
     pub async fn start_task(&self, container_id: impl Into<String>) -> Result<()> {
         self.0.start_task(container_id).await
     }
 
-    pub async fn wait_task(&self, container_id: impl Into<String>) -> Result<()> {
+    pub async fn wait_task(&self, container_id: impl Into<String>) -> Result<u32> {
         self.0.wait_task(container_id).await
     }
 
@@ -365,3 +391,28 @@ impl Clone for Client {
         Self(self.0.clone())
     }
 }
+
+fn chain_id(digests: Vec<String>) -> String {
+    let mut chain_id = digests.get(0).cloned().unwrap_or_default();
+    for digest in digests.iter().skip(1) {
+        chain_id = sha256::digest(format!("{chain_id} {digest}"));
+        chain_id = format!("sha256:{chain_id}");
+    }
+    chain_id
+}
+
+#[cfg(test)]
+mod test {
+    #[test]
+    fn chain_id_smoke() {
+        let diffs = vec![
+            String::from("sha256:6f60b56fd4d6a01ebc6ee4133eb429a00c327acc869a0c6083f0e4bc784d8d07"),
+            String::from("sha256:4d851d7c3ef9a3cb8c6553806846038c3c81498e1f6d6dc60bb03291f223b99a"),
+        ];
+        let chain_id = super::chain_id(diffs);
+        assert_eq!(
+            chain_id,
+            "sha256:0f8505411d5fe958101c5e6b6e31c61262a05f7aff548bf7742ff1ad24d6bf88"
+        );
+    }
+}

crates/stress-test/src/containerd/containerd.rs

@@ -9,10 +9,8 @@ pub struct Containerd {
 }
 
 impl Containerd {
-    pub async fn new() -> Result<Self> {
-        Ok(Self {
-            containerd: Client::default().await?,
-        })
+    pub async fn new(client: Client) -> Result<Self> {
+        Ok(Self { containerd: client })
     }
 }
 

crates/stress-test/src/containerd/task.rs

@@ -1,9 +1,7 @@
-use std::path::PathBuf;
-
-use anyhow::Result;
+use anyhow::{Result, ensure};
 use containerd_client::types::Mount;
 use oci_spec::runtime::{ProcessBuilder, RootBuilder, Spec, SpecBuilder, UserBuilder};
-use tokio::fs::canonicalize;
+use tempfile::{TempDir, tempdir};
 use tokio_async_drop::tokio_async_drop;
 
 use super::Client;
@@ -19,6 +17,7 @@ pub struct Task {
     spec: Spec,
     task_deleted: RunOnce,
     container_deleted: RunOnce,
+    dir: TempDir,
 }
 
 impl Task {
@@ -32,15 +31,9 @@ impl Task {
         let runtime = runtime.into();
 
         let id = make_task_id();
-        let entrypoint = containerd.entrypoint(&image).await?;
         let mounts = containerd.get_mounts(&id, &image).await?;
 
-        let mut args: Vec<_> = args.into_iter().map(|arg| arg.into()).collect();
-        if args.is_empty() {
-            args = entrypoint;
-        } else if let Some(argv0) = entrypoint.into_iter().next() {
-            args.insert(0, argv0);
-        }
+        let args: Vec<_> = args.into_iter().map(|arg| arg.into()).collect();
 
         let process = ProcessBuilder::default()
             .user(UserBuilder::default().build().unwrap())
@@ -71,28 +64,28 @@ impl Task {
             spec,
             task_deleted: RunOnce::new(),
             container_deleted: RunOnce::new(),
+            dir: tempdir()?,
         })
     }
 }
 
 impl crate::traits::Task for Task {
-    async fn create(&self, verbose: bool) -> Result<()> {
-        let stdout = if !verbose {
-            PathBuf::new()
-        } else if let Ok(stdout) = canonicalize("/proc/self/fd/1").await {
-            stdout
-        } else {
-            PathBuf::new()
-        };
+    async fn create(&self) -> Result<()> {
+        let stdout = self.dir.path().join("stdout");
+        let stderr = self.dir.path().join("stderr");
+
+        let _ = std::fs::write(&stdout, "");
+        let _ = std::fs::write(&stderr, "");
 
         let stdout = stdout.to_string_lossy().into_owned();
+        let stderr = stderr.to_string_lossy().into_owned();
 
         self.containerd
             .create_container(&self.id, &self.image, &self.runtime, self.spec.clone())
             .await?;
 
         self.containerd
-            .create_task(&self.id, &self.mounts[..], stdout)
+            .create_task(&self.id, &self.mounts[..], stdout, stderr)
             .await?;
 
         Ok(())
@@ -103,7 +96,14 @@ impl crate::traits::Task for Task {
     }
 
     async fn wait(&self) -> Result<()> {
-        self.containerd.wait_task(&self.id).await
+        let status = self.containerd.wait_task(&self.id).await?;
+        let stdout = std::fs::read_to_string(self.dir.path().join("stdout")).unwrap_or_default();
+        let stderr = std::fs::read_to_string(self.dir.path().join("stderr")).unwrap_or_default();
+        ensure!(
+            status == 0,
+            "Exit status {status}, stdout: {stdout:?}, stderr: {stderr:?}"
+        );
+        Ok(())
     }
 
     async fn delete(&self) -> Result<()> {