- Add support to choose the sync strategy resp. ownership per destination and switch the default behaviour from replace to apply.

alex-berger · alex-berger · commit a6fa35569d04 · 2021-08-17T10:34:32.000+02:00
- Fix a bug where `uid` and `resourceVersion` was not cleared in create requests if destination objects was deleted.
diff --git a/README.md b/README.md
@@ -105,7 +105,7 @@ spec:
 kind: ObjectSync
 apiVersion: sync.rustrial.org/v1alpha1
 metadata:
-  name: my-test-config-map-distributor
+  name: my-test-secret-map-distributor
   namespace: default
 spec:
   source:
@@ -123,7 +123,7 @@ spec:
 kind: ObjectSync
 apiVersion: sync.rustrial.org/v1alpha1
 metadata:
-  name: my-test-config-map-distributor
+  name: my-test-cronjob-map-distributor
   namespace: default
 spec:
   source:
@@ -141,7 +141,7 @@ spec:
 kind: ObjectSync
 apiVersion: sync.rustrial.org/v1alpha1
 metadata:
-  name: my-test-config-map-distributor
+  name: my-test-hr-map-distributor
   namespace: default
 spec:
   source:
@@ -191,6 +191,35 @@ make sure it obtains all `delete` events.
 
 ---
 
+## Ownership
+
+For each destination a sync strategy can be defined, which is either 
+[sever side apply](https://kubernetes.io/docs/reference/using-api/server-side-apply/) 
+for shared ownership or *replace* for exclusive ownership. 
+By default the controller uses shared ownership (*server side apply* with the `force` flag)
+to manage sync destination objects, allowing other controllers and users to manage fields
+not set in the source object.
+
+Set the strategy to `replace` if you want the controller to take *exclusive* ownership for a destination.
+
+```yaml
+kind: ObjectSync
+apiVersion: sync.rustrial.org/v1alpha1
+metadata:
+  name: my-test-config-map-distributor
+  namespace: default
+spec:
+  source:
+    group: ""
+    kind: ConfigMap
+    name: my-namespace
+  destinations:
+  - namespace: "*"
+    strategy: "replace"
+```
+
+---
+
 ## License
 
 Licensed under either of
diff --git a/charts/k8s-object-syncer/crds/crds.yaml b/charts/k8s-object-syncer/crds/crds.yaml
@@ -40,6 +40,13 @@ spec:
                       namespace:
                         description: "The destination (target) namespace, if empty `\"\"` or `\"*\"` the source object is synced to all namespaces."
                         type: string
+                      strategy:
+                        description: "The sync strategy to use for this destination, defaults to \"apply\" (server side apply)."
+                        enum:
+                          - apply
+                          - replace
+                        nullable: true
+                        type: string
                     required:
                       - namespace
                     type: object
@@ -139,6 +146,13 @@ spec:
                             nullable: true
                             type: string
                         type: object
+                      strategy:
+                        description: "The [`SyncStrategy`] applied to this destination."
+                        enum:
+                          - apply
+                          - replace
+                        nullable: true
+                        type: string
                       syncedVersion:
                         description: "The last source version syced, `None` if not yet synced."
                         nullable: true
diff --git a/rustrial-k8s-object-syncer-apis/src/lib.rs b/rustrial-k8s-object-syncer-apis/src/lib.rs
@@ -108,6 +108,26 @@ pub struct SourceObject {
     pub namespace: Option<String>,
 }
 
+/// Destination object synchronization strategy.
+#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq, Eq, Hash, JsonSchema)]
+pub enum SyncStrategy {
+    /// Use [server side apply](https://kubernetes.io/docs/reference/using-api/server-side-apply/) to
+    /// manage (sync) destination objects with a shared ownership model
+    /// (only overwriting changes in fields present in source object).
+    #[serde(rename = "apply")]
+    Apply,
+    /// Use replace to manage (sync) destination objects with exclusive ownership
+    /// (overwriting all changes made by others).
+    #[serde(rename = "replace")]
+    Replace,
+}
+
+impl Default for SyncStrategy {
+    fn default() -> Self {
+        Self::Apply
+    }
+}
+
 /// Synchronization target configuration.
 #[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Hash, Clone, JsonSchema)]
 pub struct Destination {
@@ -118,9 +138,16 @@ pub struct Destination {
     /// The destination (target) namespace, if empty `""` or `"*"` the source object
     /// is synced to all namespaces.
     pub namespace: String,
+    /// The sync strategy to use for this destination, defaults to "apply" (server side apply).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub strategy: Option<SyncStrategy>,
 }
 
 impl Destination {
+    pub fn strategy(&self) -> SyncStrategy {
+        self.strategy.unwrap_or_default()
+    }
+
     pub fn applies_to_all_namespaces(&self) -> bool {
         self.namespace.as_str() == "" || self.namespace.as_str() == "*"
     }
@@ -196,6 +223,15 @@ pub struct DestinationStatus {
     /// The last source version syced, `None` if not yet synced.
     #[serde(skip_serializing_if = "Option::is_none", rename = "syncedVersion")]
     pub synced_version: Option<ObjectRevision>,
+    /// The [`SyncStrategy`] applied to this destination.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub strategy: Option<SyncStrategy>,
+}
+
+impl DestinationStatus {
+    pub fn strategy(&self) -> SyncStrategy {
+        self.strategy.unwrap_or_default()
+    }
 }
 
 impl PartialOrd for DestinationStatus {
@@ -326,4 +362,16 @@ mod tests {
             serde_json::to_string(&p).unwrap()
         );
     }
+
+    #[test]
+    fn sync_strategy() {
+        assert_eq!(
+            r#""apply""#,
+            serde_json::to_string(&SyncStrategy::Apply).unwrap()
+        );
+        assert_eq!(
+            r#""replace""#,
+            serde_json::to_string(&SyncStrategy::Replace).unwrap()
+        );
+    }
 }
diff --git a/rustrial-k8s-object-syncer/src/resource_controller.rs b/rustrial-k8s-object-syncer/src/resource_controller.rs
@@ -13,7 +13,10 @@ use futures::{
 };
 use k8s_openapi::{api::core::v1::Namespace, chrono::Utc};
 use kube::{
-    api::{ApiResource, DynamicObject, GroupVersionKind, ListParams, PostParams, TypeMeta},
+    api::{
+        ApiResource, DynamicObject, GroupVersionKind, ListParams, Patch, PatchParams, PostParams,
+        TypeMeta,
+    },
     Api, Client, ResourceExt,
 };
 use kube_runtime::{
@@ -27,7 +30,8 @@ use opentelemetry::{
     KeyValue,
 };
 use rustrial_k8s_object_syncer_apis::{
-    Condition, DestinationStatus, ObjectRevision, ObjectSync, ObjectSyncSpec, API_GROUP,
+    Condition, DestinationStatus, ObjectRevision, ObjectSync, ObjectSyncSpec, SyncStrategy,
+    API_GROUP,
 };
 use std::{
     borrow::BorrowMut,
@@ -212,27 +216,27 @@ impl ResourceControllerImpl {
     fn expected_destinations<'a>(
         &self,
         event: &'a ObjectSyncModifications,
-    ) -> impl Iterator<Item = (String, String)> + 'a {
+    ) -> impl Iterator<Item = (String, String, Option<SyncStrategy>)> + 'a {
         let spec: &ObjectSyncSpec = &event.spec;
         let cache = self.namespace_cache.state();
         spec.destinations.iter().flat_map(move |d| {
-            let mut tmp: Vec<(String, String)> = Default::default();
+            let mut tmp: Vec<(String, String, Option<SyncStrategy>)> = Default::default();
             if d.applies_to_all_namespaces() {
                 for ns in cache.iter() {
                     // Make sure we skip deleted namespaces, as otherwise the finalizers on the synced
                     // destination objects will prevent the namespace from being deleted.
                     if ns.metadata.deletion_timestamp.is_none() {
-                        if let Some(x) = d.applies_to(event, ns.name().as_str()) {
-                            tmp.push(x);
+                        if let Some((ns, name)) = d.applies_to(event, ns.name().as_str()) {
+                            tmp.push((ns, name, d.strategy));
                         }
                     }
                 }
-            } else if let Some(x) = d.applies_to(event, d.namespace.as_str()) {
+            } else if let Some((namespace, name)) = d.applies_to(event, d.namespace.as_str()) {
                 if let Some(ns) = cache.iter().find(|ns| ns.name() == d.namespace) {
                     // Make sure we skip deleted namespaces, as otherwise the finalizers on the synced
                     // destination objects will prevent the namespace from being deleted.
                     if ns.metadata.deletion_timestamp.is_none() {
-                        tmp.push(x);
+                        tmp.push((namespace, name, d.strategy));
                     }
                 }
             }
@@ -280,8 +284,6 @@ impl ResourceControllerImpl {
             }
             template.metadata.namespace = Some(d.namespace.clone());
             template.metadata.name = Some(d.name.clone());
-            template.metadata.uid = Default::default();
-            template.metadata.resource_version = Default::default();
             template.metadata.generation = Default::default();
             template.metadata.generate_name = Default::default();
             template.metadata.managed_fields = Default::default();
@@ -293,6 +295,8 @@ impl ResourceControllerImpl {
             let api = self.namespaced_api(d.namespace.as_str());
             let mut retry_attempts = 3i32;
             while retry_attempts > 0 {
+                template.metadata.uid = Default::default();
+                template.metadata.resource_version = Default::default();
                 retry_attempts -= 1;
                 match api.get(d.name.as_str()).await {
                     Ok(mut current) => {
@@ -317,11 +321,24 @@ impl ResourceControllerImpl {
                             resource_version: current.resource_version(),
                         };
                         if &Some(dst_version) != &d.synced_version
-                            || &Some(src_version.clone()) != &d.source_version
+                            || &Some(&src_version) != &d.source_version.as_ref()
                         {
                             template.metadata.uid = current.uid();
                             template.metadata.resource_version = current.resource_version();
-                            match api.replace(d.name.as_str(), &pp, &template).await {
+
+                            let result = match &d.strategy() {
+                                SyncStrategy::Replace => {
+                                    api.replace(d.name.as_str(), &pp, &template).await
+                                }
+                                SyncStrategy::Apply => {
+                                    let mut pp = PatchParams::default();
+                                    pp.field_manager = Some(MANAGER.to_string());
+                                    pp.force = true;
+                                    api.patch(d.name.as_str(), &pp, &Patch::Apply(&template))
+                                        .await
+                                }
+                            };
+                            match result {
                                 Ok(updated) => {
                                     d.synced_version = Some(ObjectRevision {
                                         uid: updated.uid(),
@@ -410,22 +427,32 @@ impl ResourceControllerImpl {
             .flatten()
             .unwrap_or_default();
         let mut expected_destinations: Vec<DestinationStatus> = Default::default();
-        for (dst_namespace, dst_name) in self.expected_destinations(event) {
-            let expected_dst = DestinationStatus {
+        for (dst_namespace, dst_name, strategy) in self.expected_destinations(event) {
+            let mut expected_dst = DestinationStatus {
                 name: dst_name,
                 namespace: dst_namespace,
                 source_version: None,
                 synced_version: None,
                 group: self.gvk.group.clone(),
                 version: self.gvk.version.clone(),
                 kind: self.gvk.kind.clone(),
+                strategy,
             };
-            let expected_dst = stale_destinations
+            if let Some(status) = stale_destinations
                 .iter()
                 .find(|d| Self::is_same_destination(d, &expected_dst))
-                .map(|d| (*d).clone())
-                .unwrap_or(expected_dst);
-            stale_destinations.retain(|d| !Self::is_same_destination(d, &expected_dst));
+            {
+                // If strategy (sync config) changed do not set version to make sure the destination
+                // object is update. Note, this is required as we cannot track the ObjectSync's resourceVersion
+                // in its own status as this would lead to an infinit reconciliation cycle.
+                if status.strategy() == expected_dst.strategy() {
+                    expected_dst.source_version = status.source_version.clone();
+                    expected_dst.synced_version = status.synced_version.clone();
+                }
+                // As destination is in set of expected destinations, remove it from the set of
+                // stale destinations.
+                stale_destinations.retain(|d| !Self::is_same_destination(d, &expected_dst));
+            }
             expected_destinations.push(expected_dst);
         }
         // 1. Remove stale destinations.
