bump to krustlet 0.5

Signed-off-by: Matthew Fisher <[email protected]>

Cargo.toml

@@ -12,11 +12,14 @@ async-trait = "0.1"
 chrono = { version = "0.4", features = ["serde"] }
 env_logger = "0.7"
 futures = "0.3"
-k8s-openapi = { version = "0.7", features = ["v1_17"] }
-kube = "0.33"
-kubelet = {version = "0.2", features = ["cli"]}
+k8s-openapi = { version = "0.9", default-features = false, features = ["v1_17"] }
+kube = { version= "0.40", default-features = false, features = ["native-tls"] }
+kubelet = "0.5"
 log = "0.4"
-oci-distribution = "0.1"
+oci-distribution = "0.4"
+serde = "1.0"
+serde_derive = "1.0"
+serde_json = "1.0"
 tempfile = "3.1"
 tokio = { version = "0.2", features = ["fs", "stream", "macros", "io-util", "sync"] }
 wasm3 = { git = "https://github.com/Veykril/wasm3-rs.git", features = ["wasi"] }

README.md

@@ -1,3 +1,12 @@
 # krustlet-wasm3
 
 [Krustlet](https://github.com/deislabs/krustlet) provider for the [wasm3](https://github.com/wasm3/wasm3) runtime.
+
+## Prerequisites
+
+[Install Clang 3.9](https://rust-lang.github.io/rust-bindgen/requirements.html#installing-clang-39) and [rust-bindgen](https://rust-lang.github.io/rust-bindgen) prior to running `cargo build`:
+
+```console
+$ apt install llvm-dev libclang-dev clang
+$ cargo install bindgen
+```

src/provider/runtime.rs → runtime.rs

@@ -5,15 +5,16 @@ use tempfile::NamedTempFile;
 use tokio::sync::watch::{self, Sender};
 use tokio::task::JoinHandle;
 use wasm3::{Environment, Module};
-use kubelet::handle::{RuntimeHandle, Stop};
-use kubelet::status::ContainerStatus;
+use kubelet::container::Handle as ContainerHandle;
+use kubelet::container::Status as ContainerStatus;
+use kubelet::handle::StopHandler;
 
-pub struct HandleStopper {
+pub struct Runtime {
     handle: JoinHandle<anyhow::Result<()>>,
 }
 
 #[async_trait::async_trait]
-impl Stop for HandleStopper {
+impl StopHandler for Runtime {
     async fn stop(&mut self) -> anyhow::Result<()> {
         // no nothing
         Ok(())
@@ -26,13 +27,13 @@ impl Stop for HandleStopper {
 }
 
 /// A runtime context for running a wasm module with wasm3
-pub struct Runtime {
+pub struct Wasm3Runtime {
     module_bytes: Vec<u8>,
     stack_size: u32,
     output: Arc<NamedTempFile>,
 }
 
-impl Runtime {
+impl Wasm3Runtime {
     pub async fn new<L: AsRef<Path> + Send + Sync + 'static>(module_bytes: Vec<u8>, stack_size: u32, log_dir: L) -> anyhow::Result<Self> {
         let temp = tokio::task::spawn_blocking(move || -> anyhow::Result<NamedTempFile> {
             Ok(NamedTempFile::new_in(log_dir)?)
@@ -46,7 +47,7 @@ impl Runtime {
         })
     }
 
-    pub async fn start(&mut self) -> anyhow::Result<RuntimeHandle<HandleStopper, LogHandleFactory>> {
+    pub async fn start(&mut self) -> anyhow::Result<ContainerHandle<Runtime, LogHandleFactory>> {
         let temp = self.output.clone();
         let output_write = tokio::task::spawn_blocking(move || -> anyhow::Result<std::fs::File> {
             Ok(temp.reopen()?)
@@ -64,10 +65,9 @@ impl Runtime {
             temp: self.output.clone(),
         };
 
-        Ok(RuntimeHandle::new(
-            HandleStopper{handle},
+        Ok(ContainerHandle::new(
+            Runtime{handle},
             log_handle_factory,
-            status_recv,
         ))
     }
 }
@@ -77,7 +77,7 @@ pub struct LogHandleFactory {
     temp: Arc<NamedTempFile>,
 }
 
-impl kubelet::handle::LogHandleFactory<tokio::fs::File> for LogHandleFactory {
+impl kubelet::log::HandleFactory<tokio::fs::File> for LogHandleFactory {
     /// Creates `tokio::fs::File` on demand for log reading.
     fn new_handle(&self) -> tokio::fs::File {
         tokio::fs::File::from_std(self.temp.reopen().unwrap())
@@ -94,20 +94,46 @@ async fn spawn_wasm3(
     _output_write: std::fs::File, //TODO: hook this up such that log output will be written to the file
 ) -> anyhow::Result<JoinHandle<anyhow::Result<()>>> {
     let handle = tokio::task::spawn_blocking(move || -> anyhow::Result<_> {
-        let env = Environment::new().expect("cannot create environment");
+
+        let env = match Environment::new() {
+            // We can't map errors here or it moves the send channel, so we
+            // do it in a match
+            Ok(m) => m,
+            Err(e) => {
+                let message = "cannot create environment";
+                error!("{}: {:?}", message, e);
+                send(
+                    status_sender.clone(),
+                    name,
+                    Status::Terminated {
+                        failed: true,
+                        message: message.into(),
+                        timestamp: chrono::Utc::now(),
+                    },
+                    &mut cx,
+                );
+                return Err(e);
+            }
+        }
+
         let rt = env.create_runtime(stack_size).expect("cannot create runtime");
+
         let module = Module::parse(&env, &module_bytes).expect("cannot parse module");
+
         let mut module = rt.load_module(module).expect("cannot load module");
+
         module.link_wasi().expect("cannot link WASI");
+
         let func = module.find_function::<(), ()>("_start").expect("cannot find function '_start' in module");
+
         func.call().expect("cannot call '_start' in module");
-        status_sender
-        .broadcast(ContainerStatus::Terminated {
+
+        status_sender.broadcast(ContainerStatus::Terminated {
             failed: false,
             message: "Module run completed".into(),
             timestamp: chrono::Utc::now(),
-        })
-        .expect("status should be able to send");
+        }).expect("status should be able to send");
+
         Ok(())
     });
 

src/lib.rs

@@ -0,0 +1,170 @@
+//! A custom kubelet backend that can run [WASI](https://wasi.dev/) based workloads
+//!
+//! The crate provides the [`WasiProvider`] type which can be used
+//! as a provider with [`kubelet`].
+//!
+//! # Example
+//! ```rust,no_run
+//! use kubelet::{Kubelet, config::Config};
+//! use kubelet::store::oci::FileStore;
+//! use std::sync::Arc;
+//! use wasi_provider::WasiProvider;
+//!
+//! async {
+//!     // Get a configuration for the Kubelet
+//!     let kubelet_config = Config::default();
+//!     let client = oci_distribution::Client::default();
+//!     let store = Arc::new(FileStore::new(client, &std::path::PathBuf::from("")));
+//!
+//!     // Load a kubernetes configuration
+//!     let kubeconfig = kube::Config::infer().await.unwrap();
+//!
+//!     // Instantiate the provider type
+//!     let provider = WasiProvider::new(store, &kubelet_config, kubeconfig.clone()).await.unwrap();
+//!
+//!     // Instantiate the Kubelet
+//!     let kubelet = Kubelet::new(provider, kubeconfig, kubelet_config).await.unwrap();
+//!     // Start the Kubelet and block on it
+//!     kubelet.start().await.unwrap();
+//! };
+//! ```
+
+#![deny(missing_docs)]
+
+mod wasi_runtime;
+
+use std::collections::HashMap;
+use std::path::PathBuf;
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use kubelet::node::Builder;
+use kubelet::pod::{key_from_pod, pod_key, Handle, Pod};
+use kubelet::provider::{Provider, ProviderError};
+use kubelet::store::Store;
+use kubelet::volume::Ref;
+use tokio::sync::mpsc::{self, Receiver, Sender};
+use tokio::sync::RwLock;
+use wasi_runtime::Runtime;
+
+mod states;
+
+use states::registered::Registered;
+use states::terminated::Terminated;
+
+const TARGET_WASM32_WASI: &str = "wasm32-wasi";
+const LOG_DIR_NAME: &str = "wasi-logs";
+const VOLUME_DIR: &str = "volumes";
+
+/// WasiProvider provides a Kubelet runtime implementation that executes WASM
+/// binaries conforming to the WASI spec.
+#[derive(Clone)]
+pub struct WasiProvider {
+    shared: SharedPodState,
+}
+
+#[derive(Clone)]
+struct SharedPodState {
+    handles: Arc<RwLock<HashMap<String, Handle<Runtime, wasi_runtime::HandleFactory>>>>,
+    store: Arc<dyn Store + Sync + Send>,
+    log_path: PathBuf,
+    kubeconfig: kube::Config,
+    volume_path: PathBuf,
+}
+
+impl WasiProvider {
+    /// Create a new wasi provider from a module store and a kubelet config
+    pub async fn new(
+        store: Arc<dyn Store + Sync + Send>,
+        config: &kubelet::config::Config,
+        kubeconfig: kube::Config,
+    ) -> anyhow::Result<Self> {
+        let log_path = config.data_dir.join(LOG_DIR_NAME);
+        let volume_path = config.data_dir.join(VOLUME_DIR);
+        tokio::fs::create_dir_all(&log_path).await?;
+        tokio::fs::create_dir_all(&volume_path).await?;
+        Ok(Self {
+            shared: SharedPodState {
+                handles: Default::default(),
+                store,
+                log_path,
+                volume_path,
+                kubeconfig,
+            },
+        })
+    }
+}
+
+struct ModuleRunContext {
+    modules: HashMap<String, Vec<u8>>,
+    volumes: HashMap<String, Ref>,
+    status_sender: Sender<(String, kubelet::container::Status)>,
+    status_recv: Receiver<(String, kubelet::container::Status)>,
+}
+
+/// State that is shared between pod state handlers.
+pub struct PodState {
+    key: String,
+    run_context: ModuleRunContext,
+    errors: usize,
+    shared: SharedPodState,
+}
+
+// No cleanup state needed, we clean up when dropping PodState.
+#[async_trait]
+impl kubelet::state::AsyncDrop for PodState {
+    async fn async_drop(self) {
+        {
+            let mut handles = self.shared.handles.write().await;
+            handles.remove(&self.key);
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl Provider for WasiProvider {
+    type InitialState = Registered;
+    type TerminatedState = Terminated;
+    type PodState = PodState;
+
+    const ARCH: &'static str = TARGET_WASM32_WASI;
+
+    async fn node(&self, builder: &mut Builder) -> anyhow::Result<()> {
+        builder.set_architecture("wasm-wasi");
+        builder.add_taint("NoExecute", "kubernetes.io/arch", Self::ARCH);
+        Ok(())
+    }
+
+    async fn initialize_pod_state(&self, pod: &Pod) -> anyhow::Result<Self::PodState> {
+        let (tx, rx) = mpsc::channel(pod.all_containers().len());
+        let run_context = ModuleRunContext {
+            modules: Default::default(),
+            volumes: Default::default(),
+            status_sender: tx,
+            status_recv: rx,
+        };
+        let key = key_from_pod(pod);
+        Ok(PodState {
+            key,
+            run_context,
+            errors: 0,
+            shared: self.shared.clone(),
+        })
+    }
+
+    async fn logs(
+        &self,
+        namespace: String,
+        pod_name: String,
+        container_name: String,
+        sender: kubelet::log::Sender,
+    ) -> anyhow::Result<()> {
+        let mut handles = self.shared.handles.write().await;
+        let handle = handles
+            .get_mut(&pod_key(&namespace, &pod_name))
+            .ok_or_else(|| ProviderError::PodNotFound {
+                pod_name: pod_name.clone(),
+            })?;
+        handle.output(&container_name, sender).await
+    }
+}