feat: shut down the kubernetes api watcher if it recieved SIGTERM

src/kube_cache.rs

@@ -40,13 +40,16 @@ impl KubeCache {
     /// and if it is not possible returns a None.
     ///
     /// It also returns a JoinHandle which is tied to the api watcher
-    /// and if it returns that means that the kubecache is not responsive now.
+    /// and if that handle is joinable that means that `kubecache` is
+    /// not responsive after that point. (the server should be restarted)
     ///
     /// # Example:
     /// ```no_run
     /// let (kube_cache, api_task) = kube_cache::KubeCache::create().await.unwrap();
     /// ```
-    pub async fn create() -> Option<(KubeCache, JoinHandle<()>)> {
+    pub async fn create(
+        token: tokio_util::sync::CancellationToken,
+    ) -> Option<(KubeCache, JoinHandle<()>)> {
         let in_cluster = match std::env::var("KUBERNETES_SERVICE_HOST") {
             Ok(_) => true,
             Err(_) => false,
@@ -64,10 +67,16 @@ impl KubeCache {
             let infinite_watch = dep_watcher_rf
                 .applied_objects()
                 .for_each(|_o| std::future::ready(()));
-            let _res = infinite_watch.await;
+            tokio::select! {
-            tracing::error!(
+                _ = infinite_watch => {
-                "deployments watcher ended; this should not happen; (program should exit now)"
+                    tracing::error!(
-            );
+                        "deployments watcher ended; this should not happen; (program should exit now)"
+                    );
+                }
+                _ = token.cancelled() => {
+                    tracing::info!("shut down kubernetes api watcher");
+                }
+            }
         });
 
         let srv_api: Api<Service> = Api::default_namespaced(client);
@@ -208,8 +217,10 @@ impl MinecraftAPI<Server> for McApi {
 }
 
 impl McApi {
-    pub async fn create() -> Option<(Self, JoinHandle<()>)> {
+    pub async fn create(
-        let (kube_cache, kube_task) = KubeCache::create().await?;
+        token: tokio_util::sync::CancellationToken,
+    ) -> Option<(Self, JoinHandle<()>)> {
+        let (kube_cache, kube_task) = KubeCache::create(token).await?;
         Some((
             Self {
                 cache: kube_cache,

src/main.rs

@@ -27,7 +27,11 @@ async fn main() {
     let revision: &'static str = env!("COMMIT_HASH");
     tracing::info!(revision);
 
-    let (api, api_task) = kube_cache::McApi::create().await.unwrap();
+    let cancel_token = tokio_util::sync::CancellationToken::new();
+
+    let (api, api_task) = kube_cache::McApi::create(cancel_token.clone())
+        .await
+        .unwrap();
     tracing::info!("initialized kube api");
 
     let config: Config = Default::default();
@@ -43,7 +47,6 @@ async fn main() {
             return;
         }
     };
-    let cancel_token = tokio_util::sync::CancellationToken::new();
 
     let mut conn_task = proxy::start_proxy(listener, api, config, cancel_token.clone());
 
@@ -54,8 +57,8 @@ async fn main() {
         _ = &mut conn_task => {
             tracing::error!("the connection handling tokio:spawn'ed task ran to completion, which should not happen!");
         }
-        result = sigterm.recv() => {
+        _ = sigterm.recv() => {
-            tracing::info!(sigterm_signal = ?result,"SIGTERM received");
+            tracing::info!("SIGTERM received");
             cancel_token.cancel();
             let res = conn_task.await;
             tracing::info!(api_task_result = ?res, "shutdown complete");