working

Signed-off-by: Matthew Fisher <matt.fisher@microsoft.com>

Author: Matthew Fisher

src/main.rs

@@ -3,26 +3,21 @@ use kubelet::module_store::FileModuleStore;
 use kubelet::Kubelet;
 
 mod provider;
-
 use provider::Provider;
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
-    // The provider is responsible for all the "back end" logic. If you are creating
-    // a new Kubelet, all you need to implement is a provider.
     let config = Config::new_from_flags(env!("CARGO_PKG_VERSION"));
-
     let kubeconfig = kube::Config::infer().await?;
 
-    // Initialize the logger
     env_logger::init();
 
     let client = oci_distribution::Client::default();
     let mut module_store_path = config.data_dir.join(".oci");
     module_store_path.push("modules");
     let store = FileModuleStore::new(client, &module_store_path);
 
-    let provider = Provider::new(store);
+    let provider = Provider::new(store, kubeconfig.clone());
     let kubelet = Kubelet::new(provider, kubeconfig, config);
     kubelet.start().await
 }

src/provider/mod.rs

@@ -28,11 +28,15 @@
 //! };
 //! ```
 
+use k8s_openapi::api::core::v1::Pod as KubePod;
+use kube::api::DeleteParams;
+use kube::Api;
+use kube::Config as KubeConfig;
 use kubelet::handle::{key_from_pod, pod_key};
 use kubelet::module_store::ModuleStore;
 use kubelet::provider::ProviderError;
 use kubelet::Pod;
-use log::{debug, info};
+use log::{debug, error, info, trace};
 use std::collections::HashMap;
 use std::sync::Arc;
 use tokio::sync::RwLock;
@@ -47,14 +51,16 @@ const TARGET_WASM32_WASI: &str = "wasm32-wasi";
 pub struct Provider<S> {
     pods: Arc<RwLock<HashMap<String, HashMap<String, Runtime>>>>,
     store: S,
+    kubeconfig: KubeConfig,
 }
 
 impl<S: ModuleStore + Send + Sync> Provider<S> {
     /// Create a new wasi provider from a module store and a kubelet config
-    pub fn new(store: S) -> Self {
+    pub fn new(store: S, kubeconfig: KubeConfig) -> Self {
         Self {
             pods: Default::default(),
             store,
+            kubeconfig,
         }
     }
 }
@@ -79,7 +85,8 @@ impl<S: ModuleStore + Send + Sync> kubelet::Provider for Provider<S> {
                 .remove(&container.name)
                 .expect("FATAL ERROR: module map not properly populated");
 
-            let mut runtime = Runtime::new(module_data, 1 as u32);
+            // TODO: expose this as a feature flag (--stack-size)
+            let mut runtime = Runtime::new(module_data, (1024 * 60) as u32);
 
             debug!("Starting container {} on thread", container.name);
             runtime.start()?;
@@ -100,12 +107,75 @@ impl<S: ModuleStore + Send + Sync> kubelet::Provider for Provider<S> {
         Ok(())
     }
 
-    async fn modify(&self, _pod: Pod) -> anyhow::Result<()> {
-        unimplemented!()
+    async fn modify(&self, pod: Pod) -> anyhow::Result<()> {
+        // The only things we care about are:
+        // 1. metadata.deletionTimestamp => signal all containers to stop and then mark them
+        //    as terminated
+        // 2. spec.containers[*].image, spec.initContainers[*].image => stop the currently
+        //    running containers and start new ones?
+        // 3. spec.activeDeadlineSeconds => Leaving unimplemented for now
+        // TODO: Determine what the proper behavior should be if labels change
+        debug!(
+            "Got pod modified event for {} in namespace {}",
+            pod.name(),
+            pod.namespace()
+        );
+        trace!("Modified pod spec: {:#?}", pod.as_kube_pod());
+        if let Some(_timestamp) = pod.deletion_timestamp() {
+            let mut pods = self.pods.write().await;
+            match pods.get_mut(&key_from_pod(&pod)) {
+                Some(h) => {
+                    for (_name, runtime) in h {
+                        runtime.stop()?;
+                    }
+                    // Follow up with a delete when everything is stopped
+                    let dp = DeleteParams {
+                        grace_period_seconds: Some(0),
+                        ..Default::default()
+                    };
+                    let pod_client: Api<KubePod> = Api::namespaced(
+                        kube::client::Client::new(self.kubeconfig.clone()),
+                        pod.namespace(),
+                    );
+                    match pod_client.delete(pod.name(), &dp).await {
+                        Ok(_) => Ok(()),
+                        Err(e) => Err(e.into()),
+                    }
+                }
+                None => {
+                    // This isn't an error with the pod, so don't return an error (otherwise it will
+                    // get updated in its status). This is an unlikely case to get into and means
+                    // that something is likely out of sync, so just log the error
+                    error!(
+                        "Unable to find pod {} in namespace {} when trying to stop all containers",
+                        pod.name(),
+                        pod.namespace()
+                    );
+                    Ok(())
+                }
+            }
+        } else {
+            Ok(())
+        }
+        // TODO: Implement behavior for stopping old containers and restarting when the container
+        // image changes
     }
 
-    async fn delete(&self, _pod: Pod) -> anyhow::Result<()> {
-        unimplemented!()
+    async fn delete(&self, pod: Pod) -> anyhow::Result<()> {
+        let mut pods = self.pods.write().await;
+        match pods.remove(&key_from_pod(&pod)) {
+            Some(_) => debug!(
+                "Pod {} in namespace {} removed",
+                pod.name(),
+                pod.namespace()
+            ),
+            None => info!(
+                "unable to find pod {} in namespace {}, it was likely already deleted",
+                pod.name(),
+                pod.namespace()
+            ),
+        }
+        Ok(())
     }
 
     async fn logs(
@@ -116,7 +186,7 @@ impl<S: ModuleStore + Send + Sync> kubelet::Provider for Provider<S> {
         _sender: kubelet::LogSender,
     ) -> anyhow::Result<()> {
         let mut pods = self.pods.write().await;
-        let _pod = pods
+        let _containers = pods
             .get_mut(&pod_key(&namespace, &pod_name))
             .ok_or_else(|| ProviderError::PodNotFound {
                 pod_name: pod_name.clone(),

src/provider/runtime.rs

@@ -1,7 +1,7 @@
 use std::error;
 use std::fmt;
 
-use wasm3::{Environment, Module};
+use wasm3::Environment;
 
 type Result<T> = std::result::Result<T, RuntimeError>;
 
@@ -14,9 +14,9 @@ pub struct RuntimeError {
 enum RuntimeErrorKind {
     AlreadyStarted,
     CannotCreateRuntime,
-    CannotParseModule,
+    CannotLinkWASI,
     CannotLoadModule,
-    NoMainFunction,
+    NoEntrypoint,
     RunFailure,
 }
 
@@ -35,9 +35,9 @@ impl RuntimeError {
         match self.kind {
             RuntimeErrorKind::AlreadyStarted => "runtime already started",
             RuntimeErrorKind::CannotCreateRuntime => "cannot create runtime",
-            RuntimeErrorKind::CannotParseModule => "cannot parse module",
+            RuntimeErrorKind::CannotLinkWASI => "cannot link module to the WASI runtime",
             RuntimeErrorKind::CannotLoadModule => "cannot load module",
-            RuntimeErrorKind::NoMainFunction => "no function called 'main' found",
+            RuntimeErrorKind::NoEntrypoint => "no entrypoint function called '_start' found",
             RuntimeErrorKind::RunFailure => "failure during function call",
         }
     }
@@ -81,14 +81,13 @@ impl Runtime {
         let rt = env
             .create_runtime(self.stack_size)
             .map_err(|_| RuntimeError::new(RuntimeErrorKind::CannotCreateRuntime))?;
-        let module = Module::parse(&env, &self.module_bytes)
-            .map_err(|_| RuntimeError::new(RuntimeErrorKind::CannotParseModule))?;
-        let module = rt
-            .load_module(module)
+        let mut module = rt
+            .parse_and_load_module(&self.module_bytes)
             .map_err(|_| RuntimeError::new(RuntimeErrorKind::CannotLoadModule))?;
+        module.link_wasi().map_err(|_| RuntimeError::new(RuntimeErrorKind::CannotLinkWASI))?;
         let func = module
-            .find_function::<(), ()>("main")
-            .map_err(|_| RuntimeError::new(RuntimeErrorKind::NoMainFunction))?;
+            .find_function::<(), ()>("_start")
+            .map_err(|_| RuntimeError::new(RuntimeErrorKind::NoEntrypoint))?;
         self.current_status = RuntimeStatus::Running;
         // FIXME: run this in the background
         // for now, we block until the function is complete, then call .stop()
@@ -97,7 +96,7 @@ impl Runtime {
         self.stop()
     }
 
-    fn stop(&mut self) -> Result<()> {
+    pub fn stop(&mut self) -> Result<()> {
         // it is OK for the runtime to stop an already stopped module. Effectively a no-op
         self.current_status = RuntimeStatus::Stopped;
         Ok(())