feat: watch and filter

Cargo.toml

@@ -11,7 +11,7 @@ description = "Copy k8s resources (or parts thereof) across clusters"
 [dependencies]
 clap = { version = "4.4", features = ["derive", "help", "env", "std"] }
 futures = "0.3"
-kube = { version = "0.85.0", features = ["runtime", "derive"] }
+kube = { version = "0.85.0", features = ["runtime", "derive", "unstable-runtime"] }
 kube-derive = "0.84.0"
 k8s-openapi = { version = "0.19.0", features = ["v1_26"] }
 kubert = { version = "0.19.0", features = [
@@ -29,6 +29,12 @@ serde_json = "1"
 serde_yaml = "0.9.34"
 thiserror = "1"
 serde_json_path = "0.6.7"
+backoff = "0.4.*"
+tokio-context = "0.1.*"
+tokio-stream = "0.1.*"
 
 [dev-dependencies]
 rstest = "0.17.0"
+once_cell = "1.19.*"
+chrono = "0.4.*"
+rand = "0.8.5"

src/controller.rs

@@ -4,6 +4,7 @@ use futures::StreamExt;
 use k8s_openapi::apimachinery::pkg::apis::meta::v1::OwnerReference;
 use kube::api::DeleteParams;
 use kube::api::Patch::Merge;
+use kube::runtime::{predicates, reflector, WatchStreamExt};
 use kube::{
     api::{ListParams, Patch, PatchParams},
     runtime::{
@@ -13,24 +14,28 @@ use kube::{
     Api, Client, Resource, ResourceExt,
 };
 use serde_json::json;
+use tokio_context::context::RefContext;
 #[allow(unused_imports)]
 use tracing::{debug, error, info, warn};
 
 use util::{WithItemAdded, WithItemRemoved};
 
 use crate::mapping::{apply_mappings, clone_resource};
+use crate::remote_watcher_manager::RemoteWatcherManager;
 use crate::resource_extensions::NamespacedApi;
 use crate::{requeue_after, resources::ResourceSync, util, Error, Result, FINALIZER};
 
 pub struct Context {
     pub client: Client,
+    pub remote_watcher_manager: RemoteWatcherManager,
 }
 
 async fn reconcile_deleted_resource(
     resource_sync: Arc<ResourceSync>,
     name: &str,
     target_api: NamespacedApi,
     parent_api: Api<ResourceSync>,
+    ctx: Arc<Context>,
 ) -> Result<Action> {
     if !resource_sync.has_target_finalizer() {
         // We have already removed our finalizer, so nothing more needs to be done
@@ -41,19 +46,24 @@ async fn reconcile_deleted_resource(
 
     match target_api.get(target_name).await {
         Ok(target) if target.metadata.deletion_timestamp.is_some() => {
-            // Target is being deleted, wait for it to be deleted
-            // For now we need a requeue after, but in the future we should try to watch the target if we can
-            requeue_after!()
+            resource_sync
+                .start_remote_watches_if_not_watching(ctx)
+                .await;
+            Ok(Action::await_change())
         }
         Ok(_) => {
             target_api
                 .delete(target_name, &DeleteParams::foreground())
                 .await?;
-            // Deleted target, wait for it to be deleted
-            // For now we need a requeue after, but in the future we should try to watch the target if we can
-            requeue_after!()
+
+            resource_sync
+                .start_remote_watches_if_not_watching(ctx)
+                .await;
+            Ok(Action::await_change())
         }
         Err(kube::Error::Api(err)) if err.code == 404 => {
+            resource_sync.stop_remote_watches_if_watching(ctx).await;
+
             let patched_finalizers = resource_sync
                 .finalizers_clone_or_empty()
                 .with_item_removed(&FINALIZER.to_string());
@@ -95,15 +105,15 @@ async fn add_target_finalizer(
         .patch(name, &PatchParams::default(), &patch)
         .await?;
 
-    // For now we are watching all events for the ResourceSync, so the patch will trigger a reconcile
-    Ok(Action::await_change())
+    requeue_after!(Duration::from_millis(500))
 }
 
 async fn reconcile_normally(
     resource_sync: Arc<ResourceSync>,
     name: &str,
     source_api: NamespacedApi,
     target_api: NamespacedApi,
+    ctx: Arc<Context>,
 ) -> Result<Action> {
     let target_namespace = &target_api.namespace;
     let target_ar = &target_api.ar;
@@ -157,11 +167,19 @@ async fn reconcile_normally(
         .patch(&target_ref.name, &ssapply, &Patch::Apply(&target))
         .await?;
 
+    resource_sync
+        .start_remote_watches_if_not_watching(ctx)
+        .await;
+
     info!(?name, ?target_ref, "successfully reconciled");
 
-    requeue_after!()
+    Ok(Action::await_change())
 }
 
+// TODO: Until CDC finalizer is implemented (and even after that if a CDC is deleted using foreground propagation) CDC deletion could cause the cluster to be deleted or unreachable before we have a chance to clean up the resource(s) owned by the ResourceSync, we should be able to detect and handle this scenario gracefully
+// TODO: Immutability on source/target (via CEL?)
+// TODO: If secrets for remote clusters on target and source (when applicable) no longer exist then simply allow the ResourceSync to be deleted by removing the finalizer
+
 async fn reconcile(resource_sync: Arc<ResourceSync>, ctx: Arc<Context>) -> Result<Action> {
     let name = resource_sync
         .metadata
@@ -176,26 +194,27 @@ async fn reconcile(resource_sync: Arc<ResourceSync>, ctx: Arc<Context>) -> Resul
     let target_api = resource_sync
         .spec
         .target
-        .api_for(Arc::clone(&ctx), &local_ns)
+        .api_for(ctx.client.clone(), &local_ns)
         .await?;
     let source_api = resource_sync
         .spec
         .source
-        .api_for(Arc::clone(&ctx), &local_ns)
+        .api_for(ctx.client.clone(), &local_ns)
         .await?;
-    let parent_api = resource_sync.api(Arc::clone(&ctx));
+    let parent_api = resource_sync.api(ctx.client.clone());
 
     match resource_sync {
         resource_sync if resource_sync.has_been_deleted() => {
-            reconcile_deleted_resource(resource_sync, &name, target_api, parent_api).await
+            reconcile_deleted_resource(resource_sync, &name, target_api, parent_api, ctx).await
         }
         resource_sync if !resource_sync.has_target_finalizer() => {
             add_target_finalizer(resource_sync, &name, parent_api).await
         }
-        _ => reconcile_normally(resource_sync, &name, source_api, target_api).await,
+        _ => reconcile_normally(resource_sync, &name, source_api, target_api, ctx).await,
     }
 }
 
+// TODO: Exponential Backoff using DefaultBackoff for watcher
 fn error_policy(resource_sync: Arc<ResourceSync>, error: &Error, _ctx: Arc<Context>) -> Action {
     let name = resource_sync.name_any();
     warn!(?name, %error, "reconcile failed");
@@ -209,13 +228,36 @@ pub async fn run(client: Client) -> Result<()> {
         error!("CRD is not queryable; {e:?}. Is the CRD installed?");
         std::process::exit(1);
     }
-    Controller::new(docs, watcher::Config::default().any_semantic())
+
+    let (reader, writer) = reflector::store();
+    let resource_syncs = watcher(docs, watcher::Config::default().any_semantic())
+        .default_backoff()
+        .reflect(writer)
+        .applied_objects()
+        .predicate_filter(predicates::generation);
+
+    let (ctx, handle) = RefContext::new();
+
+    let (remote_watcher_manager, remote_objects_trigger) =
+        RemoteWatcherManager::new(ctx, client.clone());
+
+    Controller::for_stream(resource_syncs, reader)
+        .reconcile_on(remote_objects_trigger)
         .shutdown_on_signal()
-        .run(reconcile, error_policy, Arc::new(Context { client }))
+        .run(
+            reconcile,
+            error_policy,
+            Arc::new(Context {
+                client,
+                remote_watcher_manager,
+            }),
+        )
         .filter_map(|x| async move { Result::ok(x) })
         .for_each(|_| futures::future::ready(()))
         .await;
 
+    handle.cancel();
+
     Ok(())
 }