refactor: introduce ValidatedRoleConfig

Author: maltesander

rust/operator-binary/src/controller.rs

@@ -27,7 +27,6 @@ use stackable_operator::{
     },
     kvp::Labels,
     logging::controller::ReconcilerError,
-    role_utils::GenericRoleConfig,
     shared::time::Duration,
     status::condition::{
         compute_conditions, operations::ClusterOperationsConditionBuilder,
@@ -119,17 +118,21 @@ pub struct ValidatedCluster {
     /// value.
     pub product_version: ProductVersion,
     pub cluster_config: ValidatedClusterConfig,
+    /// Per-role configuration (e.g. the Pod disruption budget), keyed by role.
+    pub role_configs: BTreeMap<KafkaRole, ValidatedRoleConfig>,
     pub role_group_configs: BTreeMap<KafkaRole, BTreeMap<RoleGroupName, ValidatedRoleGroupConfig>>,
 }
 
 impl ValidatedCluster {
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         name: ClusterName,
         namespace: NamespaceName,
         uid: Uid,
         cluster_domain: DomainName,
         image: ResolvedProductImage,
         cluster_config: ValidatedClusterConfig,
+        role_configs: BTreeMap<KafkaRole, ValidatedRoleConfig>,
         role_group_configs: BTreeMap<KafkaRole, BTreeMap<RoleGroupName, ValidatedRoleGroupConfig>>,
     ) -> Self {
         // `app_version_label_value` is constructed to be a valid label value, so it is also a
@@ -150,6 +153,7 @@ impl ValidatedCluster {
             image,
             product_version,
             cluster_config,
+            role_configs,
             role_group_configs,
         }
     }
@@ -365,6 +369,15 @@ impl ValidatedClusterConfig {
     }
 }
 
+/// Per-role configuration extracted during validation.
+///
+/// Resolved from the raw [`v1alpha1::KafkaCluster`] spec during validation so the reconcile loop
+/// never has to read it. Kafka's `GenericRoleConfig` only carries the Pod disruption budget.
+#[derive(Clone, Debug)]
+pub struct ValidatedRoleConfig {
+    pub pdb: stackable_operator::commons::pdb::PdbConfig,
+}
+
 /// Lets [`ValidatedCluster`] act as the owner [`Resource`] for child objects, so owner
 /// references are built from it (via the captured `metadata`) rather than the raw CR.
 impl Resource for ValidatedCluster {
@@ -706,11 +719,8 @@ pub async fn reconcile_kafka(
             );
         }
 
-        let role_cfg = kafka.role_config(kafka_role);
-        if let Some(GenericRoleConfig {
-            pod_disruption_budget: pdb,
-        }) = role_cfg
-            && let Some(pdb) = build_pdb(pdb, &validated_cluster, kafka_role)
+        if let Some(role_config) = validated_cluster.role_configs.get(kafka_role)
+            && let Some(pdb) = build_pdb(&role_config.pdb, &validated_cluster, kafka_role)
         {
             cluster_resources
                 .add(client, pdb)

rust/operator-binary/src/controller/validate.rs

@@ -29,8 +29,8 @@ use stackable_operator::{
 
 use crate::{
     controller::{
-        RoleGroupName, ValidatedCluster, ValidatedClusterConfig, ValidatedRoleGroupConfig,
-        dereference::DereferencedObjects,
+        RoleGroupName, ValidatedCluster, ValidatedClusterConfig, ValidatedRoleConfig,
+        ValidatedRoleGroupConfig, dereference::DereferencedObjects,
     },
     crd::{
         self, CONTAINER_IMAGE_BASE_NAME,
@@ -228,6 +228,7 @@ pub fn validate(
         .vector_aggregator_config_map_name
         .clone();
 
+    let mut role_configs: BTreeMap<KafkaRole, ValidatedRoleConfig> = BTreeMap::new();
     let mut role_group_configs: BTreeMap<
         KafkaRole,
         BTreeMap<RoleGroupName, ValidatedRoleGroupConfig>,
@@ -246,10 +247,16 @@ pub fn validate(
         validate_broker_logging,
         &vector_aggregator_config_map_name,
     )?;
+    role_configs.insert(
+        KafkaRole::Broker,
+        ValidatedRoleConfig {
+            pdb: broker_role.role_config.pod_disruption_budget.clone(),
+        },
+    );
     role_group_configs.insert(KafkaRole::Broker, broker_groups);
 
-    // Controllers are optional: ZooKeeper-mode clusters have none, and `controller_role()`
-    // errors when `controllers` is unset, which would stop their reconciliation.
+    // Controllers are optional: ZooKeeper-mode clusters have none, in which case they are simply
+    // absent from both maps and not reconciled.
     if let Some(controller_role) = kafka.spec.controllers.as_ref() {
         let controller_groups = validate_role_group_configs(
             controller_role,
@@ -260,6 +267,12 @@ pub fn validate(
             validate_controller_logging,
             &vector_aggregator_config_map_name,
         )?;
+        role_configs.insert(
+            KafkaRole::Controller,
+            ValidatedRoleConfig {
+                pdb: controller_role.role_config.pod_disruption_budget.clone(),
+            },
+        );
         role_group_configs.insert(KafkaRole::Controller, controller_groups);
     }
 
@@ -292,6 +305,7 @@ pub fn validate(
                 .broker_id_pod_config_map_name
                 .clone(),
         },
+        role_configs,
         role_group_configs,
     ))
 }

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

@@ -321,13 +321,6 @@ impl v1alpha1::KafkaCluster {
             })
     }
 
-    pub fn role_config(&self, role: &KafkaRole) -> Option<&GenericRoleConfig> {
-        match role {
-            KafkaRole::Broker => self.spec.brokers.as_ref().map(|b| &b.role_config),
-            KafkaRole::Controller => self.spec.controllers.as_ref().map(|b| &b.role_config),
-        }
-    }
-
     pub fn broker_role(&self) -> Result<&BrokerRole, Error> {
         self.spec.brokers.as_ref().context(MissingRoleSnafu {
             role: KafkaRole::Broker.to_string(),