feat: Support objectOverrides

crates/stackable-operator/CHANGELOG.md

@@ -4,6 +4,21 @@ All notable changes to this project will be documented in this file.
 
 ## [Unreleased]
 
+### Added
+
+- Support `objectOverrides` ([#1118]).
+
+### Changed
+
+- BREAKING: `ClusterResources` now requires the objects added to implement `DeepMerge`.
+  This is very likely a stackable-operator internal change, but technically breaking ([#1118]).
+
+### Removed
+
+- BREAKING: `ClusterResources` no longer derives `Eq` and `PartialEq` ([#1118]).
+
+[#1118]: https://github.com/stackabletech/operator-rs/pull/1118
+
 ## [0.100.3] - 2025-10-31
 
 ### Changed

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,
@@ -42,6 +42,7 @@ use crate::{
         Label, LabelError, Labels,
         consts::{K8S_APP_INSTANCE_KEY, K8S_APP_MANAGED_BY_KEY, K8S_APP_NAME_KEY},
     },
+    patchinator::{self, ObjectOverrides, apply_patches},
     utils::format_full_controller_name,
 };
 
@@ -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>
@@ -332,6 +340,7 @@ impl ClusterResource for Deployment {
 /// use serde::{Deserialize, Serialize};
 /// use stackable_operator::client::Client;
 /// use stackable_operator::cluster_resources::{self, ClusterResourceApplyStrategy, ClusterResources};
+/// use stackable_operator::patchinator::ObjectOverrides;
 /// use stackable_operator::product_config_utils::ValidatedRoleConfigByPropertyKind;
 /// use stackable_operator::role_utils::Role;
 /// use std::sync::Arc;
@@ -348,7 +357,10 @@ impl ClusterResource for Deployment {
 ///     plural = "AppClusters",
 ///     namespaced,
 /// )]
-/// struct AppClusterSpec {}
+/// struct AppClusterSpec {
+///     #[serde(flatten)]
+///     pub object_overrides: ObjectOverrides,
+/// }
 ///
 /// enum Error {
 ///     CreateClusterResources {
@@ -371,6 +383,7 @@ impl ClusterResource for Deployment {
 ///         CONTROLLER_NAME,
 ///         &app.object_ref(&()),
 ///         ClusterResourceApplyStrategy::Default,
+///         &app.spec.object_overrides,
 ///     )
 ///     .map_err(|source| Error::CreateClusterResources { source })?;
 ///
@@ -413,8 +426,8 @@ impl ClusterResource for Deployment {
 ///     Ok(Action::await_change())
 /// }
 /// ```
-#[derive(Debug, Eq, PartialEq)]
-pub struct ClusterResources {
+#[derive(Debug)]
+pub struct ClusterResources<'a> {
     /// The namespace of the cluster
     namespace: String,
 
@@ -442,9 +455,12 @@ 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: &'a ObjectOverrides,
 }
 
-impl ClusterResources {
+impl<'a> ClusterResources<'a> {
     /// Constructs new `ClusterResources`.
     ///
     /// # Arguments
@@ -470,6 +486,7 @@ impl ClusterResources {
         controller_name: &str,
         cluster: &ObjectReference,
         apply_strategy: ClusterResourceApplyStrategy,
+        object_overrides: &'a ObjectOverrides,
     ) -> Result<Self> {
         let namespace = cluster
             .namespace
@@ -494,6 +511,7 @@ impl ClusterResources {
             manager: format_full_controller_name(operator_name, controller_name),
             resource_ids: Default::default(),
             apply_strategy,
+            object_overrides,
         })
     }
 
@@ -563,7 +581,10 @@ 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).context(ApplyObjectOverridesSnafu)?;
 
         let patched_resource = self
             .apply_strategy

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

@@ -1,7 +1,148 @@
-use crate::crd::listener::listeners::v1alpha1::ListenerSpec;
+use k8s_openapi::{DeepMerge, merge_strategies};
+
+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 DeepMerge for Listener {
+    fn merge_from(&mut self, other: Self) {
+        DeepMerge::merge_from(&mut self.metadata, other.metadata);
+        DeepMerge::merge_from(&mut self.spec, other.spec);
+        DeepMerge::merge_from(&mut self.status, other.status);
+    }
+}
+
+impl DeepMerge for ListenerSpec {
+    fn merge_from(&mut self, other: Self) {
+        DeepMerge::merge_from(&mut self.class_name, other.class_name);
+        merge_strategies::map::granular(
+            &mut self.extra_pod_selector_labels,
+            other.extra_pod_selector_labels,
+            |current_item, other_item| {
+                DeepMerge::merge_from(current_item, other_item);
+            },
+        );
+        merge_strategies::list::map(
+            &mut self.ports,
+            other.ports,
+            &[|lhs, rhs| lhs.name == rhs.name],
+            |current_item, other_item| {
+                DeepMerge::merge_from(current_item, other_item);
+            },
+        );
+        DeepMerge::merge_from(
+            &mut self.publish_not_ready_addresses,
+            other.publish_not_ready_addresses,
+        );
+    }
+}
+
+impl DeepMerge for ListenerStatus {
+    fn merge_from(&mut self, other: Self) {
+        DeepMerge::merge_from(&mut self.service_name, other.service_name);
+        merge_strategies::list::map(
+            &mut self.ingress_addresses,
+            other.ingress_addresses,
+            &[|lhs, rhs| lhs.address == rhs.address],
+            |current_item, other_item| {
+                DeepMerge::merge_from(current_item, other_item);
+            },
+        );
+        merge_strategies::map::granular(
+            &mut self.node_ports,
+            other.node_ports,
+            |current_item, other_item| {
+                DeepMerge::merge_from(current_item, other_item);
+            },
+        );
+    }
+}
+
+impl DeepMerge for ListenerIngress {
+    fn merge_from(&mut self, other: Self) {
+        DeepMerge::merge_from(&mut self.address, other.address);
+        self.address_type = other.address_type;
+        merge_strategies::map::granular(
+            &mut self.ports,
+            other.ports,
+            |current_item, other_item| {
+                DeepMerge::merge_from(current_item, other_item);
+            },
+        );
+    }
+}
+
+impl DeepMerge for ListenerPort {
+    fn merge_from(&mut self, other: Self) {
+        DeepMerge::merge_from(&mut self.name, other.name);
+        DeepMerge::merge_from(&mut self.port, other.port);
+        DeepMerge::merge_from(&mut self.protocol, other.protocol);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn deep_merge_listener() {
+        let mut base: ListenerSpec = serde_yaml::from_str(
+            "
+className: my-listener-class
+extraPodSelectorLabels:
+  foo: bar
+ports:
+  - name: http
+    port: 8080
+    protocol: http
+  - name: https
+    port: 8080
+    protocol: https
+# publishNotReadyAddresses defaults to true
+",
+        )
+        .unwrap();
+
+        let patch: ListenerSpec = serde_yaml::from_str(
+            "
+className: custom-listener-class
+extraPodSelectorLabels:
+  foo: overridden
+  extra: label
+ports:
+  - name: https
+    port: 8443
+publishNotReadyAddresses: false
+",
+        )
+        .unwrap();
+
+        base.merge_from(patch);
+
+        let expected: ListenerSpec = serde_yaml::from_str(
+            "
+className: custom-listener-class
+extraPodSelectorLabels:
+  foo: overridden
+  extra: label
+ports:
+  - name: http
+    port: 8080
+    protocol: http
+  - name: https
+    port: 8443 # overridden
+    protocol: https
+publishNotReadyAddresses: false
+",
+        )
+        .unwrap();
+
+        assert_eq!(base, expected);
+    }
+}

crates/stackable-operator/src/lib.rs

@@ -22,6 +22,7 @@ pub mod kvp;
 pub mod logging;
 pub mod memory;
 pub mod namespace;
+pub mod patchinator;
 pub mod pod_utils;
 pub mod product_config_utils;
 pub mod product_logging;

crates/stackable-operator/src/patchinator/crd.rs

@@ -0,0 +1,14 @@
+use kube::api::DynamicObject;
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+
+use crate::utils::crds::raw_object_list_schema;
+
+#[derive(Clone, Debug, Deserialize, JsonSchema, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ObjectOverrides {
+    /// TODO docs
+    #[serde(default)]
+    #[schemars(schema_with = "raw_object_list_schema")]
+    pub object_overrides: Vec<DynamicObject>,
+}