feat: Support objectOverrides

Author: sbernauer

crates/stackable-operator/src/cluster_resources.rs

@@ -8,7 +8,7 @@ use std::{
 #[cfg(doc)]
 use k8s_openapi::api::core::v1::{NodeSelector, Pod};
 use k8s_openapi::{
-    NamespaceResourceScope,
+    DeepMerge, NamespaceResourceScope,
     api::{
         apps::v1::{
             DaemonSet, DaemonSetSpec, Deployment, DeploymentSpec, StatefulSet, StatefulSetSpec,
@@ -22,9 +22,10 @@ use k8s_openapi::{
     },
     apimachinery::pkg::apis::meta::v1::{LabelSelector, LabelSelectorRequirement},
 };
-use kube::{Resource, ResourceExt, core::ErrorResponse};
+use kube::{Resource, ResourceExt, api::DynamicObject, core::ErrorResponse};
 use serde::{Serialize, de::DeserializeOwned};
 use snafu::{OptionExt, ResultExt, Snafu};
+use stackable_shared::patchinator::{self, apply_patches, parse_patches};
 use strum::Display;
 use tracing::{debug, info, warn};
 
@@ -87,6 +88,12 @@ pub enum Error {
         #[snafu(source(from(crate::client::Error, Box::new)))]
         source: Box<crate::client::Error>,
     },
+
+    #[snafu(display("failed to parse user-provided object overrides"))]
+    ParseObjectOverrides { source: patchinator::Error },
+
+    #[snafu(display("failed to apply user-provided object overrides"))]
+    ApplyObjectOverrides { source: patchinator::Error },
 }
 
 /// A cluster resource handled by [`ClusterResources`].
@@ -97,6 +104,7 @@ pub enum Error {
 /// it must be added to [`ClusterResources::delete_orphaned_resources`] as well.
 pub trait ClusterResource:
     Clone
+    + DeepMerge
     + Debug
     + DeserializeOwned
     + Resource<DynamicType = (), Scope = NamespaceResourceScope>
@@ -413,7 +421,7 @@ impl ClusterResource for Deployment {
 ///     Ok(Action::await_change())
 /// }
 /// ```
-#[derive(Debug, Eq, PartialEq)]
+#[derive(Debug)]
 pub struct ClusterResources {
     /// The namespace of the cluster
     namespace: String,
@@ -442,6 +450,9 @@ pub struct ClusterResources {
     /// Strategy to manage how cluster resources are applied. Resources could be patched, merged
     /// or not applied at all depending on the strategy.
     apply_strategy: ClusterResourceApplyStrategy,
+
+    /// Arbitrary Kubernetes object overrides specified by the user via the CRD.
+    object_overrides: Vec<DynamicObject>,
 }
 
 impl ClusterResources {
@@ -470,6 +481,7 @@ impl ClusterResources {
         controller_name: &str,
         cluster: &ObjectReference,
         apply_strategy: ClusterResourceApplyStrategy,
+        object_overrides: Option<impl AsRef<str>>,
     ) -> Result<Self> {
         let namespace = cluster
             .namespace
@@ -483,6 +495,12 @@ impl ClusterResources {
             .uid
             .clone()
             .context(MissingObjectKeySnafu { key: "uid" })?;
+        let object_overrides = match object_overrides {
+            Some(object_overrides) => {
+                parse_patches(object_overrides).context(ParseObjectOverridesSnafu)?
+            }
+            None => vec![],
+        };
 
         Ok(ClusterResources {
             namespace,
@@ -494,6 +512,7 @@ impl ClusterResources {
             manager: format_full_controller_name(operator_name, controller_name),
             resource_ids: Default::default(),
             apply_strategy,
+            object_overrides,
         })
     }
 
@@ -563,7 +582,11 @@ impl ClusterResources {
                 .unwrap_or_else(|err| warn!("{}", err));
         }
 
-        let mutated = resource.maybe_mutate(&self.apply_strategy);
+        let mut mutated = resource.maybe_mutate(&self.apply_strategy);
+
+        // We apply the object overrides of the user at the very last to offer maximum flexibility.
+        apply_patches(&mut mutated, self.object_overrides.iter())
+            .context(ApplyObjectOverridesSnafu)?;
 
         let patched_resource = self
             .apply_strategy

crates/stackable-operator/src/crd/listener/listeners/v1alpha1_impl.rs

@@ -1,7 +1,94 @@
-use crate::crd::listener::listeners::v1alpha1::ListenerSpec;
+use crate::crd::listener::listeners::v1alpha1::{
+    Listener, ListenerIngress, ListenerPort, ListenerSpec, ListenerStatus,
+};
 
 impl ListenerSpec {
     pub(super) const fn default_publish_not_ready_addresses() -> Option<bool> {
         Some(true)
     }
 }
+
+impl k8s_openapi::DeepMerge for Listener {
+    fn merge_from(&mut self, other: Self) {
+        k8s_openapi::DeepMerge::merge_from(&mut self.metadata, other.metadata);
+        k8s_openapi::DeepMerge::merge_from(&mut self.spec, other.spec);
+        k8s_openapi::DeepMerge::merge_from(&mut self.status, other.status);
+    }
+}
+
+impl k8s_openapi::DeepMerge for ListenerSpec {
+    fn merge_from(&mut self, other: Self) {
+        k8s_openapi::DeepMerge::merge_from(&mut self.class_name, other.class_name);
+        k8s_openapi::merge_strategies::map::granular(
+            &mut self.extra_pod_selector_labels,
+            other.extra_pod_selector_labels,
+            |current_item, other_item| {
+                k8s_openapi::DeepMerge::merge_from(current_item, other_item);
+            },
+        );
+        k8s_openapi::merge_strategies::list::map(
+            &mut self.ports,
+            other.ports,
+            &[|lhs, rhs| lhs.name == rhs.name],
+            |current_item, other_item| {
+                k8s_openapi::DeepMerge::merge_from(current_item, other_item);
+            },
+        );
+        k8s_openapi::DeepMerge::merge_from(
+            &mut self.publish_not_ready_addresses,
+            other.publish_not_ready_addresses,
+        );
+        todo!()
+    }
+}
+
+impl k8s_openapi::DeepMerge for ListenerStatus {
+    fn merge_from(&mut self, other: Self) {
+        k8s_openapi::DeepMerge::merge_from(&mut self.service_name, other.service_name);
+        k8s_openapi::merge_strategies::list::map(
+            &mut self.ingress_addresses,
+            other.ingress_addresses,
+            &[|lhs, rhs| lhs.address == rhs.address],
+            |current_item, other_item| {
+                k8s_openapi::DeepMerge::merge_from(current_item, other_item);
+            },
+        );
+        k8s_openapi::merge_strategies::map::granular(
+            &mut self.node_ports,
+            other.node_ports,
+            |current_item, other_item| {
+                k8s_openapi::DeepMerge::merge_from(current_item, other_item);
+            },
+        );
+    }
+}
+
+impl k8s_openapi::DeepMerge for ListenerIngress {
+    fn merge_from(&mut self, other: Self) {
+        k8s_openapi::DeepMerge::merge_from(&mut self.address, other.address);
+        self.address_type = other.address_type;
+        k8s_openapi::merge_strategies::map::granular(
+            &mut self.ports,
+            other.ports,
+            |current_item, other_item| {
+                k8s_openapi::DeepMerge::merge_from(current_item, other_item);
+            },
+        );
+    }
+}
+
+impl k8s_openapi::DeepMerge for ListenerPort {
+    fn merge_from(&mut self, other: Self) {
+        k8s_openapi::DeepMerge::merge_from(&mut self.name, other.name);
+        k8s_openapi::DeepMerge::merge_from(&mut self.port, other.port);
+        k8s_openapi::DeepMerge::merge_from(&mut self.protocol, other.protocol);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #[test]
+    fn deep_merge_listener() {
+        todo!("Add some basic tests for merging");
+    }
+}

crates/stackable-shared/src/lib.rs

@@ -2,7 +2,7 @@
 //! workspace.
 
 pub mod crd;
+pub mod patchinator;
 pub mod secret;
-
 pub mod time;
 pub mod yaml;