diff --git a/CHANGELOG.md b/CHANGELOG.md index b4598838..207a0c37 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 dependent on static inputs ([#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 diff --git a/rust/operator-binary/src/controller/build/kerberos.rs b/rust/operator-binary/src/controller/build/kerberos.rs index 624525f7..7e2ebebe 100644 --- a/rust/operator-binary/src/controller/build/kerberos.rs +++ b/rust/operator-binary/src/controller/build/kerberos.rs @@ -15,7 +15,10 @@ use stackable_operator::{ }, commons::secret_class::SecretClassVolumeProvisionParts, constant, - v2::builder::pod::container::{EnvVarName, EnvVarSet}, + v2::{ + builder::pod::container::{EnvVarName, EnvVarSet}, + types::kubernetes::VolumeName, + }, }; use crate::{ @@ -26,6 +29,8 @@ use crate::{ }, }; +constant!(KERBEROS_VOLUME_NAME: VolumeName = "kerberos"); + #[derive(Snafu, Debug)] pub enum Error { #[snafu(display("failed to add Kerberos secret volume"))] @@ -35,13 +40,15 @@ pub enum Error { #[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, - }, } +/// 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 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, @@ -56,21 +63,23 @@ 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)?; pb.add_volume( - VolumeBuilder::new("kerberos") + VolumeBuilder::new(&*KERBEROS_VOLUME_NAME) .ephemeral(kerberos_secret_operator_volume) .build(), ) .context(AddVolumeSnafu)?; 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.", + ); } } @@ -108,5 +117,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/listener.rs b/rust/operator-binary/src/controller/build/resource/listener.rs index d3cc299c..a7a9be6f 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::{ @@ -24,6 +25,11 @@ 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 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, @@ -31,19 +37,27 @@ 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 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." + 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: 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. + // 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; let resource_names = ResourceNames { cluster_name: cluster_name.clone(), @@ -58,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, @@ -110,3 +124,35 @@ 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() { + // 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. + 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() + ); + } + } +} diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs index 063d6232..d166d035 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}, }, }; @@ -147,11 +144,6 @@ pub enum Error { 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, @@ -172,12 +164,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 +192,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(BrokerContainer::KcatProber.name()); + let mut cb_kafka = new_container_builder(BrokerContainer::Kafka.name()); let mut pod_builder = PodBuilder::new(); @@ -240,12 +217,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() { @@ -296,21 +271,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 +307,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 { @@ -363,7 +338,7 @@ 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, ) @@ -376,14 +351,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)?; 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 @@ -404,7 +379,7 @@ pub fn build_broker_rolegroup_statefulset( add_vector_container( &mut pod_builder, - &container_name(BrokerContainer::Vector), + BrokerContainer::Vector.name(), &validated_rg.config.logging, resolved_product_image, &resource_names, @@ -465,11 +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 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(ControllerContainer::Kafka.name()); let mut pod_builder = PodBuilder::new(); @@ -517,14 +488,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 { @@ -590,7 +561,7 @@ pub fn build_controller_rolegroup_statefulset( add_vector_container( &mut pod_builder, - &container_name(ControllerContainer::Vector), + ControllerContainer::Vector.name(), &validated_rg.config.logging, resolved_product_image, &resource_names, @@ -725,7 +696,7 @@ 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(), ) @@ -751,7 +722,7 @@ fn add_common_pod_config( }) .context(AddVolumeSnafu)? .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], )), @@ -773,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/controller/build/security.rs b/rust/operator-binary/src/controller/build/security.rs index e36191c1..865be8a7 100644 --- a/rust/operator-binary/src/controller/build/security.rs +++ b/rust/operator-binary/src/controller/build/security.rs @@ -3,7 +3,7 @@ //! //! 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::{ @@ -16,9 +16,11 @@ use stackable_operator::{ }, }, 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 +41,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,12 +52,12 @@ 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)] @@ -68,11 +70,6 @@ pub enum Error { #[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, @@ -223,6 +220,11 @@ 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 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, @@ -235,52 +237,52 @@ pub fn add_broker_volume_and_volume_mounts( // 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)?; 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)?; 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)?; 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, @@ -300,6 +302,11 @@ 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 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, @@ -308,7 +315,7 @@ pub fn add_controller_volume_and_volume_mounts( ) -> 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(), @@ -328,10 +335,10 @@ pub fn add_controller_volume_and_volume_mounts( .context(AddVolumeSnafu)?; 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."); Ok(()) } @@ -562,7 +569,7 @@ 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 { @@ -586,7 +593,7 @@ fn create_kcat_tls_volume( /// 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 { @@ -598,8 +605,8 @@ 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) @@ -763,6 +770,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 c2d11825..5005122f 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 c1eafa34..7ff41f9b 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,15 +7,19 @@ 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::ListenerClassName, + v2::types::kubernetes::{ContainerName, ListenerClassName}, }; 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, @@ -35,6 +41,23 @@ 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 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, + BrokerContainer::Kafka => &KAFKA_CONTAINER_NAME, + } + } +} + #[derive(Clone, Debug, PartialEq, Fragment, JsonSchema)] #[fragment_attrs( derive( @@ -70,16 +93,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 +116,28 @@ 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 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() { + assert_eq!(container.name().to_string(), container.to_string()); + } + } +} diff --git a/rust/operator-binary/src/crd/role/commons.rs b/rust/operator-binary/src/crd/role/commons.rs index 1ef6f3dc..32ac469f 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/role/controller.rs b/rust/operator-binary/src/crd/role/controller.rs index ec025eab..39c98e1e 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::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,21 @@ 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 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, + } + } +} + #[derive(Clone, Debug, Default, PartialEq, Fragment, JsonSchema)] #[fragment_attrs( derive( @@ -83,3 +102,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 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() { + assert_eq!(container.name().to_string(), container.to_string()); + } + } +} diff --git a/rust/operator-binary/src/crd/tls.rs b/rust/operator-binary/src/crd/tls.rs index 82650799..161f02aa 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; + } +}