feat: Add graceful shutdown

CHANGELOG.md

@@ -12,6 +12,7 @@ All notable changes to this project will be documented in this file.
 
 ### Changed
 
+- Gracefully shutdown all concurrent tasks by forwarding the SIGTERM signal ([#366]).
 - OLM deployer doesn't add owner references to cluster scoped objects anymore ([#360]).
   Owner references ensure that objects are garbage collected by OpenShift upon operator removal but they cause problems when the operator is updated.
   This means that cluster wide objects are not removed anymore when the operator is uninstalled.
@@ -22,6 +23,7 @@ All notable changes to this project will be documented in this file.
 [#363]: https://github.com/stackabletech/listener-operator/pull/363
 [#364]: https://github.com/stackabletech/listener-operator/pull/364
 [#365]: https://github.com/stackabletech/listener-operator/pull/365
+[#366]: https://github.com/stackabletech/listener-operator/pull/366
 
 ## [25.11.0] - 2025-11-07
 

Cargo.toml

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

rust/operator-binary/src/listener_controller.rs

@@ -1,5 +1,6 @@
 use std::{
     collections::{BTreeMap, BTreeSet},
+    future::Future,
     sync::Arc,
 };
 
@@ -48,7 +49,10 @@ const OPERATOR_NAME: &str = "listeners.stackable.tech";
 const CONTROLLER_NAME: &str = "listener";
 pub const FULL_CONTROLLER_NAME: &str = concatcp!(CONTROLLER_NAME, '.', OPERATOR_NAME);
 
-pub async fn run(client: stackable_operator::client::Client) {
+pub async fn run<F>(client: stackable_operator::client::Client, shutdown_signal: F)
+where
+    F: Future<Output = ()> + Send + Sync + 'static,
+{
     let controller = controller::Controller::new(
         client.get_all_api::<DeserializeGuard<listener::v1alpha1::Listener>>(),
         watcher::Config::default(),
@@ -119,7 +123,7 @@ pub async fn run(client: stackable_operator::client::Client) {
                     })
             },
         )
-        .shutdown_on_signal()
+        .graceful_shutdown_on(shutdown_signal)
         .run(
             reconcile,
             error_policy,

rust/operator-binary/src/main.rs

@@ -23,8 +23,8 @@ use stackable_operator::{
     eos::EndOfSupportChecker,
     shared::yaml::SerializeOptions,
     telemetry::Tracing,
+    utils::signal::SignalWatcher,
 };
-use tokio::signal::unix::{SignalKind, signal};
 use tokio_stream::wrappers::UnixListenerStream;
 use tonic::transport::Server;
 use utils::unix_stream::{TonicUnixStream, uds_bind_private};
@@ -116,9 +116,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(
@@ -134,9 +138,9 @@ async fn main() -> anyhow::Result<()> {
                 let _ = std::fs::remove_file(&csi_endpoint);
             }
 
-            let mut sigterm = signal(SignalKind::terminate())?;
             let csi_listener =
                 UnixListenerStream::new(uds_bind_private(csi_endpoint)?).map_ok(TonicUnixStream);
+
             let csi_server = Server::builder()
                 .add_service(
                     tonic_reflection::server::Builder::configure()
@@ -152,9 +156,10 @@ async fn main() -> anyhow::Result<()> {
                         .add_service(ControllerServer::new(ListenerOperatorController {
                             client: client.clone(),
                         }))
-                        .serve_with_incoming_shutdown(csi_listener, sigterm.recv().map(|_| ()))
+                        .serve_with_incoming_shutdown(csi_listener, sigterm_watcher.handle())
                         .map_err(|err| anyhow!(err).context("failed to run csi server"));
-                    let controller = listener_controller::run(client).map(anyhow::Ok);
+                    let controller =
+                        listener_controller::run(client, sigterm_watcher.handle()).map(anyhow::Ok);
 
                     futures::try_join!(csi_server, controller, eos_checker)?;
                 }
@@ -165,7 +170,7 @@ async fn main() -> anyhow::Result<()> {
                             client: client.clone(),
                             node_name: node_name.to_owned(),
                         }))
-                        .serve_with_incoming_shutdown(csi_listener, sigterm.recv().map(|_| ()))
+                        .serve_with_incoming_shutdown(csi_listener, sigterm_watcher.handle())
                         .map_err(|err| anyhow!(err).context("failed to run csi server"));
 
                     futures::try_join!(csi_server, eos_checker)?;