Merge branch 'main' into dev2

Signed-off-by: jokemanfire <[email protected]>

Author: jokemanfire

.github/workflows/ci.yml

@@ -222,7 +222,7 @@ jobs:
           # run the example
           cargo run -p containerd-shim --example skeleton -- -namespace default -id 1234 -address "\\.\pipe\containerd-containerd"  -publish-binary ./bin/containerd start
           ps skeleton
-          cargo run -p containerd-shim-protos --example shim-proto-connect \\.\pipe\containerd-shim-17630016127144989388-pipe
+          cargo run -p containerd-shim-protos --example shim-proto-connect \\.\pipe\containerd-shim-bc764c65e177434fcefe8257dc440be8b8acf7c96156320d965938f7e9ae1a35-pipe
           $skeleton = get-process skeleton -ErrorAction SilentlyContinue
           if ($skeleton) { exit 1 }
       - name: Run client

Cargo.toml

@@ -32,7 +32,7 @@ futures = "0.3.19"
 libc = "0.2.112"
 log = {version = "0.4.2", features=["kv_unstable"]}
 nix = "0.29"
-oci-spec = "0.6"
+oci-spec = "0.7"
 os_pipe = "1.1"
 prctl = "1.0.0"
 prost = "0.13"

crates/runc-shim/src/cgroup_memory.rs

@@ -93,15 +93,15 @@ pub async fn register_memory_event(
     let path = cg_dir.join(event_name);
     let event_file = fs::File::open(path.clone())
         .await
-        .map_err(other_error!(e, "Error get path:"))?;
+        .map_err(other_error!("Error get path:"))?;
 
     let eventfd = EventFd::from_value_and_flags(0, EfdFlags::EFD_CLOEXEC)?;
 
     let event_control_path = cg_dir.join("cgroup.event_control");
     let data = format!("{} {}", eventfd.as_raw_fd(), event_file.as_raw_fd());
     fs::write(&event_control_path, data.clone())
         .await
-        .map_err(other_error!(e, "Error write eventfd:"))?;
+        .map_err(other_error!("Error write eventfd:"))?;
 
     let mut buf = [0u8; 8];
 

crates/runc-shim/src/common.rs

@@ -192,7 +192,7 @@ pub fn create_runc(
     }
     gopts
         .build()
-        .map_err(other_error!(e, "unable to create runc instance"))
+        .map_err(other_error!("unable to create runc instance"))
 }
 
 #[derive(Default)]

crates/runc-shim/src/container.rs

@@ -56,6 +56,8 @@ pub trait Container {
     async fn stats(&self) -> Result<Metrics>;
     async fn all_processes(&self) -> Result<Vec<ProcessInfo>>;
     async fn close_io(&mut self, exec_id: Option<&str>) -> Result<()>;
+    async fn pause(&mut self) -> Result<()>;
+    async fn resume(&mut self) -> Result<()>;
 }
 
 #[async_trait]
@@ -212,6 +214,14 @@ where
         let process = self.get_mut_process(exec_id)?;
         process.close_io().await
     }
+
+    async fn pause(&mut self) -> Result<()> {
+        self.init.pause().await
+    }
+
+    async fn resume(&mut self) -> Result<()> {
+        self.init.resume().await
+    }
 }
 
 impl<T, E, P> ContainerTemplate<T, E, P>

crates/runc-shim/src/processes.rs

@@ -55,6 +55,8 @@ pub trait Process {
     async fn stats(&self) -> Result<Metrics>;
     async fn ps(&self) -> Result<Vec<ProcessInfo>>;
     async fn close_io(&mut self) -> Result<()>;
+    async fn pause(&mut self) -> Result<()>;
+    async fn resume(&mut self) -> Result<()>;
 }
 
 #[async_trait]
@@ -65,6 +67,8 @@ pub trait ProcessLifecycle<P: Process> {
     async fn update(&self, p: &mut P, resources: &LinuxResources) -> Result<()>;
     async fn stats(&self, p: &P) -> Result<Metrics>;
     async fn ps(&self, p: &P) -> Result<Vec<ProcessInfo>>;
+    async fn pause(&self, p: &mut P) -> Result<()>;
+    async fn resume(&self, p: &mut P) -> Result<()>;
 }
 
 pub struct ProcessTemplate<S> {
@@ -200,4 +204,12 @@ where
         }
         Ok(())
     }
+
+    async fn pause(&mut self) -> Result<()> {
+        self.lifecycle.clone().pause(self).await
+    }
+
+    async fn resume(&mut self) -> Result<()> {
+        self.lifecycle.clone().resume(self).await
+    }
 }

crates/runc-shim/src/runc.rs

@@ -360,7 +360,7 @@ impl ProcessLifecycle<InitProcess> for RuncInitLifecycle {
             .runtime
             .ps(&p.id)
             .await
-            .map_err(other_error!(e, "failed to execute runc ps"))?;
+            .map_err(other_error!("failed to execute runc ps"))?;
         Ok(pids
             .iter()
             .map(|&x| ProcessInfo {
@@ -369,6 +369,46 @@ impl ProcessLifecycle<InitProcess> for RuncInitLifecycle {
             })
             .collect())
     }
+
+    #[cfg(target_os = "linux")]
+    async fn pause(&self, p: &mut InitProcess) -> Result<()> {
+        match p.state {
+            Status::RUNNING => {
+                p.state = Status::PAUSING;
+                if let Err(e) = self.runtime.pause(p.id.as_str()).await {
+                    p.state = Status::RUNNING;
+                    return Err(runtime_error(&self.bundle, e, "OCI runtime pause failed").await);
+                }
+                p.state = Status::PAUSED;
+                Ok(())
+            }
+            _ => Err(other!("cannot pause when in {:?} state", p.state)),
+        }
+    }
+
+    #[cfg(not(target_os = "linux"))]
+    async fn pause(&self, _p: &mut InitProcess) -> Result<()> {
+        Err(Error::Unimplemented("pause".to_string()))
+    }
+
+    #[cfg(target_os = "linux")]
+    async fn resume(&self, p: &mut InitProcess) -> Result<()> {
+        match p.state {
+            Status::PAUSED => {
+                if let Err(e) = self.runtime.resume(p.id.as_str()).await {
+                    return Err(runtime_error(&self.bundle, e, "OCI runtime pause failed").await);
+                }
+                p.state = Status::RUNNING;
+                Ok(())
+            }
+            _ => Err(other!("cannot resume when in {:?} state", p.state)),
+        }
+    }
+
+    #[cfg(not(target_os = "linux"))]
+    async fn resume(&self, _p: &mut InitProcess) -> Result<()> {
+        Err(Error::Unimplemented("resume".to_string()))
+    }
 }
 
 impl RuncInitLifecycle {
@@ -495,6 +535,14 @@ impl ProcessLifecycle<ExecProcess> for RuncExecLifecycle {
     async fn ps(&self, _p: &ExecProcess) -> Result<Vec<ProcessInfo>> {
         Err(Error::Unimplemented("exec ps".to_string()))
     }
+
+    async fn pause(&self, _p: &mut ExecProcess) -> Result<()> {
+        Err(Error::Unimplemented("exec pause".to_string()))
+    }
+
+    async fn resume(&self, _p: &mut ExecProcess) -> Result<()> {
+        Err(Error::Unimplemented("exec resume".to_string()))
+    }
 }
 
 async fn copy_console(

crates/runc-shim/src/service.rs

@@ -27,12 +27,12 @@ use containerd_shim::{
     event::Event,
     io_error,
     monitor::{Subject, Topic},
-    protos::{events::task::TaskExit, protobuf::MessageDyn},
+    protos::{events::task::TaskExit, protobuf::MessageDyn, ttrpc::context::with_timeout},
     util::{
         convert_to_timestamp, read_options, read_pid_from_file, read_runtime, read_spec, timestamp,
         write_str_to_file,
     },
-    Config, Context, DeleteResponse, Error, Flags, StartOpts,
+    Config, DeleteResponse, Error, Flags, StartOpts,
 };
 use log::{debug, error, warn};
 use tokio::sync::mpsc::{channel, Receiver, Sender};
@@ -218,8 +218,11 @@ async fn forward(
 ) {
     tokio::spawn(async move {
         while let Some((topic, e)) = rx.recv().await {
+            // While ttrpc push the event,give it a 5 seconds timeout.
+            // Prevent event reporting from taking too long time.
+            // Learnd from goshim's containerd/runtime/v2/shim/publisher.go
             publisher
-                .publish(Context::default(), &topic, &ns, e)
+                .publish(with_timeout(5000000000), &topic, &ns, e)
                 .await
                 .unwrap_or_else(|e| warn!("publish {} to containerd: {}", topic, e));
         }

crates/runc-shim/src/task.rs

@@ -50,6 +50,10 @@ use std::path::Path;
 
 #[cfg(target_os = "linux")]
 use cgroups_rs::hierarchies::is_cgroup2_unified_mode;
+use containerd_shim::{
+    api::{PauseRequest, ResumeRequest},
+    protos::events::task::{TaskPaused, TaskResumed},
+};
 #[cfg(target_os = "linux")]
 use containerd_shim::{
     error::{Error, Result},
@@ -143,7 +147,7 @@ async fn monitor_oom(id: &String, pid: u32, tx: EventSender) -> Result<()> {
             "memory.oom_control",
         )
         .await
-        .map_err(other_error!(e, "register_memory_event failed:"))?;
+        .map_err(other_error!("register_memory_event failed:"))?;
 
         run_oom_monitor(rx, id.to_string(), tx);
     }
@@ -288,6 +292,32 @@ where
         })
     }
 
+    async fn pause(&self, _ctx: &TtrpcContext, req: PauseRequest) -> TtrpcResult<Empty> {
+        info!("pause request for {:?}", req);
+        let mut container = self.get_container(req.id()).await?;
+        container.pause().await?;
+        self.send_event(TaskPaused {
+            container_id: req.id.to_string(),
+            ..Default::default()
+        })
+        .await;
+        info!("pause request for {:?} returns successfully", req);
+        Ok(Empty::new())
+    }
+
+    async fn resume(&self, _ctx: &TtrpcContext, req: ResumeRequest) -> TtrpcResult<Empty> {
+        info!("resume request for {:?}", req);
+        let mut container = self.get_container(req.id()).await?;
+        container.resume().await?;
+        self.send_event(TaskResumed {
+            container_id: req.id.to_string(),
+            ..Default::default()
+        })
+        .await;
+        info!("resume request for {:?} returns successfully", req);
+        Ok(Empty::new())
+    }
+
     async fn kill(&self, _ctx: &TtrpcContext, req: KillRequest) -> TtrpcResult<Empty> {
         info!("Kill request for {:?}", req);
         let mut container = self.get_container(req.id()).await?;

crates/runc/src/io.rs

@@ -25,6 +25,7 @@ use std::{
     sync::Mutex,
 };
 
+
 #[cfg(feature = "async")]
 use tokio::io::{AsyncRead, AsyncWrite};
 use tokio::net::unix::pipe::{self, Receiver, Sender};