push pod_descriptors calc down to build_rolegroup_config_map

Author: adwk67

rust/operator-binary/src/controller.rs

@@ -22,6 +22,7 @@ use stackable_operator::{
         compute_conditions, operations::ClusterOperationsConditionBuilder,
         statefulset::StatefulSetConditionBuilder,
     },
+    utils::cluster_info::KubernetesClusterInfo,
 };
 use strum::{EnumDiscriminants, IntoStaticStr};
 
@@ -65,9 +66,6 @@ pub enum Error {
     #[snafu(display("failed to validate cluster"))]
     ValidateCluster { source: validate::Error },
 
-    #[snafu(display("failed to build pod descriptors"))]
-    BuildPodDescriptors { source: crate::crd::Error },
-
     #[snafu(display("invalid kafka listeners"))]
     InvalidKafkaListeners {
         source: crate::crd::listener::KafkaListenerError,
@@ -205,7 +203,6 @@ impl ReconcilerError for Error {
             Error::BuildService { .. } => None,
             Error::BuildListener { .. } => None,
             Error::InvalidKafkaListeners { .. } => None,
-            Error::BuildPodDescriptors { .. } => None,
         }
     }
 }
@@ -222,6 +219,7 @@ pub struct ValidatedKafkaCluster {
     // classes — rejected because nothing downstream needs them beyond kafka_security.
     pub authorization_config: Option<KafkaAuthorizationConfig>,
     pub role_groups: BTreeMap<KafkaRole, BTreeMap<String, ValidatedRoleGroupConfig>>,
+    pub kubernetes_cluster_info: KubernetesClusterInfo,
 }
 
 pub struct ValidatedRoleGroupConfig {
@@ -327,21 +325,12 @@ pub async fn reconcile_kafka(
             )
             .context(InvalidKafkaListenersSnafu)?;
 
-            let pod_descriptors = kafka
-                .pod_descriptors(
-                    None,
-                    &client.kubernetes_cluster_info,
-                    validated_cluster.kafka_security.client_port(),
-                )
-                .context(BuildPodDescriptorsSnafu)?;
-
             let rg_configmap = build::config_map::build_rolegroup_config_map(
                 kafka,
                 &validated_cluster,
                 &rolegroup_ref,
                 validated_rg,
                 &kafka_listeners,
-                &pod_descriptors,
             )
             .context(BuildConfigMapSnafu)?;
 

rust/operator-binary/src/controller/build/config_map.rs

@@ -12,8 +12,8 @@ use stackable_operator::{
 use crate::{
     controller::{KAFKA_CONTROLLER_NAME, ValidatedKafkaCluster, ValidatedRoleGroupConfig},
     crd::{
-        JVM_SECURITY_PROPERTIES_FILE, KafkaPodDescriptor, MetadataManager,
-        STACKABLE_LISTENER_BOOTSTRAP_DIR, STACKABLE_LISTENER_BROKER_DIR,
+        JVM_SECURITY_PROPERTIES_FILE, MetadataManager, STACKABLE_LISTENER_BOOTSTRAP_DIR,
+        STACKABLE_LISTENER_BROKER_DIR,
         listener::{KafkaListenerConfig, node_address_cmd},
         role::AnyConfig,
         v1alpha1,
@@ -66,6 +66,9 @@ pub enum Error {
 
     #[snafu(display("failed to build jaas configuration file for {rolegroup}"))]
     BuildJaasConfig { rolegroup: String },
+
+    #[snafu(display("failed to build pod descriptors"))]
+    BuildPodDescriptors { source: crate::crd::Error },
 }
 
 /// The rolegroup [`ConfigMap`] configures the rolegroup based on the configuration given by the administrator
@@ -75,7 +78,6 @@ pub fn build_rolegroup_config_map(
     rolegroup: &RoleGroupRef<v1alpha1::KafkaCluster>,
     validated_rg: &ValidatedRoleGroupConfig,
     listener_config: &KafkaListenerConfig,
-    pod_descriptors: &[KafkaPodDescriptor],
 ) -> Result<ConfigMap, Error> {
     let kafka_security = &validated_cluster.kafka_security;
     let resolved_product_image = &validated_cluster.image;
@@ -87,6 +89,14 @@ pub fn build_rolegroup_config_map(
         .as_ref()
         .map(|auth_config| auth_config.opa_connect.clone());
 
+    let pod_descriptors = &kafka
+        .pod_descriptors(
+            None,
+            &validated_cluster.kubernetes_cluster_info,
+            validated_cluster.kafka_security.client_port(),
+        )
+        .context(BuildPodDescriptorsSnafu)?;
+
     let metadata_manager = kafka
         .effective_metadata_manager()
         .context(InvalidMetadataManagerSnafu)?;

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

@@ -9,7 +9,7 @@
 //! and stays here as-is.
 
 use snafu::{ResultExt, Snafu};
-use stackable_operator::client::Client;
+use stackable_operator::{client::Client, utils::cluster_info::KubernetesClusterInfo};
 
 use crate::crd::{
     authentication::{self, ResolvedAuthenticationClasses},
@@ -33,6 +33,7 @@ type Result<T, E = Error> = std::result::Result<T, E>;
 pub struct DereferencedObjects {
     pub authentication_classes: ResolvedAuthenticationClasses,
     pub authorization_config: Option<KafkaAuthorizationConfig>,
+    pub kubernetes_cluster_info: KubernetesClusterInfo,
 }
 
 /// Fetches all Kubernetes objects referenced from the [`v1alpha1::KafkaCluster`] spec.
@@ -59,5 +60,6 @@ pub async fn dereference(
     Ok(DereferencedObjects {
         authentication_classes,
         authorization_config,
+        kubernetes_cluster_info: client.kubernetes_cluster_info.clone(),
     })
 }

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

@@ -146,6 +146,7 @@ pub fn validate(
         kafka_security,
         authorization_config: dereferenced_objects.authorization_config,
         role_groups,
+        kubernetes_cluster_info: dereferenced_objects.kubernetes_cluster_info,
     })
 }