From cd70a6b20d0997dbecde7dbf71d802ce166ff2f8 Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Fri, 28 Aug 2026 17:54:27 +0200 Subject: [PATCH 01/10] use expect where possible and remove unecessary enums/Results --- .../src/controller/build/kerberos.rs | 57 +++---- .../controller/build/resource/config_map.rs | 12 +- .../controller/build/resource/discovery.rs | 7 +- .../controller/build/resource/statefulset.rs | 158 ++++++------------ .../src/controller/build/security.rs | 138 +++++++-------- rust/operator-binary/src/crd/mod.rs | 39 +++-- rust/operator-binary/src/crd/role/broker.rs | 29 ++-- rust/operator-binary/src/crd/role/commons.rs | 6 +- rust/operator-binary/src/crd/tls.rs | 17 +- 9 files changed, 192 insertions(+), 271 deletions(-) diff --git a/rust/operator-binary/src/controller/build/kerberos.rs b/rust/operator-binary/src/controller/build/kerberos.rs index 624525f71..bee698226 100644 --- a/rust/operator-binary/src/controller/build/kerberos.rs +++ b/rust/operator-binary/src/controller/build/kerberos.rs @@ -1,21 +1,17 @@ use std::str::FromStr; -use snafu::{ResultExt, Snafu}; use stackable_operator::{ - builder::{ - self, - pod::{ - PodBuilder, - container::ContainerBuilder, - volume::{ - SecretOperatorVolumeSourceBuilder, SecretOperatorVolumeSourceBuilderError, - VolumeBuilder, - }, - }, + builder::pod::{ + PodBuilder, + container::ContainerBuilder, + volume::{SecretOperatorVolumeSourceBuilder, VolumeBuilder}, }, commons::secret_class::SecretClassVolumeProvisionParts, constant, - v2::builder::pod::container::{EnvVarName, EnvVarSet}, + v2::{ + builder::pod::container::{EnvVarName, EnvVarSet}, + types::kubernetes::VolumeName, + }, }; use crate::{ @@ -26,21 +22,7 @@ use crate::{ }, }; -#[derive(Snafu, Debug)] -pub enum Error { - #[snafu(display("failed to add Kerberos secret volume"))] - KerberosSecretVolume { - source: SecretOperatorVolumeSourceBuilderError, - }, - - #[snafu(display("failed to add needed volume"))] - AddVolume { source: builder::pod::Error }, - - #[snafu(display("failed to add needed volumeMount"))] - AddVolumeMount { - source: builder::pod::container::Error, - }, -} +constant!(KERBEROS_VOLUME_NAME: VolumeName = "kerberos"); pub fn add_kerberos_pod_config( kafka_security: &ValidatedKafkaSecurity, @@ -48,7 +30,7 @@ pub fn add_kerberos_pod_config( cb_kcat_prober: &mut ContainerBuilder, cb_kafka: &mut ContainerBuilder, pb: &mut PodBuilder, -) -> Result<(), Error> { +) { if let Some(kerberos_secret_class) = kafka_security.kerberos_secret_class() { // Mount keytab let kerberos_secret_operator_volume = SecretOperatorVolumeSourceBuilder::new( @@ -56,25 +38,25 @@ pub fn add_kerberos_pod_config( // We need both public (krb5.conf) and private (keytab) parts. SecretClassVolumeProvisionParts::PublicPrivate, ) - .with_listener_volume_scope(LISTENER_BROKER_VOLUME_NAME) - .with_listener_volume_scope(LISTENER_BOOTSTRAP_VOLUME_NAME) + .with_listener_volume_scope(&*LISTENER_BROKER_VOLUME_NAME) + .with_listener_volume_scope(&*LISTENER_BOOTSTRAP_VOLUME_NAME) .with_kerberos_service_name(role.kerberos_service_name()) .build() - .context(KerberosSecretVolumeSnafu)?; + .expect("The annotation keys are static and annotation values cannot be invalid."); pb.add_volume( - VolumeBuilder::new("kerberos") + VolumeBuilder::new(&*KERBEROS_VOLUME_NAME) .ephemeral(kerberos_secret_operator_volume) .build(), ) - .context(AddVolumeSnafu)?; + .expect("The volume names are statically defined and there should be no duplicates."); for cb in [cb_kafka, cb_kcat_prober] { - cb.add_volume_mount("kerberos", STACKABLE_KERBEROS_DIR) - .context(AddVolumeMountSnafu)?; + cb.add_volume_mount(&*KERBEROS_VOLUME_NAME, STACKABLE_KERBEROS_DIR) + .expect( + "The mount paths are statically defined and there should be no duplicates.", + ); } } - - Ok(()) } constant!(KRB5_CONFIG: EnvVarName = "KRB5_CONFIG"); @@ -108,5 +90,6 @@ mod tests { // Test that dereferencing the constants does not panic. let _ = *KRB5_CONFIG; let _ = *KAFKA_OPTS; + let _ = *KERBEROS_VOLUME_NAME; } } diff --git a/rust/operator-binary/src/controller/build/resource/config_map.rs b/rust/operator-binary/src/controller/build/resource/config_map.rs index a1be2a907..1393eba1b 100644 --- a/rust/operator-binary/src/controller/build/resource/config_map.rs +++ b/rust/operator-binary/src/controller/build/resource/config_map.rs @@ -30,12 +30,6 @@ use crate::{ #[derive(Snafu, Debug)] pub enum Error { - #[snafu(display("failed to build ConfigMap for role group {role_group}"))] - BuildRoleGroupConfig { - source: stackable_operator::builder::configmap::Error, - role_group: RoleGroupName, - }, - #[snafu(display( "failed to serialize [{}] for role group {role_group}", ConfigFileName::Security @@ -194,11 +188,9 @@ pub fn build_rolegroup_config_map( cm_builder.add_data(VECTOR_CONFIG_FILE, vector_config); } - cm_builder + Ok(cm_builder .build() - .with_context(|_| BuildRoleGroupConfigSnafu { - role_group: role_group_name.clone(), - }) + .expect("The ConfigMap metadata is set in this function.")) } // Generate JAAS configuration file for Kerberos authentication diff --git a/rust/operator-binary/src/controller/build/resource/discovery.rs b/rust/operator-binary/src/controller/build/resource/discovery.rs index 2dd1c7fda..58528695a 100644 --- a/rust/operator-binary/src/controller/build/resource/discovery.rs +++ b/rust/operator-binary/src/controller/build/resource/discovery.rs @@ -17,11 +17,6 @@ use crate::{ pub enum Error { #[snafu(display("nodePort was out of range"))] InvalidNodePort { source: TryFromIntError }, - - #[snafu(display("failed to build ConfigMap"))] - BuildConfigMap { - source: stackable_operator::builder::configmap::Error, - }, } /// Build a discovery [`ConfigMap`] containing information about how to connect to a certain @@ -76,7 +71,7 @@ pub fn build_discovery_configmap(validated_cluster: &ValidatedCluster) -> Result ) .add_data("KAFKA", bootstrap_servers) .build() - .context(BuildConfigMapSnafu)?; + .expect("The ConfigMap metadata is set in this function."); Ok(discovery_cm) } diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs index 063d62325..244d3abb2 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -5,11 +5,8 @@ use stackable_operator::{ builder::{ meta::ObjectMetaBuilder, pod::{ - PodBuilder, - container::{ContainerBuilder, FieldPathEnvVar}, - resources::ResourceRequirementsBuilder, - security::PodSecurityContextBuilder, - volume::VolumeBuilder, + PodBuilder, container::FieldPathEnvVar, resources::ResourceRequirementsBuilder, + security::PodSecurityContextBuilder, volume::VolumeBuilder, }, }, commons::product_image_selection::ResolvedProductImage, @@ -30,7 +27,7 @@ use stackable_operator::{ builder::{ meta::ownerreference_from_resource, pod::{ - container::{EnvVarName, EnvVarSet}, + container::{EnvVarName, EnvVarSet, new_container_builder}, volume::{ListenerReference, listener_operator_volume_source_builder_build_pvc}, }, }, @@ -39,7 +36,7 @@ use stackable_operator::{ STACKABLE_LOG_DIR, ValidatedContainerLogConfigChoice, vector_container, }, role_group_utils::ResourceNames, - types::kubernetes::{ConfigMapKey, ContainerName, PersistentVolumeClaimName, VolumeName}, + types::kubernetes::{ConfigMapKey, ContainerName, VolumeName}, }, }; @@ -132,31 +129,6 @@ const POD_MANAGEMENT_POLICY_PARALLEL: &str = "Parallel"; #[derive(Snafu, Debug)] pub enum Error { - #[snafu(display("failed to add kerberos config"))] - AddKerberosConfig { - source: crate::controller::build::kerberos::Error, - }, - - #[snafu(display("failed to add listener volume"))] - AddListenerVolume { - source: stackable_operator::builder::pod::Error, - }, - - #[snafu(display("failed to add Secret Volumes and VolumeMounts"))] - AddVolumesAndVolumeMounts { - source: crate::controller::build::security::Error, - }, - - #[snafu(display("failed to add needed volumeMount"))] - AddVolumeMount { - source: stackable_operator::builder::pod::container::Error, - }, - - #[snafu(display("failed to add needed volume"))] - AddVolume { - source: stackable_operator::builder::pod::Error, - }, - #[snafu(display("failed to build pod descriptors"))] BuildPodDescriptors { source: crate::controller::PodDescriptorsError, @@ -172,12 +144,6 @@ pub enum Error { source: crate::controller::build::graceful_shutdown::Error, }, - #[snafu(display("invalid Container name [{name}]"))] - InvalidContainerName { - name: String, - source: stackable_operator::builder::pod::container::Error, - }, - #[snafu(display("missing secret lifetime"))] MissingSecretLifetime, } @@ -206,17 +172,8 @@ pub fn build_broker_rolegroup_statefulset( role_group_name, ); - let kcat_prober_container_name = BrokerContainer::KcatProber.to_string(); - let mut cb_kcat_prober = - ContainerBuilder::new(&kcat_prober_container_name).context(InvalidContainerNameSnafu { - name: kcat_prober_container_name.clone(), - })?; - - let kafka_container_name = BrokerContainer::Kafka.to_string(); - let mut cb_kafka = - ContainerBuilder::new(&kafka_container_name).context(InvalidContainerNameSnafu { - name: kafka_container_name.clone(), - })?; + let mut cb_kcat_prober = new_container_builder(&container_name(BrokerContainer::KcatProber)); + let mut cb_kafka = new_container_builder(&container_name(BrokerContainer::Kafka)); let mut pod_builder = PodBuilder::new(); @@ -231,8 +188,7 @@ pub fn build_broker_rolegroup_statefulset( &mut cb_kcat_prober, &mut cb_kafka, &requested_secret_lifetime, - ) - .context(AddVolumesAndVolumeMountsSnafu)?; + ); let mut pvcs = merged_config.resources().storage.build_pvcs(); @@ -240,12 +196,10 @@ pub fn build_broker_rolegroup_statefulset( // main broker listener is an ephemeral PVC instead let bootstrap_listener_name = validated_cluster.bootstrap_listener_name(kafka_role, role_group_name); - let bootstrap_pvc_name = PersistentVolumeClaimName::from_str(LISTENER_BOOTSTRAP_VOLUME_NAME) - .expect("the bootstrap listener volume name is a valid PVC name"); pvcs.push(listener_operator_volume_source_builder_build_pvc( &ListenerReference::Listener(bootstrap_listener_name), &unversioned_recommended_labels, - &bootstrap_pvc_name, + &LISTENER_BOOTSTRAP_VOLUME_NAME, )); if kafka_security.has_kerberos_enabled() { @@ -255,8 +209,7 @@ pub fn build_broker_rolegroup_statefulset( &mut cb_kcat_prober, &mut cb_kafka, &mut pod_builder, - ) - .context(AddKerberosConfigSnafu)?; + ); } // Operator-set env vars first; the user's `envOverrides` are merged on top last and win. @@ -296,21 +249,21 @@ pub fn build_broker_rolegroup_statefulset( cb_kafka .add_env_vars(env) .add_container_ports(container_ports(kafka_security)) - .add_volume_mount(LOG_DIRS_VOLUME_NAME, STACKABLE_DATA_DIR) - .context(AddVolumeMountSnafu)? - .add_volume_mount(STACKABLE_CONFIG_DIR_NAME, STACKABLE_CONFIG_DIR) - .context(AddVolumeMountSnafu)? + .add_volume_mount(&*LOG_DIRS_VOLUME_NAME, STACKABLE_DATA_DIR) + .expect("The mount paths are statically defined and there should be no duplicates.") + .add_volume_mount(&*STACKABLE_CONFIG_DIR_NAME, STACKABLE_CONFIG_DIR) + .expect("The mount paths are statically defined and there should be no duplicates.") .add_volume_mount( - LISTENER_BOOTSTRAP_VOLUME_NAME, + &*LISTENER_BOOTSTRAP_VOLUME_NAME, STACKABLE_LISTENER_BOOTSTRAP_DIR, ) - .context(AddVolumeMountSnafu)? - .add_volume_mount(LISTENER_BROKER_VOLUME_NAME, STACKABLE_LISTENER_BROKER_DIR) - .context(AddVolumeMountSnafu)? - .add_volume_mount(STACKABLE_LOG_CONFIG_DIR_NAME, STACKABLE_LOG_CONFIG_DIR) - .context(AddVolumeMountSnafu)? - .add_volume_mount(STACKABLE_LOG_DIR_NAME, STACKABLE_LOG_DIR) - .context(AddVolumeMountSnafu)? + .expect("The mount paths are statically defined and there should be no duplicates.") + .add_volume_mount(&*LISTENER_BROKER_VOLUME_NAME, STACKABLE_LISTENER_BROKER_DIR) + .expect("The mount paths are statically defined and there should be no duplicates.") + .add_volume_mount(&*STACKABLE_LOG_CONFIG_DIR_NAME, STACKABLE_LOG_CONFIG_DIR) + .expect("The mount paths are statically defined and there should be no duplicates.") + .add_volume_mount(&*STACKABLE_LOG_DIR_NAME, STACKABLE_LOG_DIR) + .expect("The mount paths are statically defined and there should be no duplicates.") .resources(merged_config.resources().clone().into()); // Use kcat sidecar for probing container status rather than the official Kafka tools, since they incur a lot of @@ -332,12 +285,12 @@ pub fn build_broker_rolegroup_statefulset( .build(), ) .add_volume_mount( - LISTENER_BOOTSTRAP_VOLUME_NAME, + &*LISTENER_BOOTSTRAP_VOLUME_NAME, STACKABLE_LISTENER_BOOTSTRAP_DIR, ) - .context(AddVolumeMountSnafu)? - .add_volume_mount(LISTENER_BROKER_VOLUME_NAME, STACKABLE_LISTENER_BROKER_DIR) - .context(AddVolumeMountSnafu)? + .expect("The mount paths are statically defined and there should be no duplicates.") + .add_volume_mount(&*LISTENER_BROKER_VOLUME_NAME, STACKABLE_LISTENER_BROKER_DIR) + .expect("The mount paths are statically defined and there should be no duplicates.") // Only allow the global load balancing service to send traffic to pods that are members of the quorum // This also acts as a hint to the StatefulSet controller to wait for each pod to enter quorum before taking down the next .readiness_probe(Probe { @@ -354,7 +307,7 @@ pub fn build_broker_rolegroup_statefulset( &mut pod_builder, &validated_rg.config.logging, &resource_names, - )?; + ); let metadata = ObjectMetaBuilder::new() .with_labels(recommended_labels.clone()) @@ -363,11 +316,11 @@ pub fn build_broker_rolegroup_statefulset( if let Some(listener_class) = merged_config.listener_class() { pod_builder .add_listener_volume_by_listener_class( - LISTENER_BROKER_VOLUME_NAME, + LISTENER_BROKER_VOLUME_NAME.as_ref(), listener_class.as_ref(), &recommended_labels, ) - .context(AddListenerVolumeSnafu)?; + .expect("The annotation keys are static, annotation values cannot be invalid, and the volume name is statically defined."); } if let Some(broker_id_config_map_name) = &validated_cluster @@ -376,14 +329,14 @@ pub fn build_broker_rolegroup_statefulset( { pod_builder .add_volume( - VolumeBuilder::new(BROKER_ID_POD_MAP_DIR_NAME) + VolumeBuilder::new(&*BROKER_ID_POD_MAP_DIR_NAME) .with_config_map(broker_id_config_map_name) .build(), ) - .context(AddVolumeSnafu)?; + .expect("The volume names are statically defined and there should be no duplicates."); cb_kafka - .add_volume_mount(BROKER_ID_POD_MAP_DIR_NAME, BROKER_ID_POD_MAP_DIR) - .context(AddVolumeMountSnafu)?; + .add_volume_mount(&*BROKER_ID_POD_MAP_DIR_NAME, BROKER_ID_POD_MAP_DIR) + .expect("The mount paths are statically defined and there should be no duplicates."); } pod_builder @@ -400,7 +353,7 @@ pub fn build_broker_rolegroup_statefulset( .cluster_resource_names() .service_account_name() .as_ref(), - )?; + ); add_vector_container( &mut pod_builder, @@ -465,11 +418,7 @@ pub fn build_controller_rolegroup_statefulset( let recommended_labels = recommended_labels_for_role_group_resources(validated_cluster, kafka_role, role_group_name); - let kafka_container_name = ControllerContainer::Kafka.to_string(); - let mut cb_kafka = - ContainerBuilder::new(&kafka_container_name).context(InvalidContainerNameSnafu { - name: kafka_container_name.clone(), - })?; + let mut cb_kafka = new_container_builder(&container_name(ControllerContainer::Kafka)); let mut pod_builder = PodBuilder::new(); @@ -517,14 +466,14 @@ pub fn build_controller_rolegroup_statefulset( cb_kafka .add_env_vars(env) .add_container_ports(container_ports(kafka_security)) - .add_volume_mount(LOG_DIRS_VOLUME_NAME, STACKABLE_DATA_DIR) - .context(AddVolumeMountSnafu)? - .add_volume_mount(STACKABLE_CONFIG_DIR_NAME, STACKABLE_CONFIG_DIR) - .context(AddVolumeMountSnafu)? - .add_volume_mount(STACKABLE_LOG_CONFIG_DIR_NAME, STACKABLE_LOG_CONFIG_DIR) - .context(AddVolumeMountSnafu)? - .add_volume_mount(STACKABLE_LOG_DIR_NAME, STACKABLE_LOG_DIR) - .context(AddVolumeMountSnafu)? + .add_volume_mount(&*LOG_DIRS_VOLUME_NAME, STACKABLE_DATA_DIR) + .expect("The mount paths are statically defined and there should be no duplicates.") + .add_volume_mount(&*STACKABLE_CONFIG_DIR_NAME, STACKABLE_CONFIG_DIR) + .expect("The mount paths are statically defined and there should be no duplicates.") + .add_volume_mount(&*STACKABLE_LOG_CONFIG_DIR_NAME, STACKABLE_LOG_CONFIG_DIR) + .expect("The mount paths are statically defined and there should be no duplicates.") + .add_volume_mount(&*STACKABLE_LOG_DIR_NAME, STACKABLE_LOG_DIR) + .expect("The mount paths are statically defined and there should be no duplicates.") .resources(merged_config.resources().clone().into()) // TODO: improve probes .liveness_probe(Probe { @@ -552,7 +501,7 @@ pub fn build_controller_rolegroup_statefulset( &mut pod_builder, &validated_rg.config.logging, &resource_names, - )?; + ); let metadata = ObjectMetaBuilder::new() .with_labels(recommended_labels.clone()) @@ -568,8 +517,7 @@ pub fn build_controller_rolegroup_statefulset( &mut pod_builder, &mut cb_kafka, &requested_secret_lifetime, - ) - .context(AddVolumesAndVolumeMountsSnafu)?; + ); let kafka_container = cb_kafka.build(); @@ -586,7 +534,7 @@ pub fn build_controller_rolegroup_statefulset( .cluster_resource_names() .service_account_name() .as_ref(), - )?; + ); add_vector_container( &mut pod_builder, @@ -716,7 +664,7 @@ fn add_log_config_volume( pod_builder: &mut PodBuilder, logging: &ValidatedLogging, resource_names: &ResourceNames, -) -> Result<(), Error> { +) { let config_map = match &logging.kafka_container { ValidatedContainerLogConfigChoice::Custom(config_map_name) => config_map_name.to_string(), ValidatedContainerLogConfigChoice::Automatic(_) => { @@ -725,12 +673,11 @@ fn add_log_config_volume( }; pod_builder .add_volume( - VolumeBuilder::new(STACKABLE_LOG_CONFIG_DIR_NAME) + VolumeBuilder::new(&*STACKABLE_LOG_CONFIG_DIR_NAME) .with_config_map(config_map) .build(), ) - .context(AddVolumeSnafu)?; - Ok(()) + .expect("The volume names are statically defined and there should be no duplicates."); } /// Adds the `config` volume, the `log` emptyDir, the service account and the pod security @@ -739,7 +686,7 @@ fn add_common_pod_config( pod_builder: &mut PodBuilder, resource_names: &ResourceNames, service_account_name: &str, -) -> Result<(), Error> { +) { pod_builder .add_volume(Volume { name: STACKABLE_CONFIG_DIR_NAME.to_string(), @@ -749,21 +696,20 @@ fn add_common_pod_config( }), ..Volume::default() }) - .context(AddVolumeSnafu)? + .expect("The volume names are statically defined and there should be no duplicates.") .add_empty_dir_volume( - STACKABLE_LOG_DIR_NAME, + &*STACKABLE_LOG_DIR_NAME, Some(product_logging::framework::calculate_log_volume_size_limit( &[MAX_KAFKA_LOG_FILES_SIZE], )), ) - .context(AddVolumeSnafu)? + .expect("The volume names are statically defined and there should be no duplicates.") .service_account_name(service_account_name) .security_context( PodSecurityContextBuilder::with_stackable_defaults() .fs_group(1000) .build(), ); - Ok(()) } /// Adds the Vector log-aggregation sidecar container, when the Vector agent is enabled. diff --git a/rust/operator-binary/src/controller/build/security.rs b/rust/operator-binary/src/controller/build/security.rs index e36191c10..c2155d1b1 100644 --- a/rust/operator-binary/src/controller/build/security.rs +++ b/rust/operator-binary/src/controller/build/security.rs @@ -3,22 +3,20 @@ //! //! These consume the validated security inputs and produce build artifacts; they must not perform //! any validation themselves. -use std::collections::BTreeMap; +use std::{collections::BTreeMap, str::FromStr}; -use snafu::{ResultExt, Snafu}; use stackable_operator::{ - builder::{ - self, - pod::{ - PodBuilder, - container::ContainerBuilder, - volume::{SecretFormat, SecretOperatorVolumeSourceBuilder, VolumeBuilder}, - }, + builder::pod::{ + PodBuilder, + container::ContainerBuilder, + volume::{SecretFormat, SecretOperatorVolumeSourceBuilder, VolumeBuilder}, }, commons::secret_class::SecretClassVolumeProvisionParts, + constant, crd::authentication::core, k8s_openapi::api::core::v1::Volume, shared::time::Duration, + v2::types::kubernetes::VolumeName, }; use crate::{ @@ -39,7 +37,7 @@ const INTER_BROKER_LISTENER_NAME: &str = "inter.broker.listener.name"; const KEYSTORE_P12_FILE_NAME: &str = "keystore.p12"; const OPA_TLS_MOUNT_PATH: &str = "/stackable/tls-opa"; // opa -const OPA_TLS_VOLUME_NAME: &str = "tls-opa"; +constant!(OPA_TLS_VOLUME_NAME: VolumeName = "tls-opa"); const SSL_STORE_PASSWORD: &str = ""; const SSL_STORE_TYPE_PKCS12: &str = "PKCS12"; const SSL_CLIENT_AUTH_REQUIRED: &str = "required"; @@ -50,35 +48,14 @@ const PROPERTY_SASL_ENABLED_MECHANISMS: &str = "sasl.enabled.mechanisms"; const PROPERTY_SASL_KERBEROS_SERVICE_NAME: &str = "sasl.kerberos.service.name"; const PROPERTY_SASL_INTER_BROKER_MECHANISM: &str = "sasl.mechanism.inter.broker.protocol"; const STACKABLE_TLS_KAFKA_INTERNAL_DIR: &str = "/stackable/tls-kafka-internal"; -const STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME: &str = "tls-kafka-internal"; +constant!(STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME: VolumeName = "tls-kafka-internal"); const STACKABLE_TLS_KAFKA_SERVER_DIR: &str = "/stackable/tls-kafka-server"; -const STACKABLE_TLS_KAFKA_SERVER_VOLUME_NAME: &str = "tls-kafka-server"; +constant!(STACKABLE_TLS_KAFKA_SERVER_VOLUME_NAME: VolumeName = "tls-kafka-server"); // directories const STACKABLE_TLS_KCAT_DIR: &str = "/stackable/tls-kcat"; -const STACKABLE_TLS_KCAT_VOLUME_NAME: &str = "tls-kcat"; +constant!(STACKABLE_TLS_KCAT_VOLUME_NAME: VolumeName = "tls-kcat"); const TRUSTSTORE_P12_FILE_NAME: &str = "truststore.p12"; -#[derive(Snafu, Debug)] -pub enum Error { - #[snafu(display("failed to build the secret operator Volume"))] - SecretVolumeBuild { - source: stackable_operator::builder::pod::volume::SecretOperatorVolumeSourceBuilderError, - }, - - #[snafu(display("failed to add needed volume"))] - AddVolume { source: builder::pod::Error }, - - #[snafu(display("failed to add needed volumeMount"))] - AddVolumeMount { - source: builder::pod::container::Error, - }, - - #[snafu(display("failed to build OPA TLS certificate volume"))] - OpaTlsCertSecretClassVolumeBuild { - source: stackable_operator::builder::pod::volume::SecretOperatorVolumeSourceBuilderError, - }, -} - pub fn copy_opa_tls_cert_command(security: &ValidatedKafkaSecurity) -> String { match security.opa_secret_class().is_some() { true => format!( @@ -229,58 +206,58 @@ pub fn add_broker_volume_and_volume_mounts( cb_kcat_prober: &mut ContainerBuilder, cb_kafka: &mut ContainerBuilder, requested_secret_lifetime: &Duration, -) -> Result<(), Error> { +) { // add tls (server or client authentication volumes) if required if let Some(tls_server_secret_class) = tls_secret_class(security) { // We have to mount tls pem files for kcat (the mount can be used directly) pod_builder .add_volume(create_kcat_tls_volume( - STACKABLE_TLS_KCAT_VOLUME_NAME, + &STACKABLE_TLS_KCAT_VOLUME_NAME, tls_server_secret_class, requested_secret_lifetime, - )?) - .context(AddVolumeSnafu)?; + )) + .expect("The volume names are statically defined and there should be no duplicates."); cb_kcat_prober - .add_volume_mount(STACKABLE_TLS_KCAT_VOLUME_NAME, STACKABLE_TLS_KCAT_DIR) - .context(AddVolumeMountSnafu)?; + .add_volume_mount(&*STACKABLE_TLS_KCAT_VOLUME_NAME, STACKABLE_TLS_KCAT_DIR) + .expect("The mount paths are statically defined and there should be no duplicates."); // Keystores fore the kafka container pod_builder .add_volume(create_tls_keystore_volume( - STACKABLE_TLS_KAFKA_SERVER_VOLUME_NAME, + &STACKABLE_TLS_KAFKA_SERVER_VOLUME_NAME, tls_server_secret_class, requested_secret_lifetime, - )?) - .context(AddVolumeSnafu)?; + )) + .expect("The volume names are statically defined and there should be no duplicates."); cb_kafka .add_volume_mount( - STACKABLE_TLS_KAFKA_SERVER_VOLUME_NAME, + &*STACKABLE_TLS_KAFKA_SERVER_VOLUME_NAME, STACKABLE_TLS_KAFKA_SERVER_DIR, ) - .context(AddVolumeMountSnafu)?; + .expect("The mount paths are statically defined and there should be no duplicates."); } pod_builder .add_volume(create_tls_keystore_volume( - STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME, + &STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME, security.tls_internal_secret_class(), requested_secret_lifetime, - )?) - .context(AddVolumeSnafu)?; + )) + .expect("The volume names are statically defined and there should be no duplicates."); cb_kafka .add_volume_mount( - STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME, + &*STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME, STACKABLE_TLS_KAFKA_INTERNAL_DIR, ) - .context(AddVolumeMountSnafu)?; + .expect("The mount paths are statically defined and there should be no duplicates."); if let Some(secret_class) = security.opa_secret_class() { cb_kafka - .add_volume_mount(OPA_TLS_VOLUME_NAME, OPA_TLS_MOUNT_PATH) - .context(AddVolumeMountSnafu)?; + .add_volume_mount(&*OPA_TLS_VOLUME_NAME, OPA_TLS_MOUNT_PATH) + .expect("The mount paths are statically defined and there should be no duplicates."); pod_builder .add_volume( - VolumeBuilder::new(OPA_TLS_VOLUME_NAME) + VolumeBuilder::new(&*OPA_TLS_VOLUME_NAME) .ephemeral( SecretOperatorVolumeSourceBuilder::new( secret_class, @@ -288,14 +265,12 @@ pub fn add_broker_volume_and_volume_mounts( SecretClassVolumeProvisionParts::Public, ) .build() - .context(OpaTlsCertSecretClassVolumeBuildSnafu)?, + .expect("The annotation keys are static and annotation values cannot be invalid."), ) .build(), ) - .context(AddVolumeSnafu)?; + .expect("The volume names are statically defined and there should be no duplicates."); } - - Ok(()) } /// Adds required volumes and volume mounts to the controller pod and container builders @@ -305,10 +280,10 @@ pub fn add_controller_volume_and_volume_mounts( pod_builder: &mut PodBuilder, cb_kafka: &mut ContainerBuilder, requested_secret_lifetime: &Duration, -) -> Result<(), Error> { +) { pod_builder .add_volume( - VolumeBuilder::new(STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME) + VolumeBuilder::new(&*STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME) .ephemeral( SecretOperatorVolumeSourceBuilder::new( security.tls_internal_secret_class(), @@ -321,19 +296,19 @@ pub fn add_controller_volume_and_volume_mounts( .with_auto_tls_cert_lifetime(*requested_secret_lifetime) .with_auto_tls_cert_domain_components_in_subject_dn(true) .build() - .context(SecretVolumeBuildSnafu)?, + .expect( + "The annotation keys are static and annotation values cannot be invalid.", + ), ) .build(), ) - .context(AddVolumeSnafu)?; + .expect("The volume names are statically defined and there should be no duplicates."); cb_kafka .add_volume_mount( - STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME, + &*STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME, STACKABLE_TLS_KAFKA_INTERNAL_DIR, ) - .context(AddVolumeMountSnafu)?; - - Ok(()) + .expect("The mount paths are statically defined and there should be no duplicates."); } /// Inserts the `listener..ssl.{keystore,truststore}.{location,password,type}` @@ -562,11 +537,11 @@ fn tls_secret_class(security: &ValidatedKafkaSecurity) -> Option<&str> { /// Creates ephemeral volumes to mount the `SecretClass` into the Pods for kcat client fn create_kcat_tls_volume( - volume_name: &str, + volume_name: &VolumeName, secret_class_name: &str, requested_secret_lifetime: &Duration, -) -> Result { - Ok(VolumeBuilder::new(volume_name) +) -> Volume { + VolumeBuilder::new(volume_name) .ephemeral( SecretOperatorVolumeSourceBuilder::new( secret_class_name, @@ -579,18 +554,18 @@ fn create_kcat_tls_volume( .with_auto_tls_cert_lifetime(*requested_secret_lifetime) .with_auto_tls_cert_domain_components_in_subject_dn(true) .build() - .context(SecretVolumeBuildSnafu)?, + .expect("The annotation keys are static and annotation values cannot be invalid."), ) - .build()) + .build() } /// Creates ephemeral volumes to mount the `SecretClass` into the Pods as keystores fn create_tls_keystore_volume( - volume_name: &str, + volume_name: &VolumeName, secret_class_name: &str, requested_secret_lifetime: &Duration, -) -> Result { - Ok(VolumeBuilder::new(volume_name) +) -> Volume { + VolumeBuilder::new(volume_name) .ephemeral( SecretOperatorVolumeSourceBuilder::new( secret_class_name, @@ -598,15 +573,15 @@ fn create_tls_keystore_volume( SecretClassVolumeProvisionParts::PublicPrivate, ) .with_pod_scope() - .with_listener_volume_scope(LISTENER_BROKER_VOLUME_NAME) - .with_listener_volume_scope(LISTENER_BOOTSTRAP_VOLUME_NAME) + .with_listener_volume_scope(&*LISTENER_BROKER_VOLUME_NAME) + .with_listener_volume_scope(&*LISTENER_BOOTSTRAP_VOLUME_NAME) .with_format(SecretFormat::TlsPkcs12) .with_auto_tls_cert_lifetime(*requested_secret_lifetime) .with_auto_tls_cert_domain_components_in_subject_dn(true) .build() - .context(SecretVolumeBuildSnafu)?, + .expect("The annotation keys are static and annotation values cannot be invalid."), ) - .build()) + .build() } fn kcat_client_auth_ssl(cert_directory: &str) -> Vec { @@ -763,6 +738,15 @@ mod tests { // ---- kcat_prober_container_commands ---- + #[test] + fn test_constants() { + // Test that dereferencing the constants does not panic. + let _ = *OPA_TLS_VOLUME_NAME; + let _ = *STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME; + let _ = *STACKABLE_TLS_KAFKA_SERVER_VOLUME_NAME; + let _ = *STACKABLE_TLS_KCAT_VOLUME_NAME; + } + #[test] fn kcat_prober_plaintext_targets_insecure_client_port() { let commands = kcat_prober_container_commands(&plaintext()); diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs index c2d118256..5005122fa 100644 --- a/rust/operator-binary/src/crd/mod.rs +++ b/rust/operator-binary/src/crd/mod.rs @@ -28,7 +28,10 @@ use stackable_operator::{ role_utils::{JavaCommonConfig, Role}, types::{ common::Port, - kubernetes::{ConfigMapName, NamespaceName, ServiceName, StatefulSetName}, + kubernetes::{ + ConfigMapName, NamespaceName, PersistentVolumeClaimName, ServiceName, + StatefulSetName, VolumeName, + }, }, }, versioned::versioned, @@ -51,24 +54,27 @@ pub const METRICS_PORT: Port = Port(9606); // env vars constant!(pub KAFKA_HEAP_OPTS: EnvVarName = "KAFKA_HEAP_OPTS"); // server_properties -pub const LOG_DIRS_VOLUME_NAME: &str = "log-dirs"; +// The log-dirs PVC (a volumeClaimTemplate) and the volume mount referencing it share this name. +constant!(pub LOG_DIRS_VOLUME_NAME: PersistentVolumeClaimName = "log-dirs"); // directories -pub const LISTENER_BROKER_VOLUME_NAME: &str = "listener-broker"; -pub const LISTENER_BOOTSTRAP_VOLUME_NAME: &str = "listener-bootstrap"; +constant!(pub LISTENER_BROKER_VOLUME_NAME: VolumeName = "listener-broker"); +// The bootstrap listener PVC (a volumeClaimTemplate) and the volume mount referencing it share +// this name. +constant!(pub LISTENER_BOOTSTRAP_VOLUME_NAME: PersistentVolumeClaimName = "listener-bootstrap"); pub const STACKABLE_LISTENER_BROKER_DIR: &str = "/stackable/listener-broker"; pub const STACKABLE_LISTENER_BOOTSTRAP_DIR: &str = "/stackable/listener-bootstrap"; pub const STACKABLE_DATA_DIR: &str = "/stackable/data"; pub const STACKABLE_CONFIG_DIR: &str = "/stackable/config"; -pub const STACKABLE_CONFIG_DIR_NAME: &str = "config"; +constant!(pub STACKABLE_CONFIG_DIR_NAME: VolumeName = "config"); // kerberos pub const STACKABLE_KERBEROS_DIR: &str = "/stackable/kerberos"; pub const STACKABLE_KERBEROS_KRB5_PATH: &str = "/stackable/kerberos/krb5.conf"; // logging pub const STACKABLE_LOG_CONFIG_DIR: &str = "/stackable/log_config"; -pub const STACKABLE_LOG_CONFIG_DIR_NAME: &str = "log-config"; -pub const STACKABLE_LOG_DIR_NAME: &str = "log"; +constant!(pub STACKABLE_LOG_CONFIG_DIR_NAME: VolumeName = "log-config"); +constant!(pub STACKABLE_LOG_DIR_NAME: VolumeName = "log"); pub const BROKER_ID_POD_MAP_DIR: &str = "/stackable/broker-id-pod-map"; -pub const BROKER_ID_POD_MAP_DIR_NAME: &str = "broker-id-pod-map-dir"; +constant!(pub BROKER_ID_POD_MAP_DIR_NAME: VolumeName = "broker-id-pod-map-dir"); #[derive(Snafu, Debug)] pub enum Error { @@ -76,16 +82,6 @@ pub enum Error { "The ZooKeeper metadata manager is not supported for Kafka version 4 and higher" ))] Kafka4RequiresKraftMetadataManager, - - #[snafu(display( - "Kafka version 4 and higher requires a Kraft controller (configured via `spec.controller`)" - ))] - Kafka4RequiresKraft, - - #[snafu(display( - "Kraft controller (`spec.controller`) and ZooKeeper (`spec.clusterConfig.zookeeperConfigMapName`) are configured. Please only choose one" - ))] - KraftAndZookeeperConfigured, } pub type BrokerRole = Role< @@ -413,6 +409,13 @@ mod tests { fn test_constants() { // Test that dereferencing the constants does not panic. let _ = *KAFKA_HEAP_OPTS; + let _ = *LOG_DIRS_VOLUME_NAME; + let _ = *LISTENER_BROKER_VOLUME_NAME; + let _ = *LISTENER_BOOTSTRAP_VOLUME_NAME; + let _ = *STACKABLE_CONFIG_DIR_NAME; + let _ = *STACKABLE_LOG_CONFIG_DIR_NAME; + let _ = *STACKABLE_LOG_DIR_NAME; + let _ = *BROKER_ID_POD_MAP_DIR_NAME; } fn get_server_secret_class(kafka: &v1alpha1::KafkaCluster) -> Option { diff --git a/rust/operator-binary/src/crd/role/broker.rs b/rust/operator-binary/src/crd/role/broker.rs index c1eafa342..54e5dfb91 100644 --- a/rust/operator-binary/src/crd/role/broker.rs +++ b/rust/operator-binary/src/crd/role/broker.rs @@ -1,3 +1,5 @@ +use std::str::FromStr; + use serde::{Deserialize, Serialize}; use stackable_operator::{ commons::resources::{ @@ -5,6 +7,7 @@ use stackable_operator::{ PvcConfigFragment, Resources, ResourcesFragment, }, config::{fragment::Fragment, merge::Merge}, + constant, k8s_openapi::apimachinery::pkg::api::resource::Quantity, product_logging::{self, spec::Logging}, schemars::{self, JsonSchema}, @@ -14,6 +17,9 @@ use strum::{Display, EnumIter}; use crate::crd::role::commons::{CommonConfig, Storage, StorageFragment}; +// The default listener class for both the bootstrap and the broker listeners. +constant!(DEFAULT_LISTENER_CLASS: ListenerClassName = "cluster-internal"); + #[derive( Clone, Debug, @@ -70,16 +76,8 @@ impl BrokerConfig { pub fn default_config(cluster_name: &str, role: &str) -> BrokerConfigFragment { BrokerConfigFragment { common_config: CommonConfig::default_config(cluster_name, role), - bootstrap_listener_class: Some( - "cluster-internal" - .parse() - .expect("\"cluster-internal\" is a valid listener class name"), - ), - broker_listener_class: Some( - "cluster-internal" - .parse() - .expect("\"cluster-internal\" is a valid listener class name"), - ), + bootstrap_listener_class: Some(DEFAULT_LISTENER_CLASS.clone()), + broker_listener_class: Some(DEFAULT_LISTENER_CLASS.clone()), logging: product_logging::spec::default_logging(), resources: ResourcesFragment { cpu: CpuLimitsFragment { @@ -101,3 +99,14 @@ impl BrokerConfig { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_constants() { + // Test that dereferencing the constants does not panic. + let _ = *DEFAULT_LISTENER_CLASS; + } +} diff --git a/rust/operator-binary/src/crd/role/commons.rs b/rust/operator-binary/src/crd/role/commons.rs index 1ef6f3dc4..32ac469f0 100644 --- a/rust/operator-binary/src/crd/role/commons.rs +++ b/rust/operator-binary/src/crd/role/commons.rs @@ -7,7 +7,7 @@ use stackable_operator::{ shared::time::Duration, }; -use crate::crd::affinity::get_affinity; +use crate::crd::{LOG_DIRS_VOLUME_NAME, affinity::get_affinity}; #[derive(Clone, Debug, Default, PartialEq, Fragment, JsonSchema)] #[fragment_attrs( @@ -29,12 +29,10 @@ pub struct Storage { } impl Storage { - pub const LOG_DIRS_VOLUME_NAME: &str = "log-dirs"; - pub fn build_pvcs(&self) -> Vec { let data_pvc = self .log_dirs - .build_pvc(Self::LOG_DIRS_VOLUME_NAME, Some(vec!["ReadWriteOnce"])); + .build_pvc(LOG_DIRS_VOLUME_NAME.as_ref(), Some(vec!["ReadWriteOnce"])); vec![data_pvc] } } diff --git a/rust/operator-binary/src/crd/tls.rs b/rust/operator-binary/src/crd/tls.rs index 82650799e..161f02aa2 100644 --- a/rust/operator-binary/src/crd/tls.rs +++ b/rust/operator-binary/src/crd/tls.rs @@ -2,11 +2,12 @@ use std::str::FromStr; use serde::{Deserialize, Serialize}; use stackable_operator::{ + constant, schemars::{self, JsonSchema}, v2::types::kubernetes::SecretClassName, }; -const TLS_DEFAULT_SECRET_CLASS: &str = "tls"; +constant!(TLS_DEFAULT_SECRET_CLASS: SecretClassName = "tls"); #[derive(Clone, Deserialize, Debug, Eq, JsonSchema, PartialEq, Serialize)] #[serde(rename_all = "camelCase")] @@ -44,8 +45,7 @@ pub fn default_kafka_tls() -> Option { /// The `tls` default secret class as a typed name. fn default_secret_class() -> SecretClassName { - SecretClassName::from_str(TLS_DEFAULT_SECRET_CLASS) - .expect("the default secret class name is valid") + TLS_DEFAULT_SECRET_CLASS.clone() } /// Helper methods to provide defaults in the CRDs and tests @@ -57,3 +57,14 @@ pub fn internal_tls_default() -> SecretClassName { pub fn server_tls_default() -> Option { Some(default_secret_class()) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_constants() { + // Test that dereferencing the constants does not panic. + let _ = *TLS_DEFAULT_SECRET_CLASS; + } +} From 33511ea4710a86a24dc1cccc302896e927848d4b Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Fri, 28 Aug 2026 17:57:55 +0200 Subject: [PATCH 02/10] changelog --- CHANGELOG.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index b4598838b..92948d7a4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,7 @@ All notable changes to this project will be documented in this file. Broker StatefulSets created by older operator versions cannot be updated in place: after the operator upgrade, delete each broker StatefulSet so that the operator immediately recreates it with the new labels ([#1011]). +- Make operations infallible where appropriate ([#1017]). ### Fixed @@ -41,6 +42,7 @@ All notable changes to this project will be documented in this file. [#1000]: https://github.com/stackabletech/kafka-operator/pull/1000 [#1011]: https://github.com/stackabletech/kafka-operator/pull/1011 [#1014]: https://github.com/stackabletech/kafka-operator/pull/1014 +[#1017]: https://github.com/stackabletech/kafka-operator/pull/1017 ## [26.7.0] - 2026-07-21 From 2ab931789cedeae03190a87f4c05acec0ac08fcf Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Mon, 7 Sep 2026 16:28:27 +0200 Subject: [PATCH 03/10] add panic docs to helper functions --- .../src/controller/build/kerberos.rs | 8 +++++++ .../controller/build/resource/statefulset.rs | 12 ++++++++++ .../src/controller/build/security.rs | 22 +++++++++++++++++++ 3 files changed, 42 insertions(+) diff --git a/rust/operator-binary/src/controller/build/kerberos.rs b/rust/operator-binary/src/controller/build/kerberos.rs index bee698226..17a627fdf 100644 --- a/rust/operator-binary/src/controller/build/kerberos.rs +++ b/rust/operator-binary/src/controller/build/kerberos.rs @@ -24,6 +24,14 @@ use crate::{ constant!(KERBEROS_VOLUME_NAME: VolumeName = "kerberos"); +/// Adds the Kerberos keytab and `krb5.conf` volume to the pod builder and mounts it into the +/// Kafka and kcat-prober containers, when Kerberos is enabled. +/// +/// # Panics +/// +/// Panics if the volumes or volume mounts cannot be added to the builders. Only call this +/// on builders whose volume names and mount paths are still distinct from the ones added +/// here. pub fn add_kerberos_pod_config( kafka_security: &ValidatedKafkaSecurity, role: &KafkaRole, diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs index 244d3abb2..bf4aa2b3e 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -660,6 +660,12 @@ fn common_kafka_env( /// Adds the `log-config` volume, sourced either from the user-supplied custom log config /// `ConfigMap` or the rolegroup `ConfigMap` (which carries the operator-generated config). /// Branches on the *validated* Kafka-container logging choice. +/// +/// # Panics +/// +/// Panics if the volumes or volume mounts cannot be added to the builders. Only call this +/// on builders whose volume names and mount paths are still distinct from the ones added +/// here. fn add_log_config_volume( pod_builder: &mut PodBuilder, logging: &ValidatedLogging, @@ -682,6 +688,12 @@ fn add_log_config_volume( /// Adds the `config` volume, the `log` emptyDir, the service account and the pod security /// context that the broker and controller pods share. +/// +/// # Panics +/// +/// Panics if the volumes or volume mounts cannot be added to the builders. Only call this +/// on builders whose volume names and mount paths are still distinct from the ones added +/// here. fn add_common_pod_config( pod_builder: &mut PodBuilder, resource_names: &ResourceNames, diff --git a/rust/operator-binary/src/controller/build/security.rs b/rust/operator-binary/src/controller/build/security.rs index c2155d1b1..d466313b8 100644 --- a/rust/operator-binary/src/controller/build/security.rs +++ b/rust/operator-binary/src/controller/build/security.rs @@ -200,6 +200,12 @@ pub fn client_properties(security: &ValidatedKafkaSecurity) -> Vec<(String, Opti /// Adds required volumes and volume mounts to the broker pod and container builders /// depending on the tls and authentication settings. +/// +/// # Panics +/// +/// Panics if the volumes or volume mounts cannot be added to the builders. Only call this +/// on builders whose volume names and mount paths are still distinct from the ones added +/// here. pub fn add_broker_volume_and_volume_mounts( security: &ValidatedKafkaSecurity, pod_builder: &mut PodBuilder, @@ -275,6 +281,12 @@ pub fn add_broker_volume_and_volume_mounts( /// Adds required volumes and volume mounts to the controller pod and container builders /// depending on the tls and authentication settings. +/// +/// # Panics +/// +/// Panics if the volumes or volume mounts cannot be added to the builders. Only call this +/// on builders whose volume names and mount paths are still distinct from the ones added +/// here. pub fn add_controller_volume_and_volume_mounts( security: &ValidatedKafkaSecurity, pod_builder: &mut PodBuilder, @@ -536,6 +548,11 @@ fn tls_secret_class(security: &ValidatedKafkaSecurity) -> Option<&str> { } /// Creates ephemeral volumes to mount the `SecretClass` into the Pods for kcat client +/// +/// # Panics +/// +/// Panics if the volume source cannot be built, which cannot happen because the annotation +/// keys are static and annotation values cannot be invalid. fn create_kcat_tls_volume( volume_name: &VolumeName, secret_class_name: &str, @@ -560,6 +577,11 @@ fn create_kcat_tls_volume( } /// Creates ephemeral volumes to mount the `SecretClass` into the Pods as keystores +/// +/// # Panics +/// +/// Panics if the volume source cannot be built, which cannot happen because the annotation +/// keys are static and annotation values cannot be invalid. fn create_tls_keystore_volume( volume_name: &VolumeName, secret_class_name: &str, From a2ea17a742f0254f26ba79c68c864dc02961597d Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Mon, 7 Sep 2026 16:42:45 +0200 Subject: [PATCH 04/10] remove panics doc where it make no sense --- rust/operator-binary/src/controller/build/security.rs | 10 ---------- 1 file changed, 10 deletions(-) diff --git a/rust/operator-binary/src/controller/build/security.rs b/rust/operator-binary/src/controller/build/security.rs index d466313b8..6570ba7f6 100644 --- a/rust/operator-binary/src/controller/build/security.rs +++ b/rust/operator-binary/src/controller/build/security.rs @@ -548,11 +548,6 @@ fn tls_secret_class(security: &ValidatedKafkaSecurity) -> Option<&str> { } /// Creates ephemeral volumes to mount the `SecretClass` into the Pods for kcat client -/// -/// # Panics -/// -/// Panics if the volume source cannot be built, which cannot happen because the annotation -/// keys are static and annotation values cannot be invalid. fn create_kcat_tls_volume( volume_name: &VolumeName, secret_class_name: &str, @@ -577,11 +572,6 @@ fn create_kcat_tls_volume( } /// Creates ephemeral volumes to mount the `SecretClass` into the Pods as keystores -/// -/// # Panics -/// -/// Panics if the volume source cannot be built, which cannot happen because the annotation -/// keys are static and annotation values cannot be invalid. fn create_tls_keystore_volume( volume_name: &VolumeName, secret_class_name: &str, From 54babc9ae5322f5c2ab169e06c58cbb51fd4c744 Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Tue, 8 Sep 2026 17:17:37 +0200 Subject: [PATCH 05/10] revert expects where checked data is not static/explicit --- CHANGELOG.md | 2 +- .../src/controller/build/kerberos.rs | 39 ++++++--- .../controller/build/resource/config_map.rs | 12 ++- .../controller/build/resource/discovery.rs | 7 +- .../controller/build/resource/statefulset.rs | 65 ++++++++------ .../src/controller/build/security.rs | 84 ++++++++++++------- 6 files changed, 137 insertions(+), 72 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 92948d7a4..207a0c37f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,7 +25,7 @@ All notable changes to this project will be documented in this file. Broker StatefulSets created by older operator versions cannot be updated in place: after the operator upgrade, delete each broker StatefulSet so that the operator immediately recreates it with the new labels ([#1011]). -- Make operations infallible where appropriate ([#1017]). +- Make operations infallible where dependent on static inputs ([#1017]). ### Fixed diff --git a/rust/operator-binary/src/controller/build/kerberos.rs b/rust/operator-binary/src/controller/build/kerberos.rs index 17a627fdf..7e2ebebea 100644 --- a/rust/operator-binary/src/controller/build/kerberos.rs +++ b/rust/operator-binary/src/controller/build/kerberos.rs @@ -1,10 +1,17 @@ use std::str::FromStr; +use snafu::{ResultExt, Snafu}; use stackable_operator::{ - builder::pod::{ - PodBuilder, - container::ContainerBuilder, - volume::{SecretOperatorVolumeSourceBuilder, VolumeBuilder}, + builder::{ + self, + pod::{ + PodBuilder, + container::ContainerBuilder, + volume::{ + SecretOperatorVolumeSourceBuilder, SecretOperatorVolumeSourceBuilderError, + VolumeBuilder, + }, + }, }, commons::secret_class::SecretClassVolumeProvisionParts, constant, @@ -24,21 +31,31 @@ use crate::{ constant!(KERBEROS_VOLUME_NAME: VolumeName = "kerberos"); +#[derive(Snafu, Debug)] +pub enum Error { + #[snafu(display("failed to add Kerberos secret volume"))] + KerberosSecretVolume { + source: SecretOperatorVolumeSourceBuilderError, + }, + + #[snafu(display("failed to add needed volume"))] + AddVolume { source: builder::pod::Error }, +} + /// Adds the Kerberos keytab and `krb5.conf` volume to the pod builder and mounts it into the /// Kafka and kcat-prober containers, when Kerberos is enabled. /// /// # Panics /// -/// Panics if the volumes or volume mounts cannot be added to the builders. Only call this -/// on builders whose volume names and mount paths are still distinct from the ones added -/// here. +/// Panics if the volume mounts cannot be added to the container builders. Only call this on +/// container builders whose mount paths are still distinct from the ones added here. pub fn add_kerberos_pod_config( kafka_security: &ValidatedKafkaSecurity, role: &KafkaRole, cb_kcat_prober: &mut ContainerBuilder, cb_kafka: &mut ContainerBuilder, pb: &mut PodBuilder, -) { +) -> Result<(), Error> { if let Some(kerberos_secret_class) = kafka_security.kerberos_secret_class() { // Mount keytab let kerberos_secret_operator_volume = SecretOperatorVolumeSourceBuilder::new( @@ -50,13 +67,13 @@ pub fn add_kerberos_pod_config( .with_listener_volume_scope(&*LISTENER_BOOTSTRAP_VOLUME_NAME) .with_kerberos_service_name(role.kerberos_service_name()) .build() - .expect("The annotation keys are static and annotation values cannot be invalid."); + .context(KerberosSecretVolumeSnafu)?; pb.add_volume( VolumeBuilder::new(&*KERBEROS_VOLUME_NAME) .ephemeral(kerberos_secret_operator_volume) .build(), ) - .expect("The volume names are statically defined and there should be no duplicates."); + .context(AddVolumeSnafu)?; for cb in [cb_kafka, cb_kcat_prober] { cb.add_volume_mount(&*KERBEROS_VOLUME_NAME, STACKABLE_KERBEROS_DIR) @@ -65,6 +82,8 @@ pub fn add_kerberos_pod_config( ); } } + + Ok(()) } constant!(KRB5_CONFIG: EnvVarName = "KRB5_CONFIG"); diff --git a/rust/operator-binary/src/controller/build/resource/config_map.rs b/rust/operator-binary/src/controller/build/resource/config_map.rs index 1393eba1b..a1be2a907 100644 --- a/rust/operator-binary/src/controller/build/resource/config_map.rs +++ b/rust/operator-binary/src/controller/build/resource/config_map.rs @@ -30,6 +30,12 @@ use crate::{ #[derive(Snafu, Debug)] pub enum Error { + #[snafu(display("failed to build ConfigMap for role group {role_group}"))] + BuildRoleGroupConfig { + source: stackable_operator::builder::configmap::Error, + role_group: RoleGroupName, + }, + #[snafu(display( "failed to serialize [{}] for role group {role_group}", ConfigFileName::Security @@ -188,9 +194,11 @@ pub fn build_rolegroup_config_map( cm_builder.add_data(VECTOR_CONFIG_FILE, vector_config); } - Ok(cm_builder + cm_builder .build() - .expect("The ConfigMap metadata is set in this function.")) + .with_context(|_| BuildRoleGroupConfigSnafu { + role_group: role_group_name.clone(), + }) } // Generate JAAS configuration file for Kerberos authentication diff --git a/rust/operator-binary/src/controller/build/resource/discovery.rs b/rust/operator-binary/src/controller/build/resource/discovery.rs index 58528695a..2dd1c7fda 100644 --- a/rust/operator-binary/src/controller/build/resource/discovery.rs +++ b/rust/operator-binary/src/controller/build/resource/discovery.rs @@ -17,6 +17,11 @@ use crate::{ pub enum Error { #[snafu(display("nodePort was out of range"))] InvalidNodePort { source: TryFromIntError }, + + #[snafu(display("failed to build ConfigMap"))] + BuildConfigMap { + source: stackable_operator::builder::configmap::Error, + }, } /// Build a discovery [`ConfigMap`] containing information about how to connect to a certain @@ -71,7 +76,7 @@ pub fn build_discovery_configmap(validated_cluster: &ValidatedCluster) -> Result ) .add_data("KAFKA", bootstrap_servers) .build() - .expect("The ConfigMap metadata is set in this function."); + .context(BuildConfigMapSnafu)?; Ok(discovery_cm) } diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs index bf4aa2b3e..b82a9c908 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -129,6 +129,26 @@ const POD_MANAGEMENT_POLICY_PARALLEL: &str = "Parallel"; #[derive(Snafu, Debug)] pub enum Error { + #[snafu(display("failed to add kerberos config"))] + AddKerberosConfig { + source: crate::controller::build::kerberos::Error, + }, + + #[snafu(display("failed to add listener volume"))] + AddListenerVolume { + source: stackable_operator::builder::pod::Error, + }, + + #[snafu(display("failed to add Secret Volumes and VolumeMounts"))] + AddVolumesAndVolumeMounts { + source: crate::controller::build::security::Error, + }, + + #[snafu(display("failed to add needed volume"))] + AddVolume { + source: stackable_operator::builder::pod::Error, + }, + #[snafu(display("failed to build pod descriptors"))] BuildPodDescriptors { source: crate::controller::PodDescriptorsError, @@ -188,7 +208,8 @@ pub fn build_broker_rolegroup_statefulset( &mut cb_kcat_prober, &mut cb_kafka, &requested_secret_lifetime, - ); + ) + .context(AddVolumesAndVolumeMountsSnafu)?; let mut pvcs = merged_config.resources().storage.build_pvcs(); @@ -209,7 +230,8 @@ pub fn build_broker_rolegroup_statefulset( &mut cb_kcat_prober, &mut cb_kafka, &mut pod_builder, - ); + ) + .context(AddKerberosConfigSnafu)?; } // Operator-set env vars first; the user's `envOverrides` are merged on top last and win. @@ -307,7 +329,7 @@ pub fn build_broker_rolegroup_statefulset( &mut pod_builder, &validated_rg.config.logging, &resource_names, - ); + )?; let metadata = ObjectMetaBuilder::new() .with_labels(recommended_labels.clone()) @@ -320,7 +342,7 @@ pub fn build_broker_rolegroup_statefulset( listener_class.as_ref(), &recommended_labels, ) - .expect("The annotation keys are static, annotation values cannot be invalid, and the volume name is statically defined."); + .context(AddListenerVolumeSnafu)?; } if let Some(broker_id_config_map_name) = &validated_cluster @@ -333,7 +355,7 @@ pub fn build_broker_rolegroup_statefulset( .with_config_map(broker_id_config_map_name) .build(), ) - .expect("The volume names are statically defined and there should be no duplicates."); + .context(AddVolumeSnafu)?; cb_kafka .add_volume_mount(&*BROKER_ID_POD_MAP_DIR_NAME, BROKER_ID_POD_MAP_DIR) .expect("The mount paths are statically defined and there should be no duplicates."); @@ -353,7 +375,7 @@ pub fn build_broker_rolegroup_statefulset( .cluster_resource_names() .service_account_name() .as_ref(), - ); + )?; add_vector_container( &mut pod_builder, @@ -501,7 +523,7 @@ pub fn build_controller_rolegroup_statefulset( &mut pod_builder, &validated_rg.config.logging, &resource_names, - ); + )?; let metadata = ObjectMetaBuilder::new() .with_labels(recommended_labels.clone()) @@ -517,7 +539,8 @@ pub fn build_controller_rolegroup_statefulset( &mut pod_builder, &mut cb_kafka, &requested_secret_lifetime, - ); + ) + .context(AddVolumesAndVolumeMountsSnafu)?; let kafka_container = cb_kafka.build(); @@ -534,7 +557,7 @@ pub fn build_controller_rolegroup_statefulset( .cluster_resource_names() .service_account_name() .as_ref(), - ); + )?; add_vector_container( &mut pod_builder, @@ -660,17 +683,11 @@ fn common_kafka_env( /// Adds the `log-config` volume, sourced either from the user-supplied custom log config /// `ConfigMap` or the rolegroup `ConfigMap` (which carries the operator-generated config). /// Branches on the *validated* Kafka-container logging choice. -/// -/// # Panics -/// -/// Panics if the volumes or volume mounts cannot be added to the builders. Only call this -/// on builders whose volume names and mount paths are still distinct from the ones added -/// here. fn add_log_config_volume( pod_builder: &mut PodBuilder, logging: &ValidatedLogging, resource_names: &ResourceNames, -) { +) -> Result<(), Error> { let config_map = match &logging.kafka_container { ValidatedContainerLogConfigChoice::Custom(config_map_name) => config_map_name.to_string(), ValidatedContainerLogConfigChoice::Automatic(_) => { @@ -683,22 +700,17 @@ fn add_log_config_volume( .with_config_map(config_map) .build(), ) - .expect("The volume names are statically defined and there should be no duplicates."); + .context(AddVolumeSnafu)?; + Ok(()) } /// Adds the `config` volume, the `log` emptyDir, the service account and the pod security /// context that the broker and controller pods share. -/// -/// # Panics -/// -/// Panics if the volumes or volume mounts cannot be added to the builders. Only call this -/// on builders whose volume names and mount paths are still distinct from the ones added -/// here. fn add_common_pod_config( pod_builder: &mut PodBuilder, resource_names: &ResourceNames, service_account_name: &str, -) { +) -> Result<(), Error> { pod_builder .add_volume(Volume { name: STACKABLE_CONFIG_DIR_NAME.to_string(), @@ -708,20 +720,21 @@ fn add_common_pod_config( }), ..Volume::default() }) - .expect("The volume names are statically defined and there should be no duplicates.") + .context(AddVolumeSnafu)? .add_empty_dir_volume( &*STACKABLE_LOG_DIR_NAME, Some(product_logging::framework::calculate_log_volume_size_limit( &[MAX_KAFKA_LOG_FILES_SIZE], )), ) - .expect("The volume names are statically defined and there should be no duplicates.") + .context(AddVolumeSnafu)? .service_account_name(service_account_name) .security_context( PodSecurityContextBuilder::with_stackable_defaults() .fs_group(1000) .build(), ); + Ok(()) } /// Adds the Vector log-aggregation sidecar container, when the Vector agent is enabled. diff --git a/rust/operator-binary/src/controller/build/security.rs b/rust/operator-binary/src/controller/build/security.rs index 6570ba7f6..865be8a76 100644 --- a/rust/operator-binary/src/controller/build/security.rs +++ b/rust/operator-binary/src/controller/build/security.rs @@ -5,11 +5,15 @@ //! any validation themselves. use std::{collections::BTreeMap, str::FromStr}; +use snafu::{ResultExt, Snafu}; use stackable_operator::{ - builder::pod::{ - PodBuilder, - container::ContainerBuilder, - volume::{SecretFormat, SecretOperatorVolumeSourceBuilder, VolumeBuilder}, + builder::{ + self, + pod::{ + PodBuilder, + container::ContainerBuilder, + volume::{SecretFormat, SecretOperatorVolumeSourceBuilder, VolumeBuilder}, + }, }, commons::secret_class::SecretClassVolumeProvisionParts, constant, @@ -56,6 +60,22 @@ const STACKABLE_TLS_KCAT_DIR: &str = "/stackable/tls-kcat"; constant!(STACKABLE_TLS_KCAT_VOLUME_NAME: VolumeName = "tls-kcat"); const TRUSTSTORE_P12_FILE_NAME: &str = "truststore.p12"; +#[derive(Snafu, Debug)] +pub enum Error { + #[snafu(display("failed to build the secret operator Volume"))] + SecretVolumeBuild { + source: stackable_operator::builder::pod::volume::SecretOperatorVolumeSourceBuilderError, + }, + + #[snafu(display("failed to add needed volume"))] + AddVolume { source: builder::pod::Error }, + + #[snafu(display("failed to build OPA TLS certificate volume"))] + OpaTlsCertSecretClassVolumeBuild { + source: stackable_operator::builder::pod::volume::SecretOperatorVolumeSourceBuilderError, + }, +} + pub fn copy_opa_tls_cert_command(security: &ValidatedKafkaSecurity) -> String { match security.opa_secret_class().is_some() { true => format!( @@ -203,16 +223,15 @@ pub fn client_properties(security: &ValidatedKafkaSecurity) -> Vec<(String, Opti /// /// # Panics /// -/// Panics if the volumes or volume mounts cannot be added to the builders. Only call this -/// on builders whose volume names and mount paths are still distinct from the ones added -/// here. +/// Panics if the volume mounts cannot be added to the container builders. Only call this on +/// container builders whose mount paths are still distinct from the ones added here. pub fn add_broker_volume_and_volume_mounts( security: &ValidatedKafkaSecurity, pod_builder: &mut PodBuilder, cb_kcat_prober: &mut ContainerBuilder, cb_kafka: &mut ContainerBuilder, requested_secret_lifetime: &Duration, -) { +) -> Result<(), Error> { // add tls (server or client authentication volumes) if required if let Some(tls_server_secret_class) = tls_secret_class(security) { // We have to mount tls pem files for kcat (the mount can be used directly) @@ -221,8 +240,8 @@ pub fn add_broker_volume_and_volume_mounts( &STACKABLE_TLS_KCAT_VOLUME_NAME, tls_server_secret_class, requested_secret_lifetime, - )) - .expect("The volume names are statically defined and there should be no duplicates."); + )?) + .context(AddVolumeSnafu)?; cb_kcat_prober .add_volume_mount(&*STACKABLE_TLS_KCAT_VOLUME_NAME, STACKABLE_TLS_KCAT_DIR) .expect("The mount paths are statically defined and there should be no duplicates."); @@ -232,8 +251,8 @@ pub fn add_broker_volume_and_volume_mounts( &STACKABLE_TLS_KAFKA_SERVER_VOLUME_NAME, tls_server_secret_class, requested_secret_lifetime, - )) - .expect("The volume names are statically defined and there should be no duplicates."); + )?) + .context(AddVolumeSnafu)?; cb_kafka .add_volume_mount( &*STACKABLE_TLS_KAFKA_SERVER_VOLUME_NAME, @@ -247,8 +266,8 @@ pub fn add_broker_volume_and_volume_mounts( &STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME, security.tls_internal_secret_class(), requested_secret_lifetime, - )) - .expect("The volume names are statically defined and there should be no duplicates."); + )?) + .context(AddVolumeSnafu)?; cb_kafka .add_volume_mount( &*STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME, @@ -271,12 +290,14 @@ pub fn add_broker_volume_and_volume_mounts( SecretClassVolumeProvisionParts::Public, ) .build() - .expect("The annotation keys are static and annotation values cannot be invalid."), + .context(OpaTlsCertSecretClassVolumeBuildSnafu)?, ) .build(), ) - .expect("The volume names are statically defined and there should be no duplicates."); + .context(AddVolumeSnafu)?; } + + Ok(()) } /// Adds required volumes and volume mounts to the controller pod and container builders @@ -284,15 +305,14 @@ pub fn add_broker_volume_and_volume_mounts( /// /// # Panics /// -/// Panics if the volumes or volume mounts cannot be added to the builders. Only call this -/// on builders whose volume names and mount paths are still distinct from the ones added -/// here. +/// Panics if the volume mounts cannot be added to the container builder. Only call this on a +/// container builder whose mount paths are still distinct from the ones added here. pub fn add_controller_volume_and_volume_mounts( security: &ValidatedKafkaSecurity, pod_builder: &mut PodBuilder, cb_kafka: &mut ContainerBuilder, requested_secret_lifetime: &Duration, -) { +) -> Result<(), Error> { pod_builder .add_volume( VolumeBuilder::new(&*STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME) @@ -308,19 +328,19 @@ pub fn add_controller_volume_and_volume_mounts( .with_auto_tls_cert_lifetime(*requested_secret_lifetime) .with_auto_tls_cert_domain_components_in_subject_dn(true) .build() - .expect( - "The annotation keys are static and annotation values cannot be invalid.", - ), + .context(SecretVolumeBuildSnafu)?, ) .build(), ) - .expect("The volume names are statically defined and there should be no duplicates."); + .context(AddVolumeSnafu)?; cb_kafka .add_volume_mount( &*STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME, STACKABLE_TLS_KAFKA_INTERNAL_DIR, ) .expect("The mount paths are statically defined and there should be no duplicates."); + + Ok(()) } /// Inserts the `listener..ssl.{keystore,truststore}.{location,password,type}` @@ -552,8 +572,8 @@ fn create_kcat_tls_volume( volume_name: &VolumeName, secret_class_name: &str, requested_secret_lifetime: &Duration, -) -> Volume { - VolumeBuilder::new(volume_name) +) -> Result { + Ok(VolumeBuilder::new(volume_name) .ephemeral( SecretOperatorVolumeSourceBuilder::new( secret_class_name, @@ -566,9 +586,9 @@ fn create_kcat_tls_volume( .with_auto_tls_cert_lifetime(*requested_secret_lifetime) .with_auto_tls_cert_domain_components_in_subject_dn(true) .build() - .expect("The annotation keys are static and annotation values cannot be invalid."), + .context(SecretVolumeBuildSnafu)?, ) - .build() + .build()) } /// Creates ephemeral volumes to mount the `SecretClass` into the Pods as keystores @@ -576,8 +596,8 @@ fn create_tls_keystore_volume( volume_name: &VolumeName, secret_class_name: &str, requested_secret_lifetime: &Duration, -) -> Volume { - VolumeBuilder::new(volume_name) +) -> Result { + Ok(VolumeBuilder::new(volume_name) .ephemeral( SecretOperatorVolumeSourceBuilder::new( secret_class_name, @@ -591,9 +611,9 @@ fn create_tls_keystore_volume( .with_auto_tls_cert_lifetime(*requested_secret_lifetime) .with_auto_tls_cert_domain_components_in_subject_dn(true) .build() - .expect("The annotation keys are static and annotation values cannot be invalid."), + .context(SecretVolumeBuildSnafu)?, ) - .build() + .build()) } fn kcat_client_auth_ssl(cert_directory: &str) -> Vec { From 3aea773f7a7f34d1741c9db58baba33c798965ff Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Wed, 9 Sep 2026 16:31:32 +0200 Subject: [PATCH 06/10] add comment/test for bootstrap_listener_name --- .../src/controller/build/resource/listener.rs | 33 +++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/rust/operator-binary/src/controller/build/resource/listener.rs b/rust/operator-binary/src/controller/build/resource/listener.rs index d3cc299c5..b495200ce 100644 --- a/rust/operator-binary/src/controller/build/resource/listener.rs +++ b/rust/operator-binary/src/controller/build/resource/listener.rs @@ -24,6 +24,8 @@ use crate::{ /// A free function (rather than only a [`ValidatedCluster`] method) so the dereference step can /// compute the name from the raw cluster identity when fetching the stored `Listener`s that the /// discovery `ConfigMap` is built from. +/// +/// The returned ListenerName is a lowercase RFC 1035 label name (checked by a unit test). pub fn bootstrap_listener_name( cluster_name: &ClusterName, role: &KafkaRole, @@ -110,3 +112,34 @@ fn bootstrap_listener_ports( } }] } + +#[cfg(test)] +mod tests { + use stackable_operator::validation::RFC_1123_LABEL_MAX_LENGTH; + use strum::IntoEnumIterator; + + use super::*; + + #[test] + fn bootstrap_listener_name_is_rfc_1035_label_name() { + // Every ClusterName is a valid RFC 1035 label name, so we use just some string with maximum + // length. The role group name is user-provided, so use the maximum length of an RFC 1123 + // label there as well; operator-rs then hash-truncates the qualified role group name. + let _ = ClusterName::IS_RFC_1035_LABEL_NAME; + let cluster_name = ClusterName::from_str(&"a".repeat(ClusterName::MAX_LENGTH)) + .expect("is a valid ClusterName"); + let role_group_name = RoleGroupName::from_str(&"g".repeat(RFC_1123_LABEL_MAX_LENGTH)) + .expect("is a valid RoleGroupName"); + + for role in KafkaRole::iter() { + let bootstrap_listener_name = + bootstrap_listener_name(&cluster_name, &role, &role_group_name); + assert!( + stackable_operator::validation::is_lowercase_rfc_1035_label( + bootstrap_listener_name.as_ref() + ) + .is_ok() + ); + } + } +} From 53f294c28dd236ca613d6b465d5eb3afe3eb3b23 Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Thu, 10 Sep 2026 10:17:53 +0200 Subject: [PATCH 07/10] add a check for maximum length for an RFC 1035 label name --- .../src/controller/build/resource/listener.rs | 29 ++++++++++++------- 1 file changed, 18 insertions(+), 11 deletions(-) diff --git a/rust/operator-binary/src/controller/build/resource/listener.rs b/rust/operator-binary/src/controller/build/resource/listener.rs index b495200ce..b56539b07 100644 --- a/rust/operator-binary/src/controller/build/resource/listener.rs +++ b/rust/operator-binary/src/controller/build/resource/listener.rs @@ -8,6 +8,7 @@ use stackable_operator::{ role_group_utils::{QualifiedRoleGroupName, ResourceNames}, types::{kubernetes::ListenerName, operator::ClusterName}, }, + validation::RFC_1035_LABEL_MAX_LENGTH, }; use crate::{ @@ -25,7 +26,9 @@ use crate::{ /// compute the name from the raw cluster identity when fetching the stored `Listener`s that the /// discovery `ConfigMap` is built from. /// -/// The returned ListenerName is a lowercase RFC 1035 label name (checked by a unit test). +/// The returned ListenerName is a lowercase RFC 1035 label name. Its length is ensured at compile +/// time; the character class follows from [`QualifiedRoleGroupName`] being an RFC 1035 label name +/// and is additionally checked by a unit test. pub fn bootstrap_listener_name( cluster_name: &ClusterName, role: &KafkaRole, @@ -33,19 +36,22 @@ pub fn bootstrap_listener_name( ) -> ListenerName { const BOOTSTRAP_SUFFIX: &str = "-bootstrap"; - // Compile-time checks that `-bootstrap` is a valid ListenerName, so - // the `expect` below cannot fire. + // Compile-time checks that `-bootstrap` is an RFC 1035 label name + // (and therefore also a valid ListenerName), so the `expect` below cannot fire. // - // Length: the qualified role group name plus the suffix stays within the ListenerName limit. + // Length: the qualified role group name plus the suffix stays within the RFC 1035 label limit. const _: () = assert!( - QualifiedRoleGroupName::MAX_LENGTH + BOOTSTRAP_SUFFIX.len() <= ListenerName::MAX_LENGTH, - "The string `-bootstrap` must not exceed the limit of Listener \ - names." + QualifiedRoleGroupName::MAX_LENGTH + BOOTSTRAP_SUFFIX.len() <= RFC_1035_LABEL_MAX_LENGTH, + "The string `-bootstrap` must not exceed the limit of an \ + RFC 1035 label name." ); - // Characters: a ListenerName is an RFC 1123 DNS subdomain. The qualified role group name is an - // RFC 1123 label name (which is a subdomain of a single label); appending `-bootstrap` keeps it - // one, as the name still starts and ends with an alphanumeric character and adds no invalid ones. - let _ = QualifiedRoleGroupName::IS_RFC_1123_SUBDOMAIN_NAME; + // Characters: the qualified role group name is an RFC 1035 label name, i.e. it starts with a + // letter and consists of lowercase alphanumeric characters and dashes. Appending `-bootstrap` + // adds only such characters and ends with a letter, so the result is still an RFC 1035 label + // name. Every RFC 1035 label name is also an RFC 1123 DNS subdomain name, which is what a + // ListenerName requires. + let _ = QualifiedRoleGroupName::IS_RFC_1035_LABEL_NAME; + let _ = ListenerName::IS_RFC_1123_SUBDOMAIN_NAME; let resource_names = ResourceNames { cluster_name: cluster_name.clone(), @@ -122,6 +128,7 @@ mod tests { #[test] fn bootstrap_listener_name_is_rfc_1035_label_name() { + // The length is already ensured at compile time; this test covers the character class. // Every ClusterName is a valid RFC 1035 label name, so we use just some string with maximum // length. The role group name is user-provided, so use the maximum length of an RFC 1123 // label there as well; operator-rs then hash-truncates the qualified role group name. From 21f5321d730de6d80bc9171421ad757cbf543590 Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Thu, 10 Sep 2026 10:29:10 +0200 Subject: [PATCH 08/10] add deref impls and tests for both containers --- .../controller/build/resource/statefulset.rs | 17 +++---- rust/operator-binary/src/crd/role/broker.rs | 37 +++++++++++++++- .../src/crd/role/controller.rs | 44 +++++++++++++++++++ 3 files changed, 84 insertions(+), 14 deletions(-) diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs index b82a9c908..abe167f04 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -192,8 +192,8 @@ pub fn build_broker_rolegroup_statefulset( role_group_name, ); - let mut cb_kcat_prober = new_container_builder(&container_name(BrokerContainer::KcatProber)); - let mut cb_kafka = new_container_builder(&container_name(BrokerContainer::Kafka)); + let mut cb_kcat_prober = new_container_builder(&BrokerContainer::KcatProber); + let mut cb_kafka = new_container_builder(&BrokerContainer::Kafka); let mut pod_builder = PodBuilder::new(); @@ -379,7 +379,7 @@ pub fn build_broker_rolegroup_statefulset( add_vector_container( &mut pod_builder, - &container_name(BrokerContainer::Vector), + &BrokerContainer::Vector, &validated_rg.config.logging, resolved_product_image, &resource_names, @@ -440,7 +440,7 @@ pub fn build_controller_rolegroup_statefulset( let recommended_labels = recommended_labels_for_role_group_resources(validated_cluster, kafka_role, role_group_name); - let mut cb_kafka = new_container_builder(&container_name(ControllerContainer::Kafka)); + let mut cb_kafka = new_container_builder(&ControllerContainer::Kafka); let mut pod_builder = PodBuilder::new(); @@ -561,7 +561,7 @@ pub fn build_controller_rolegroup_statefulset( add_vector_container( &mut pod_builder, - &container_name(ControllerContainer::Vector), + &ControllerContainer::Vector, &validated_rg.config.logging, resolved_product_image, &resource_names, @@ -744,13 +744,6 @@ fn add_common_pod_config( /// [`ValidatedLogging`]. The container mounts the /// static `vector.yaml` from the `config` volume and is driven by the env vars the /// [`vector_container`] sets. -/// The [`ContainerName`] for a role container, derived from its `Display` name so the -/// Vector sidecar's container name always matches that container's logging-config key. -fn container_name(container: impl std::fmt::Display) -> ContainerName { - ContainerName::from_str(&container.to_string()) - .expect("a container enum variant is always a valid ContainerName") -} - fn add_vector_container( pod_builder: &mut PodBuilder, vector_container_name: &ContainerName, diff --git a/rust/operator-binary/src/crd/role/broker.rs b/rust/operator-binary/src/crd/role/broker.rs index 54e5dfb91..1639b7541 100644 --- a/rust/operator-binary/src/crd/role/broker.rs +++ b/rust/operator-binary/src/crd/role/broker.rs @@ -1,4 +1,4 @@ -use std::str::FromStr; +use std::{ops::Deref, str::FromStr}; use serde::{Deserialize, Serialize}; use stackable_operator::{ @@ -11,7 +11,7 @@ use stackable_operator::{ k8s_openapi::apimachinery::pkg::api::resource::Quantity, product_logging::{self, spec::Logging}, schemars::{self, JsonSchema}, - v2::types::kubernetes::ListenerClassName, + v2::types::kubernetes::{ContainerName, ListenerClassName}, }; use strum::{Display, EnumIter}; @@ -41,6 +41,24 @@ pub enum BrokerContainer { Kafka, } +// Typed container names. They must match the strum `Display` (kebab-case) of the variants above, +// which is pinned by a unit test. +constant!(VECTOR_CONTAINER_NAME: ContainerName = "vector"); +constant!(KCAT_PROBER_CONTAINER_NAME: ContainerName = "kcat-prober"); +constant!(KAFKA_CONTAINER_NAME: ContainerName = "kafka"); + +impl Deref for BrokerContainer { + type Target = ContainerName; + + fn deref(&self) -> &Self::Target { + match self { + BrokerContainer::Vector => &VECTOR_CONTAINER_NAME, + BrokerContainer::KcatProber => &KCAT_PROBER_CONTAINER_NAME, + BrokerContainer::Kafka => &KAFKA_CONTAINER_NAME, + } + } +} + #[derive(Clone, Debug, PartialEq, Fragment, JsonSchema)] #[fragment_attrs( derive( @@ -102,11 +120,26 @@ impl BrokerConfig { #[cfg(test)] mod tests { + use strum::IntoEnumIterator; + use super::*; #[test] fn test_constants() { // Test that dereferencing the constants does not panic. let _ = *DEFAULT_LISTENER_CLASS; + let _ = *VECTOR_CONTAINER_NAME; + let _ = *KCAT_PROBER_CONTAINER_NAME; + let _ = *KAFKA_CONTAINER_NAME; + } + + /// The typed container names behind `BrokerContainer`'s `Deref` must agree with its strum + /// `Display`, which the logging configuration still uses as the per-container key. + #[test] + fn container_names_match_display() { + for container in BrokerContainer::iter() { + let container_name: &ContainerName = &container; + assert_eq!(container_name.to_string(), container.to_string()); + } } } diff --git a/rust/operator-binary/src/crd/role/controller.rs b/rust/operator-binary/src/crd/role/controller.rs index ec025eabc..ed1caa30e 100644 --- a/rust/operator-binary/src/crd/role/controller.rs +++ b/rust/operator-binary/src/crd/role/controller.rs @@ -1,3 +1,5 @@ +use std::{ops::Deref, str::FromStr}; + use serde::{Deserialize, Serialize}; use stackable_operator::{ commons::resources::{ @@ -5,9 +7,11 @@ use stackable_operator::{ PvcConfigFragment, Resources, ResourcesFragment, }, config::{fragment::Fragment, merge::Merge}, + constant, k8s_openapi::apimachinery::pkg::api::resource::Quantity, product_logging::{self, spec::Logging}, schemars::{self, JsonSchema}, + v2::types::kubernetes::ContainerName, }; use strum::{Display, EnumIter}; @@ -33,6 +37,22 @@ pub enum ControllerContainer { Kafka, } +// Typed container names. They must match the strum `Display` (kebab-case) of the variants above, +// which is pinned by a unit test. +constant!(VECTOR_CONTAINER_NAME: ContainerName = "vector"); +constant!(KAFKA_CONTAINER_NAME: ContainerName = "kafka"); + +impl Deref for ControllerContainer { + type Target = ContainerName; + + fn deref(&self) -> &Self::Target { + match self { + ControllerContainer::Vector => &VECTOR_CONTAINER_NAME, + ControllerContainer::Kafka => &KAFKA_CONTAINER_NAME, + } + } +} + #[derive(Clone, Debug, Default, PartialEq, Fragment, JsonSchema)] #[fragment_attrs( derive( @@ -83,3 +103,27 @@ impl ControllerConfig { } } } + +#[cfg(test)] +mod tests { + use strum::IntoEnumIterator; + + use super::*; + + #[test] + fn test_constants() { + // Test that dereferencing the constants does not panic. + let _ = *VECTOR_CONTAINER_NAME; + let _ = *KAFKA_CONTAINER_NAME; + } + + /// The typed container names behind `ControllerContainer`'s `Deref` must agree with its strum + /// `Display`, which the logging configuration still uses as the per-container key. + #[test] + fn container_names_match_display() { + for container in ControllerContainer::iter() { + let container_name: &ContainerName = &container; + assert_eq!(container_name.to_string(), container.to_string()); + } + } +} From 8a80e05d272f011426e3259340c14057535601de Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Thu, 10 Sep 2026 11:53:05 +0200 Subject: [PATCH 09/10] tighten up compile time check and associated comments --- .../src/controller/build/resource/listener.rs | 30 +++++++++++-------- 1 file changed, 18 insertions(+), 12 deletions(-) diff --git a/rust/operator-binary/src/controller/build/resource/listener.rs b/rust/operator-binary/src/controller/build/resource/listener.rs index b56539b07..a7a9be6fc 100644 --- a/rust/operator-binary/src/controller/build/resource/listener.rs +++ b/rust/operator-binary/src/controller/build/resource/listener.rs @@ -26,9 +26,10 @@ use crate::{ /// compute the name from the raw cluster identity when fetching the stored `Listener`s that the /// discovery `ConfigMap` is built from. /// -/// The returned ListenerName is a lowercase RFC 1035 label name. Its length is ensured at compile -/// time; the character class follows from [`QualifiedRoleGroupName`] being an RFC 1035 label name -/// and is additionally checked by a unit test. +/// The returned name is both a valid [`ListenerName`] and a lowercase RFC 1035 label name. The +/// length is ensured at compile time for both; the character class follows from +/// [`QualifiedRoleGroupName`] being an RFC 1035 label name and is additionally checked by a unit +/// test. pub fn bootstrap_listener_name( cluster_name: &ClusterName, role: &KafkaRole, @@ -36,20 +37,25 @@ pub fn bootstrap_listener_name( ) -> ListenerName { const BOOTSTRAP_SUFFIX: &str = "-bootstrap"; - // Compile-time checks that `-bootstrap` is an RFC 1035 label name - // (and therefore also a valid ListenerName), so the `expect` below cannot fire. + // Compile-time checks that `-bootstrap` is both a valid ListenerName + // and an RFC 1035 label name, so the `expect` below cannot fire. // + // Length: the qualified role group name plus the suffix stays within the ListenerName limit. + const _: () = assert!( + QualifiedRoleGroupName::MAX_LENGTH + BOOTSTRAP_SUFFIX.len() <= ListenerName::MAX_LENGTH, + "The string `-bootstrap` must not exceed the limit of Listener \ + names." + ); // Length: the qualified role group name plus the suffix stays within the RFC 1035 label limit. const _: () = assert!( QualifiedRoleGroupName::MAX_LENGTH + BOOTSTRAP_SUFFIX.len() <= RFC_1035_LABEL_MAX_LENGTH, "The string `-bootstrap` must not exceed the limit of an \ RFC 1035 label name." ); - // Characters: the qualified role group name is an RFC 1035 label name, i.e. it starts with a - // letter and consists of lowercase alphanumeric characters and dashes. Appending `-bootstrap` - // adds only such characters and ends with a letter, so the result is still an RFC 1035 label - // name. Every RFC 1035 label name is also an RFC 1123 DNS subdomain name, which is what a - // ListenerName requires. + // Characters: the qualified role group name is an RFC 1123 DNS subdomain name (which a + // ListenerName requires) and an RFC 1035 label name. Appending `-bootstrap` adds only lowercase + // letters and a dash and ends with a letter, so the result is still both. + let _ = QualifiedRoleGroupName::IS_RFC_1123_SUBDOMAIN_NAME; let _ = QualifiedRoleGroupName::IS_RFC_1035_LABEL_NAME; let _ = ListenerName::IS_RFC_1123_SUBDOMAIN_NAME; @@ -66,8 +72,8 @@ pub fn bootstrap_listener_name( .expect("is a valid Listener name") } -/// Kafka clients will use the load-balanced bootstrap listener to get a list of broker addresses and will use those to -/// transmit data to the correct broker. +/// Kafka clients will use the load-balanced bootstrap listener to get a list of broker addresses +/// and will use those to transmit data to the correct broker. // TODO (@NickLarsenNZ): Move shared functionality to stackable-operator pub fn build_broker_rolegroup_bootstrap_listener( validated_cluster: &ValidatedCluster, From ef5e9b4e4d64d1a88f58b0c8dae230ce0cc50036 Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Thu, 10 Sep 2026 13:59:10 +0200 Subject: [PATCH 10/10] replace deref with name() accessor --- .../controller/build/resource/statefulset.rs | 10 +++++----- rust/operator-binary/src/crd/role/broker.rs | 16 +++++++--------- rust/operator-binary/src/crd/role/controller.rs | 17 ++++++++--------- 3 files changed, 20 insertions(+), 23 deletions(-) diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs index abe167f04..d166d0352 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -192,8 +192,8 @@ pub fn build_broker_rolegroup_statefulset( role_group_name, ); - let mut cb_kcat_prober = new_container_builder(&BrokerContainer::KcatProber); - let mut cb_kafka = new_container_builder(&BrokerContainer::Kafka); + let mut cb_kcat_prober = new_container_builder(BrokerContainer::KcatProber.name()); + let mut cb_kafka = new_container_builder(BrokerContainer::Kafka.name()); let mut pod_builder = PodBuilder::new(); @@ -379,7 +379,7 @@ pub fn build_broker_rolegroup_statefulset( add_vector_container( &mut pod_builder, - &BrokerContainer::Vector, + BrokerContainer::Vector.name(), &validated_rg.config.logging, resolved_product_image, &resource_names, @@ -440,7 +440,7 @@ pub fn build_controller_rolegroup_statefulset( let recommended_labels = recommended_labels_for_role_group_resources(validated_cluster, kafka_role, role_group_name); - let mut cb_kafka = new_container_builder(&ControllerContainer::Kafka); + let mut cb_kafka = new_container_builder(ControllerContainer::Kafka.name()); let mut pod_builder = PodBuilder::new(); @@ -561,7 +561,7 @@ pub fn build_controller_rolegroup_statefulset( add_vector_container( &mut pod_builder, - &ControllerContainer::Vector, + ControllerContainer::Vector.name(), &validated_rg.config.logging, resolved_product_image, &resource_names, diff --git a/rust/operator-binary/src/crd/role/broker.rs b/rust/operator-binary/src/crd/role/broker.rs index 1639b7541..7ff41f9b7 100644 --- a/rust/operator-binary/src/crd/role/broker.rs +++ b/rust/operator-binary/src/crd/role/broker.rs @@ -1,4 +1,4 @@ -use std::{ops::Deref, str::FromStr}; +use std::str::FromStr; use serde::{Deserialize, Serialize}; use stackable_operator::{ @@ -47,10 +47,9 @@ constant!(VECTOR_CONTAINER_NAME: ContainerName = "vector"); constant!(KCAT_PROBER_CONTAINER_NAME: ContainerName = "kcat-prober"); constant!(KAFKA_CONTAINER_NAME: ContainerName = "kafka"); -impl Deref for BrokerContainer { - type Target = ContainerName; - - fn deref(&self) -> &Self::Target { +impl BrokerContainer { + /// The typed container name of this variant. + pub fn name(&self) -> &'static ContainerName { match self { BrokerContainer::Vector => &VECTOR_CONTAINER_NAME, BrokerContainer::KcatProber => &KCAT_PROBER_CONTAINER_NAME, @@ -133,13 +132,12 @@ mod tests { let _ = *KAFKA_CONTAINER_NAME; } - /// The typed container names behind `BrokerContainer`'s `Deref` must agree with its strum - /// `Display`, which the logging configuration still uses as the per-container key. + /// The typed container names returned by `name` must agree with the strum `Display` + /// of `BrokerContainer`, which the logging configuration still uses as the per-container key. #[test] fn container_names_match_display() { for container in BrokerContainer::iter() { - let container_name: &ContainerName = &container; - assert_eq!(container_name.to_string(), container.to_string()); + assert_eq!(container.name().to_string(), container.to_string()); } } } diff --git a/rust/operator-binary/src/crd/role/controller.rs b/rust/operator-binary/src/crd/role/controller.rs index ed1caa30e..39c98e1ef 100644 --- a/rust/operator-binary/src/crd/role/controller.rs +++ b/rust/operator-binary/src/crd/role/controller.rs @@ -1,4 +1,4 @@ -use std::{ops::Deref, str::FromStr}; +use std::str::FromStr; use serde::{Deserialize, Serialize}; use stackable_operator::{ @@ -42,10 +42,9 @@ pub enum ControllerContainer { constant!(VECTOR_CONTAINER_NAME: ContainerName = "vector"); constant!(KAFKA_CONTAINER_NAME: ContainerName = "kafka"); -impl Deref for ControllerContainer { - type Target = ContainerName; - - fn deref(&self) -> &Self::Target { +impl ControllerContainer { + /// The typed container name of this variant. + pub fn name(&self) -> &'static ContainerName { match self { ControllerContainer::Vector => &VECTOR_CONTAINER_NAME, ControllerContainer::Kafka => &KAFKA_CONTAINER_NAME, @@ -117,13 +116,13 @@ mod tests { let _ = *KAFKA_CONTAINER_NAME; } - /// The typed container names behind `ControllerContainer`'s `Deref` must agree with its strum - /// `Display`, which the logging configuration still uses as the per-container key. + /// The typed container names returned by `name` must agree with the strum `Display` + /// of `ControllerContainer`, which the logging configuration still uses as the per-container + /// key. #[test] fn container_names_match_display() { for container in ControllerContainer::iter() { - let container_name: &ContainerName = &container; - assert_eq!(container_name.to_string(), container.to_string()); + assert_eq!(container.name().to_string(), container.to_string()); } } }