kube/test-deployment.yaml

@@ -15,7 +15,7 @@ spec:
         app: minecraft-ingress
     spec:
       serviceAccountName: minecraft-ingress
-      terminationGracePeriodSeconds: 28800 # This is 8 hours
+      terminationGracePeriodSeconds: 5
       containers:
       - name: minecraft-ingress
         image: git.tami.moe/tamipes/minecraft-ingress:testing

src/kube_cache.rs

@@ -40,16 +40,13 @@ 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 that handle is joinable that means that `kubecache` is
-    /// not responsive after that point. (the server should be restarted)
+    /// and if it returns that means that the kubecache is not responsive now.
     ///
     /// # Example:
     /// ```no_run
     /// let (kube_cache, api_task) = kube_cache::KubeCache::create().await.unwrap();
     /// ```
-    pub async fn create(
-        token: tokio_util::sync::CancellationToken,
-    ) -> Option<(KubeCache, JoinHandle<()>)> {
+    pub async fn create() -> Option<(KubeCache, JoinHandle<()>)> {
         let in_cluster = match std::env::var("KUBERNETES_SERVICE_HOST") {
             Ok(_) => true,
             Err(_) => false,
@@ -67,16 +64,10 @@ impl KubeCache {
             let infinite_watch = dep_watcher_rf
                 .applied_objects()
                 .for_each(|_o| std::future::ready(()));
-            tokio::select! {
-                _ = infinite_watch => {
+            let _res = infinite_watch.await;
             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);
@@ -217,10 +208,8 @@ impl MinecraftAPI<Server> for McApi {
 }
 
 impl McApi {
-    pub async fn create(
-        token: tokio_util::sync::CancellationToken,
-    ) -> Option<(Self, JoinHandle<()>)> {
-        let (kube_cache, kube_task) = KubeCache::create(token).await?;
+    pub async fn create() -> Option<(Self, JoinHandle<()>)> {
+        let (kube_cache, kube_task) = KubeCache::create().await?;
         Some((
             Self {
                 cache: kube_cache,

src/main.rs

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