Integrate namespace annotation patch into sync workflow

src/api/capi_cluster.rs

@@ -7,6 +7,7 @@ use fleet_api_rs::{
     },
     fleet_clustergroup::{ClusterGroupSelector, ClusterGroupSpec},
 };
+use k8s_openapi::api::core::v1::Namespace;
 use kube::{
     CustomResource, Resource, ResourceExt as _,
     api::{ObjectMeta, TypeMeta},
@@ -26,6 +27,9 @@ use super::{
 #[cfg(feature = "agent-initiated")]
 use super::fleet_cluster_registration_token::ClusterRegistrationToken;
 
+pub static FLEET_WORKSPACE_ANNOTATION: &str =
+    "field.cattle.io/allow-fleetworkspace-creation-for-existing-namespace";
+
 /// `ClusterProxy` defines the desired state of the CAPI Cluster.
 #[derive(CustomResource, Serialize, Deserialize, Clone, Debug, JsonSchema)]
 #[kube(
@@ -204,6 +208,21 @@ impl Cluster {
         .into()
     }
 
+    pub(crate) fn to_namespace(self: &Cluster) -> Namespace {
+        Namespace {
+            metadata: ObjectMeta {
+                name: self.namespace(),
+                annotations: Some({
+                    let mut map = BTreeMap::new();
+                    map.insert(FLEET_WORKSPACE_ANNOTATION.to_string(), "true".to_string());
+                    map
+                }),
+                ..Default::default()
+            },
+            ..Default::default()
+        }
+    }
+
     pub(crate) fn cluster_class_namespace(&self) -> Option<&str> {
         self.spec
             .proxy

src/api/comparable.rs

@@ -1,6 +1,32 @@
+use std::collections::HashSet;
+
+use k8s_openapi::api::core::v1::Namespace;
+
 // Trait for resources that can be compared
 pub(crate) trait ResourceDiff: kube::ResourceExt {
     fn diff(&self, other: &Self) -> bool {
-        self.meta() != other.meta()
+        let annotations_equal = self
+            .annotations()
+            .iter()
+            .all(|(k, v)| other.annotations().get(k) == Some(v));
+        let labels_equal = self
+            .labels()
+            .iter()
+            .all(|(k, v)| other.labels().get(k) == Some(v));
+
+        let owner_uids: HashSet<String> = other
+            .owner_references()
+            .iter()
+            .map(|r| &r.uid)
+            .cloned()
+            .collect();
+        let owner_references_equal = self
+            .owner_references()
+            .iter()
+            .all(|self_ref| owner_uids.contains(&self_ref.uid));
+
+        !annotations_equal || !labels_equal || !owner_references_equal
     }
 }
+
+impl ResourceDiff for Namespace {}

src/controller.rs

@@ -417,4 +417,4 @@ fn error_policy(doc: Arc<impl kube::Resource>, error: &Error, ctx: Arc<Context>)
     warn!("reconcile failed: {:?}", error);
     ctx.metrics.reconcile_failure(doc, error);
     Action::requeue(Duration::from_secs(10))
-}
+}

src/controllers/cluster.rs

@@ -1,5 +1,5 @@
 use crate::api::bundle_namespace_mapping::BundleNamespaceMapping;
-use crate::api::capi_cluster::Cluster;
+use crate::api::capi_cluster::{Cluster, FLEET_WORKSPACE_ANNOTATION};
 
 use crate::api::fleet_addon_config::FleetAddonConfig;
 use crate::api::fleet_cluster::{self};
@@ -8,6 +8,7 @@ use crate::api::fleet_cluster::{self};
 use crate::api::fleet_cluster_registration_token::ClusterRegistrationToken;
 use crate::api::fleet_clustergroup::ClusterGroup;
 use crate::controllers::addon_config::to_dynamic_event;
+use crate::controllers::controller::GetApi;
 use futures::StreamExt as _;
 use k8s_openapi::api::core::v1::Namespace;
 use kube::api::{
@@ -17,10 +18,14 @@ use kube::api::{
 use kube::client::scope;
 use kube::runtime::watcher::{self, Config};
 use kube::{Api, Client};
-use kube::{Resource, api::{ResourceExt, Patch}, runtime::controller::Action};
+use kube::{
+    Resource,
+    api::{Patch, ResourceExt},
+    runtime::controller::Action,
+};
 use serde::Serialize;
-use serde_json::{Value,json};
-use tracing::{info,debug};
+use serde_json::{Value, json};
+use tracing::{debug, info};
 
 use std::sync::Arc;
 
@@ -30,9 +35,9 @@ use super::controller::{
 use super::{BundleResult, ClusterSyncError, ClusterSyncResult};
 
 pub static CONTROLPLANE_READY_CONDITION: &str = "ControlPlaneReady";
-pub static FLEET_WORKSPACE_ANNOTATION: &str = "field.cattle.io/allow-fleetworkspace-creation-for-existing-namespace";
 
 pub struct FleetClusterBundle {
+    namespace: Namespace,
     template_sources: TemplateSources,
     fleet: fleet_cluster::Cluster,
     fleet_group: Option<ClusterGroup>,
@@ -178,17 +183,18 @@ impl FleetBundle for FleetClusterBundle {
         }
 
         // Ensure the fleet workspace annotation is present.
-        let patch = json!({
-                        "metadata": {
-                            "annotations": {
-                                FLEET_WORKSPACE_ANNOTATION: "true"
-                            }
-                        }
-                    });
-        let namespace_name = self.fleet.namespace().unwrap_or_default();
-        let namespaces = Api::<Namespace>::all(ctx.client.clone());
-        namespaces.patch_metadata(&namespace_name,&PatchParams::default(), &Patch::Merge(&patch)).await?;
-        debug!("Added fleet annotation to namespace {}.", namespace_name);
+        patch(
+            ctx.clone(),
+            &mut self.namespace,
+            &PatchParams::apply("namespace-addon-provider-fleet"),
+        )
+        .await
+        .map_err(ClusterSyncError::NamespacePatchError)?;
+
+        debug!(
+            "Added fleet annotation to namespace {}.",
+            self.fleet.get_namespace()
+        );
 
         Ok(Action::await_change())
     }
@@ -214,17 +220,18 @@ impl FleetBundle for FleetClusterBundle {
                 return Ok(Action::await_change());
             }
 
-            Api::<BundleNamespaceMapping>::namespaced(ctx.client.clone(), &ns.unwrap_or_default())
+            BundleNamespaceMapping::get_api(ctx.client.clone(), mapping.get_namespace())
                 .delete(&mapping.name_any(), &DeleteParams::default())
                 .await?;
         }
 
         // List all other clusters in this namespace
-        let namespaces = Api::<Namespace>::all(ctx.client.clone());
-        let namespace_name = self.fleet.namespace().unwrap_or_default();
-        let clusters = Api::<Cluster>::namespaced(ctx.client.clone(), &namespace_name);
-        let other_clusters = clusters
-            .list(&ListParams::default().fields(&format!("metadata.namespace={},metadata.name!={}", namespace_name, self.fleet.name_any())))
+        let other_clusters = Cluster::get_api(ctx.client.clone(), self.fleet.get_namespace())
+            .list(
+                &ListParams::default()
+                    .fields(&format!("metadata.name!={}", self.fleet.name_any()))
+                    .limit(1),
+            )
             .await?;
         // If no other clusters are found in this namespace, remove the fleet workspace annotation.
         if other_clusters.items.is_empty() {
@@ -235,8 +242,17 @@ impl FleetBundle for FleetClusterBundle {
                     }
                 }
             });
-            namespaces.patch_metadata(&namespace_name,&PatchParams::default(), &Patch::Merge(&patch)).await?;
-            debug!("Removed fleet annotation from namespace {}.", namespace_name);
+            Namespace::get_api(ctx.client.clone(), &())
+                .patch_metadata(
+                    self.fleet.get_namespace(),
+                    &PatchParams::default(),
+                    &Patch::Merge(&patch),
+                )
+                .await?;
+            debug!(
+                "Removed fleet annotation from namespace {}.",
+                self.fleet.get_namespace()
+            );
         }
 
         Ok(Action::await_change())
@@ -266,6 +282,7 @@ impl FleetController for Cluster {
             cluster_registration_token: self
                 .to_cluster_registration_token(config.spec.cluster.as_ref()),
             config,
+            namespace: self.to_namespace(),
         }))
     }
 }
@@ -319,4 +336,4 @@ impl Cluster {
 
         Ok(Action::await_change())
     }
-}
+}

src/controllers/cluster_group.rs

@@ -1,9 +1,10 @@
 use crate::api::capi_clusterclass::ClusterClass;
 use crate::api::fleet_clustergroup::ClusterGroup;
+use crate::controllers::controller::GetApi;
 
+use kube::ResourceExt;
 use kube::api::{Patch, PatchParams};
 use kube::runtime::controller::Action;
-use kube::{Api, ResourceExt};
 use serde_json::json;
 
 use std::ops::Deref;
@@ -32,7 +33,6 @@ impl ClusterGroup {
                     .iter()
                     .map(|(k, v)| (k.to_string(), v.to_string())),
             );
-
             patch(
                 ctx.clone(),
                 self,
@@ -43,8 +43,7 @@ impl ClusterGroup {
 
         if self.finalizers().iter().any(|f| f == FLEET_FINALIZER) {
             self.finalizers_mut().retain(|f| f != FLEET_FINALIZER);
-            let api: Api<Self> =
-                Api::namespaced(ctx.client.clone(), &self.namespace().unwrap_or_default());
+            let api = Self::get_api(ctx.client.clone(), self.get_namespace());
             api.patch(
                 &self.name_any(),
                 &PatchParams::default(),