feat: Support objectOverrides (#1118)

* feat: Support `objectOverrides`

* refactor: Switch to a lis of objects (as opposed to a big string field)

* changelog

* Add TODO for docs

* Add a test for Listener merging

* Fix doctests

* Improve CRD docs

* Remove unused error variant

* Add to DummyCluster

* Improve CRD docs

* Link to concepts page

* Derive PartialEq again

* Move import

* Take owned value

* Update crates/stackable-operator/src/cluster_resources.rs

Co-authored-by: Techassi <[email protected]>

* PartialEq again

* Improve changelog

* Add a comment in DeepMerge impl

* Add some rustdocs

* patchinator -> deep_merger

* Fix remaining "patch" mentions

* Fixup accidential change

* Rename patch -> merge

* Use indoc for tests

* refactor: Move stuff into ObjectOverrides::apply_to

* refactor: Switch ObjectOverrides to unit struct

* fix tests

* Update crates/stackable-operator/CHANGELOG.md

* Merge the merge into the onto

---------

Co-authored-by: Techassi <[email protected]>

Author: sbernauer

Cargo.lock

@@ -4,6 +4,22 @@ All notable changes to this project will be documented in this file.
 
 ## [Unreleased]
 
+### Added
+
+- Support `objectOverrides`, a list of generic Kubernetes objects, which are merged into the objects created by the operator.
+  Alongside, a `deep_merger` module was added, which takes a Kubernetes object and a list of overrides and merges them into the provided object ([#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` ([#1118]).
+
+[#1118]: https://github.com/stackabletech/operator-rs/pull/1118
+
 ## [0.100.3] - 2025-10-31
 
 ### Changed

crates/stackable-operator/CHANGELOG.md

@@ -55,5 +55,6 @@ tracing-subscriber.workspace = true
 url.workspace = true
 
 [dev-dependencies]
+indoc.workspace = true
 rstest.workspace = true
 tempfile.workspace = true

crates/stackable-operator/Cargo.toml

@@ -634,6 +634,20 @@ spec:
                 required:
                 - roleGroups
                 type: object
+              objectOverrides:
+                default: []
+                description: |-
+                  A list of generic Kubernetes objects, which are merged into the objects that the operator
+                  creates.
+
+                  List entries are arbitrary YAML objects, which need to be valid Kubernetes objects.
+
+                  Read the [Object overrides documentation](https://docs.stackable.tech/home/nightly/concepts/overrides#object-overrides)
+                  for more information.
+                items:
+                  type: object
+                  x-kubernetes-preserve-unknown-fields: true
+                type: array
               opaConfig:
                 description: |-
                   Configure the OPA stacklet [discovery ConfigMap](https://docs.stackable.tech/home/nightly/concepts/service_discovery)

crates/stackable-operator/crds/DummyCluster.yaml

@@ -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,
@@ -38,6 +38,7 @@ use crate::{
         },
     },
     crd::listener,
+    deep_merger::{self, ObjectOverrides},
     kvp::{
         Label, LabelError, Labels,
         consts::{K8S_APP_INSTANCE_KEY, K8S_APP_MANAGED_BY_KEY, K8S_APP_NAME_KEY},
@@ -87,6 +88,9 @@ pub enum Error {
         #[snafu(source(from(crate::client::Error, Box::new)))]
         source: Box<crate::client::Error>,
     },
+
+    #[snafu(display("failed to apply user-provided object overrides"))]
+    ApplyObjectOverrides { source: deep_merger::Error },
 }
 
 /// A cluster resource handled by [`ClusterResources`].
@@ -97,6 +101,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 +337,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::deep_merger::ObjectOverrides;
 /// use stackable_operator::product_config_utils::ValidatedRoleConfigByPropertyKind;
 /// use stackable_operator::role_utils::Role;
 /// use std::sync::Arc;
@@ -348,7 +354,10 @@ impl ClusterResource for Deployment {
 ///     plural = "AppClusters",
 ///     namespaced,
 /// )]
-/// struct AppClusterSpec {}
+/// struct AppClusterSpec {
+///     #[serde(default)]
+///     pub object_overrides: ObjectOverrides,
+/// }
 ///
 /// enum Error {
 ///     CreateClusterResources {
@@ -371,6 +380,7 @@ impl ClusterResource for Deployment {
 ///         CONTROLLER_NAME,
 ///         &app.object_ref(&()),
 ///         ClusterResourceApplyStrategy::Default,
+///         app.spec.object_overrides.clone(),
 ///     )
 ///     .map_err(|source| Error::CreateClusterResources { source })?;
 ///
@@ -413,7 +423,7 @@ impl ClusterResource for Deployment {
 ///     Ok(Action::await_change())
 /// }
 /// ```
-#[derive(Debug, Eq, PartialEq)]
+#[derive(Debug, PartialEq)]
 pub struct ClusterResources {
     /// The namespace of the cluster
     namespace: String,
@@ -442,6 +452,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: ObjectOverrides,
 }
 
 impl ClusterResources {
@@ -470,6 +483,7 @@ impl ClusterResources {
         controller_name: &str,
         cluster: &ObjectReference,
         apply_strategy: ClusterResourceApplyStrategy,
+        object_overrides: ObjectOverrides,
     ) -> Result<Self> {
         let namespace = cluster
             .namespace
@@ -494,6 +508,7 @@ impl ClusterResources {
             manager: format_full_controller_name(operator_name, controller_name),
             resource_ids: Default::default(),
             apply_strategy,
+            object_overrides,
         })
     }
 
@@ -563,7 +578,12 @@ 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 end to offer maximum flexibility.
+        self.object_overrides
+            .apply_to(&mut mutated)
+            .context(ApplyObjectOverridesSnafu)?;
 
         let patched_resource = self
             .apply_strategy

crates/stackable-operator/src/cluster_resources.rs

@@ -1,7 +1,145 @@
-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,
+            // The unique thing identifying a port is it's name
+            &[|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,
+            // The unique thing identifying an ingress address is it's address
+            &[|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 indoc::indoc;
+
+    use super::*;
+
+    #[test]
+    fn deep_merge_listener() {
+        let mut base: ListenerSpec = serde_yaml::from_str(indoc! {"
+            className: my-listener-class
+            extraPodSelectorLabels:
+            foo: bar
+            ports:
+              - name: http
+                port: 8080
+                protocol: http
+              - name: https
+                port: 8080
+                protocol: https
+            # publishNotReadyAddresses defaults to true
+        "})
+        .expect("test YAML is valid");
+
+        let merge: ListenerSpec = serde_yaml::from_str(indoc! {"
+            className: custom-listener-class
+            extraPodSelectorLabels:
+            foo: overridden
+            extra: label
+            ports:
+              - name: https
+                port: 8443
+            publishNotReadyAddresses: false
+        "})
+        .expect("test YAML is valid");
+
+        base.merge_from(merge);
+        let expected: ListenerSpec = serde_yaml::from_str(indoc! {"
+            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
+        "})
+        .expect("test YAML is valid");
+
+        assert_eq!(base, expected);
+    }
+}

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

@@ -0,0 +1,39 @@
+use k8s_openapi::DeepMerge;
+use kube::api::DynamicObject;
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize, de::DeserializeOwned};
+
+use super::apply_deep_merge;
+use crate::utils::crds::raw_object_list_schema;
+
+#[derive(Clone, Debug, Deserialize, Default, JsonSchema, Serialize, PartialEq)]
+pub struct ObjectOverrides(
+    /// A list of generic Kubernetes objects, which are merged into the objects that the operator
+    /// creates.
+    ///
+    /// List entries are arbitrary YAML objects, which need to be valid Kubernetes objects.
+    ///
+    /// Read the [Object overrides documentation](DOCS_BASE_URL_PLACEHOLDER/concepts/overrides#object-overrides)
+    /// for more information.
+    //
+    // Remember to use `#[serde(default)]` when including this into a CRD!
+    #[schemars(schema_with = "raw_object_list_schema")]
+    Vec<DynamicObject>,
+);
+
+impl ObjectOverrides {
+    /// Takes an arbitrary Kubernetes object (`base`) and applies the configured list of deep merges
+    /// to it.
+    ///
+    /// Merges are only applied to objects that have the same apiVersion, kind, name
+    /// and namespace.
+    pub fn apply_to<R>(&self, base: &mut R) -> Result<(), super::Error>
+    where
+        R: kube::Resource<DynamicType = ()> + DeepMerge + DeserializeOwned,
+    {
+        for object_override in &self.0 {
+            apply_deep_merge(base, object_override)?;
+        }
+        Ok(())
+    }
+}