Evict pods based on expiry annotations (#20)

## Description

This is the flip side of stackabletech/secret-operator#114, and implements the actual eviction of the pods once they expire.

- [X] Time-based eviction (`restarter.stackable.tech/expires-at.{TAG}` annotation)
- [ ] Dependency-based eviction (like the current STS eviction)

Author: nightkr

Cargo.lock

@@ -4,4 +4,4 @@
 * Concepts
 ** xref:authenticationclass.adoc[]
 ** xref:tls.adoc[]
-** xref:reloader.adoc[]
+** xref:restarter.adoc[]

docs/modules/ROOT/nav.adoc

@@ -0,0 +1,35 @@
+= Restarter
+
+The Stackable Commons Operator can automatically restart `Pod` objects based on certain criteria. This can be applied to
+either the `Pod` or certain controller objects (such as `StatefulSet`).
+
+== `Pod`
+
+Pods are evicted when any of their restart criteria (listed below) expire, with the expectation that their owning controller
+is then responsible for restarting them.
+
+Because they are evicted rather than deleted, this process should respect `PodDisruptionBudget` constraints, allowing users
+to ensure that clusters are restarted gracefully.
+
+=== Expiration date
+
+Annotation:: `restarter.stackable.tech/expires-at.\{tag\}`
+
+Pods can be configured to expire at a certain point in time. In this case, the `Pod` should have the annotation
+`restarter.stackable.tech/expires-at.\{tag\}` set to a datetime formatted according to RFC 3339 (such as
+`"2022-04-21T13:24:15.225774724+00:00"`).
+`\{tag\}` should be a deterministic but unique ID identifying the reason for the expiry.
+
+Multiple `expires-at` annotations can be set on the same `Pod`, in which case the *earliest* expiration datetime
+takes precedence.
+
+== `StatefulSet`
+
+StatefulSets are rolling-restarted when any of their restart criteria (listed below) expire.
+
+=== Stale configuration
+
+Label:: `restarter.stackable.tech/enabled`
+
+StatefulSets can be configured to restart when any referenced configuration object (`ConfigMap` or `Secret`) changes.
+To enable this, set the `restarter.stackable.tech/enabled` label on the `StatefulSet` to `true`.

docs/modules/ROOT/pages/reloader.adoc

@@ -14,8 +14,8 @@ Multiple operators use this CRD as a way to express the authentication of the pr
 |xref:tls.adoc[]
 |Section of a CRD describing how to connect to a TLS enabled system
 
-|xref:reloader.adoc[]
-|An operator that watches `StatefulSets` and restarts them if mounted `ConfigMap` or `Secret` changes
+|xref:restarter.adoc[]
+|An operator that watches `Pod` objects and their controllers and restarts them when required
 |===
 
 The following diagram describes the relationship between the CRDs

docs/modules/ROOT/pages/restarter.adoc

@@ -18,6 +18,7 @@ snafu = "0.7"
 stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "0.16.0" }
 strum = { version = "0.24", features = ["derive"] }
 tokio = { version = "1.17", features = ["macros", "rt-multi-thread"] }
+tracing = "0.1.34"
 
 [build-dependencies]
 built = { version =  "0.5", features = ["chrono", "git2"] }

docs/modules/ROOT/pages/usage.adoc

@@ -1,5 +1,6 @@
 mod restart_controller;
 
+use futures::pin_mut;
 use stackable_operator::cli::{Command, ProductOperatorRun};
 
 use clap::Parser;
@@ -46,7 +47,10 @@ async fn main() -> anyhow::Result<()> {
             ))
             .await?;
 
-            restart_controller::start(&client).await?
+            let sts_restart_controller = restart_controller::statefulset::start(&client);
+            let pod_restart_controller = restart_controller::pod::start(&client);
+            pin_mut!(sts_restart_controller, pod_restart_controller);
+            futures::future::select(sts_restart_controller, pod_restart_controller).await;
         }
     }
 

rust/operator-binary/Cargo.toml

@@ -0,0 +1,2 @@
+pub mod pod;
+pub mod statefulset;

rust/operator-binary/src/main.rs

@@ -0,0 +1,127 @@
+use std::{sync::Arc, time::Duration};
+
+use futures::StreamExt;
+use snafu::{OptionExt, ResultExt, Snafu};
+use stackable_operator::{
+    client::Client,
+    k8s_openapi::{
+        api::core::v1::Pod,
+        chrono::{self, DateTime, FixedOffset, Utc},
+    },
+    kube::{
+        self,
+        api::{EvictParams, ListParams},
+        core::DynamicObject,
+        runtime::{
+            controller::{Action, Context},
+            reflector::ObjectRef,
+            Controller,
+        },
+    },
+    logging::controller::{report_controller_reconciled, ReconcilerError},
+};
+use strum::{EnumDiscriminants, IntoStaticStr};
+
+struct Ctx {
+    client: Client,
+}
+
+#[derive(Snafu, Debug, EnumDiscriminants)]
+#[strum_discriminants(derive(IntoStaticStr))]
+enum Error {
+    #[snafu(display("Pod has no name"))]
+    PodHasNoName,
+    #[snafu(display(
+        "failed to parse expiry timestamp annotation ({annotation:?}: {value:?}) as RFC 3999"
+    ))]
+    UnparseableExpiryTimestamp {
+        source: chrono::ParseError,
+        annotation: String,
+        value: String,
+    },
+    #[snafu(display("failed to evict Pod"))]
+    EvictPod { source: kube::Error },
+}
+
+impl ReconcilerError for Error {
+    fn category(&self) -> &'static str {
+        ErrorDiscriminants::from(self).into()
+    }
+
+    fn secondary_object(&self) -> Option<ObjectRef<DynamicObject>> {
+        match self {
+            Error::PodHasNoName => None,
+            Error::UnparseableExpiryTimestamp {
+                source: _,
+                annotation: _,
+                value: _,
+            } => None,
+            Error::EvictPod { source: _ } => None,
+        }
+    }
+}
+
+pub async fn start(client: &Client) {
+    let controller = Controller::new(client.get_all_api::<Pod>(), ListParams::default());
+    controller
+        .run(
+            reconcile,
+            error_policy,
+            Context::new(Ctx {
+                client: client.clone(),
+            }),
+        )
+        .for_each(|res| async move {
+            report_controller_reconciled(client, "pod.restarter.commons.stackable.tech", &res)
+        })
+        .await;
+}
+
+async fn reconcile(pod: Arc<Pod>, ctx: Context<Ctx>) -> Result<Action, Error> {
+    if pod.metadata.deletion_timestamp.is_some() {
+        // Object is already being deleted, no point trying again
+        return Ok(Action::await_change());
+    }
+
+    let pod_expires_at = pod
+        .metadata
+        .annotations
+        .iter()
+        .flatten()
+        .filter(|(k, _)| k.starts_with("restarter.stackable.tech/expires-at."))
+        .map(|(k, v)| {
+            DateTime::parse_from_rfc3339(v).context(UnparseableExpiryTimestampSnafu {
+                annotation: k,
+                value: v,
+            })
+        })
+        .min_by_key(|res| {
+            // Prefer propagating errors over successful cases
+            (res.is_ok(), res.as_ref().ok().cloned())
+        })
+        .transpose()?;
+
+    let now = DateTime::<FixedOffset>::from(Utc::now());
+    let time_until_pod_expires = pod_expires_at.map(|expires_at| (expires_at - now).to_std());
+    match time_until_pod_expires {
+        Some(Err(_has_already_expired)) => {
+            let pods = ctx
+                .get_ref()
+                .client
+                .get_api::<Pod>(pod.metadata.namespace.as_deref());
+            pods.evict(
+                pod.metadata.name.as_deref().context(PodHasNoNameSnafu)?,
+                &EvictParams::default(),
+            )
+            .await
+            .context(EvictPodSnafu)?;
+            Ok(Action::await_change())
+        }
+        Some(Ok(time_until_pod_expires)) => Ok(Action::requeue(time_until_pod_expires)),
+        None => Ok(Action::await_change()),
+    }
+}
+
+fn error_policy(_error: &Error, _ctx: Context<Ctx>) -> Action {
+    Action::requeue(Duration::from_secs(5))
+}

rust/operator-binary/src/restart_controller/mod.rs

@@ -60,8 +60,8 @@ impl ReconcilerError for Error {
     }
 }
 
-pub async fn start(client: &Client) -> anyhow::Result<()> {
-    let kube = kube::Client::try_default().await?;
+pub async fn start(client: &Client) {
+    let kube = client.as_kube_client();
     let stses = kube::Api::<StatefulSet>::all(kube.clone());
     let cms = kube::Api::<ConfigMap>::all(kube.clone());
     let secrets = kube::Api::<Secret>::all(kube.clone());
@@ -114,11 +114,9 @@ pub async fn start(client: &Client) -> anyhow::Result<()> {
         ),
     )
     .for_each(|res| async move {
-        report_controller_reconciled(client, "commons.superset.stackable.tech", &res)
+        report_controller_reconciled(client, "statefulset.restarter.commons.stackable.tech", &res)
     })
     .await;
-
-    Ok(())
 }
 
 fn trigger_all<S, K>(