refactor: extract build logic from KafkaSecurity

Author: maltesander

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

@@ -9,7 +9,7 @@ use stackable_operator::{
 
 use super::properties::ConfigFileName;
 use crate::{
-    controller::security::ValidatedKafkaSecurity,
+    controller::{build::security::copy_opa_tls_cert_command, security::ValidatedKafkaSecurity},
     crd::{
         BROKER_ID_POD_MAP_DIR, KafkaPodDescriptor, STACKABLE_CONFIG_DIR,
         STACKABLE_KERBEROS_KRB5_PATH, STACKABLE_LOG_CONFIG_DIR,
@@ -64,7 +64,7 @@ pub fn broker_kafka_container_commands(
             true => format!("export KERBEROS_REALM=$(grep -oP 'default_realm = \\K.*' {STACKABLE_KERBEROS_KRB5_PATH})"),
             false => "".to_string(),
         },
-        import_opa_tls_cert = kafka_security.copy_opa_tls_cert_command(),
+        import_opa_tls_cert = copy_opa_tls_cert_command(kafka_security),
         broker_start_command = broker_start_command(kraft_mode, controller_descriptors, product_version),
     }
 }

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

@@ -6,3 +6,4 @@ pub mod jvm;
 pub mod kerberos;
 pub mod properties;
 pub mod resource;
+pub mod security;

rust/operator-binary/src/controller/build/properties/broker_properties.rs

@@ -3,7 +3,11 @@ use std::collections::BTreeMap;
 use super::kraft_controllers;
 use crate::{
     controller::{
-        ValidatedClusterConfig, build::graceful_shutdown::graceful_shutdown_config_properties,
+        ValidatedClusterConfig,
+        build::{
+            graceful_shutdown::graceful_shutdown_config_properties,
+            security::broker_config_settings,
+        },
     },
     crd::{
         KafkaPodDescriptor,
@@ -106,7 +110,7 @@ pub fn build(
         ]);
     }
 
-    result.extend(cluster_config.kafka_security.broker_config_settings());
+    result.extend(broker_config_settings(&cluster_config.kafka_security));
     result.extend(graceful_shutdown_config_properties());
     result.extend(overrides);
 

rust/operator-binary/src/controller/build/properties/controller_properties.rs

@@ -3,7 +3,11 @@ use std::collections::BTreeMap;
 use super::kraft_controllers;
 use crate::{
     controller::{
-        ValidatedClusterConfig, build::graceful_shutdown::graceful_shutdown_config_properties,
+        ValidatedClusterConfig,
+        build::{
+            graceful_shutdown::graceful_shutdown_config_properties,
+            security::controller_config_settings,
+        },
     },
     crd::{
         KafkaPodDescriptor,
@@ -65,7 +69,7 @@ pub fn build(
         );
     }
 
-    result.extend(cluster_config.kafka_security.controller_config_settings());
+    result.extend(controller_config_settings(&cluster_config.kafka_security));
     result.extend(graceful_shutdown_config_properties());
     result.extend(overrides);
 

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

@@ -13,8 +13,11 @@ use stackable_operator::{
 use crate::{
     controller::{
         RoleGroupName, ValidatedCluster, ValidatedRoleGroupConfig,
-        build::properties::{
-            ConfigFileName, config_file_name, product_logging::role_group_config_map_data,
+        build::{
+            properties::{
+                ConfigFileName, config_file_name, product_logging::role_group_config_map_data,
+            },
+            security::client_properties,
         },
     },
     crd::{
@@ -152,8 +155,7 @@ pub fn build_rolegroup_config_map(
         .add_data(
             ConfigFileName::Client.to_string(),
             to_java_properties_string(
-                kafka_security
-                    .client_properties()
+                client_properties(kafka_security)
                     .iter()
                     .filter_map(|(k, v)| v.as_ref().map(|v| (k, v))),
             )

rust/operator-binary/src/controller/build/resource/statefulset.rs

@@ -53,6 +53,10 @@ use crate::{
             graceful_shutdown::add_graceful_shutdown_config,
             kerberos::add_kerberos_pod_config,
             properties::product_logging::MAX_KAFKA_LOG_FILES_SIZE,
+            security::{
+                add_broker_volume_and_volume_mounts, add_controller_volume_and_volume_mounts,
+                kcat_prober_container_commands,
+            },
         },
         node_id_hasher::node_id_hash32_offset,
         security::ValidatedKafkaSecurity,
@@ -91,7 +95,7 @@ pub enum Error {
 
     #[snafu(display("failed to add Secret Volumes and VolumeMounts"))]
     AddVolumesAndVolumeMounts {
-        source: crate::controller::security::Error,
+        source: crate::controller::build::security::Error,
     },
 
     #[snafu(display("failed to add needed volumeMount"))]
@@ -168,14 +172,14 @@ pub fn build_broker_rolegroup_statefulset(
         .deref()
         .requested_secret_lifetime
         .context(MissingSecretLifetimeSnafu)?;
-    kafka_security
-        .add_broker_volume_and_volume_mounts(
-            &mut pod_builder,
-            &mut cb_kcat_prober,
-            &mut cb_kafka,
-            &requested_secret_lifetime,
-        )
-        .context(AddVolumesAndVolumeMountsSnafu)?;
+    add_broker_volume_and_volume_mounts(
+        kafka_security,
+        &mut pod_builder,
+        &mut cb_kcat_prober,
+        &mut cb_kafka,
+        &requested_secret_lifetime,
+    )
+    .context(AddVolumesAndVolumeMountsSnafu)?;
 
     let mut pvcs = merged_config.resources().storage.build_pvcs();
 
@@ -323,7 +327,7 @@ pub fn build_broker_rolegroup_statefulset(
         .readiness_probe(Probe {
             exec: Some(ExecAction {
                 // If the broker is able to get its fellow cluster members then it has at least completed basic registration at some point
-                command: Some(kafka_security.kcat_prober_container_commands()),
+                command: Some(kcat_prober_container_commands(kafka_security)),
             }),
             timeout_seconds: Some(5),
             period_seconds: Some(2),
@@ -576,13 +580,13 @@ pub fn build_controller_rolegroup_statefulset(
         .deref()
         .requested_secret_lifetime
         .context(MissingSecretLifetimeSnafu)?;
-    kafka_security
-        .add_controller_volume_and_volume_mounts(
-            &mut pod_builder,
-            &mut cb_kafka,
-            &requested_secret_lifetime,
-        )
-        .context(AddVolumesAndVolumeMountsSnafu)?;
+    add_controller_volume_and_volume_mounts(
+        kafka_security,
+        &mut pod_builder,
+        &mut cb_kafka,
+        &requested_secret_lifetime,
+    )
+    .context(AddVolumesAndVolumeMountsSnafu)?;
 
     let kafka_container = cb_kafka.build();