Adds monitoring of systemd units (#53)

Added monitoring of systemd units to notice when they fail.

During startup, every unit will be checked once every second for 10 seconds, after this it will be considered as "running".

When a unit is considered as "running" the state will be checked once every 10 seconds.

This is a simpler version of what we want to provide in a final implementation, where we listen to changes on dbus instead of polling periodically.

fixes #10

Author: soenkeliebau

Cargo.lock

@@ -1,17 +1,19 @@
+use anyhow::anyhow;
 use k8s_openapi::api::core::v1::{
     ContainerState, ContainerStateRunning, ContainerStatus as KubeContainerStatus, PodCondition,
 };
 use kubelet::pod::Pod;
 use kubelet::state::prelude::*;
 use kubelet::state::{State, Transition};
-use log::{debug, trace};
+use log::{debug, info, trace};
 
 use crate::provider::states::failed::Failed;
 use crate::provider::states::installing::Installing;
 use crate::provider::states::make_status_with_containers_and_condition;
 use crate::provider::PodState;
 use k8s_openapi::apimachinery::pkg::apis::meta::v1::Time;
 use k8s_openapi::chrono;
+use tokio::time::Duration;
 
 #[derive(Debug, TransitionTo)]
 #[transition_to(Failed, Running, Installing)]
@@ -34,17 +36,57 @@ impl State<PodState> for Running {
         pod_state: &mut PodState,
         _pod: &Pod,
     ) -> Transition<PodState> {
+        // We loop here indefinitely and "wake up" periodically to check if the service is still
+        // up and running
+        // Interruption of this loop is triggered externally by the Krustlet code when
+        //   - the pod which this state machine refers to gets deleted
+        //   - Krustlet shuts down
         loop {
-            tokio::select! {
-                _ = tokio::time::delay_for(std::time::Duration::from_secs(10))  => {
-                    trace!("Checking if service {} is still running.", &pod_state.service_name);
+            tokio::time::delay_for(Duration::from_secs(10)).await;
+            trace!(
+                "Checking if service {} is still running.",
+                &pod_state.service_name
+            );
+
+            // Iterate over all units and check their state
+            // if the [`service_units`] Option is a None variant, return a failed state
+            // as we need to run something otherwise we are not doing anything
+            let systemd_units = match &pod_state.service_units {
+                Some(units) => units,
+                None => return Transition::Complete(Err(anyhow!(format!("No systemd units found for service [{}], this should not happen, please report a bug for this!", pod_state.service_name)))),
+            };
+
+            for unit in systemd_units {
+                match pod_state.systemd_manager.is_running(&unit.get_name()) {
+                    Ok(true) => trace!(
+                        "Unit [{}] of service [{}] still running ...",
+                        &unit.get_name(),
+                        pod_state.service_name
+                    ),
+                    Ok(false) => {
+                        info!("Unit [{}] for service [{}] failed unexpectedly, transitioning to failed state.", pod_state.service_name, unit.get_name());
+                        return Transition::next(
+                            self,
+                            Failed {
+                                message: "".to_string(),
+                            },
+                        );
+                    }
+                    Err(dbus_error) => {
+                        info!(
+                            "Error querying ActiveState for Unit [{}] of service [{}]: [{}].",
+                            pod_state.service_name,
+                            unit.get_name(),
+                            dbus_error
+                        );
+                        return Transition::Complete(Err(dbus_error));
+                    }
                 }
             }
-            // TODO: We are not watching the service yet, need to subscribe to events and
-            // react to those
         }
     }
 
+    // test
     async fn json_status(
         &self,
         pod_state: &mut PodState,

src/provider/states/running.rs

@@ -6,7 +6,10 @@ use crate::provider::states::failed::Failed;
 use crate::provider::states::running::Running;
 use crate::provider::states::setup_failed::SetupFailed;
 use crate::provider::PodState;
-use log::{error, info, warn};
+use anyhow::anyhow;
+use log::{debug, error, info, warn};
+use std::time::Instant;
+use tokio::time::Duration;
 
 #[derive(Default, Debug, TransitionTo)]
 #[transition_to(Running, Failed, SetupFailed)]
@@ -36,6 +39,37 @@ impl State<PodState> for Starting {
                     );
                     return Transition::Complete(Err(enable_error));
                 }
+
+                let start_time = Instant::now();
+                // TODO: does this need to be configurable, or ar we happy with a hard coded value
+                //  for now. I've briefly looked at the podspec and couldn't identify a good field
+                //  to use for this - also, currently this starts containers (= systemd units) in
+                //  order and waits 10 seconds for every unit, so a service with five containers
+                //  would take 50 seconds until it reported running - which is totally fine in case
+                //  the units actually depend on each other, but a case could be made for waiting
+                //  once at the end
+                while start_time.elapsed().as_secs() < 10 {
+                    tokio::time::delay_for(Duration::from_secs(1)).await;
+                    debug!(
+                        "Checking if unit [{}] is still up and running.",
+                        &unit.get_name()
+                    );
+                    match pod_state.systemd_manager.is_running(&unit.get_name()) {
+                        Ok(true) => debug!(
+                            "Service [{}] still running after [{}] seconds",
+                            &unit.get_name(),
+                            start_time.elapsed().as_secs()
+                        ),
+                        Ok(false) => {
+                            return Transition::Complete(Err(anyhow!(format!(
+                                "Unit [{}] stopped unexpectedly during startup after [{}] seconds.",
+                                &unit.get_name(),
+                                start_time.elapsed().as_secs()
+                            ))))
+                        }
+                        Err(dbus_error) => return Transition::Complete(Err(dbus_error)),
+                    }
+                }
             }
         } else {
             warn!(

src/provider/states/starting.rs

@@ -5,7 +5,7 @@
 //!
 use crate::provider::systemdmanager::systemdunit::SystemDUnit;
 use anyhow::anyhow;
-use dbus::arg::{AppendAll, ReadAll};
+use dbus::arg::{AppendAll, ReadAll, Variant};
 use dbus::blocking::SyncConnection;
 use dbus::strings::Member;
 use dbus::Path;
@@ -25,6 +25,7 @@ pub enum UnitTypes {
 const SYSTEMD_DESTINATION: &str = "org.freedesktop.systemd1";
 const SYSTEMD_NODE: &str = "/org/freedesktop/systemd1";
 const SYSTEMD_MANAGER_INTERFACE: &str = "org.freedesktop.systemd1.Manager";
+const DBUS_PROPERTIES_INTERFACE: &str = "org.freedesktop.DBus.Properties";
 
 /// The main way of interacting with this module, this struct offers
 /// the public methods for managing service units.
@@ -171,10 +172,7 @@ impl SystemdManager {
         }
 
         let unit_file = self.units_directory.join(&unit_name);
-        if linked_unit_file
-            && unit_file.exists()
-            && unit_file.symlink_metadata()?.file_type().is_file()
-        {
+        if unit_file.exists() && unit_file.symlink_metadata()?.file_type().is_file() {
             // Handle the special case where we need to replace an actual file with a symlink
             // This only occurs when switching from writing the file
             // directly into the units folder to using a linked file - should not happen in practice
@@ -357,6 +355,28 @@ impl SystemdManager {
         }
     }
 
+    pub fn is_running(&self, unit: &str) -> Result<bool, anyhow::Error> {
+        let unit_node = self
+            .method_call("GetUnit", (&unit,))
+            .map(|r: (Path,)| r.0)?;
+
+        let proxy = self
+            .connection
+            .with_proxy(SYSTEMD_DESTINATION, &unit_node, self.timeout);
+
+        let active_state = proxy
+            .method_call(
+                DBUS_PROPERTIES_INTERFACE,
+                "Get",
+                ("org.freedesktop.systemd1.Unit", "ActiveState"),
+            )
+            .map(|r: (Variant<String>,)| r.0)?;
+
+        // TODO: I think this can panic, there should be a get() method on Variant that returns
+        //   an option, but I've not yet been able to get that to work
+        Ok(active_state.0 == "active")
+    }
+
     // Symlink a unit file into the systemd unit folder
     // This is not public on purpose, as [create] should be the normal way to link unit files
     // when using this crate