From cd70a6b20d0997dbecde7dbf71d802ce166ff2f8 Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Fri, 28 Aug 2026 17:54:27 +0200 Subject: [PATCH 1/2] 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 2/2] 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