refactor: Remove product-config, build config from typed ValidatedCluster

Replace the product-config-based validation pipeline with a typed, framework-
based one mirroring trino-operator:

- validate() now produces a ValidatedCluster { name, image, cluster_config,
  role_group_configs } via framework::role_utils::with_validated_config, instead
  of the product-config PropertyNameKind map. server.<myid> quorum entries are
  precomputed into cluster_config so the build step never needs the CRD.
- Add zk_controller/build/config_map.rs which builds zoo.cfg and
  security.properties directly from the typed config, referencing the
  ZookeeperCluster only for the owner reference. The operator defaults that
  product-config used to inject from deploy/config-spec/properties.yaml
  (admin.serverPort, clientPort, dataDir, initLimit, syncLimit, tickTime,
  metricsProvider.*, networkaddress.cache.*) are now seeded here, byte-identical
  to before (pinned by the kuttl ConfigMap snapshot).
- Drop the product_config Configuration impl from the CRD; add a manual Merge
  impl for ZookeeperConfigOverrides (CRD schema unchanged) and Ord for
  ZookeeperRole.
- Minimal consumer changes: the StatefulSet derives MYID_OFFSET/ZOOCFGDIR from
  the merged config plus envOverrides; the metrics Service takes a resolved port.
- Remove ProductConfigManager from main.rs/Ctx and the product-config dependency
  from both Cargo.toml. It remains only as a transitive dependency of
  stackable-operator itself.

Unit tests now exercise the product-config-free path and assert the seeded
defaults and security.properties byte-for-byte. The generated CRD is unchanged.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: maltesander

Cargo.lock

@@ -10,7 +10,6 @@ edition = "2024"
 repository = "https://github.com/stackabletech/zookeeper-operator"
 
 [workspace.dependencies]
-product-config = { git = "https://github.com/stackabletech/product-config.git", tag = "0.8.0" }
 stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "stackable-operator-0.111.1", features = ["webhook"] }
 
 anyhow = "1.0"

Cargo.toml

@@ -10,7 +10,6 @@ publish = false
 build = "build.rs"
 
 [dependencies]
-product-config.workspace = true
 stackable-operator.workspace = true
 
 anyhow.workspace = true

rust/operator-binary/Cargo.toml

@@ -25,7 +25,6 @@ use stackable_operator::{
     },
     kube::{CustomResource, ResourceExt, runtime::reflector::ObjectRef},
     memory::{BinaryMultiple, MemoryQuantity},
-    product_config_utils::{self, Configuration},
     product_logging::{self, spec::Logging},
     role_utils::{GenericRoleConfig, JavaCommonConfig, Role, RoleGroup, RoleGroupRef},
     schemars::{self, JsonSchema},
@@ -346,6 +345,34 @@ pub mod versioned {
     }
 }
 
+impl Merge for v1alpha1::ZookeeperConfigOverrides {
+    /// Merges `defaults` (role level) into `self` (role-group level). Keys present
+    /// in `self` win; keys only in `defaults` are added. Mirrors how the previous
+    /// product-config pipeline merged role and role-group `configOverrides`.
+    fn merge(&mut self, defaults: &Self) {
+        merge_key_value_overrides(&mut self.zoo_cfg, &defaults.zoo_cfg);
+        merge_key_value_overrides(&mut self.security_properties, &defaults.security_properties);
+    }
+}
+
+/// Merges two optional [`KeyValueConfigOverrides`], with `this` taking precedence.
+fn merge_key_value_overrides(
+    this: &mut Option<KeyValueConfigOverrides>,
+    defaults: &Option<KeyValueConfigOverrides>,
+) {
+    match (this.as_mut(), defaults) {
+        (Some(this), Some(defaults)) => {
+            for (key, value) in &defaults.overrides {
+                this.overrides
+                    .entry(key.clone())
+                    .or_insert_with(|| value.clone());
+            }
+        }
+        (None, Some(defaults)) => *this = Some(defaults.clone()),
+        (Some(_), None) | (None, None) => {}
+    }
+}
+
 impl KeyValueOverridesProvider for v1alpha1::ZookeeperConfigOverrides {
     fn get_key_value_overrides(&self, file: &str) -> BTreeMap<String, Option<String>> {
         let field = match file {
@@ -368,7 +395,9 @@ impl KeyValueOverridesProvider for v1alpha1::ZookeeperConfigOverrides {
     Eq,
     Hash,
     JsonSchema,
+    Ord,
     PartialEq,
+    PartialOrd,
     Serialize,
     EnumString,
 )]
@@ -417,7 +446,7 @@ impl v1alpha1::ZookeeperConfig {
     pub const SYNC_LIMIT: &'static str = "syncLimit";
     pub const TICK_TIME: &'static str = "tickTime";
 
-    fn default_server_config(
+    pub(crate) fn default_server_config(
         cluster_name: &str,
         role: &ZookeeperRole,
     ) -> v1alpha1::ZookeeperConfigFragment {
@@ -451,89 +480,6 @@ impl v1alpha1::ZookeeperConfig {
     }
 }
 
-impl Configuration for v1alpha1::ZookeeperConfigFragment {
-    type Configurable = v1alpha1::ZookeeperCluster;
-
-    fn compute_env(
-        &self,
-        resource: &Self::Configurable,
-        _role_name: &str,
-    ) -> Result<BTreeMap<String, Option<String>>, product_config_utils::Error> {
-        Ok([
-            (
-                v1alpha1::ZookeeperConfig::MYID_OFFSET.to_string(),
-                self.myid_offset
-                    .or(v1alpha1::ZookeeperConfig::default_server_config(
-                        &resource.name_any(),
-                        &ZookeeperRole::Server,
-                    )
-                    .myid_offset)
-                    .map(|myid_offset| myid_offset.to_string()),
-            ),
-            // This is used by zkEnv.sh and for the shell scripts in bin/
-            // If unset it tries to find the conf directory automatically and that fails
-            (
-                "ZOOCFGDIR".to_string(),
-                Some(STACKABLE_RW_CONFIG_DIR.to_string()),
-            ),
-        ]
-        .into())
-    }
-
-    fn compute_cli(
-        &self,
-        _resource: &Self::Configurable,
-        _role_name: &str,
-    ) -> Result<BTreeMap<String, Option<String>>, product_config_utils::Error> {
-        Ok(BTreeMap::new())
-    }
-
-    fn compute_files(
-        &self,
-        _resource: &Self::Configurable,
-        _role_name: &str,
-        file: &str,
-    ) -> Result<BTreeMap<String, Option<String>>, product_config_utils::Error> {
-        let mut result = BTreeMap::new();
-        if file == ZOOKEEPER_PROPERTIES_FILE {
-            if let Some(init_limit) = self.init_limit {
-                result.insert(
-                    v1alpha1::ZookeeperConfig::INIT_LIMIT.to_string(),
-                    Some(init_limit.to_string()),
-                );
-            }
-            if let Some(sync_limit) = self.sync_limit {
-                result.insert(
-                    v1alpha1::ZookeeperConfig::SYNC_LIMIT.to_string(),
-                    Some(sync_limit.to_string()),
-                );
-            }
-            if let Some(tick_time) = self.tick_time {
-                result.insert(
-                    v1alpha1::ZookeeperConfig::TICK_TIME.to_string(),
-                    Some(tick_time.to_string()),
-                );
-            }
-            result.insert(
-                v1alpha1::ZookeeperConfig::DATA_DIR.to_string(),
-                Some(STACKABLE_DATA_DIR.to_string()),
-            );
-            result.insert(
-                "metricsProvider.className".to_string(),
-                Some(
-                    "org.apache.zookeeper.metrics.prometheus.PrometheusMetricsProvider".to_string(),
-                ),
-            );
-            result.insert(
-                METRICS_PROVIDER_HTTP_PORT_KEY.to_string(),
-                Some(METRICS_PROVIDER_HTTP_PORT.to_string()),
-            );
-        }
-
-        Ok(result)
-    }
-}
-
 impl ZookeeperRole {
     pub fn roles() -> Vec<String> {
         let mut roles = vec![];

rust/operator-binary/src/crd/mod.rs

@@ -77,7 +77,7 @@ async fn main() -> anyhow::Result<()> {
         Command::Run(RunArguments {
             operator_environment,
             watch_namespace,
-            product_config,
+            product_config: _,
             maintenance,
             common,
         }) => {
@@ -124,11 +124,6 @@ async fn main() -> anyhow::Result<()> {
                 .run(sigterm_watcher.handle())
                 .map_err(|err| anyhow!(err).context("failed to run webhook server"));
 
-            let product_config = product_config.load(&[
-                "deploy/config-spec/properties.yaml",
-                "/etc/stackable/zookeeper-operator/config-spec/properties.yaml",
-            ])?;
-
             let zk_controller = Controller::new(
                 watch_namespace.get_api::<DeserializeGuard<v1alpha1::ZookeeperCluster>>(&client),
                 watcher::Config::default(),
@@ -161,7 +156,6 @@ async fn main() -> anyhow::Result<()> {
                     Arc::new(zk_controller::Ctx {
                         operator_environment: operator_environment.clone(),
                         client: client.clone(),
-                        product_config,
                     }),
                 )
                 // We can let the reporting happen in the background

rust/operator-binary/src/main.rs

@@ -1,10 +1,4 @@
-use std::{
-    collections::{BTreeMap, HashMap},
-    str::FromStr,
-};
-
-use product_config::types::PropertyNameKind;
-use snafu::{OptionExt, ResultExt, Snafu};
+use snafu::{ResultExt, Snafu};
 use stackable_operator::{
     builder::meta::ObjectMetaBuilder,
     commons::product_image_selection::ResolvedProductImage,
@@ -15,10 +9,9 @@ use stackable_operator::{
 
 use crate::{
     crd::{
-        APP_NAME, JMX_METRICS_PORT, JMX_METRICS_PORT_NAME, METRICS_PROVIDER_HTTP_PORT,
-        METRICS_PROVIDER_HTTP_PORT_KEY, METRICS_PROVIDER_HTTP_PORT_NAME, ZOOKEEPER_ELECTION_PORT,
-        ZOOKEEPER_ELECTION_PORT_NAME, ZOOKEEPER_LEADER_PORT, ZOOKEEPER_LEADER_PORT_NAME,
-        ZOOKEEPER_PROPERTIES_FILE, v1alpha1,
+        APP_NAME, JMX_METRICS_PORT, JMX_METRICS_PORT_NAME, METRICS_PROVIDER_HTTP_PORT_NAME,
+        ZOOKEEPER_ELECTION_PORT, ZOOKEEPER_ELECTION_PORT_NAME, ZOOKEEPER_LEADER_PORT,
+        ZOOKEEPER_LEADER_PORT_NAME, v1alpha1,
     },
     utils::build_recommended_labels,
     zk_controller::ZK_CONTROLLER_NAME,
@@ -40,12 +33,6 @@ pub enum Error {
     BuildLabel {
         source: stackable_operator::kvp::LabelError,
     },
-
-    #[snafu(display("missing zookeeper properties file {ZOOKEEPER_PROPERTIES_FILE} in config"))]
-    MissingPropertiesFile,
-
-    #[snafu(display("missing provider http port key {METRICS_PROVIDER_HTTP_PORT_KEY} in config"))]
-    MissingProviderHttpPortKey,
 }
 
 /// The rolegroup [`Service`] is a headless service that allows internal access to the instances of a certain rolegroup
@@ -110,10 +97,8 @@ pub(crate) fn build_server_rolegroup_metrics_service(
     zk: &v1alpha1::ZookeeperCluster,
     rolegroup: &RoleGroupRef<v1alpha1::ZookeeperCluster>,
     resolved_product_image: &ResolvedProductImage,
-    rolegroup_config: &HashMap<PropertyNameKind, BTreeMap<String, String>>,
+    metrics_port: u16,
 ) -> Result<Service, Error> {
-    let metrics_port = metrics_port_from_rolegroup_config(rolegroup_config)?;
-
     let metadata = ObjectMetaBuilder::new()
         .name_and_namespace(zk)
         .name(rolegroup.rolegroup_metrics_service_name())
@@ -166,29 +151,6 @@ pub(crate) fn build_server_rolegroup_metrics_service(
     })
 }
 
-pub(crate) fn metrics_port_from_rolegroup_config(
-    rolegroup_config: &HashMap<PropertyNameKind, BTreeMap<String, String>>,
-) -> Result<u16, Error> {
-    let metrics_port = rolegroup_config
-        .get(&PropertyNameKind::File(
-            ZOOKEEPER_PROPERTIES_FILE.to_string(),
-        ))
-        .context(MissingPropertiesFileSnafu)?
-        .get(METRICS_PROVIDER_HTTP_PORT_KEY)
-        .context(MissingProviderHttpPortKeySnafu)?;
-
-    let port = match u16::from_str(metrics_port) {
-        Ok(port) => port,
-        Err(err) => {
-            tracing::error!("{err}");
-            tracing::info!("Defaulting to using {METRICS_PROVIDER_HTTP_PORT} as metrics port.");
-            METRICS_PROVIDER_HTTP_PORT
-        }
-    };
-
-    Ok(port)
-}
-
 /// Common labels for Prometheus
 fn prometheus_labels() -> Labels {
     Labels::try_from([("prometheus.io/scrape", "true")]).expect("should be a valid label")