Initial shim bootstrapper implementation

Signed-off-by: Maksym Pavlenko <pavlenko.maksym@gmail.com>

crates/protos/src/lib.rs

@@ -1,8 +1,8 @@
 // Supress warning: redundant field names in struct initialization
 #![allow(clippy::redundant_field_names)]
 
-/// Propagate protobuf module we've used in this crate.
 pub use protobuf;
+pub use ttrpc;
 
 #[rustfmt::skip]
 pub mod events;

crates/shim/Cargo.toml

@@ -5,4 +5,7 @@ authors = ["Maksym Pavlenko <pavlenko.maksym@gmail.com>"]
 edition = "2018"
 
 [dependencies]
+go-flag = "0.1.0"
+thiserror = "1.0"
+
 containerd-protos = { path = "../protos" }

crates/shim/src/args.rs

@@ -0,0 +1,142 @@
+use go_flag::{self, FlagError};
+use std::ffi::OsStr;
+use thiserror::Error;
+
+/// Flags to be passed from containerd daemon to a shim binary.
+/// Reflects https://github.com/containerd/containerd/blob/master/runtime/v2/shim/shim.go#L100
+#[derive(Debug, Default)]
+pub struct Flags {
+    /// Enable debug output in logs.
+    pub debug: bool,
+    /// Namespace that owns the shim.
+    pub namespace: String,
+    /// Id of the task.
+    pub id: String,
+    /// Abstract socket path to serve.
+    pub socket: String,
+    /// Path to the bundle if not workdir.
+    pub bundle: String,
+    /// GRPC address back to main containerd.
+    pub address: String,
+    /// Path to publish binary (used for publishing events).
+    pub publish_binary: String,
+    /// Shim action (start / delete).
+    /// See https://github.com/containerd/containerd/blob/master/runtime/v2/shim/shim.go#L191
+    pub action: String,
+}
+
+#[derive(Debug, Error, PartialEq)]
+pub enum Error {
+    /// Either bad or unknown flag.
+    #[error("Invalid arg: {0}")]
+    InvalidArg(String),
+    /// Required flag is missing.
+    #[error("Missing arg: {0}")]
+    MissingArg(String),
+    /// Syntax error.
+    #[error("Parse failed: {0}")]
+    ParseError(String),
+}
+
+/// Parses command line arguments passed to the shim.
+/// This func replicates https://github.com/containerd/containerd/blob/master/runtime/v2/shim/shim.go#L110
+pub fn parse<S: AsRef<OsStr>>(args: &[S]) -> Result<Flags, Error> {
+    let mut flags = Flags::default();
+
+    let args: Vec<String> = go_flag::parse_args(args, |f| {
+        f.add_flag("debug", &mut flags.debug);
+        f.add_flag("namespace", &mut flags.namespace);
+        f.add_flag("id", &mut flags.id);
+        f.add_flag("socket", &mut flags.socket);
+        f.add_flag("bundle", &mut flags.bundle);
+        f.add_flag("address", &mut flags.address);
+        f.add_flag("publish-binary", &mut flags.publish_binary);
+    })
+    .map_err(|e| match e {
+        FlagError::BadFlag { flag } => Error::InvalidArg(flag),
+        FlagError::UnknownFlag { name } => Error::InvalidArg(name),
+        FlagError::ArgumentNeeded { name } => Error::MissingArg(name),
+        FlagError::ParseError { error } => Error::ParseError(format!("{:?}", error)),
+    })?;
+
+    if let Some(action) = args.get(0) {
+        flags.action = action.into();
+    }
+
+    if flags.namespace.is_empty() {
+        return Err(Error::MissingArg(String::from(
+            "Shim namespace cannot be empty",
+        )));
+    }
+
+    Ok(flags)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parse_all() {
+        let args = [
+            "-debug",
+            "-id",
+            "123",
+            "-namespace",
+            "default",
+            "-socket",
+            "/path/to/socket",
+            "-publish-binary",
+            "/path/to/binary",
+            "-bundle",
+            "bundle",
+            "-address",
+            "address",
+            "delete",
+        ];
+
+        let flags = parse(&args).unwrap();
+
+        assert_eq!(flags.debug, true);
+        assert_eq!(flags.id, "123");
+        assert_eq!(flags.namespace, "default");
+        assert_eq!(flags.socket, "/path/to/socket");
+        assert_eq!(flags.publish_binary, "/path/to/binary");
+        assert_eq!(flags.bundle, "bundle");
+        assert_eq!(flags.address, "address");
+        assert_eq!(flags.action, "delete");
+    }
+
+    #[test]
+    fn parse_flags() {
+        let args = ["-id", "123", "-namespace", "default"];
+
+        let flags = parse(&args).unwrap();
+
+        assert_eq!(flags.debug, false);
+        assert_eq!(flags.id, "123");
+        assert_eq!(flags.namespace, "default");
+        assert_eq!(flags.action, "");
+    }
+
+    #[test]
+    fn parse_action() {
+        let args = ["-namespace", "1", "start"];
+
+        let flags = parse(&args).unwrap();
+        assert_eq!(flags.action, "start");
+        assert_eq!(flags.id, "");
+    }
+
+    #[test]
+    fn no_namespace() {
+        let empty: [String; 0] = [];
+        let result = parse(&empty).err();
+        assert_eq!(
+            result,
+            Some(Error::MissingArg(
+                "Shim namespace cannot be empty".to_owned()
+            ))
+        )
+    }
+}

crates/shim/src/lib.rs

@@ -1,4 +1,16 @@
-use containerd_protos::shim::shim_ttrpc::Task;
+use containerd_protos::protobuf::Message;
+use containerd_protos::shim::{shim::DeleteResponse, shim_ttrpc::create_task, shim_ttrpc::Task};
+use containerd_protos::ttrpc::Server;
+use std::env;
+use std::error;
+use std::io::{self, Write};
+use std::process;
+use std::sync::Arc;
+use thiserror::Error;
+
+mod args;
+
+pub use containerd_protos as protos;
 
 #[derive(Debug, Default)]
 pub struct StartOpts {
@@ -9,15 +21,85 @@ pub struct StartOpts {
 }
 
 pub trait Shim: Task {
-    fn new(id: String, namespace: String) -> Self;
+    fn new(id: &str, namespace: &str) -> Self;
 
-    fn start_shim(opts: StartOpts);
-    fn cleanup();
+    fn start_shim(&mut self, opts: StartOpts) -> Result<String, Box<dyn error::Error>>;
+    fn cleanup(&mut self) -> Result<DeleteResponse, Box<dyn error::Error>>;
 }
 
-pub fn run<T>(id: String)
+pub fn run<T>(id: &str)
 where
-    T: Shim,
+    T: Shim + Send + Sync + 'static,
 {
-    let _shim = T::new(id, "".to_string());
+    if let Some(err) = bootstrap::<T>(id).err() {
+        eprintln!("{}: {:?}", id, err);
+        process::exit(1);
+    }
+}
+
+fn bootstrap<T>(id: &str) -> Result<(), Error>
+where
+    T: Shim + Send + Sync + 'static,
+{
+    let os_args: Vec<_> = env::args_os().collect();
+    let flags = args::parse(&os_args[1..])?;
+
+    let ttrpc_address = env::var("TTRPC_ADDRESS")?;
+
+    let mut shim = T::new(id, &flags.namespace);
+
+    match flags.action.as_str() {
+        "start" => {
+            let args = StartOpts {
+                id: id.into(),
+                containerd_binary: flags.publish_binary,
+                address: flags.address,
+                ttrpc_address,
+            };
+
+            let address = shim.start_shim(args).map_err(Error::Start)?;
+            io::stdout().lock().write_fmt(format_args!("{}", address))?;
+
+            Ok(())
+        }
+        "delete" => {
+            let response = shim.cleanup().map_err(Error::Cleanup)?;
+
+            let stdout = io::stdout();
+            let mut locked = stdout.lock();
+            response.write_to_writer(&mut locked)?;
+
+            Ok(())
+        }
+        _ => {
+            let task_service = create_task(Arc::new(Box::new(shim)));
+
+            let host = format!("unix://{}", flags.socket);
+            let mut server = Server::new().bind(&host)?.register_service(task_service);
+
+            server.start()?;
+
+            Ok(())
+        }
+    }
+}
+
+#[derive(Debug, Error)]
+pub enum Error {
+    /// Invalid command line arguments.
+    #[error("Failed to parse command line")]
+    Flags(#[from] args::Error),
+    /// TTRPC specific error.
+    #[error("TTRPC error")]
+    Ttrpc(#[from] containerd_protos::ttrpc::Error),
+    #[error("Protobuf error")]
+    Protobuf(#[from] containerd_protos::protobuf::error::ProtobufError),
+    #[error("IO error")]
+    Io(#[from] io::Error),
+    #[error("Env error")]
+    Env(#[from] env::VarError),
+    #[error("Failed to start shim")]
+    Start(Box<dyn error::Error>),
+    #[error("Shim cleanup failed")]
+    Cleanup(Box<dyn error::Error>),
 }