Add functionality to stop and restart processes (#25)

Add functionality to stop processes, which in turn enables restarting on recreated pods.
Also contains change to add pod-uid to the config directory name.

Author: soenkeliebau

src/provider/error.rs

@@ -43,4 +43,6 @@ pub enum StackableError {
         "The following config maps were specified in a pod but not found: {missing_config_maps:?}"
     )]
     MissingConfigMapsError { missing_config_maps: Vec<String> },
+    #[error("Object is missing key: {key}")]
+    MissingObjectKey { key: &'static str },
 }

src/provider/mod.rs

@@ -13,7 +13,9 @@ use kubelet::provider::Provider;
 use log::{debug, error};
 
 use crate::provider::error::StackableError;
-use crate::provider::error::StackableError::{CrdMissing, KubeError, PodValidationError};
+use crate::provider::error::StackableError::{
+    CrdMissing, KubeError, MissingObjectKey, PodValidationError,
+};
 use crate::provider::repository::package::Package;
 use crate::provider::states::downloading::Downloading;
 use crate::provider::states::terminated::Terminated;
@@ -40,13 +42,15 @@ pub struct PodState {
     log_directory: PathBuf,
     package_download_backoff_strategy: ExponentialBackoffStrategy,
     service_name: String,
+    service_uid: String,
     package: Package,
     process_handle: Option<Child>,
 }
 
 impl PodState {
     pub fn get_service_config_directory(&self) -> PathBuf {
-        self.config_directory.join(&self.service_name)
+        self.config_directory
+            .join(format!("{}-{}", &self.service_name, &self.service_uid))
     }
 
     pub fn get_service_package_directory(&self) -> PathBuf {
@@ -153,6 +157,17 @@ impl Provider for StackableProvider {
 
     async fn initialize_pod_state(&self, pod: &Pod) -> anyhow::Result<Self::PodState> {
         let service_name = pod.name();
+
+        // Extract uid from pod object, if this fails we return an error -
+        // this should not happen, as all objects we get from Kubernetes should have
+        // a uid set!
+        let service_uid = if let Some(uid) = pod.as_kube_pod().metadata.uid.as_ref() {
+            uid.to_string()
+        } else {
+            return Err(anyhow::Error::new(MissingObjectKey {
+                key: ".metadata.uid",
+            }));
+        };
         let parcel_directory = self.parcel_directory.clone();
         // TODO: make this configurable
         let download_directory = parcel_directory.join("_download");
@@ -175,6 +190,7 @@ impl Provider for StackableProvider {
             config_directory: self.config_directory.clone(),
             package_download_backoff_strategy: ExponentialBackoffStrategy::default(),
             service_name: String::from(service_name),
+            service_uid,
             package,
             process_handle: None,
         })

src/provider/states/running.rs

@@ -1,5 +1,6 @@
-use std::process::Child;
-
+use k8s_openapi::api::core::v1::{
+    ContainerState, ContainerStateRunning, ContainerStatus as KubeContainerStatus,
+};
 use kubelet::pod::Pod;
 use kubelet::state::prelude::*;
 use kubelet::state::{State, Transition};
@@ -9,15 +10,10 @@ use crate::provider::states::failed::Failed;
 use crate::provider::states::installing::Installing;
 use crate::provider::states::stopping::Stopping;
 use crate::provider::PodState;
-use k8s_openapi::api::core::v1::{
-    ContainerState, ContainerStateRunning, ContainerStatus as KubeContainerStatus,
-};
 
 #[derive(Debug, TransitionTo)]
 #[transition_to(Stopping, Failed, Running, Installing)]
-pub struct Running {
-    pub process_handle: Option<Child>,
-}
+pub struct Running {}
 
 #[async_trait::async_trait]
 impl State<PodState> for Running {
@@ -26,17 +22,32 @@ impl State<PodState> for Running {
         pod_state: &mut PodState,
         _pod: &Pod,
     ) -> Transition<PodState> {
-        debug!("waiting");
-        let mut handle = std::mem::replace(&mut self.process_handle, None).unwrap();
-
         loop {
             tokio::select! {
                 _ = tokio::time::delay_for(std::time::Duration::from_secs(1))  => {
                     trace!("Checking if service {} is still running.", &pod_state.service_name);
                 }
             }
-            match handle.try_wait() {
-                Ok(None) => debug!("Service {} is still running", &pod_state.service_name),
+
+            // Obtain a mutable reference to the process handle
+            let child = if let Some(testproc) = pod_state.process_handle.as_mut() {
+                testproc
+            } else {
+                return Transition::next(
+                    self,
+                    Failed {
+                        message: "Unable to obtain process handle from podstate!".to_string(),
+                    },
+                );
+            };
+
+            // Check if an exit code is available for the process - if yes, it exited
+            match child.try_wait() {
+                Ok(None) => debug!(
+                    "Service {} is still running with pid {}",
+                    &pod_state.service_name,
+                    child.id()
+                ),
                 _ => {
                     error!(
                         "Service {} died unexpectedly, moving to failed state",

src/provider/states/starting.rs

@@ -115,13 +115,10 @@ impl State<PodState> for Starting {
                                 );
                             }
                         }
-                        //pod_state.process_handle = Some(child);
-                        return Transition::next(
-                            self,
-                            Running {
-                                process_handle: Some(child),
-                            },
-                        );
+                        // Store the child handle in the podstate so that later states
+                        // can use it
+                        pod_state.process_handle = Some(child);
+                        return Transition::next(self, Running {});
                     }
                     Err(error) => {
                         let error_message = format!("Failed to start process with error {}", error);

src/provider/states/stopping.rs

@@ -15,10 +15,10 @@ pub struct Stopping;
 impl State<PodState> for Stopping {
     async fn next(self: Box<Self>, pod_state: &mut PodState, _pod: &Pod) -> Transition<PodState> {
         if let Some(child) = &pod_state.process_handle {
-            let pid = child.id();
             info!(
                 "Received stop command for service {}, stopping process with pid {}",
-                pod_state.service_name, pid
+                pod_state.service_name,
+                child.id()
             );
         }
         Transition::next(self, Stopped)

src/provider/states/terminated.rs

@@ -1,5 +1,6 @@
+use anyhow::anyhow;
 use kubelet::state::prelude::*;
-use log::info;
+use log::{debug, error, info};
 
 use crate::provider::PodState;
 
@@ -12,8 +13,31 @@ pub struct Terminated {
 #[async_trait::async_trait]
 impl State<PodState> for Terminated {
     async fn next(self: Box<Self>, pod_state: &mut PodState, _pod: &Pod) -> Transition<PodState> {
-        info!("Service {} was terminated!", &pod_state.service_name);
-        Transition::Complete(Ok(()))
+        info!(
+            "Pod {} was terminated, stopping process!",
+            &pod_state.service_name
+        );
+        // Obtain a mutable reference to the process handle
+        let child = if let Some(testproc) = pod_state.process_handle.as_mut() {
+            testproc
+        } else {
+            return Transition::Complete(Err(anyhow!("Unable to retrieve process handle")));
+        };
+
+        return match child.kill() {
+            Ok(()) => {
+                debug!("Successfully killed process {}", pod_state.service_name);
+                Transition::Complete(Ok(()))
+            }
+            Err(e) => {
+                error!(
+                    "Failed to stop process with pid {} due to: {:?}",
+                    child.id(),
+                    e
+                );
+                Transition::Complete(Err(anyhow::Error::new(e)))
+            }
+        };
     }
 
     async fn json_status(