runk: Support exec sub-command

`exec` will execute a command inside a container which exists and is not
frozon or stopped. *Inside* means that the new process share namespaces
and cgroup with the container init process. Command can be specified by
`--process` parameter to read from a file, or from other parameters such
as arg, env, etc. In order to be compatible with `create`/`run`
commands, I refactor libcontainer. `Container` in builder.rs is divided
into `InitContainer` and `ActivatedContainer`. `InitContainer` is used
for `create`/`run` command. It will load spec from given bundle path.
`ActivatedContainer` is used by `exec` command, and will read the
container's status file, which stores the spec and `CreateOpt` for
creating the rustjail::LinuxContainer. Adapt the spec by replacing the
process with given options and updating the namesapces with some paths
to join the container. I also rename the `ContainerContext` as
`ContainerLauncher`, which is only used to spawn process now. It uses
the `LinuxContaier` in rustjail as the runner. For `create`/`run`, the
`launch` method will create a new container and run the first process.
For `exec`, the `launch` method will spawn a process which joins a
container.

Fixes #4363

Signed-off-by: Chen Yiyang <cyyzero@qq.com>

src/agent/rustjail/src/container.rs

@@ -1171,7 +1171,7 @@ fn do_exec(args: &[String]) -> ! {
     unreachable!()
 }
 
-fn update_namespaces(logger: &Logger, spec: &mut Spec, init_pid: RawFd) -> Result<()> {
+pub fn update_namespaces(logger: &Logger, spec: &mut Spec, init_pid: RawFd) -> Result<()> {
     info!(logger, "updating namespaces");
     let linux = spec
         .linux

src/tools/runk/Cargo.lock

@@ -543,6 +543,7 @@ dependencies = [
  "nix 0.23.1",
  "oci",
  "rustjail",
+ "scopeguard",
  "serde",
  "serde_json",
  "slog",
@@ -986,6 +987,7 @@ dependencies = [
  "slog",
  "slog-async",
  "tabwriter",
+ "tempfile",
  "tokio",
  "users",
 ]

src/tools/runk/Cargo.toml

@@ -25,6 +25,9 @@ serde_json = "1.0.74"
 users = "0.11.0"
 tabwriter = "1.2.1"
 
+[dev-dependencies]
+tempfile = "3.3.0"
+
 [workspace]
 members = [
     "libcontainer"

src/tools/runk/libcontainer/Cargo.toml

@@ -18,6 +18,7 @@ slog = "2.7.0"
 chrono = { version = "0.4.19", features = ["serde"] }
 serde = { version = "1.0.133", features = ["derive"] }
 serde_json = "1.0.74"
+scopeguard = "1.1.0"
 
 [dev-dependencies]
 tempfile = "3.3.0"

src/tools/runk/libcontainer/src/builder.rs

@@ -3,22 +3,53 @@
 // SPDX-License-Identifier: Apache-2.0
 //
 
-use crate::container::{get_config_path, ContainerContext};
+use crate::container::{get_config_path, ContainerLauncher};
+use crate::{
+    status::{get_current_container_state, Status},
+    utils::validate_process_spec,
+};
 use anyhow::{anyhow, Result};
 use derive_builder::Builder;
-use oci::Spec;
+use oci::{ContainerState, Process as OCIProcess, Spec};
+use rustjail::container::update_namespaces;
+use rustjail::{container::LinuxContainer, specconv::CreateOpts};
+use slog::{debug, Logger};
+use std::fs::File;
 use std::path::{Path, PathBuf};
 
-#[derive(Default, Builder, Debug)]
-pub struct Container {
+/// Used for create and run commands. It will prepare the options used for creating a new container.
+#[derive(Default, Builder, Debug, Clone)]
+#[builder(build_fn(validate = "Self::validate"))]
+pub struct InitContainer {
     id: String,
     bundle: PathBuf,
     root: PathBuf,
     console_socket: Option<PathBuf>,
+    pid_file: Option<PathBuf>,
 }
 
-impl Container {
-    pub fn create_ctx(self) -> Result<ContainerContext> {
+impl InitContainerBuilder {
+    /// Pre-validate before building InitContainer
+    fn validate(&self) -> Result<(), String> {
+        // ensure container hasn't already been created
+        let id = self.id.as_ref().unwrap();
+        let root = self.root.as_ref().unwrap();
+        let path = root.join(id);
+        if path.as_path().exists() {
+            return Err(format!(
+                "container {} already exists at path {:?}",
+                id, root
+            ));
+        }
+        Ok(())
+    }
+}
+
+impl InitContainer {
+    /// Create ContainerLauncher that can be used to launch a new container.
+    /// It will read the spec under bundle path.
+    pub fn create_launcher(self, logger: &Logger) -> Result<ContainerLauncher> {
+        debug!(logger, "enter InitContainer::create_launcher {:?}", self);
         let bundle_canon = self.bundle.canonicalize()?;
         let config_path = get_config_path(&bundle_canon);
         let mut spec = Spec::load(
@@ -26,58 +57,227 @@ impl Container {
                 .to_str()
                 .ok_or_else(|| anyhow!("invalid config path"))?,
         )?;
+        // Only absolute rootfs path is valid when creating LinuxContainer later.
+        canonicalize_spec_root(&mut spec, &bundle_canon)?;
+        debug!(logger, "load spec from config file: {:?}", spec);
+        validate_spec(&spec, &self.console_socket)?;
 
-        if spec.root.is_some() {
+        let config = CreateOpts {
+            cgroup_name: "".to_string(),
+            use_systemd_cgroup: false,
+            // TODO: liboci-cli does not support --no-pivot option for create and run command.
+            // After liboci-cli supports the option, we will change the following code.
+            // no_pivot_root: self.no_pivot,
+            no_pivot_root: false,
+            no_new_keyring: false,
+            spec: Some(spec),
+            rootless_euid: false,
+            rootless_cgroup: false,
+        };
+        debug!(logger, "create LinuxContainer with config: {:?}", config);
+        let container =
+            create_linux_container(&self.id, &self.root, config, self.console_socket, logger)?;
+
+        Ok(ContainerLauncher::new(
+            &self.id,
+            &bundle_canon,
+            &self.root,
+            true,
+            container,
+            self.pid_file,
+        ))
+    }
+}
+
+/// Used for exec command. It will prepare the options for joining an existing container.
+#[derive(Default, Builder, Debug, Clone)]
+#[builder(build_fn(validate = "Self::validate"))]
+pub struct ActivatedContainer {
+    pub id: String,
+    pub root: PathBuf,
+    pub console_socket: Option<PathBuf>,
+    pub pid_file: Option<PathBuf>,
+    pub tty: bool,
+    pub cwd: Option<PathBuf>,
+    pub env: Vec<(String, String)>,
+    pub no_new_privs: bool,
+    pub args: Vec<String>,
+    pub process: Option<PathBuf>,
+}
+
+impl ActivatedContainerBuilder {
+    /// pre-validate before building ActivatedContainer
+    fn validate(&self) -> Result<(), String> {
+        // ensure container exists
+        let id = self.id.as_ref().unwrap();
+        let root = self.root.as_ref().unwrap();
+        let path = root.join(id);
+        if !path.as_path().exists() {
+            return Err(format!(
+                "container {} does not exist at path {:?}",
+                id, root
+            ));
+        }
+
+        // ensure argv will not be empty in process exec phase later
+        let process = self.process.as_ref().unwrap();
+        let args = self.args.as_ref().unwrap();
+        if process.is_none() && args.is_empty() {
+            return Err("process and args cannot be all empty".to_string());
+        }
+        Ok(())
+    }
+}
+
+impl ActivatedContainer {
+    /// Create ContainerLauncher that can be used to spawn a process in an existing container.
+    /// It reads the spec from status file of an existing container, and adapted it with given process file
+    /// or other options like args, env, etc. It also changes the namespace in spec to join the container.
+    pub fn create_launcher(self, logger: &Logger) -> Result<ContainerLauncher> {
+        debug!(
+            logger,
+            "enter ActivatedContainer::create_launcher {:?}", self
+        );
+        let status = Status::load(&self.root, &self.id)?;
+        let state = get_current_container_state(&status)?;
+
+        // If state is Created or Running, we can execute the process.
+        if state != ContainerState::Created && state != ContainerState::Running {
+            return Err(anyhow!("cannot exec in a stopped or paused container"));
+        }
+
+        let mut config = status.config;
+        let spec = config.spec.as_mut().unwrap();
+        self.adapt_exec_spec(spec, status.pid, logger)?;
+        debug!(logger, "adapted spec: {:?}", spec);
+        validate_spec(spec, &self.console_socket)?;
+
+        debug!(logger, "create LinuxContainer with config: {:?}", config);
+        // Maybe we should move some properties from status into LinuxContainer,
+        // like pid, process_start_time, created, cgroup_manager, etc. But it works now.
+        let container =
+            create_linux_container(&self.id, &self.root, config, self.console_socket, logger)?;
+
+        Ok(ContainerLauncher::new(
+            &self.id,
+            &status.bundle,
+            &self.root,
+            false,
+            container,
+            self.pid_file,
+        ))
+    }
+
+    /// Adapt spec to execute a new process which will join the container.
+    fn adapt_exec_spec(&self, spec: &mut Spec, pid: i32, logger: &Logger) -> Result<()> {
+        // If with --process, load process from file.
+        // Otherwise, update process with args and other options.
+        if let Some(process_path) = self.process.as_ref() {
+            spec.process = Some(Self::get_process(process_path)?);
+        } else if let Some(process) = spec.process.as_mut() {
+            self.update_process(process)?;
+        } else {
+            return Err(anyhow!("process is empty in spec"));
+        };
+        // Exec process will join the container's namespaces
+        update_namespaces(logger, spec, pid)?;
+        Ok(())
+    }
+
+    /// Update process with args and other options.
+    fn update_process(&self, process: &mut OCIProcess) -> Result<()> {
+        process.args = self.args.clone();
+        process.no_new_privileges = self.no_new_privs;
+        process.terminal = self.tty;
+        if let Some(cwd) = self.cwd.as_ref() {
+            process.cwd = cwd.as_path().display().to_string();
+        }
+        process
+            .env
+            .extend(self.env.iter().map(|kv| format!("{}={}", kv.0, kv.1)));
+        Ok(())
+    }
+
+    /// Read and parse OCI Process from path
+    fn get_process(process_path: &Path) -> Result<OCIProcess> {
+        let f = File::open(process_path)?;
+        Ok(serde_json::from_reader(f)?)
+    }
+}
+
+/// If root in spec is a relative path, make it absolute.
+fn canonicalize_spec_root(spec: &mut Spec, bundle_canon: &Path) -> Result<()> {
     let mut spec_root = spec
         .root
         .as_mut()
         .ok_or_else(|| anyhow!("root config was not present in the spec file"))?;
     let rootfs_path = Path::new(&spec_root.path);
-
-            // If the rootfs path in the spec file is a relative path,
-            // convert it into a canonical path to pass validation of rootfs in the agent.
-            if !&rootfs_path.is_absolute() {
+    if !rootfs_path.is_absolute() {
         spec_root.path = bundle_canon
             .join(rootfs_path)
             .canonicalize()?
             .to_str()
             .map(|s| s.to_string())
-                    .ok_or_else(|| {
-                        anyhow!("failed to convert a rootfs path into a canonical path")
-                    })?;
-            }
+            .ok_or_else(|| anyhow!("failed to convert a rootfs path into a canonical path"))?;
     }
+    Ok(())
+}
 
+fn create_linux_container(
+    id: &str,
+    root: &Path,
+    config: CreateOpts,
+    console_socket: Option<PathBuf>,
+    logger: &Logger,
+) -> Result<LinuxContainer> {
+    let mut container = LinuxContainer::new(
+        id,
+        root.to_str()
+            .map(|s| s.to_string())
+            .ok_or_else(|| anyhow!("failed to convert bundle path"))?
+            .as_str(),
+        config,
+        logger,
+    )?;
+    if let Some(socket_path) = console_socket.as_ref() {
+        container.set_console_socket(socket_path)?;
+    }
+    Ok(container)
+}
+
+/// Check whether spec is valid. Now runk only support detach mode.
+pub fn validate_spec(spec: &Spec, console_socket: &Option<PathBuf>) -> Result<()> {
+    validate_process_spec(&spec.process)?;
     if let Some(process) = spec.process.as_ref() {
         // runk always launches containers with detached mode, so users have to
         // use a console socket with run or create operation when a terminal is used.
-            if process.terminal && self.console_socket.is_none() {
+        if process.terminal && console_socket.is_none() {
             return Err(anyhow!(
                 "cannot allocate a pseudo-TTY without setting a console socket"
             ));
         }
     }
-
-        Ok(ContainerContext {
-            id: self.id,
-            bundle: bundle_canon,
-            state_root: self.root,
-            spec,
-            // TODO: liboci-cli does not support --no-pivot option for create and run command.
-            // After liboci-cli supports the option, we will change the following code.
-            // no_pivot_root: self.no_pivot,
-            no_pivot_root: false,
-            console_socket: self.console_socket,
-        })
-    }
+    Ok(())
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
     use crate::container::CONFIG_FILE_NAME;
-    use oci::{self, Spec};
-    use std::{fs::File, path::PathBuf};
+    use crate::utils::test_utils::TEST_ROOTFS_PATH;
+    use chrono::DateTime;
+    use nix::unistd::getpid;
+    use oci::{self, Root, Spec};
+    use oci::{Linux, LinuxNamespace, User};
+    use rustjail::cgroups::fs::Manager;
+    use rustjail::container::TYPETONAME;
+    use slog::o;
+    use std::fs::create_dir;
+    use std::time::SystemTime;
+    use std::{
+        fs::{create_dir_all, File},
+        path::PathBuf,
+    };
     use tempfile::tempdir;
 
     #[derive(Debug)]
@@ -86,56 +286,88 @@ mod tests {
         bundle: PathBuf,
         root: PathBuf,
         console_socket: Option<PathBuf>,
-        spec: Spec,
-        no_pivot_root: bool,
+        pid_file: Option<PathBuf>,
+        config: CreateOpts,
     }
 
     #[test]
-    fn test_create_ctx() {
+    fn test_init_container_validate() {
+        let root = tempdir().unwrap();
+        let id = "test".to_string();
+        Status::create_dir(root.path(), id.as_str()).unwrap();
+        let result = InitContainerBuilder::default()
+            .id(id)
+            .root(root.path().to_path_buf())
+            .bundle(PathBuf::from("test"))
+            .pid_file(None)
+            .console_socket(None)
+            .build();
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_init_container_create_launcher() {
+        let logger = slog::Logger::root(slog::Discard, o!());
+        let root_dir = tempdir().unwrap();
         let bundle_dir = tempdir().unwrap();
+        // create dummy rootfs
+        create_dir(bundle_dir.path().join("rootfs")).unwrap();
         let config_file = bundle_dir.path().join(CONFIG_FILE_NAME);
-        let spec = Spec::default();
+        let mut spec = create_dummy_spec();
         let file = File::create(config_file).unwrap();
         serde_json::to_writer(&file, &spec).unwrap();
 
+        spec.root.as_mut().unwrap().path = bundle_dir
+            .path()
+            .join(TEST_ROOTFS_PATH)
+            .to_string_lossy()
+            .to_string();
         let test_data = TestData {
             id: String::from("test"),
-            bundle: PathBuf::from(bundle_dir.into_path()),
-            root: PathBuf::from("test"),
+            bundle: bundle_dir.path().to_path_buf(),
+            root: root_dir.into_path(),
             console_socket: Some(PathBuf::from("test")),
-            spec: Spec::default(),
-            no_pivot_root: false,
+            config: CreateOpts {
+                spec: Some(spec),
+                ..Default::default()
+            },
+            pid_file: Some(PathBuf::from("test")),
         };
 
-        let test_ctx = ContainerContext {
-            id: test_data.id.clone(),
-            bundle: test_data.bundle.clone(),
-            state_root: test_data.root.clone(),
-            spec: test_data.spec.clone(),
-            no_pivot_root: test_data.no_pivot_root,
-            console_socket: test_data.console_socket.clone(),
-        };
-
-        let ctx = ContainerBuilder::default()
+        let launcher = InitContainerBuilder::default()
             .id(test_data.id.clone())
             .bundle(test_data.bundle.clone())
             .root(test_data.root.clone())
             .console_socket(test_data.console_socket.clone())
+            .pid_file(test_data.pid_file.clone())
             .build()
             .unwrap()
-            .create_ctx()
+            .create_launcher(&logger)
             .unwrap();
 
-        assert_eq!(test_ctx, ctx);
+        // LinuxContainer doesn't impl PartialEq, so we need to compare the fields manually.
+        assert!(launcher.init);
+        assert_eq!(launcher.bundle, test_data.bundle);
+        assert_eq!(launcher.state_root, test_data.root);
+        assert_eq!(launcher.pid_file, test_data.pid_file);
+        assert_eq!(launcher.runner.id, test_data.id);
+        assert_eq!(launcher.runner.config.spec, test_data.config.spec);
+        assert_eq!(
+            Some(launcher.runner.console_socket),
+            test_data.console_socket
+        );
     }
 
     #[test]
-    fn test_create_ctx_tty_err() {
+    fn test_init_container_tty_err() {
+        let logger = slog::Logger::root(slog::Discard, o!());
         let bundle_dir = tempdir().unwrap();
         let config_file = bundle_dir.path().join(CONFIG_FILE_NAME);
 
-        let mut spec = Spec::default();
-        spec.process = Some(oci::Process::default());
+        let mut spec = oci::Spec {
+            process: Some(oci::Process::default()),
+            ..Default::default()
+        };
         spec.process.as_mut().unwrap().terminal = true;
 
         let file = File::create(config_file).unwrap();
@@ -143,22 +375,263 @@ mod tests {
 
         let test_data = TestData {
             id: String::from("test"),
-            bundle: PathBuf::from(bundle_dir.into_path()),
-            root: PathBuf::from("test"),
+            bundle: bundle_dir.into_path(),
+            root: tempdir().unwrap().into_path(),
             console_socket: None,
-            spec: Spec::default(),
-            no_pivot_root: false,
+            config: CreateOpts {
+                spec: Some(spec),
+                ..Default::default()
+            },
+            pid_file: None,
         };
 
-        let ctx = ContainerBuilder::default()
+        let result = InitContainerBuilder::default()
             .id(test_data.id.clone())
             .bundle(test_data.bundle.clone())
             .root(test_data.root.clone())
             .console_socket(test_data.console_socket.clone())
+            .pid_file(test_data.pid_file)
             .build()
             .unwrap()
-            .create_ctx();
+            .create_launcher(&logger);
 
-        assert!(ctx.is_err());
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_canonicalize_spec_root() {
+        let gen_spec = |p: &str| -> Spec {
+            Spec {
+                root: Some(Root {
+                    path: p.to_string(),
+                    readonly: false,
+                }),
+                ..Default::default()
+            }
+        };
+
+        let rootfs_name = TEST_ROOTFS_PATH;
+        let temp_dir = tempdir().unwrap();
+        let bundle_dir = temp_dir.path();
+        let abs_root = bundle_dir.join(rootfs_name);
+        create_dir_all(abs_root.clone()).unwrap();
+        let mut spec = gen_spec(abs_root.to_str().unwrap());
+        assert!(canonicalize_spec_root(&mut spec, bundle_dir).is_ok());
+        assert_eq!(spec.root.unwrap().path, abs_root.to_str().unwrap());
+        let mut spec = gen_spec(rootfs_name);
+        assert!(canonicalize_spec_root(&mut spec, bundle_dir).is_ok());
+        assert_eq!(spec.root.unwrap().path, abs_root.to_str().unwrap());
+    }
+
+    fn create_dummy_spec() -> Spec {
+        let linux = oci::Linux {
+            namespaces: TYPETONAME
+                .iter()
+                .filter(|&(_, &name)| name != "user")
+                .map(|ns| LinuxNamespace {
+                    r#type: ns.0.to_string(),
+                    path: "".to_string(),
+                })
+                .collect(),
+            ..Default::default()
+        };
+        Spec {
+            version: "1.0".to_string(),
+            process: Some(OCIProcess {
+                args: vec!["sleep".to_string(), "10".to_string()],
+                env: vec!["PATH=/bin:/usr/bin".to_string()],
+                cwd: "/".to_string(),
+                ..Default::default()
+            }),
+            hostname: "runk".to_string(),
+            root: Some(Root {
+                path: TEST_ROOTFS_PATH.to_string(),
+                readonly: false,
+            }),
+            linux: Some(linux),
+            ..Default::default()
+        }
+    }
+
+    fn create_dummy_status(id: &str, pid: i32, root: &Path, spec: &Spec) -> Status {
+        Status {
+            oci_version: spec.version.clone(),
+            id: id.to_string(),
+            pid,
+            root: root.to_path_buf(),
+            bundle: PathBuf::from("/tmp"),
+            rootfs: TEST_ROOTFS_PATH.to_string(),
+            process_start_time: 0,
+            created: DateTime::from(SystemTime::now()),
+            cgroup_manager: Manager::new("test").unwrap(),
+            config: CreateOpts {
+                spec: Some(spec.clone()),
+                ..Default::default()
+            },
+        }
+    }
+
+    fn create_activated_dirs(root: &Path, id: &str, bundle: &Path) {
+        Status::create_dir(root, id).unwrap();
+        create_dir_all(bundle.join(TEST_ROOTFS_PATH)).unwrap();
+    }
+
+    #[test]
+    fn test_activated_container_validate() {
+        let root = tempdir().unwrap();
+        let id = "test".to_string();
+        Status::create_dir(root.path(), &id).unwrap();
+        let result = ActivatedContainerBuilder::default()
+            .id(id)
+            .root(root.into_path())
+            .console_socket(None)
+            .pid_file(None)
+            .tty(false)
+            .cwd(None)
+            .env(Vec::new())
+            .no_new_privs(false)
+            .process(None)
+            .args(vec!["sleep".to_string(), "10".to_string()])
+            .build();
+        assert!(result.is_ok());
+    }
+
+    #[test]
+    fn test_activated_container_create() {
+        let logger = slog::Logger::root(slog::Discard, o!());
+        let bundle_dir = tempdir().unwrap();
+        let root = tempdir().unwrap();
+        // let bundle = temp
+        let id = "test".to_string();
+        create_activated_dirs(
+            &root.path().to_path_buf(),
+            &id,
+            &bundle_dir.path().to_path_buf(),
+        );
+        let pid = getpid().as_raw();
+
+        let mut spec = create_dummy_spec();
+        spec.root.as_mut().unwrap().path = bundle_dir
+            .path()
+            .join(TEST_ROOTFS_PATH)
+            .to_string_lossy()
+            .to_string();
+
+        let status = create_dummy_status(&id, pid, &root.path().to_path_buf(), &spec);
+        status.save().unwrap();
+
+        let result = ActivatedContainerBuilder::default()
+            .id(id)
+            .root(root.into_path())
+            .console_socket(Some(PathBuf::from("/var/run/test.sock")))
+            .pid_file(Some(PathBuf::from("test")))
+            .tty(true)
+            .cwd(Some(PathBuf::from("/tmp")))
+            .env(vec![
+                ("K1".to_string(), "V1".to_string()),
+                ("K2".to_string(), "V2".to_string()),
+            ])
+            .no_new_privs(true)
+            .process(None)
+            .args(vec!["sleep".to_string(), "10".to_string()])
+            .build()
+            .unwrap();
+
+        let linux = Linux {
+            namespaces: TYPETONAME
+                .iter()
+                .filter(|&(_, &name)| name != "user")
+                .map(|ns| LinuxNamespace {
+                    r#type: ns.0.to_string(),
+                    path: format!("/proc/{}/ns/{}", pid, ns.1),
+                })
+                .collect(),
+            ..Default::default()
+        };
+        spec.linux = Some(linux);
+        spec.process = Some(OCIProcess {
+            terminal: result.tty,
+            console_size: None,
+            user: User::default(),
+            args: result.args.clone(),
+            cwd: result.cwd.clone().unwrap().to_string_lossy().to_string(),
+            env: vec![
+                "PATH=/bin:/usr/bin".to_string(),
+                "K1=V1".to_string(),
+                "K2=V2".to_string(),
+            ],
+            capabilities: None,
+            rlimits: Vec::new(),
+            no_new_privileges: result.no_new_privs,
+            apparmor_profile: "".to_string(),
+            oom_score_adj: None,
+            selinux_label: "".to_string(),
+        });
+        let launcher = result.clone().create_launcher(&logger).unwrap();
+        assert!(!launcher.init);
+        assert_eq!(launcher.runner.config.spec.unwrap(), spec);
+        assert_eq!(
+            launcher.runner.console_socket,
+            result.console_socket.unwrap()
+        );
+        assert_eq!(launcher.pid_file, result.pid_file);
+    }
+
+    #[test]
+    fn test_activated_container_create_with_process() {
+        const PROCESS_FILE_NAME: &str = "process.json";
+        let bundle_dir = tempdir().unwrap();
+        let process_file = bundle_dir.path().join(PROCESS_FILE_NAME);
+        let process_template = OCIProcess {
+            args: vec!["sleep".to_string(), "10".to_string()],
+            cwd: "/".to_string(),
+            ..Default::default()
+        };
+        let file = File::create(process_file.clone()).unwrap();
+        serde_json::to_writer(&file, &process_template).unwrap();
+
+        let logger = slog::Logger::root(slog::Discard, o!());
+        let root = tempdir().unwrap();
+        let id = "test".to_string();
+        let pid = getpid().as_raw();
+        let mut spec = create_dummy_spec();
+        spec.root.as_mut().unwrap().path = bundle_dir
+            .path()
+            .join(TEST_ROOTFS_PATH)
+            .to_string_lossy()
+            .to_string();
+        create_activated_dirs(
+            &root.path().to_path_buf(),
+            &id,
+            &bundle_dir.path().to_path_buf(),
+        );
+
+        let status = create_dummy_status(&id, pid, &root.path().to_path_buf(), &spec);
+        status.save().unwrap();
+        let launcher = ActivatedContainerBuilder::default()
+            .id(id)
+            .root(root.into_path())
+            .console_socket(None)
+            .pid_file(None)
+            .tty(true)
+            .cwd(Some(PathBuf::from("/tmp")))
+            .env(vec![
+                ("K1".to_string(), "V1".to_string()),
+                ("K2".to_string(), "V2".to_string()),
+            ])
+            .no_new_privs(true)
+            .process(Some(process_file))
+            .args(vec!["sleep".to_string(), "10".to_string()])
+            .build()
+            .unwrap()
+            .create_launcher(&logger)
+            .unwrap();
+
+        assert!(!launcher.init);
+
+        assert_eq!(
+            launcher.runner.config.spec.unwrap().process.unwrap(),
+            process_template
+        );
     }
 }

src/tools/runk/libcontainer/src/container.rs

@@ -5,16 +5,16 @@
 
 use crate::status::Status;
 use anyhow::{anyhow, Result};
-use nix::unistd::{chdir, unlink, Pid};
-use oci::Spec;
+use nix::unistd::{chdir, unlink};
 use rustjail::{
     container::{BaseContainer, LinuxContainer, EXEC_FIFO_FILENAME},
-    process::Process,
-    specconv::CreateOpts,
+    process::{Process, ProcessOperations},
 };
-use slog::Logger;
+use scopeguard::defer;
+use slog::{debug, Logger};
 use std::{
     env::current_dir,
+    fs,
     path::{Path, PathBuf},
 };
 
@@ -26,96 +26,136 @@ pub enum ContainerAction {
     Run,
 }
 
-#[derive(Debug, Clone, PartialEq)]
-pub struct ContainerContext {
+/// Used to run a process. If init is set, it will create a container and run the process in it.
+/// If init is not set, it will run the process in an existing container.
+#[derive(Debug)]
+pub struct ContainerLauncher {
     pub id: String,
     pub bundle: PathBuf,
     pub state_root: PathBuf,
-    pub spec: Spec,
-    pub no_pivot_root: bool,
-    pub console_socket: Option<PathBuf>,
+    pub init: bool,
+    pub runner: LinuxContainer,
+    pub pid_file: Option<PathBuf>,
 }
 
-impl ContainerContext {
-    pub async fn launch(&self, action: ContainerAction, logger: &Logger) -> Result<Pid> {
-        Status::create_dir(&self.state_root, &self.id)?;
+impl ContainerLauncher {
+    pub fn new(
+        id: &str,
+        bundle: &Path,
+        state_root: &Path,
+        init: bool,
+        runner: LinuxContainer,
+        pid_file: Option<PathBuf>,
+    ) -> Self {
+        ContainerLauncher {
+            id: id.to_string(),
+            bundle: bundle.to_path_buf(),
+            state_root: state_root.to_path_buf(),
+            init,
+            runner,
+            pid_file,
+        }
+    }
 
-        let current_dir = current_dir()?;
-        chdir(&self.bundle)?;
-
-        let create_opts = CreateOpts {
-            cgroup_name: "".to_string(),
-            use_systemd_cgroup: false,
-            no_pivot_root: self.no_pivot_root,
-            no_new_keyring: false,
-            spec: Some(self.spec.clone()),
-            rootless_euid: false,
-            rootless_cgroup: false,
-        };
-
-        let mut ctr = LinuxContainer::new(
-            &self.id,
-            &self
-                .state_root
-                .to_str()
-                .map(|s| s.to_string())
-                .ok_or_else(|| anyhow!("failed to convert bundle path"))?,
-            create_opts.clone(),
-            logger,
-        )?;
-
-        let process = if self.spec.process.is_some() {
-            Process::new(
-                logger,
-                self.spec
-                    .process
-                    .as_ref()
-                    .ok_or_else(|| anyhow!("process config was not present in the spec file"))?,
-                &self.id,
-                true,
-                0,
-            )?
+    /// Launch a process. For init containers, we will create a container. For non-init, it will join an existing container.
+    pub async fn launch(&mut self, action: ContainerAction, logger: &Logger) -> Result<()> {
+        if self.init {
+            self.spawn_container(action, logger).await?;
         } else {
-            return Err(anyhow!("no process configuration"));
-        };
-
-        if let Some(ref csocket_path) = self.console_socket {
-            ctr.set_console_socket(csocket_path)?;
+            if action != ContainerAction::Run {
+                return Err(anyhow!(
+                    "ContainerAction::Create is used for init-container only"
+                ));
             }
-
-        match action {
-            ContainerAction::Create => {
-                ctr.start(process).await?;
+            self.spawn_process(ContainerAction::Run, logger).await?;
         }
-            ContainerAction::Run => {
-                ctr.run(process).await?;
-            }
-        }
-
-        let oci_state = ctr.oci_state()?;
-        let status = Status::new(
-            &self.state_root,
-            &self.bundle,
-            oci_state,
-            ctr.init_process_start_time,
-            ctr.created,
-            ctr.cgroup_manager
-                .ok_or_else(|| anyhow!("cgroup manager was not present"))?,
-            create_opts,
+        if let Some(pid_file) = self.pid_file.as_ref() {
+            fs::write(
+                pid_file,
+                format!("{}", self.runner.get_process(self.id.as_str())?.pid()),
             )?;
+        }
+        Ok(())
+    }
 
+    /// Create the container by invoking runner to spawn the first process and save status.
+    async fn spawn_container(&mut self, action: ContainerAction, logger: &Logger) -> Result<()> {
+        // State root path root/id has been created in LinuxContainer::new(),
+        // so we don't have to create it again.
+
+        self.spawn_process(action, logger).await?;
+        let status = self.get_status()?;
         status.save()?;
+        debug!(logger, "saved status is {:?}", status);
 
+        // Clean up the fifo file created by LinuxContainer, which is used for block the created process.
         if action == ContainerAction::Run {
             let fifo_path = get_fifo_path(&status);
             if fifo_path.exists() {
                 unlink(&fifo_path)?;
             }
         }
+        Ok(())
+    }
 
-        chdir(&current_dir)?;
+    /// Generate rustjail::Process from OCI::Process
+    fn get_process(&self, logger: &Logger) -> Result<Process> {
+        let spec = self.runner.config.spec.as_ref().unwrap();
+        if spec.process.is_some() {
+            Ok(Process::new(
+                logger,
+                spec.process
+                    .as_ref()
+                    .ok_or_else(|| anyhow!("process config was not present in the spec file"))?,
+                // rustjail::LinuxContainer use the exec_id to identify processes in a container,
+                // so we can get the spawned process by ctr.get_process(exec_id) later.
+                // Since LinuxContainer is temporarily created to spawn one process in each runk invocation,
+                // we can use arbitrary string as the exec_id. Here we choose the container id.
+                &self.id,
+                self.init,
+                0,
+            )?)
+        } else {
+            Err(anyhow!("no process configuration"))
+        }
+    }
 
-        Ok(Pid::from_raw(ctr.init_process_pid))
+    /// Spawn a new process in the container by invoking runner.
+    async fn spawn_process(&mut self, action: ContainerAction, logger: &Logger) -> Result<()> {
+        // Agent will chdir to bundle_path before creating LinuxContainer. Just do the same as agent.
+        let current_dir = current_dir()?;
+        chdir(&self.bundle)?;
+        defer! {
+            chdir(&current_dir).unwrap();
+        }
+
+        let process = self.get_process(logger)?;
+        match action {
+            ContainerAction::Create => {
+                self.runner.start(process).await?;
+            }
+            ContainerAction::Run => {
+                self.runner.run(process).await?;
+            }
+        }
+        Ok(())
+    }
+
+    /// Generate runk specified Status
+    fn get_status(&self) -> Result<Status> {
+        let oci_state = self.runner.oci_state()?;
+        Status::new(
+            &self.state_root,
+            &self.bundle,
+            oci_state,
+            self.runner.init_process_start_time,
+            self.runner.created,
+            self.runner
+                .cgroup_manager
+                .clone()
+                .ok_or_else(|| anyhow!("cgroup manager was not present"))?,
+            self.runner.config.clone(),
+        )
     }
 }
 

src/tools/runk/libcontainer/src/utils.rs

@@ -5,6 +5,7 @@
 
 use anyhow::{anyhow, Result};
 use nix::sys::stat::Mode;
+use oci::Process;
 use std::{
     fs::{DirBuilder, File},
     io::{prelude::*, BufReader},
@@ -33,10 +34,30 @@ pub fn create_dir_with_mode<P: AsRef<Path>>(path: P, mode: Mode, recursive: bool
         .create(path)?)
 }
 
+// Validate process just like runc, https://github.com/opencontainers/runc/pull/623
+pub fn validate_process_spec(process: &Option<Process>) -> Result<()> {
+    let process = process
+        .as_ref()
+        .ok_or_else(|| anyhow!("process property must not be empty"))?;
+    if process.cwd.is_empty() {
+        return Err(anyhow!("cwd property must not be empty"));
+    }
+    let cwd = Path::new(process.cwd.as_str());
+    if !cwd.is_absolute() {
+        return Err(anyhow!("cwd must be an absolute path"));
+    }
+    if process.args.is_empty() {
+        return Err(anyhow!("args must not be empty"));
+    }
+    Ok(())
+}
+
 #[cfg(test)]
 pub(crate) mod test_utils {
+    use super::*;
     use crate::status::Status;
     use nix::unistd::getpid;
+    use oci::Process;
     use oci::State as OCIState;
     use oci::{ContainerState, Root, Spec};
     use rustjail::cgroups::fs::Manager as CgroupManager;
@@ -57,6 +78,7 @@ pub(crate) mod test_utils {
         },
         "cpath": "test"
     }"#;
+    pub const TEST_ROOTFS_PATH: &str = "rootfs";
 
     pub fn create_dummy_opts() -> CreateOpts {
         let spec = Spec {
@@ -95,7 +117,7 @@ pub(crate) mod test_utils {
         let status = Status::new(
             Path::new(TEST_STATE_ROOT_PATH),
             Path::new(TEST_BUNDLE_PATH),
-            oci_state.clone(),
+            oci_state,
             1,
             created,
             cgm,
@@ -105,4 +127,24 @@ pub(crate) mod test_utils {
 
         status
     }
+
+    #[test]
+    pub fn test_validate_process_spec() {
+        let valid_process = Process {
+            args: vec!["test".to_string()],
+            cwd: "/".to_string(),
+            ..Default::default()
+        };
+        assert!(validate_process_spec(&None).is_err());
+        assert!(validate_process_spec(&Some(valid_process.clone())).is_ok());
+        let mut invalid_process = valid_process.clone();
+        invalid_process.args = vec![];
+        assert!(validate_process_spec(&Some(invalid_process)).is_err());
+        let mut invalid_process = valid_process.clone();
+        invalid_process.cwd = "".to_string();
+        assert!(validate_process_spec(&Some(invalid_process)).is_err());
+        let mut invalid_process = valid_process;
+        invalid_process.cwd = "test/".to_string();
+        assert!(validate_process_spec(&Some(invalid_process)).is_err());
+    }
 }

src/tools/runk/src/commands/create.rs

@@ -4,34 +4,24 @@
 //
 
 use anyhow::Result;
-use libcontainer::{builder::ContainerBuilder, container::ContainerAction};
+use libcontainer::{builder::InitContainerBuilder, container::ContainerAction};
 use liboci_cli::Create;
-use nix::unistd::Pid;
 use slog::{info, Logger};
-use std::{fs, path::Path};
+use std::path::Path;
 
 pub async fn run(opts: Create, root: &Path, logger: &Logger) -> Result<()> {
-    let ctx = ContainerBuilder::default()
+    let mut launcher = InitContainerBuilder::default()
         .id(opts.container_id)
         .bundle(opts.bundle)
         .root(root.to_path_buf())
         .console_socket(opts.console_socket)
+        .pid_file(opts.pid_file)
         .build()?
-        .create_ctx()?;
+        .create_launcher(logger)?;
 
-    let pid = ctx.launch(ContainerAction::Create, logger).await?;
-
-    if let Some(ref pid_file) = opts.pid_file {
-        create_pid_file(pid_file, pid)?;
-    }
+    launcher.launch(ContainerAction::Create, logger).await?;
 
     info!(&logger, "create command finished successfully");
 
     Ok(())
 }
-
-fn create_pid_file<P: AsRef<Path>>(pid_file: P, pid: Pid) -> Result<()> {
-    fs::write(pid_file.as_ref(), format!("{}", pid))?;
-
-    Ok(())
-}

src/tools/runk/src/commands/exec.rs

@@ -0,0 +1,32 @@
+// Copyright 2021-2022 Kata Contributors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+use anyhow::Result;
+use libcontainer::builder::ActivatedContainerBuilder;
+use libcontainer::container::ContainerAction;
+use liboci_cli::Exec;
+use slog::{info, Logger};
+use std::path::Path;
+
+pub async fn run(opts: Exec, root: &Path, logger: &Logger) -> Result<()> {
+    let mut launcher = ActivatedContainerBuilder::default()
+        .id(opts.container_id)
+        .root(root.to_path_buf())
+        .console_socket(opts.console_socket)
+        .pid_file(opts.pid_file)
+        .tty(opts.tty)
+        .cwd(opts.cwd)
+        .env(opts.env)
+        .no_new_privs(opts.no_new_privs)
+        .process(opts.process)
+        .args(opts.command)
+        .build()?
+        .create_launcher(logger)?;
+
+    launcher.launch(ContainerAction::Run, logger).await?;
+
+    info!(&logger, "exec command finished successfully");
+    Ok(())
+}

src/tools/runk/src/commands/mod.rs

@@ -5,6 +5,7 @@
 
 pub mod create;
 pub mod delete;
+pub mod exec;
 pub mod kill;
 pub mod list;
 pub mod run;

src/tools/runk/src/commands/run.rs

@@ -4,21 +4,22 @@
 //
 
 use anyhow::Result;
-use libcontainer::{builder::ContainerBuilder, container::ContainerAction};
+use libcontainer::{builder::InitContainerBuilder, container::ContainerAction};
 use liboci_cli::Run;
 use slog::{info, Logger};
 use std::path::Path;
 
 pub async fn run(opts: Run, root: &Path, logger: &Logger) -> Result<()> {
-    let ctx = ContainerBuilder::default()
+    let mut launcher = InitContainerBuilder::default()
         .id(opts.container_id)
         .bundle(opts.bundle)
         .root(root.to_path_buf())
         .console_socket(opts.console_socket)
+        .pid_file(opts.pid_file)
         .build()?
-        .create_ctx()?;
+        .create_launcher(logger)?;
 
-    ctx.launch(ContainerAction::Run, logger).await?;
+    launcher.launch(ContainerAction::Run, logger).await?;
 
     info!(&logger, "run command finished successfully");
 

src/tools/runk/src/main.rs

@@ -79,6 +79,7 @@ async fn cmd_run(subcmd: SubCommand, root_path: &Path, logger: &Logger) -> Resul
             CommonCmd::Run(run) => commands::run::run(run, root_path, logger).await,
             CommonCmd::Spec(spec) => commands::spec::run(spec, logger),
             CommonCmd::List(list) => commands::list::run(list, root_path, logger),
+            CommonCmd::Exec(exec) => commands::exec::run(exec, root_path, logger).await,
             _ => {
                 return Err(anyhow!("command is not implemented yet"));
             }