pod lifecycle tracking complete

Cargo.toml

@@ -30,6 +30,7 @@ cached = { version = "0.45.1", features = ["async"] }
 chrono = "0.4.26"
 clap = { version = "4.3.21", features = ["derive"] }
 futures = "0.3.28"
+itertools = "0.11.0"
 k8s-openapi = { version = "0.19.0", features = ["v1_27"] }
 paste = "1.0.14"
 reqwest = { version = "0.11.18", default-features = false, features = ["json", "rustls-tls"] }

ctrl/controller.rs

@@ -7,10 +7,7 @@ use k8s_openapi::apimachinery::pkg::apis::meta::v1 as metav1;
 use kube::runtime::controller::Action;
 use kube::ResourceExt;
 use reqwest::Url;
-use simkube::k8s::{
-    add_common_fields,
-    namespaced_name,
-};
+use simkube::k8s::add_common_fields;
 use simkube::prelude::*;
 use simkube::trace::storage;
 use tokio::time::Duration;
@@ -117,6 +114,6 @@ pub(crate) async fn reconcile(
 }
 
 pub(crate) fn error_policy(simulation: Arc<Simulation>, error: &ReconcileError, _: Arc<SimulationContext>) -> Action {
-    warn!("reconcile failed on simulation {}: {:?}", namespaced_name(simulation.deref()), error);
+    warn!("reconcile failed on simulation {}: {:?}", simulation.namespaced_name(), error);
     Action::requeue(Duration::from_secs(5 * 60))
 }

driver/main.rs

@@ -1,5 +1,4 @@
 use std::cmp::max;
-use std::collections::BTreeMap;
 use std::fs;
 use std::time::Duration;
 
@@ -15,6 +14,7 @@ use kube::api::{
 use kube::ResourceExt;
 use serde_json::json;
 use simkube::jsonutils;
+use simkube::k8s::macros::*;
 use simkube::k8s::{
     add_common_fields,
     prefixed_ns,
@@ -48,7 +48,7 @@ fn build_virtual_ns(sim_name: &str, ns_name: &str, sim_root: &SimulationRoot) ->
     let mut ns = corev1::Namespace {
         metadata: metav1::ObjectMeta {
             name: Some(ns_name.into()),
-            labels: Some(BTreeMap::from([(VIRTUAL_LABEL_KEY.into(), "true".into())])),
+            labels: klabel!(VIRTUAL_LABEL_KEY = "true"),
             ..Default::default()
         },
         ..Default::default()

lib/rust/errors.rs

@@ -8,40 +8,35 @@ pub(crate) use thiserror::Error;
 
 pub type EmptyResult = anyhow::Result<()>;
 
-macro_rules! err_impl_helper {
-    ($errtype:ident, $item:ident, String) => {
+macro_rules! err_impl {
+    (@hidden $errtype:ident, $item:ident, String) => {
         paste! {
             pub(crate) fn [<$item:snake>](in_: &str) -> anyhow::Error {
                 anyhow!{$errtype::$item(in_.into())}
             }
         }
     };
 
-    ($errtype:ident, $item:ident, $($dtype:tt)::+) => {
+    (@hidden $errtype:ident, $item:ident, $($dtype:tt)::+) => {
         paste! {
             pub(crate) fn [<$item:snake>](in_: &$($dtype)::+) -> anyhow::Error {
                 anyhow!{$errtype::$item(in_.clone())}
             }
         }
     };
-}
 
-macro_rules! err_impl {
     ($errtype:ident,
         $(#[$errinfo:meta] $item:ident($($dtype:tt)::+),)+
-    ) => (
+    ) => {
         #[derive(Debug, Error)]
         pub(crate) enum $errtype {
             $(#[$errinfo] $item($($dtype)::+)),+
         }
 
         impl $errtype {
-            $(err_impl_helper! {$errtype, $item, $($dtype)::+})+
+            $(err_impl! {@hidden $errtype, $item, $($dtype)::+})+
         }
-    )
+    };
 }
 
-pub(crate) use {
-    err_impl,
-    err_impl_helper,
-};
+pub(crate) use err_impl;

lib/rust/k8s/container.rs

@@ -0,0 +1,27 @@
+use super::*;
+
+impl StartEndTimeable for corev1::ContainerState {
+    fn start_ts(&self) -> anyhow::Result<Option<i64>> {
+        match self {
+            corev1::ContainerState { running: Some(r), terminated: None, waiting: None } => {
+                Ok(Some(r.started_at.as_ref().unwrap().0.timestamp()))
+            },
+            corev1::ContainerState { running: None, terminated: Some(t), waiting: None } => {
+                Ok(Some(t.started_at.as_ref().unwrap().0.timestamp()))
+            },
+            corev1::ContainerState { running: None, terminated: None, waiting: Some(_) } => Ok(None),
+            _ => Err(KubernetesError::malformed_container_state(self)),
+        }
+    }
+
+    fn end_ts(&self) -> anyhow::Result<Option<i64>> {
+        match self {
+            corev1::ContainerState { running: Some(_), terminated: None, waiting: None } => Ok(None),
+            corev1::ContainerState { running: None, terminated: Some(t), waiting: None } => {
+                Ok(Some(t.finished_at.as_ref().unwrap().0.timestamp()))
+            },
+            corev1::ContainerState { running: None, terminated: None, waiting: Some(_) } => Ok(None),
+            _ => Err(KubernetesError::malformed_container_state(self)),
+        }
+    }
+}

lib/rust/k8s/macros.rs

@@ -0,0 +1,10 @@
+#[macro_export]
+macro_rules! klabel {
+    ($($key:tt=$val:literal),+$(,)?) => {
+        Some(BTreeMap::from([$(($key.to_string(), $val.to_string())),+]))
+    };
+}
+
+pub use std::collections::BTreeMap;
+
+pub use klabel;

lib/rust/k8s/mod.rs

@@ -1,10 +1,58 @@
 mod apiset;
+mod container;
 mod gvk;
+pub mod macros;
+mod pod;
+mod pod_lifecycle;
 mod util;
 
 pub use apiset::*;
 pub use gvk::*;
+use k8s_openapi::api::core::v1 as corev1;
+use k8s_openapi::apimachinery::pkg::apis::meta::v1 as metav1;
 pub use util::*;
 
+use crate::errors::*;
+use crate::macros::partial_ord_eq_ref;
+
+const LAST_APPLIED_CONFIG_LABEL_KEY: &str = "kubectl.kubernetes.io/last-applied-configuration";
+const DEPL_REVISION_LABEL_KEY: &str = "deployment.kubernetes.io/revision";
+
+err_impl! {KubernetesError,
+    #[error("field not found in struct: {0}")]
+    FieldNotFound(String),
+
+    #[error("malformed container status: {0:?}")]
+    MalformedContainerState(corev1::ContainerState),
+
+    #[error("malformed label selector: {0:?}")]
+    MalformedLabelSelector(metav1::LabelSelectorRequirement),
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum PodLifecycleData {
+    Empty,
+    Running(i64),
+    Finished(i64, i64),
+}
+partial_ord_eq_ref!(PodLifecycleData);
+
+pub trait KubeResourceExt {
+    fn namespaced_name(&self) -> String;
+    fn matches(&self, sel: &metav1::LabelSelector) -> anyhow::Result<bool>;
+}
+
+pub trait PodExt {
+    fn spec(&self) -> anyhow::Result<&corev1::PodSpec>;
+    fn status(&self) -> anyhow::Result<&corev1::PodStatus>;
+    fn spec_mut(&mut self) -> &mut corev1::PodSpec;
+    fn status_mut(&mut self) -> &mut corev1::PodStatus;
+}
+
+trait StartEndTimeable {
+    fn start_ts(&self) -> anyhow::Result<Option<i64>>;
+    fn end_ts(&self) -> anyhow::Result<Option<i64>>;
+}
+
 #[cfg(test)]
-mod util_test;
+mod test;

lib/rust/k8s/pod.rs

@@ -0,0 +1,31 @@
+use super::*;
+
+impl PodExt for corev1::Pod {
+    fn spec(&self) -> anyhow::Result<&corev1::PodSpec> {
+        match self.spec.as_ref() {
+            None => bail!(KubernetesError::field_not_found("pod spec")),
+            Some(ps) => Ok(ps),
+        }
+    }
+
+    fn status(&self) -> anyhow::Result<&corev1::PodStatus> {
+        match self.status.as_ref() {
+            None => bail!(KubernetesError::field_not_found("pod status")),
+            Some(ps) => Ok(ps),
+        }
+    }
+
+    fn spec_mut(&mut self) -> &mut corev1::PodSpec {
+        if self.spec.is_none() {
+            self.spec = Some(Default::default());
+        }
+        self.spec.as_mut().unwrap()
+    }
+
+    fn status_mut(&mut self) -> &mut corev1::PodStatus {
+        if self.status.is_none() {
+            self.status = Some(Default::default());
+        }
+        self.status.as_mut().unwrap()
+    }
+}