feat: Add graceful shutdown

Author: Techassi

Cargo.lock

@@ -10,7 +10,7 @@ edition = "2021"
 repository = "https://github.com/stackabletech/commons-operator"
 
 [workspace.dependencies]
-stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "stackable-operator-0.101.1", features = ["telemetry", "webhook"] }
+stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "stackable-operator-0.105.0", features = ["telemetry", "webhook"] }
 
 anyhow = "1.0"
 built = { version = "0.8", features = ["chrono", "git2"] }

Cargo.nix

@@ -16,6 +16,7 @@ use stackable_operator::{
     eos::EndOfSupportChecker,
     shared::yaml::SerializeOptions,
     telemetry::Tracing,
+    utils::signal::SignalWatcher,
 };
 use webhook::create_webhook_server;
 
@@ -90,9 +91,13 @@ async fn main() -> anyhow::Result<()> {
                 description = built_info::PKG_DESCRIPTION
             );
 
+            // Watches for the SIGTERM signal and sends a signal to all receivers, which gracefully
+            // shuts down all concurrent tasks below (EoS checker, controller).
+            let sigterm_watcher = SignalWatcher::sigterm()?;
+
             let eos_checker =
                 EndOfSupportChecker::new(built_info::BUILT_TIME_UTC, maintenance.end_of_support)?
-                    .run()
+                    .run(sigterm_watcher.handle())
                     .map(anyhow::Ok);
 
             let client = stackable_operator::client::initialize_operator(
@@ -107,10 +112,13 @@ async fn main() -> anyhow::Result<()> {
                 cm_store_tx,
                 secret_store_tx,
                 &watch_namespace,
+                sigterm_watcher.handle(),
             )
             .map(anyhow::Ok);
+
             let pod_restart_controller =
-                restart_controller::pod::start(&client, &watch_namespace).map(anyhow::Ok);
+                restart_controller::pod::start(&client, &watch_namespace, sigterm_watcher.handle())
+                    .map(anyhow::Ok);
 
             let webhook_server = create_webhook_server(
                 ctx,
@@ -119,8 +127,9 @@ async fn main() -> anyhow::Result<()> {
                 client.as_kube_client(),
             )
             .await?;
+
             let webhook_server = webhook_server
-                .run()
+                .run(sigterm_watcher.handle())
                 .map_err(|err| anyhow!(err).context("failed to run webhook"));
 
             futures::try_join!(

Cargo.toml

@@ -1,4 +1,4 @@
-use std::{sync::Arc, time::Duration};
+use std::{future::Future, sync::Arc, time::Duration};
 
 use futures::StreamExt;
 use http::StatusCode;
@@ -70,7 +70,10 @@ impl ReconcilerError for Error {
     }
 }
 
-pub async fn start(client: &Client, watch_namespace: &WatchNamespace) {
+pub async fn start<F>(client: &Client, watch_namespace: &WatchNamespace, shutdown_signal: F)
+where
+    F: Future<Output = ()> + Send + Sync + 'static,
+{
     let controller = Controller::new(
         watch_namespace.get_api::<PartialObjectMeta<Pod>>(client),
         // TODO: Can we only watch a subset of Pods with a specify label, e.g.
@@ -85,6 +88,7 @@ pub async fn start(client: &Client, watch_namespace: &WatchNamespace) {
         },
     ));
     controller
+        .graceful_shutdown_on(shutdown_signal)
         .run(
             reconcile,
             error_policy,

crate-hashes.json

@@ -1,4 +1,4 @@
-use std::{collections::BTreeMap, sync::Arc, time::Duration};
+use std::{collections::BTreeMap, future::Future, sync::Arc, time::Duration};
 
 use futures::{Stream, StreamExt, TryStream, stream};
 use serde_json::json;
@@ -92,12 +92,15 @@ pub fn create_context(
     (ctx, cm_store_tx, secret_store_tx)
 }
 
-pub async fn start(
+pub async fn start<F>(
     ctx: Arc<Ctx>,
     cm_store_tx: Initializer<Store<PartialObjectMeta<ConfigMap>>>,
     secret_store_tx: Initializer<Store<PartialObjectMeta<Secret>>>,
     watch_namespace: &WatchNamespace,
-) {
+    shutdown_signal: F,
+) where
+    F: Future<Output = ()>,
+{
     let stses = watch_namespace.get_api::<DeserializeGuard<StatefulSet>>(&ctx.client);
     let cms = watch_namespace.get_api::<ConfigMap>(&ctx.client);
     let secrets = watch_namespace.get_api::<Secret>(&ctx.client);
@@ -163,7 +166,10 @@ pub async fn start(
                 .applied_objects(),
                 (),
             ),
-        ),
+        )
+        // This uses the same mechanism as kube's Controller does under the hood, see
+        // https://github.com/kube-rs/kube/blob/8bcdcb52e1e13c1c1ec59f6118fbed575ac10a4b/kube-runtime/src/controller/mod.rs#L1671
+        .take_until(shutdown_signal),
         Config::default(),
     )
     // We can let the reporting happen in the background