Dynamic watches setup: wait until the first subscriber added (#232) (#234)

Signed-off-by: Danil-Grigorev <[email protected]>

Author: Danil-Grigorev

src/api/capi_cluster.rs

@@ -1,7 +1,10 @@
 use std::collections::BTreeMap;
 
 use cluster_api_rs::capi_cluster::{ClusterSpec, ClusterStatus};
-use fleet_api_rs::{fleet_bundle_namespace_mapping::BundleNamespaceMappingNamespaceSelector, fleet_clustergroup::{ClusterGroupSelector, ClusterGroupSpec}};
+use fleet_api_rs::{
+    fleet_bundle_namespace_mapping::BundleNamespaceMappingNamespaceSelector,
+    fleet_clustergroup::{ClusterGroupSelector, ClusterGroupSpec},
+};
 use kube::{
     api::{ObjectMeta, TypeMeta},
     Resource, ResourceExt as _,

src/controller.rs

@@ -15,7 +15,6 @@ use clap::Parser;
 use futures::stream::SelectAll;
 use futures::{Stream, StreamExt};
 
-use k8s_openapi::api::core::v1::Namespace;
 use k8s_openapi::apimachinery::pkg::apis::meta::v1::{Condition, Time};
 use kube::api::{DynamicObject, Patch, PatchParams};
 use kube::core::DeserializeGuard;
@@ -33,7 +32,6 @@ use kube::{
 use kube::{Resource, ResourceExt};
 
 use std::collections::BTreeMap;
-use std::future;
 
 use std::ops::Deref;
 use std::pin::Pin;
@@ -280,10 +278,6 @@ pub async fn run_cluster_controller(state: State) {
         .expect("failed to create kube Client");
 
     let (sub, reader) = state.dispatcher.subscribe();
-    let sub = sub
-        .map(|n: Arc<Namespace>| Ok(n.deref().clone()))
-        .predicate_filter(predicates::labels)
-        .filter_map(|n| future::ready(n.ok().map(Arc::new)));
     let ns_controller = Controller::for_shared_stream(sub, reader)
         .shutdown_on_signal()
         .run(

src/controllers/controller.rs

@@ -76,7 +76,8 @@ where
         .map_err(GetOrCreateError::Create)?;
 
     info!("Created fleet object");
-    ctx.diagnostics
+    match ctx
+        .diagnostics
         .read()
         .await
         .recorder(ctx.client.clone())
@@ -95,7 +96,12 @@ where
             },
             &res.object_ref(&()),
         )
-        .await?;
+        .await
+    {
+        // Ignore forbidden errors on namespace deletion
+        Err(kube::Error::Api(e)) if &e.reason == "Forbidden" => (),
+        e => e?,
+    };
 
     Ok(Action::await_change())
 }
@@ -120,7 +126,8 @@ where
         .map_err(PatchError::Patch)?;
 
     info!("Updated fleet object");
-    ctx.diagnostics
+    match ctx
+        .diagnostics
         .read()
         .await
         .recorder(ctx.client.clone())
@@ -139,7 +146,12 @@ where
             },
             &res.object_ref(&()),
         )
-        .await?;
+        .await
+    {
+        // Ignore forbidden errors on namespace deletion
+        Err(kube::Error::Api(e)) if &e.reason == "Forbidden" => (),
+        e => e?,
+    };
 
     Ok(Action::await_change())
 }
@@ -193,7 +205,12 @@ where
     }
 
     async fn cleanup(&self, ctx: Arc<Context>) -> crate::Result<Action> {
-        ctx.diagnostics
+        if let Some(mut bundle) = self.to_bundle(ctx.clone()).await? {
+            return Ok(bundle.cleanup(ctx).await?);
+        }
+
+        match ctx
+            .diagnostics
             .read()
             .await
             .recorder(ctx.client.clone())
@@ -208,11 +225,12 @@ where
                 },
                 &self.object_ref(&()),
             )
-            .await?;
-
-        if let Some(mut bundle) = self.to_bundle(ctx.clone()).await? {
-            return Ok(bundle.cleanup(ctx).await?);
-        }
+            .await
+        {
+            // Ignore forbidden errors on namespace deletion
+            Err(kube::Error::Api(e)) if &e.reason == "Forbidden" => (),
+            e => e?,
+        };
 
         Ok(Action::await_change())
     }

src/controllers/helm/install.rs

@@ -189,7 +189,14 @@ impl FleetOptions {
     pub fn patch_fleet(&self, version: &str) -> FleetPatchResult<Child> {
         let mut upgrade = Command::new("helm");
 
-        upgrade.args(["upgrade", "fleet", "fleet/fleet", "--reuse-values", "--version", version]);
+        upgrade.args([
+            "upgrade",
+            "fleet",
+            "fleet/fleet",
+            "--reuse-values",
+            "--version",
+            version,
+        ]);
 
         if !self.namespace.is_empty() {
             upgrade.args(["--namespace", &self.namespace]);

src/multi_dispatcher.rs

@@ -3,6 +3,7 @@ use std::{
     pin::Pin,
     sync::Arc,
     task::{Context, Poll},
+    time::Duration,
 };
 
 use async_broadcast::{InactiveReceiver, Receiver, Sender};
@@ -18,6 +19,7 @@ use kube::{
 };
 use pin_project::pin_project;
 use serde::de::DeserializeOwned;
+use tokio::time::sleep;
 
 #[derive(Clone)]
 pub struct MultiDispatcher {
@@ -70,6 +72,11 @@ impl MultiDispatcher {
             }
         }
     }
+
+    // Subscribers count returns the number of current receiving streams
+    pub(crate) fn subscribers_count(&self) -> usize {
+        self.dispatch_tx.receiver_count()
+    }
 }
 
 /// `BroadcastStream` allows to stream shared list of dynamic objects,
@@ -223,6 +230,10 @@ where
     W: Stream<Item = Result<Event<DynamicObject>>> + Unpin,
 {
     stream! {
+        while writer.subscribers_count() == 0 {
+            sleep(Duration::from_millis(100)).await;
+        }
+
         while let Some(event) = broadcast.next().await {
             match event {
                 Ok(ev) => {