refactor: move security mode to controller, rename KafkaTlsSecurity -> ValidatedKafkaSecurity

Author: maltesander

rust/operator-binary/src/controller.rs

@@ -48,6 +48,7 @@ use strum::{EnumDiscriminants, IntoStaticStr};
 pub(crate) mod build;
 pub(crate) mod dereference;
 pub(crate) mod node_id_hasher;
+pub(crate) mod security;
 pub(crate) mod validate;
 
 /// The type-safe role-group name from stackable-operator. Re-exported so the rest
@@ -69,12 +70,12 @@ use crate::{
             },
         },
         node_id_hasher::node_id_hash32_offset,
+        security::ValidatedKafkaSecurity,
     },
     crd::{
         APP_NAME, KafkaClusterStatus, KafkaPodDescriptor, MetadataManager, OPERATOR_NAME,
         authorization::KafkaAuthorizationConfig,
         role::{AnyConfig, AnyConfigOverrides, KafkaRole},
-        security::KafkaTlsSecurity,
         v1alpha1,
     },
 };
@@ -334,7 +335,7 @@ pub(crate) fn controller_name() -> ControllerName {
 /// Everything the build steps need is resolved here so they never have to read the
 /// raw [`v1alpha1::KafkaCluster`] spec.
 pub struct ValidatedClusterConfig {
-    pub kafka_security: KafkaTlsSecurity,
+    pub kafka_security: ValidatedKafkaSecurity,
     pub authorization_config: Option<KafkaAuthorizationConfig>,
     pub metadata_manager: MetadataManager,
 

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

@@ -8,9 +8,12 @@ use stackable_operator::{
 };
 
 use super::properties::ConfigFileName;
-use crate::crd::{
-    BROKER_ID_POD_MAP_DIR, KafkaPodDescriptor, STACKABLE_CONFIG_DIR, STACKABLE_KERBEROS_KRB5_PATH,
-    STACKABLE_LOG_CONFIG_DIR, security::KafkaTlsSecurity,
+use crate::{
+    controller::security::ValidatedKafkaSecurity,
+    crd::{
+        BROKER_ID_POD_MAP_DIR, KafkaPodDescriptor, STACKABLE_CONFIG_DIR,
+        STACKABLE_KERBEROS_KRB5_PATH, STACKABLE_LOG_CONFIG_DIR,
+    },
 };
 
 /// The JVM options selecting the Kafka log4j/log4j2 config file. Kafka 3.x uses log4j,
@@ -38,7 +41,7 @@ pub fn kafka_log_opts_env_var() -> String {
 pub fn broker_kafka_container_commands(
     kraft_mode: bool,
     controller_descriptors: Vec<KafkaPodDescriptor>,
-    kafka_security: &KafkaTlsSecurity,
+    kafka_security: &ValidatedKafkaSecurity,
     product_version: &str,
 ) -> String {
     formatdoc! {"

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

@@ -14,9 +14,12 @@ use stackable_operator::{
     commons::secret_class::SecretClassVolumeProvisionParts,
 };
 
-use crate::crd::{
-    LISTENER_BOOTSTRAP_VOLUME_NAME, LISTENER_BROKER_VOLUME_NAME, STACKABLE_KERBEROS_DIR,
-    STACKABLE_KERBEROS_KRB5_PATH, role::KafkaRole, security::KafkaTlsSecurity,
+use crate::{
+    controller::security::ValidatedKafkaSecurity,
+    crd::{
+        LISTENER_BOOTSTRAP_VOLUME_NAME, LISTENER_BROKER_VOLUME_NAME, STACKABLE_KERBEROS_DIR,
+        STACKABLE_KERBEROS_KRB5_PATH, role::KafkaRole,
+    },
 };
 
 #[derive(Snafu, Debug)]
@@ -36,7 +39,7 @@ pub enum Error {
 }
 
 pub fn add_kerberos_pod_config(
-    kafka_security: &KafkaTlsSecurity,
+    kafka_security: &ValidatedKafkaSecurity,
     role: &KafkaRole,
     cb_kcat_prober: &mut ContainerBuilder,
     cb_kafka: &mut ContainerBuilder,

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

@@ -11,21 +11,20 @@ use stackable_operator::{
 };
 
 use crate::{
-    controller::{RoleGroupName, ValidatedCluster},
+    controller::{RoleGroupName, ValidatedCluster, security::ValidatedKafkaSecurity},
     crd::{
         STACKABLE_LISTENER_BROKER_DIR,
         listener::{
             KafkaListener, KafkaListenerConfig, KafkaListenerName, KafkaListenerProtocol,
             LISTENER_LOCAL_ADDRESS, node_address_cmd, node_port_cmd,
         },
         role::KafkaRole,
-        security::KafkaTlsSecurity,
     },
 };
 
 pub fn get_kafka_listener_config(
     validated_cluster: &ValidatedCluster,
-    kafka_security: &KafkaTlsSecurity,
+    kafka_security: &ValidatedKafkaSecurity,
     role: &KafkaRole,
     role_group_name: &RoleGroupName,
     cluster_info: &KubernetesClusterInfo,
@@ -49,7 +48,7 @@ pub fn get_kafka_listener_config(
         listeners.push(KafkaListener {
             name: KafkaListenerName::Client,
             host: LISTENER_LOCAL_ADDRESS.to_string(),
-            port: KafkaTlsSecurity::SECURE_CLIENT_PORT.to_string(),
+            port: ValidatedKafkaSecurity::SECURE_CLIENT_PORT.to_string(),
         });
         advertised_listeners.push(KafkaListener {
             name: KafkaListenerName::Client,
@@ -85,7 +84,7 @@ pub fn get_kafka_listener_config(
         listeners.push(KafkaListener {
             name: KafkaListenerName::Client,
             host: LISTENER_LOCAL_ADDRESS.to_string(),
-            port: KafkaTlsSecurity::CLIENT_PORT.to_string(),
+            port: ValidatedKafkaSecurity::CLIENT_PORT.to_string(),
         });
         advertised_listeners.push(KafkaListener {
             name: KafkaListenerName::Client,
@@ -106,12 +105,12 @@ pub fn get_kafka_listener_config(
         listeners.push(KafkaListener {
             name: KafkaListenerName::Internal,
             host: LISTENER_LOCAL_ADDRESS.to_string(),
-            port: KafkaTlsSecurity::SECURE_INTERNAL_PORT.to_string(),
+            port: ValidatedKafkaSecurity::SECURE_INTERNAL_PORT.to_string(),
         });
         advertised_listeners.push(KafkaListener {
             name: KafkaListenerName::Internal,
             host: pod_fqdn.to_string(),
-            port: KafkaTlsSecurity::SECURE_INTERNAL_PORT.to_string(),
+            port: ValidatedKafkaSecurity::SECURE_INTERNAL_PORT.to_string(),
         });
         listener_security_protocol_map
             .insert(KafkaListenerName::Internal, KafkaListenerProtocol::Ssl);
@@ -220,7 +219,7 @@ mod tests {
         "#;
         let kafka = minimal_kafka(kafka_cluster);
         let validated = validated_cluster(&kafka);
-        let kafka_security = KafkaTlsSecurity::new(
+        let kafka_security = ValidatedKafkaSecurity::new(
             ResolvedAuthenticationClasses::new(vec![core::v1alpha1::AuthenticationClass {
                 metadata: ObjectMetaBuilder::new().name("auth-class").build(),
                 spec: core::v1alpha1::AuthenticationClassSpec {
@@ -294,7 +293,7 @@ mod tests {
             )
         );
 
-        let kafka_security = KafkaTlsSecurity::new(
+        let kafka_security = ValidatedKafkaSecurity::new(
             ResolvedAuthenticationClasses::new(vec![]),
             Some("tls".parse().unwrap()),
             Some("tls".parse().unwrap()),
@@ -357,8 +356,12 @@ mod tests {
             )
         );
 
-        let kafka_security =
-            KafkaTlsSecurity::new(ResolvedAuthenticationClasses::new(vec![]), None, None, None);
+        let kafka_security = ValidatedKafkaSecurity::new(
+            ResolvedAuthenticationClasses::new(vec![]),
+            None,
+            None,
+            None,
+        );
 
         let config = get_kafka_listener_config(
             &validated,
@@ -441,7 +444,7 @@ mod tests {
         "#;
         let kafka = minimal_kafka(kafka_cluster);
         let validated = validated_cluster(&kafka);
-        let kafka_security = KafkaTlsSecurity::new(
+        let kafka_security = ValidatedKafkaSecurity::new(
             ResolvedAuthenticationClasses::new(vec![core::v1alpha1::AuthenticationClass {
                 metadata: ObjectMetaBuilder::new().name("auth-class").build(),
                 spec: core::v1alpha1::AuthenticationClassSpec {

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

@@ -4,11 +4,8 @@ use stackable_operator::{
 };
 
 use crate::{
-    controller::{RoleGroupName, ValidatedCluster},
-    crd::{
-        role::{KafkaRole, broker::BrokerConfig},
-        security::KafkaTlsSecurity,
-    },
+    controller::{RoleGroupName, ValidatedCluster, security::ValidatedKafkaSecurity},
+    crd::role::{KafkaRole, broker::BrokerConfig},
 };
 
 /// Kafka clients will use the load-balanced bootstrap listener to get a list of broker addresses and will use those to
@@ -43,7 +40,7 @@ pub fn build_broker_rolegroup_bootstrap_listener(
 }
 
 fn bootstrap_listener_ports(
-    kafka_security: &KafkaTlsSecurity,
+    kafka_security: &ValidatedKafkaSecurity,
 ) -> Vec<listener::v1alpha1::ListenerPort> {
     vec![if kafka_security.has_kerberos_enabled() {
         listener::v1alpha1::ListenerPort {

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

@@ -6,8 +6,8 @@ use stackable_operator::{
 };
 
 use crate::{
-    controller::{RoleGroupName, ValidatedCluster},
-    crd::{METRICS_PORT, METRICS_PORT_NAME, role::KafkaRole, security::KafkaTlsSecurity},
+    controller::{RoleGroupName, ValidatedCluster, security::ValidatedKafkaSecurity},
+    crd::{METRICS_PORT, METRICS_PORT_NAME, role::KafkaRole},
 };
 
 /// The rolegroup [`Service`] is a headless service that allows direct access to the instances of a certain rolegroup
@@ -17,7 +17,7 @@ pub fn build_rolegroup_headless_service(
     validated_cluster: &ValidatedCluster,
     role: &KafkaRole,
     role_group_name: &RoleGroupName,
-    kafka_security: &KafkaTlsSecurity,
+    kafka_security: &ValidatedKafkaSecurity,
 ) -> Service {
     Service {
         metadata: ObjectMetaBuilder::new()
@@ -100,7 +100,7 @@ fn metrics_ports() -> Vec<ServicePort> {
     }]
 }
 
-fn headless_ports(kafka_security: &KafkaTlsSecurity) -> Vec<ServicePort> {
+fn headless_ports(kafka_security: &ValidatedKafkaSecurity) -> Vec<ServicePort> {
     vec![ServicePort {
         name: Some(kafka_security.client_port_name().into()),
         port: kafka_security.client_port().into(),

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

@@ -55,6 +55,7 @@ use crate::{
             properties::product_logging::MAX_KAFKA_LOG_FILES_SIZE,
         },
         node_id_hasher::node_id_hash32_offset,
+        security::ValidatedKafkaSecurity,
         validate::ValidatedLogging,
     },
     crd::{
@@ -67,7 +68,6 @@ use crate::{
             AnyConfig, KAFKA_NODE_ID_OFFSET, KafkaRole, broker::BrokerContainer,
             controller::ControllerContainer,
         },
-        security::KafkaTlsSecurity,
     },
 };
 
@@ -90,7 +90,9 @@ pub enum Error {
     },
 
     #[snafu(display("failed to add Secret Volumes and VolumeMounts"))]
-    AddVolumesAndVolumeMounts { source: crate::crd::security::Error },
+    AddVolumesAndVolumeMounts {
+        source: crate::controller::security::Error,
+    },
 
     #[snafu(display("failed to add needed volumeMount"))]
     AddVolumeMount {
@@ -643,7 +645,7 @@ pub fn build_controller_rolegroup_statefulset(
 }
 
 /// We only expose client HTTP / HTTPS and Metrics ports.
-fn container_ports(kafka_security: &KafkaTlsSecurity) -> Vec<ContainerPort> {
+fn container_ports(kafka_security: &ValidatedKafkaSecurity) -> Vec<ContainerPort> {
     let mut ports = vec![
         ContainerPort {
             name: Some(METRICS_PORT_NAME.to_string()),

rust/operator-binary/src/crd/security.rs …erator-binary/src/controller/security.rsrust/operator-binary/src/crd/security.rs renamed to rust/operator-binary/src/controller/security.rs rust/operator-binary/src/crd/security.rs renamed to rust/operator-binary/src/controller/security.rs

@@ -23,12 +23,13 @@ use stackable_operator::{
     v2::types::{common::Port, kubernetes::SecretClassName},
 };
 
-use super::listener::KafkaListenerProtocol;
 use crate::crd::{
     LISTENER_BOOTSTRAP_VOLUME_NAME, LISTENER_BROKER_VOLUME_NAME, STACKABLE_KERBEROS_KRB5_PATH,
     STACKABLE_LISTENER_BROKER_DIR,
     authentication::ResolvedAuthenticationClasses,
-    listener::{self, KafkaListenerName, node_address_cmd_env, node_port_cmd_env},
+    listener::{
+        self, KafkaListenerName, KafkaListenerProtocol, node_address_cmd_env, node_port_cmd_env,
+    },
     role::KafkaRole,
     tls, v1alpha1,
 };
@@ -58,14 +59,14 @@ pub enum Error {
 }
 
 /// Helper struct combining TLS settings for server and internal with the resolved AuthenticationClasses
-pub struct KafkaTlsSecurity {
+pub struct ValidatedKafkaSecurity {
     resolved_authentication_classes: ResolvedAuthenticationClasses,
     internal_secret_class: Option<SecretClassName>,
     server_secret_class: Option<SecretClassName>,
     opa_secret_class: Option<SecretClassName>,
 }
 
-impl KafkaTlsSecurity {
+impl ValidatedKafkaSecurity {
     pub const BOOTSTRAP_PORT: Port = Port(9094);
     // bootstrap: we will have a single named port with different values for
     // secure (9095) and insecure (9094). The bootstrap listener is needed to
@@ -114,7 +115,7 @@ impl KafkaTlsSecurity {
         }
     }
 
-    /// Build a [`KafkaTlsSecurity`] from already-resolved authentication classes.
+    /// Build a [`ValidatedKafkaSecurity`] from already-resolved authentication classes.
     ///
     /// The async retrieval of [`ResolvedAuthenticationClasses`] now happens in the dereference
     /// step of the controller; this constructor only reads TLS settings from the spec.
@@ -123,7 +124,7 @@ impl KafkaTlsSecurity {
         resolved_authentication_classes: ResolvedAuthenticationClasses,
         opa_secret_class: Option<SecretClassName>,
     ) -> Self {
-        KafkaTlsSecurity {
+        ValidatedKafkaSecurity {
             resolved_authentication_classes,
             internal_secret_class: kafka
                 .spec

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

@@ -30,7 +30,9 @@ use stackable_operator::{
 use crate::{
     controller::{
         RoleGroupName, ValidatedCluster, ValidatedClusterConfig, ValidatedRoleConfig,
-        ValidatedRoleGroupConfig, dereference::DereferencedObjects,
+        ValidatedRoleGroupConfig,
+        dereference::DereferencedObjects,
+        security::{self, ValidatedKafkaSecurity},
     },
     crd::{
         self, CONTAINER_IMAGE_BASE_NAME,
@@ -40,7 +42,6 @@ use crate::{
             broker::{BrokerConfig, BrokerContainer},
             controller::{ControllerConfig, ControllerContainer},
         },
-        security::{self, KafkaTlsSecurity},
         v1alpha1,
     },
 };
@@ -210,8 +211,11 @@ pub fn validate(
         .as_ref()
         .and_then(|cfg| cfg.secret_class.clone());
 
-    let kafka_security =
-        KafkaTlsSecurity::new_from_kafka_cluster(kafka, authentication_classes, opa_secret_class);
+    let kafka_security = ValidatedKafkaSecurity::new_from_kafka_cluster(
+        kafka,
+        authentication_classes,
+        opa_secret_class,
+    );
 
     kafka_security
         .validate_authentication_methods()

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

@@ -3,7 +3,6 @@ pub mod authentication;
 pub mod authorization;
 pub mod listener;
 pub mod role;
-pub mod security;
 pub mod tls;
 
 use authentication::KafkaAuthentication;