refactor: use ConfigMapName, SecretName, ListenerName in CRD

Author: maltesander

extra/crds.yaml

@@ -79,11 +79,17 @@ spec:
                         type: object
                       bootstrapListenerClass:
                         description: The ListenerClass used for bootstrapping new clients. Should use a stable ListenerClass to avoid unnecessary client restarts (such as `cluster-internal` or `external-stable`).
+                        maxLength: 253
+                        minLength: 1
                         nullable: true
+                        pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$
                         type: string
                       brokerListenerClass:
                         description: The ListenerClass used for connecting to brokers. Should use a direct connection ListenerClass to minimize cost and minimize performance overhead (such as `cluster-internal` or `external-unstable`).
+                        maxLength: 253
+                        minLength: 1
                         nullable: true
+                        pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$
                         type: string
                       gracefulShutdownTimeout:
                         description: Time period Pods have to gracefully shut down, e.g. `30m`, `1h` or `2d`. Consult the operator documentation for details.
@@ -608,11 +614,17 @@ spec:
                               type: object
                             bootstrapListenerClass:
                               description: The ListenerClass used for bootstrapping new clients. Should use a stable ListenerClass to avoid unnecessary client restarts (such as `cluster-internal` or `external-stable`).
+                              maxLength: 253
+                              minLength: 1
                               nullable: true
+                              pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$
                               type: string
                             brokerListenerClass:
                               description: The ListenerClass used for connecting to brokers. Should use a direct connection ListenerClass to minimize cost and minimize performance overhead (such as `cluster-internal` or `external-unstable`).
+                              maxLength: 253
+                              minLength: 1
                               nullable: true
+                              pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$
                               type: string
                             gracefulShutdownTimeout:
                               description: Time period Pods have to gracefully shut down, e.g. `30m`, `1h` or `2d`. Consult the operator documentation for details.
@@ -1153,7 +1165,10 @@ spec:
                       ```
                       This is necessary when migrating from ZooKeeper to Kraft mode to retain existing broker IDs
                       because previously broker ids were generated by Kafka and not the operator.
+                    maxLength: 253
+                    minLength: 1
                     nullable: true
+                    pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$
                     type: string
                   metadataManager:
                     description: |-
@@ -1204,6 +1219,11 @@ spec:
                           - Which ca.crt to use when validating the other brokers
 
                           Defaults to `tls`
+                          Set to `null` to disable internal TLS (resulting in a plaintext cluster).
+                        maxLength: 253
+                        minLength: 1
+                        nullable: true
+                        pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$
                         type: string
                       serverSecretClass:
                         default: tls
@@ -1214,7 +1234,10 @@ spec:
                           - Which cert the servers should use to authenticate themselves against the client
 
                           Defaults to `tls`.
+                        maxLength: 253
+                        minLength: 1
                         nullable: true
+                        pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$
                         type: string
                     type: object
                   vectorAggregatorConfigMapName:
@@ -1223,7 +1246,10 @@ spec:
                       It must contain the key `ADDRESS` with the address of the Vector aggregator.
                       Follow the [logging tutorial](https://docs.stackable.tech/home/nightly/tutorials/logging-vector-aggregator)
                       to learn how to configure log aggregation with Vector.
+                    maxLength: 253
+                    minLength: 1
                     nullable: true
+                    pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$
                     type: string
                   zookeeperConfigMapName:
                     description: |-
@@ -1232,7 +1258,10 @@ spec:
                       to deploy a ZooKeeper cluster, this will simply be the name of your ZookeeperCluster resource.
                       This can only be used up to Kafka version 3.9.x. Since Kafka 4.0.0, ZooKeeper support was dropped.
                       Please use the 'controller' role instead.
+                    maxLength: 253
+                    minLength: 1
                     nullable: true
+                    pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$
                     type: string
                 type: object
               clusterOperation:

rust/operator-binary/src/controller.rs

@@ -42,7 +42,7 @@ use stackable_operator::{
         kvp::label::{recommended_labels, role_group_selector},
         role_group_utils::ResourceNames,
         types::{
-            kubernetes::{NamespaceName, Uid},
+            kubernetes::{ConfigMapName, NamespaceName, Uid},
             operator::{ClusterName, ControllerName, OperatorName, ProductName, ProductVersion},
         },
     },
@@ -342,11 +342,11 @@ pub struct ValidatedClusterConfig {
     /// The discovery `ConfigMap` providing the ZooKeeper connection string, if the cluster
     /// is connected to a ZooKeeper ensemble. Resolved from the raw spec during validation so
     /// the build steps never have to read it.
-    pub zookeeper_config_map_name: Option<String>,
+    pub zookeeper_config_map_name: Option<ConfigMapName>,
 
     /// The `ConfigMap` mapping pods to broker ids, if the user supplied one. Resolved from the
     /// raw spec during validation so the build steps never have to read it.
-    pub broker_id_pod_config_map_name: Option<String>,
+    pub broker_id_pod_config_map_name: Option<ConfigMapName>,
 }
 
 impl ValidatedClusterConfig {

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

@@ -231,8 +231,8 @@ mod tests {
                     ),
                 },
             }]),
-            "internalTls".to_string(),
-            Some("tls".to_string()),
+            Some("internal-tls".parse().unwrap()),
+            Some("tls".parse().unwrap()),
             None,
         );
         let cluster_info = default_cluster_info();
@@ -296,8 +296,8 @@ mod tests {
 
         let kafka_security = KafkaTlsSecurity::new(
             ResolvedAuthenticationClasses::new(vec![]),
-            "tls".to_string(),
-            Some("tls".to_string()),
+            Some("tls".parse().unwrap()),
+            Some("tls".parse().unwrap()),
             None,
         );
         let config = get_kafka_listener_config(
@@ -357,12 +357,8 @@ mod tests {
             )
         );
 
-        let kafka_security = KafkaTlsSecurity::new(
-            ResolvedAuthenticationClasses::new(vec![]),
-            "".to_string(),
-            None,
-            None,
-        );
+        let kafka_security =
+            KafkaTlsSecurity::new(ResolvedAuthenticationClasses::new(vec![]), None, None, None);
 
         let config = get_kafka_listener_config(
             &validated,
@@ -456,8 +452,8 @@ mod tests {
                     ),
                 },
             }]),
-            "tls".to_string(),
-            Some("tls".to_string()),
+            Some("tls".parse().unwrap()),
+            Some("tls".parse().unwrap()),
             None,
         );
         let cluster_info = default_cluster_info();

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

@@ -34,7 +34,7 @@ pub fn build_broker_rolegroup_bootstrap_listener(
             .with_labels(validated_cluster.recommended_labels(role, role_group_name))
             .build(),
         spec: listener::v1alpha1::ListenerSpec {
-            class_name: Some(merged_config.bootstrap_listener_class.clone()),
+            class_name: Some(merged_config.bootstrap_listener_class.to_string()),
             ports: Some(bootstrap_listener_ports(kafka_security)),
             ..listener::v1alpha1::ListenerSpec::default()
         },

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

@@ -335,7 +335,7 @@ pub fn build_broker_rolegroup_statefulset(
         pod_builder
             .add_listener_volume_by_listener_class(
                 LISTENER_BROKER_VOLUME_NAME,
-                listener_class,
+                listener_class.as_ref(),
                 &recommended_labels,
             )
             .context(AddListenerVolumeSnafu)?;

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

@@ -105,11 +105,6 @@ pub enum Error {
         "the Vector aggregator discovery ConfigMap name is required when the Vector agent is enabled"
     ))]
     MissingVectorAggregatorConfigMapName,
-
-    #[snafu(display("the Vector aggregator discovery ConfigMap name is invalid"))]
-    ParseVectorAggregatorConfigMapName {
-        source: stackable_operator::v2::macros::attributed_string_type::Error,
-    },
 }
 
 /// Validated logging configuration for a Kafka role group's Kafka and (optional) Vector
@@ -224,17 +219,14 @@ pub fn validate(
 
     let cluster_id = kafka.cluster_id();
 
-    // The Vector aggregator discovery ConfigMap name, validated up-front so an invalid name
-    // fails reconciliation here rather than at resource-build time. It is only required (per
-    // role group) when the Vector agent is enabled; see [`validate_logging`].
+    // The Vector aggregator discovery ConfigMap name. Validity is enforced by the `ConfigMapName`
+    // type on the CRD field. It is only required (per role group) when the Vector agent is
+    // enabled; see [`validate_logging`].
     let vector_aggregator_config_map_name = kafka
         .spec
         .cluster_config
         .vector_aggregator_config_map_name
-        .as_deref()
-        .map(ConfigMapName::from_str)
-        .transpose()
-        .context(ParseVectorAggregatorConfigMapNameSnafu)?;
+        .clone();
 
     let mut role_group_configs: BTreeMap<
         KafkaRole,

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

@@ -1,10 +1,13 @@
+use std::str::FromStr;
+
 use serde::{Deserialize, Serialize};
 use snafu::{OptionExt, ResultExt, Snafu};
 use stackable_operator::{
     client::Client,
     commons::opa::{OpaApiVersion, OpaConfig},
     k8s_openapi::api::core::v1::ConfigMap,
     schemars::{self, JsonSchema},
+    v2::types::kubernetes::SecretClassName,
 };
 
 use crate::crd::v1alpha1;
@@ -23,6 +26,11 @@ pub enum Error {
         source: stackable_operator::commons::opa::Error,
     },
 
+    #[snafu(display("the OPA TLS secret class name (key `OPA_SECRET_CLASS`) is invalid"))]
+    ParseOpaSecretClass {
+        source: stackable_operator::v2::macros::attributed_string_type::Error,
+    },
+
     #[snafu(display("object defines no namespace"))]
     ObjectHasNoNamespace,
 }
@@ -36,7 +44,7 @@ pub struct KafkaAuthorization {
 
 pub struct KafkaAuthorizationConfig {
     pub opa_connect: String,
-    pub secret_class: Option<String>,
+    pub secret_class: Option<SecretClassName>,
 }
 
 impl KafkaAuthorization {
@@ -60,7 +68,10 @@ impl KafkaAuthorization {
                     namespace,
                 })?
                 .data
-                .and_then(|mut data| data.remove("OPA_SECRET_CLASS"));
+                .and_then(|mut data| data.remove("OPA_SECRET_CLASS"))
+                .map(|secret_class| SecretClassName::from_str(&secret_class))
+                .transpose()
+                .context(ParseOpaSecretClassSnafu)?;
             let opa_connect = opa
                 .full_document_url_from_config_map(client, kafka, Some("allow"), &OpaApiVersion::V1)
                 .await

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

@@ -20,7 +20,10 @@ use stackable_operator::{
     role_utils::{GenericRoleConfig, Role, RoleGroupRef},
     schemars::{self, JsonSchema},
     status::condition::{ClusterCondition, HasStatusCondition},
-    v2::{config_overrides::KeyValueConfigOverrides, role_utils::JavaCommonConfig},
+    v2::{
+        config_overrides::KeyValueConfigOverrides, role_utils::JavaCommonConfig,
+        types::kubernetes::ConfigMapName,
+    },
     versioned::versioned,
 };
 use strum::{Display, EnumIter, EnumString};
@@ -163,14 +166,14 @@ pub mod versioned {
         /// Follow the [logging tutorial](DOCS_BASE_URL_PLACEHOLDER/tutorials/logging-vector-aggregator)
         /// to learn how to configure log aggregation with Vector.
         #[serde(skip_serializing_if = "Option::is_none")]
-        pub vector_aggregator_config_map_name: Option<String>,
+        pub vector_aggregator_config_map_name: Option<ConfigMapName>,
 
         /// Provide the name of the ZooKeeper [discovery ConfigMap](DOCS_BASE_URL_PLACEHOLDER/concepts/service_discovery)
         /// here. When using the [Stackable operator for Apache ZooKeeper](DOCS_BASE_URL_PLACEHOLDER/zookeeper/)
         /// to deploy a ZooKeeper cluster, this will simply be the name of your ZookeeperCluster resource.
         /// This can only be used up to Kafka version 3.9.x. Since Kafka 4.0.0, ZooKeeper support was dropped.
         /// Please use the 'controller' role instead.
-        pub zookeeper_config_map_name: Option<String>,
+        pub zookeeper_config_map_name: Option<ConfigMapName>,
 
         /// Metadata manager to use for the Kafka cluster.
         ///
@@ -221,7 +224,7 @@ pub mod versioned {
         /// because previously broker ids were generated by Kafka and not the operator.
         ///
         #[serde(skip_serializing_if = "Option::is_none")]
-        pub broker_id_pod_config_map_name: Option<String>,
+        pub broker_id_pod_config_map_name: Option<ConfigMapName>,
     }
 
     #[derive(Clone, Debug, Default, Deserialize, JsonSchema, Merge, PartialEq, Serialize)]
@@ -424,11 +427,13 @@ pub enum MetadataManager {
 #[cfg(test)]
 mod tests {
     use rstest::rstest;
-    use stackable_operator::versioned::test_utils::RoundtripTestData;
+    use stackable_operator::{
+        v2::types::kubernetes::SecretClassName, versioned::test_utils::RoundtripTestData,
+    };
 
     use super::*;
 
-    fn get_server_secret_class(kafka: &v1alpha1::KafkaCluster) -> Option<String> {
+    fn get_server_secret_class(kafka: &v1alpha1::KafkaCluster) -> Option<SecretClassName> {
         kafka
             .spec
             .cluster_config
@@ -437,7 +442,7 @@ mod tests {
             .and_then(|tls| tls.server_secret_class.clone())
     }
 
-    fn get_internal_secret_class(kafka: &v1alpha1::KafkaCluster) -> String {
+    fn get_internal_secret_class(kafka: &v1alpha1::KafkaCluster) -> Option<SecretClassName> {
         kafka
             .spec
             .cluster_config
@@ -486,8 +491,8 @@ mod tests {
         let kafka: v1alpha1::KafkaCluster =
             serde_yaml::from_str(input).expect("illegal test input");
         assert_eq!(
-            get_server_secret_class(&kafka).unwrap(),
-            "simple-kafka-server-tls".to_string()
+            get_server_secret_class(&kafka).unwrap().to_string(),
+            "simple-kafka-server-tls"
         );
         assert_eq!(
             get_internal_secret_class(&kafka),
@@ -533,8 +538,8 @@ mod tests {
             serde_yaml::from_str(input).expect("illegal test input");
         assert_eq!(get_server_secret_class(&kafka), tls::server_tls_default());
         assert_eq!(
-            get_internal_secret_class(&kafka),
-            "simple-kafka-internal-tls".to_string()
+            get_internal_secret_class(&kafka).map(|s| s.to_string()),
+            Some("simple-kafka-internal-tls".to_string())
         );
     }
 
@@ -576,8 +581,8 @@ mod tests {
             serde_yaml::from_str(input).expect("illegal test input");
         assert_eq!(get_server_secret_class(&kafka), tls::server_tls_default());
         assert_eq!(
-            get_internal_secret_class(&kafka),
-            "simple-kafka-internal-tls".to_string()
+            get_internal_secret_class(&kafka).map(|s| s.to_string()),
+            Some("simple-kafka-internal-tls".to_string())
         );
 
         let input = r#"
@@ -596,7 +601,7 @@ mod tests {
         let kafka: v1alpha1::KafkaCluster =
             serde_yaml::from_str(input).expect("illegal test input");
         assert_eq!(
-            get_server_secret_class(&kafka),
+            get_server_secret_class(&kafka).map(|s| s.to_string()),
             Some("simple-kafka-server-tls".to_string())
         );
         assert_eq!(