bug: Add controller readiness barrier (#307)

Introduce a `tokio::sync::Barrier` to the shared controller state.
Each main controller waits on this barrier after completing its initial setup, including dynamic watch registration and initial reconciliation.
This allows external components or health checks to synchronize and wait for all core controllers to be ready before proceeding.

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

Author: Danil-Grigorev

src/controller.rs

@@ -29,6 +29,7 @@ use kube::{
     },
 };
 use kube::{Resource, ResourceExt};
+use tokio::sync::Barrier;
 
 use std::collections::BTreeMap;
 
@@ -56,6 +57,9 @@ pub struct State {
 
     // k8s api server minor version
     pub version: u32,
+
+    // Controller readiness barrier
+    pub barrier: Arc<Barrier>,
 }
 
 #[derive(Parser, Debug, Clone, Default)]
@@ -76,6 +80,7 @@ impl State {
             diagnostics: Default::default(),
             stream: BroadcastStream::new(Default::default()),
             version,
+            barrier: Arc::new(Barrier::new(3)),
         }
     }
 
@@ -98,6 +103,7 @@ impl State {
             dispatcher: self.dispatcher.clone(),
             stream: self.stream.clone(),
             version: self.version,
+            barrier: self.barrier.clone(),
         })
     }
 }
@@ -181,6 +187,9 @@ pub async fn run_fleet_addon_config_controller(state: State) {
     .await
     .expect("Initial dynamic watches setup to succeed");
 
+    // Signal that this controller is ready
+    state.barrier.wait().await;
+
     tokio::select! {
         _ = watcher => {panic!("This should not happen before controllers exit")},
         _ = futures::future::join(dynamic_watches_controller, config_controller) => {}
@@ -324,6 +333,9 @@ pub async fn run_cluster_controller(state: State) {
         .default_backoff()
         .for_each(|_| futures::future::ready(()));
 
+    // Signal that this controller is ready
+    state.barrier.wait().await;
+
     tokio::join!(clusters, ns_controller);
 }
 
@@ -374,6 +386,9 @@ pub async fn run_cluster_class_controller(state: State) {
         .default_backoff()
         .for_each(|_| futures::future::ready(()));
 
+    // Signal that this controller is ready
+    state.barrier.wait().await;
+
     tokio::join!(group_controller, cluster_class_controller);
 }
 

src/controllers/controller.rs

@@ -23,7 +23,7 @@ use tracing::field::display;
 use std::fmt::Debug;
 use std::pin::Pin;
 use std::sync::Arc;
-use tokio::sync::RwLock;
+use tokio::sync::{Barrier, RwLock};
 use tracing::{self, debug, info, instrument, Span};
 
 use super::{
@@ -51,6 +51,8 @@ pub struct Context {
     pub stream: BroadcastStream<DynamicStream>,
     // k8s minor version
     pub version: u32,
+    // Controller readiness barrier
+    pub barrier: Arc<Barrier>,
 }
 
 #[instrument(skip_all, fields(name = res.name_any(), namespace = res.namespace(), api_version = typed_gvk::<R>(()).api_version(), kind = R::kind(&()).to_string()), err)]

src/multi_dispatcher.rs

@@ -3,7 +3,6 @@ use std::{
     pin::Pin,
     sync::Arc,
     task::{Context, Poll},
-    time::Duration,
 };
 
 use async_broadcast::{InactiveReceiver, Receiver, Sender};
@@ -19,7 +18,6 @@ use kube::{
 };
 use pin_project::pin_project;
 use serde::de::DeserializeOwned;
-use tokio::time::sleep;
 
 #[derive(Clone)]
 pub struct MultiDispatcher {
@@ -72,11 +70,6 @@ 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,
@@ -230,10 +223,6 @@ 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) => {