Use shared stream/cache

Author: yiannistri

src/controller.rs

@@ -36,7 +36,7 @@ use std::collections::BTreeMap;
 use std::ops::Deref;
 use std::sync::Arc;
 use tokio::{sync::RwLock, time::Duration};
-use tracing::{self, warn};
+use tracing::{self, info, warn};
 
 /// State shared between the controller and the web server
 #[derive(Clone)]
@@ -349,8 +349,23 @@ pub async fn run_cluster_controller(state: State) {
         .default_backoff()
         .for_each(|_| futures::future::ready(()));
 
-    let cluster_ns_controller = Controller::new(
-        Api::<Cluster>::all(client.clone()),Default::default())
+    let capi_clusters = metadata_watcher(
+        Api::<Cluster>::all(client.clone()), 
+        Config::default().any_semantic(),
+    )
+    .default_handling();
+
+    let (sub, reader) = state.dispatcher.subscribe();
+    let cluster_ns_controller = Controller::for_shared_stream(sub, reader.clone())
+        .watches_stream(capi_clusters, move |_cluster| {
+            reader
+                .state()
+                .into_iter()
+                .filter_map(move |c: Arc<Cluster>| {
+                    info!("Reconcile CAPI cluster: {}", c.name_any());
+                    Some(ObjectRef::from_obj(&*c))
+                })
+        })
         .shutdown_on_signal()
         .run(
             Cluster::reconcile_fleet_annotation_in_capi_ns,