From b16b748723a59acc975519a2392db7b3ebd5898a Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Sat, 8 Aug 2026 09:51:15 +0200 Subject: [PATCH 1/6] add apply and update_status steps --- rust/operator-binary/src/controller.rs | 194 +++++------------- rust/operator-binary/src/controller/apply.rs | 130 ++++++++++++ .../src/controller/build/mod.rs | 66 +++++- .../controller/build/resource/discovery.rs | 166 ++++++++++++++- .../src/controller/build/resource/listener.rs | 32 ++- .../src/controller/dereference.rs | 77 ++++++- .../src/controller/update_status.rs | 55 +++++ .../src/controller/validate.rs | 1 + 8 files changed, 558 insertions(+), 163 deletions(-) create mode 100644 rust/operator-binary/src/controller/apply.rs create mode 100644 rust/operator-binary/src/controller/update_status.rs diff --git a/rust/operator-binary/src/controller.rs b/rust/operator-binary/src/controller.rs index 5b3cea97..259f44aa 100644 --- a/rust/operator-binary/src/controller.rs +++ b/rust/operator-binary/src/controller.rs @@ -8,6 +8,7 @@ use std::{ borrow::Cow, collections::{BTreeMap, HashMap}, + marker::PhantomData, str::FromStr, sync::Arc, }; @@ -34,13 +35,8 @@ use stackable_operator::{ kvp::Labels, logging::controller::ReconcilerError, shared::time::Duration, - status::condition::{ - compute_conditions, operations::ClusterOperationsConditionBuilder, - statefulset::StatefulSetConditionBuilder, - }, v2::{ HasName, HasUid, NameIsValidLabelValue, - cluster_resources::cluster_resources_new, kvp::label::{recommended_labels, role_group_selector}, role_group_utils::ResourceNames, role_utils, @@ -52,10 +48,12 @@ use stackable_operator::{ }; use strum::{EnumDiscriminants, IntoStaticStr}; +pub(crate) mod apply; pub(crate) mod build; pub(crate) mod dereference; pub(crate) mod node_id_hasher; pub(crate) mod security; +pub(crate) mod update_status; pub(crate) mod validate; /// The type-safe role-group name from stackable-operator. Re-exported so the rest @@ -63,9 +61,12 @@ pub(crate) mod validate; pub use stackable_operator::v2::types::operator::{RoleGroupName, RoleName}; use crate::{ - controller::{node_id_hasher::node_id_hash32_offset, security::ValidatedKafkaSecurity}, + controller::{ + apply::Applier, node_id_hasher::node_id_hash32_offset, security::ValidatedKafkaSecurity, + update_status::update_status, + }, crd::{ - APP_NAME, KafkaClusterStatus, KafkaPodDescriptor, MetadataManager, OPERATOR_NAME, + APP_NAME, KafkaPodDescriptor, MetadataManager, OPERATOR_NAME, authorization::KafkaAuthorizationConfig, role::{AnyConfig, AnyConfigOverrides, KafkaRole}, v1alpha1, @@ -91,12 +92,18 @@ pub enum PodDescriptorsError { }, } +/// Marker for prepared Kubernetes resources which are not applied yet. +pub struct Prepared; + +/// Marker for applied Kubernetes resources. +pub struct Applied; + /// Every Kubernetes resource produced by the [`build`] step. /// -/// The discovery `ConfigMap` is not part of this: it depends on the applied bootstrap -/// [`Listener`](listener)s' status and is therefore built in [`reconcile_kafka`] after they are -/// applied. -pub struct KubernetesResources { +/// This includes the discovery `ConfigMap` (in [`Self::config_maps`]): it is built from the +/// bootstrap [`Listener`](listener)s as fetched in the dereference step, and is absent while +/// they have no ingress addresses yet. +pub struct KubernetesResources { pub stateful_sets: Vec, pub services: Vec, pub listeners: Vec, @@ -104,6 +111,7 @@ pub struct KubernetesResources { pub pod_disruption_budgets: Vec, pub service_accounts: Vec, pub role_bindings: Vec, + pub status: PhantomData, } /// The validated cluster. Carries everything the build steps need, resolved once @@ -131,6 +139,11 @@ pub struct ValidatedCluster { /// Per-role configuration (e.g. the Pod disruption budget), keyed by role. pub role_configs: BTreeMap, pub role_group_configs: BTreeMap>, + /// The broker role groups' bootstrap `Listener`s as currently stored in the cluster (fetched + /// in the dereference step), from which the discovery `ConfigMap` is built. Missing or still + /// address-less around the first reconcile runs; the listener-operator populates the ingress + /// addresses and the `Listener` watch triggers a new run once it does. + pub bootstrap_listeners: Vec, } impl ValidatedCluster { @@ -144,6 +157,7 @@ impl ValidatedCluster { cluster_config: ValidatedClusterConfig, role_configs: BTreeMap, role_group_configs: BTreeMap>, + bootstrap_listeners: Vec, ) -> Self { // `app_version_label_value` is constructed to be a valid label value, so it is also a // valid `ProductVersion`. @@ -165,6 +179,7 @@ impl ValidatedCluster { cluster_config, role_configs, role_group_configs, + bootstrap_listeners, } } @@ -303,12 +318,7 @@ impl ValidatedCluster { role: &KafkaRole, role_group_name: &RoleGroupName, ) -> ListenerName { - ListenerName::from_str(&format!( - "{}-bootstrap", - self.role_group_resource_names(role, role_group_name) - .stateful_set_name() - )) - .expect("the bootstrap listener name is a valid Listener name") + build::resource::listener::bootstrap_listener_name(&self.name, role, role_group_name) } } @@ -449,6 +459,12 @@ pub struct Ctx { #[strum_discriminants(derive(IntoStaticStr))] #[allow(clippy::enum_variant_names)] pub enum Error { + #[snafu(display("failed to apply the Kubernetes resources"))] + ApplyResources { source: apply::Error }, + + #[snafu(display("failed to update the cluster status"))] + UpdateStatus { source: update_status::Error }, + #[snafu(display("failed to dereference resources"))] Dereference { source: dereference::Error }, @@ -458,31 +474,6 @@ pub enum Error { #[snafu(display("failed to build the Kubernetes resources"))] BuildResources { source: build::Error }, - #[snafu(display("failed to apply Kubernetes resource"))] - ApplyResource { - source: stackable_operator::cluster_resources::Error, - }, - - #[snafu(display("failed to build discovery ConfigMap"))] - BuildDiscoveryConfig { - source: build::resource::discovery::Error, - }, - - #[snafu(display("failed to apply discovery ConfigMap"))] - ApplyDiscoveryConfig { - source: stackable_operator::cluster_resources::Error, - }, - - #[snafu(display("failed to delete orphaned resources"))] - DeleteOrphans { - source: stackable_operator::cluster_resources::Error, - }, - - #[snafu(display("failed to update status"))] - ApplyStatus { - source: stackable_operator::client::Error, - }, - #[snafu(display("KafkaCluster object is invalid"))] InvalidKafkaCluster { source: error_boundary::InvalidObject, @@ -500,11 +491,8 @@ impl ReconcilerError for Error { Error::Dereference { .. } => None, Error::ValidateCluster { .. } => None, Error::BuildResources { .. } => None, - Error::ApplyResource { .. } => None, - Error::BuildDiscoveryConfig { .. } => None, - Error::ApplyDiscoveryConfig { .. } => None, - Error::DeleteOrphans { .. } => None, - Error::ApplyStatus { .. } => None, + Error::ApplyResources { .. } => None, + Error::UpdateStatus { .. } => None, Error::InvalidKafkaCluster { .. } => None, } } @@ -534,17 +522,6 @@ pub async fn reconcile_kafka( validate::validate(kafka, dereferenced_objects, &ctx.operator_environment) .context(ValidateClusterSnafu)?; - let mut cluster_resources = cluster_resources_new( - &product_name(), - &operator_name(), - &controller_name(), - &validated_cluster.name, - &validated_cluster.namespace, - &validated_cluster.uid, - ClusterResourceApplyStrategy::from(&kafka.spec.cluster_operation), - &kafka.spec.object_overrides, - ); - tracing::debug!( kerberos_enabled = validated_cluster.cluster_config.kafka_security.has_kerberos_enabled(), kerberos_secret_class = ?validated_cluster.cluster_config.kafka_security.kerberos_secret_class(), @@ -553,99 +530,25 @@ pub async fn reconcile_kafka( "The following security settings are used" ); - let mut ss_cond_builder = StatefulSetConditionBuilder::default(); - - // Build every Kubernetes resource up front (client-free). The discovery ConfigMap is not part - // of this, as it depends on the applied bootstrap Listeners' status (see below). + // build (no client required) let resources = build::build(&validated_cluster).context(BuildResourcesSnafu)?; - // Apply order: Services, then Listeners (collecting the applied bootstrap Listeners for the - // discovery ConfigMap), then ConfigMaps, then PodDisruptionBudgets, and finally the - // StatefulSets. The StatefulSets must be applied after all ConfigMaps and Secrets they mount to - // prevent unnecessary Pod restarts. - // See https://github.com/stackabletech/commons-operator/issues/111 for details. - for service_account in resources.service_accounts { - cluster_resources - .add(client, service_account) - .await - .context(ApplyResourceSnafu)?; - } - for role_binding in resources.role_bindings { - cluster_resources - .add(client, role_binding) - .await - .context(ApplyResourceSnafu)?; - } - for service in resources.services { - cluster_resources - .add(client, service) - .await - .context(ApplyResourceSnafu)?; - } - - let mut bootstrap_listeners = Vec::::new(); - for rg_listener in resources.listeners { - bootstrap_listeners.push( - cluster_resources - .add(client, rg_listener) - .await - .context(ApplyResourceSnafu)?, - ); - } - - for config_map in resources.config_maps { - cluster_resources - .add(client, config_map) - .await - .context(ApplyResourceSnafu)?; - } - - for pdb in resources.pod_disruption_budgets { - cluster_resources - .add(client, pdb) - .await - .context(ApplyResourceSnafu)?; - } - - for stateful_set in resources.stateful_sets { - ss_cond_builder.add( - cluster_resources - .add(client, stateful_set) - .await - .context(ApplyResourceSnafu)?, - ); - } - - // The discovery ConfigMap reports the bootstrap Listeners' ingress addresses, which are only - // populated on the applied Listener objects (by the Listener operator), so it is built here - // rather than in the client-free build() step. - let discovery_cm = build::resource::discovery::build_discovery_configmap( + // apply (client required) + let applier = Applier::new( + client, &validated_cluster, - &bootstrap_listeners, - ) - .context(BuildDiscoveryConfigSnafu)?; - - cluster_resources - .add(client, discovery_cm) - .await - .context(ApplyDiscoveryConfigSnafu)?; - - let cluster_operation_cond_builder = - ClusterOperationsConditionBuilder::new(&kafka.spec.cluster_operation); - - let status = KafkaClusterStatus { - conditions: compute_conditions(kafka, &[&ss_cond_builder, &cluster_operation_cond_builder]), - }; - - cluster_resources - .delete_orphaned_resources(client) + ClusterResourceApplyStrategy::from(&kafka.spec.cluster_operation), + &kafka.spec.object_overrides, + ); + let applied = applier + .apply(resources) .await - .context(DeleteOrphansSnafu)?; + .context(ApplyResourcesSnafu)?; - client - .apply_patch_status(OPERATOR_NAME, kafka, &status) + // update status (client required) + update_status(client, kafka, &applied) .await - .context(ApplyStatusSnafu)?; + .context(UpdateStatusSnafu)?; Ok(Action::await_change()) } @@ -710,6 +613,7 @@ pub(crate) mod test_support { authentication_classes: ResolvedAuthenticationClasses::new(Vec::new()), authorization_config: None, kubernetes_cluster_info: cluster_info(), + bootstrap_listeners: Vec::new(), }, &operator_environment(), ) diff --git a/rust/operator-binary/src/controller/apply.rs b/rust/operator-binary/src/controller/apply.rs new file mode 100644 index 00000000..74370814 --- /dev/null +++ b/rust/operator-binary/src/controller/apply.rs @@ -0,0 +1,130 @@ +//! The apply step in the KafkaCluster controller. + +use std::marker::PhantomData; + +use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + client::Client, + cluster_resources::{ClusterResource, ClusterResourceApplyStrategy, ClusterResources}, + deep_merger::ObjectOverrides, + v2::cluster_resources::cluster_resources_new, +}; +use strum::{EnumDiscriminants, IntoStaticStr}; + +use crate::controller::{ + Applied, KubernetesResources, Prepared, ValidatedCluster, controller_name, operator_name, + product_name, +}; + +#[derive(Snafu, Debug, EnumDiscriminants)] +#[strum_discriminants(derive(IntoStaticStr))] +pub enum Error { + #[snafu(display("failed to apply Kubernetes resource"))] + ApplyResource { + source: stackable_operator::cluster_resources::Error, + }, + + #[snafu(display("failed to delete orphaned resources"))] + DeleteOrphanedResources { + source: stackable_operator::cluster_resources::Error, + }, +} + +type Result = std::result::Result; + +/// Applier for the Kubernetes resource specifications produced by this controller. +/// +/// The implementation is not tied to this controller and could theoretically be moved to +/// stackable_operator if [`KubernetesResources`] would contain all possible resource types. +pub struct Applier<'a> { + client: &'a Client, + cluster_resources: ClusterResources<'a>, +} + +impl<'a> Applier<'a> { + pub fn new( + client: &'a Client, + cluster: &ValidatedCluster, + apply_strategy: ClusterResourceApplyStrategy, + object_overrides: &'a ObjectOverrides, + ) -> Applier<'a> { + let cluster_resources = cluster_resources_new( + &product_name(), + &operator_name(), + &controller_name(), + &cluster.name, + &cluster.namespace, + &cluster.uid, + apply_strategy, + object_overrides, + ); + + Applier { + client, + cluster_resources, + } + } + + pub async fn apply( + mut self, + resources: KubernetesResources, + ) -> Result> { + // Destructured without `..`, so adding a field to [`KubernetesResources`] fails to + // compile here instead of silently never being applied. + let KubernetesResources { + stateful_sets, + services, + listeners, + config_maps, + pod_disruption_budgets, + service_accounts, + role_bindings, + status: _, + } = resources; + + // Apply order is: StatefulSets last (a changed mounted ConfigMap/Secret + // must exist first, else Pods restart -- commons-operator#111). The ServiceAccount comes + // first because the Pods reference it at creation time. + let service_accounts = self.add_resources(service_accounts).await?; + let role_bindings = self.add_resources(role_bindings).await?; + let services = self.add_resources(services).await?; + let listeners = self.add_resources(listeners).await?; + let config_maps = self.add_resources(config_maps).await?; + let pod_disruption_budgets = self.add_resources(pod_disruption_budgets).await?; + let stateful_sets = self.add_resources(stateful_sets).await?; + + self.cluster_resources + .delete_orphaned_resources(self.client) + .await + .context(DeleteOrphanedResourcesSnafu)?; + + Ok(KubernetesResources { + stateful_sets, + services, + listeners, + config_maps, + pod_disruption_budgets, + service_accounts, + role_bindings, + status: PhantomData, + }) + } + + async fn add_resources( + &mut self, + resources: Vec, + ) -> Result> { + let mut applied_resources = vec![]; + + for resource in resources { + let applied_resource = self + .cluster_resources + .add(self.client, resource) + .await + .context(ApplyResourceSnafu)?; + applied_resources.push(applied_resource); + } + + Ok(applied_resources) + } +} diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs index 93e7996c..5cdd1544 100644 --- a/rust/operator-binary/src/controller/build/mod.rs +++ b/rust/operator-binary/src/controller/build/mod.rs @@ -1,16 +1,19 @@ //! Builders that assemble Kubernetes resources for kafka rolegroups. +use std::marker::PhantomData; + use snafu::{ResultExt, Snafu}; use crate::{ controller::{ - KubernetesResources, RoleGroupName, ValidatedCluster, + KubernetesResources, Prepared, RoleGroupName, ValidatedCluster, build::{ properties::{ listener::get_kafka_listener_config, product_logging::vector_config_file_content, }, resource::{ config_map::build_rolegroup_config_map, + discovery::build_discovery_configmap, listener::build_broker_rolegroup_bootstrap_listener, pdb::build_pdb, rbac::{build_role_binding, build_service_account}, @@ -45,6 +48,9 @@ pub enum Error { source: resource::statefulset::Error, role_group: RoleGroupName, }, + + #[snafu(display("failed to build discovery ConfigMap"))] + DiscoveryConfigMap { source: resource::discovery::Error }, } /// Builds every Kubernetes resource for the given validated cluster. @@ -52,10 +58,11 @@ pub enum Error { /// Does not need a Kubernetes client: every external reference is already dereferenced and /// validated by this point, so the only errors are resource-assembly failures. /// -/// The discovery `ConfigMap` is intentionally excluded: it reports the applied bootstrap -/// `Listener`s' ingress addresses (populated by the Listener operator only after apply), so it is -/// built in the reconcile step once those `Listener`s exist. -pub fn build(cluster: &ValidatedCluster) -> Result { +/// This includes the discovery `ConfigMap`, built from the bootstrap `Listener`s fetched in the +/// dereference step. It is skipped while no `Listener` has an ingress address yet (only the +/// listener-operator writes them); the `Listener` watch triggers a new reconcile run once the +/// addresses are set. +pub fn build(cluster: &ValidatedCluster) -> Result, Error> { let mut stateful_sets = vec![]; let mut services = vec![]; let mut listeners = vec![]; @@ -138,6 +145,12 @@ pub fn build(cluster: &ValidatedCluster) -> Result { } } + if let Some(discovery_cm) = + build_discovery_configmap(cluster).context(DiscoveryConfigMapSnafu)? + { + config_maps.push(discovery_cm); + } + Ok(KubernetesResources { stateful_sets, services, @@ -146,6 +159,7 @@ pub fn build(cluster: &ValidatedCluster) -> Result { pod_disruption_budgets, service_accounts: vec![build_service_account(cluster)], role_bindings: vec![build_role_binding(cluster)], + status: PhantomData, }) } @@ -275,6 +289,48 @@ mod tests { ); } + /// Once a bootstrap Listener (fetched in the dereference step) carries an ingress address, + /// `build()` emits the discovery ConfigMap, named after the cluster, alongside the rolegroup + /// ConfigMaps. The other tests run without bootstrap Listeners and therefore prove the + /// skip path (no `simple-kafka` entry in their `config_maps` assertions). + #[test] + fn build_emits_the_discovery_configmap_once_a_listener_has_an_address() { + use std::collections::BTreeMap; + + use stackable_operator::crd::listener; + + let mut cluster = kraft_cluster(); + let port_name = cluster + .cluster_config + .kafka_security + .client_port_name() + .to_owned(); + cluster.bootstrap_listeners = vec![listener::v1alpha1::Listener { + metadata: Default::default(), + spec: Default::default(), + status: Some(listener::v1alpha1::ListenerStatus { + service_name: None, + ingress_addresses: Some(vec![listener::v1alpha1::ListenerIngress { + address: "host1".to_owned(), + address_type: listener::v1alpha1::AddressType::Hostname, + ports: BTreeMap::from([(port_name, 9093)]), + }]), + node_ports: None, + }), + }]; + + let resources = build(&cluster).expect("build succeeds"); + + assert_eq!( + sorted_names(&resources.config_maps), + [ + "simple-kafka", + "simple-kafka-broker-default", + "simple-kafka-controller-default" + ] + ); + } + /// ZooKeeper mode has no `controller` role, so `build()` emits no controller resources while /// still producing the broker's bootstrap Listener. #[test] diff --git a/rust/operator-binary/src/controller/build/resource/discovery.rs b/rust/operator-binary/src/controller/build/resource/discovery.rs index de25fd3a..d19b4664 100644 --- a/rust/operator-binary/src/controller/build/resource/discovery.rs +++ b/rust/operator-binary/src/controller/build/resource/discovery.rs @@ -29,10 +29,17 @@ pub enum Error { /// Build a discovery [`ConfigMap`] containing information about how to connect to a certain /// `v1alpha1::KafkaCluster`. +/// +/// The bootstrap servers are read from the bootstrap `Listener`s' ingress addresses (carried on +/// [`ValidatedCluster::bootstrap_listeners`], fetched in the dereference step), which only the +/// listener-operator writes. Around the first reconcile runs no address exists yet; `Ok(None)` is +/// returned then instead of failing the run -- the `Listener` watch triggers a new run once the +/// addresses are set. In that window a previously tracked discovery `ConfigMap` would be deleted +/// as an orphan and re-created later, but the window only occurs while no address (and therefore +/// no usable `ConfigMap` content) exists at all. pub fn build_discovery_configmap( validated_cluster: &ValidatedCluster, - listeners: &[listener::v1alpha1::Listener], -) -> Result { +) -> Result, Error> { let kafka_security = &validated_cluster.cluster_config.kafka_security; let port_name = if kafka_security.has_kerberos_enabled() { @@ -41,14 +48,22 @@ pub fn build_discovery_configmap( kafka_security.client_port_name() }; + let hosts = listener_hosts(&validated_cluster.bootstrap_listeners, port_name)?; + if hosts.is_empty() { + tracing::debug!( + "no bootstrap Listener has an ingress address yet, skipping the discovery ConfigMap" + ); + return Ok(None); + } + // Write a list of bootstrap servers in the format that Kafka clients: // "{host1}:{port1},{host2:port2},..." - let bootstrap_servers = listener_hosts(listeners, port_name)? + let bootstrap_servers = hosts .into_iter() .map(|(host, port)| format!("{}:{}", host, port)) .collect::>() .join(","); - ConfigMapBuilder::new() + let discovery_cm = ConfigMapBuilder::new() .metadata( ObjectMetaBuilder::new() .name_and_namespace(validated_cluster) @@ -68,13 +83,15 @@ pub fn build_discovery_configmap( ) .add_data("KAFKA", bootstrap_servers) .build() - .context(BuildConfigMapSnafu) + .context(BuildConfigMapSnafu)?; + + Ok(Some(discovery_cm)) } fn listener_hosts( listeners: &[listener::v1alpha1::Listener], port_name: &str, -) -> Result + use<>, Error> { +) -> Result, Error> { listeners .iter() .flat_map(|listener| { @@ -97,3 +114,140 @@ fn listener_hosts( }) .collect::, _>>() } + +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; + + use stackable_operator::crd::listener; + + use super::build_discovery_configmap; + use crate::controller::{ + ValidatedCluster, + test_support::{minimal_kafka, validated_cluster}, + }; + + /// A ZooKeeper-mode cluster with a single `broker` role group and default (TLS) security. + fn broker_cluster() -> ValidatedCluster { + let kafka = minimal_kafka( + r#" + apiVersion: kafka.stackable.tech/v1alpha1 + kind: KafkaCluster + metadata: + name: simple-kafka + namespace: default + uid: 12345678-1234-1234-1234-123456789012 + spec: + image: + productVersion: 3.9.2 + clusterConfig: + zookeeperConfigMapName: xyz + brokers: + roleGroups: + default: + replicas: 1 + "#, + ); + validated_cluster(&kafka) + } + + fn bootstrap_listener( + ingress_addresses: Option>, + ) -> listener::v1alpha1::Listener { + listener::v1alpha1::Listener { + metadata: Default::default(), + spec: Default::default(), + status: Some(listener::v1alpha1::ListenerStatus { + service_name: None, + ingress_addresses, + node_ports: None, + }), + } + } + + fn ingress_address( + address: &str, + port_name: &str, + port: i32, + ) -> listener::v1alpha1::ListenerIngress { + listener::v1alpha1::ListenerIngress { + address: address.to_owned(), + address_type: listener::v1alpha1::AddressType::Hostname, + ports: BTreeMap::from([(port_name.to_owned(), port)]), + } + } + + #[test] + fn no_bootstrap_listeners_yield_no_configmap() { + let cluster = broker_cluster(); + + let discovery_cm = + build_discovery_configmap(&cluster).expect("discovery ConfigMap build should succeed"); + + assert!(discovery_cm.is_none()); + } + + #[test] + fn addressless_bootstrap_listeners_yield_no_configmap() { + let mut cluster = broker_cluster(); + cluster.bootstrap_listeners = vec![ + // Not yet reconciled by the listener-operator at all. + listener::v1alpha1::Listener { + status: None, + ..bootstrap_listener(None) + }, + // Reconciled, but no ingress addresses assigned yet. + bootstrap_listener(Some(Vec::new())), + ]; + + let discovery_cm = + build_discovery_configmap(&cluster).expect("discovery ConfigMap build should succeed"); + + assert!(discovery_cm.is_none()); + } + + #[test] + fn listener_addresses_are_written_to_the_configmap() { + let mut cluster = broker_cluster(); + // The fixture keeps the default TLS settings, so the client port is the TLS one. + let port_name = cluster + .cluster_config + .kafka_security + .client_port_name() + .to_owned(); + cluster.bootstrap_listeners = vec![ + bootstrap_listener(Some(vec![ingress_address("host1", &port_name, 9093)])), + bootstrap_listener(Some(vec![ingress_address("host2", &port_name, 31234)])), + ]; + + let discovery_cm = build_discovery_configmap(&cluster) + .expect("discovery ConfigMap build should succeed") + .expect("the listeners have ingress addresses, so a ConfigMap should be built"); + + assert_eq!( + discovery_cm.metadata.name.as_deref(), + Some("simple-kafka"), + "the discovery ConfigMap must be named after the cluster" + ); + let data = discovery_cm + .data + .expect("the discovery ConfigMap should carry data"); + assert_eq!( + data.get("KAFKA").map(String::as_str), + Some("host1:9093,host2:31234") + ); + } + + #[test] + fn address_without_the_client_port_is_an_error() { + let mut cluster = broker_cluster(); + cluster.bootstrap_listeners = vec![bootstrap_listener(Some(vec![ingress_address( + "host1", + "not-the-client-port", + 9093, + )]))]; + + build_discovery_configmap(&cluster) + .expect_err("an ingress address without the client port must fail the build"); + } +} diff --git a/rust/operator-binary/src/controller/build/resource/listener.rs b/rust/operator-binary/src/controller/build/resource/listener.rs index 6e5adc88..8648d020 100644 --- a/rust/operator-binary/src/controller/build/resource/listener.rs +++ b/rust/operator-binary/src/controller/build/resource/listener.rs @@ -1,6 +1,13 @@ +use std::str::FromStr; + use stackable_operator::{ - builder::meta::ObjectMetaBuilder, crd::listener, - v2::builder::meta::ownerreference_from_resource, + builder::meta::ObjectMetaBuilder, + crd::listener, + v2::{ + builder::meta::ownerreference_from_resource, + role_group_utils::ResourceNames, + types::{kubernetes::ListenerName, operator::ClusterName}, + }, }; use crate::{ @@ -8,6 +15,27 @@ use crate::{ crd::role::{KafkaRole, broker::BrokerConfig}, }; +/// The name of a broker role group's bootstrap [`Listener`](listener::v1alpha1::Listener), +/// `---bootstrap`. +/// +/// 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. +pub fn bootstrap_listener_name( + cluster_name: &ClusterName, + role: &KafkaRole, + role_group_name: &RoleGroupName, +) -> ListenerName { + let resource_names = ResourceNames { + cluster_name: cluster_name.clone(), + role_name: role.into(), + role_group_name: role_group_name.clone(), + }; + + ListenerName::from_str(&format!("{}-bootstrap", resource_names.stateful_set_name())) + .expect("the bootstrap listener name 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. // TODO (@NickLarsenNZ): Move shared functionality to stackable-operator diff --git a/rust/operator-binary/src/controller/dereference.rs b/rust/operator-binary/src/controller/dereference.rs index c90d1258..ed83395c 100644 --- a/rust/operator-binary/src/controller/dereference.rs +++ b/rust/operator-binary/src/controller/dereference.rs @@ -8,13 +8,27 @@ //! `KafkaAuthorization::get_opa_config` is a pure fetch + URL assembly (no validation to peel off) //! and stays here as-is. +use std::str::FromStr; + use snafu::{ResultExt, Snafu}; -use stackable_operator::{client::Client, utils::cluster_info::KubernetesClusterInfo}; +use stackable_operator::{ + client::Client, + crd::listener, + utils::cluster_info::KubernetesClusterInfo, + v2::{ + controller_utils::{get_cluster_name, get_namespace}, + types::kubernetes::ListenerName, + }, +}; -use crate::crd::{ - authentication::{self, ResolvedAuthenticationClasses}, - authorization::{self, KafkaAuthorizationConfig}, - v1alpha1, +use crate::{ + controller::{RoleGroupName, build::resource::listener::bootstrap_listener_name}, + crd::{ + authentication::{self, ResolvedAuthenticationClasses}, + authorization::{self, KafkaAuthorizationConfig}, + role::KafkaRole, + v1alpha1, + }, }; #[derive(Snafu, Debug)] @@ -24,6 +38,28 @@ pub enum Error { #[snafu(display("failed to get OPA config"))] GetOpaConfig { source: authorization::Error }, + + #[snafu(display("failed to resolve the cluster name"))] + ResolveClusterName { + source: stackable_operator::v2::controller_utils::Error, + }, + + #[snafu(display("failed to resolve the cluster namespace"))] + ResolveNamespace { + source: stackable_operator::v2::controller_utils::Error, + }, + + #[snafu(display("the role group name {role_group_name:?} is invalid"))] + ParseRoleGroupName { + source: stackable_operator::v2::macros::attributed_string_type::Error, + role_group_name: String, + }, + + #[snafu(display("failed to fetch bootstrap Listener {listener_name}"))] + FetchBootstrapListener { + source: stackable_operator::client::Error, + listener_name: ListenerName, + }, } type Result = std::result::Result; @@ -34,6 +70,12 @@ pub struct DereferencedObjects { pub authentication_classes: ResolvedAuthenticationClasses, pub authorization_config: Option, pub kubernetes_cluster_info: KubernetesClusterInfo, + /// The broker role groups' bootstrap `Listener`s as currently stored in the cluster. Unlike + /// the other fields they are not referenced from the spec but created by this operator itself + /// in a previous reconcile run. They are fetched because the discovery `ConfigMap` is built + /// from their ingress addresses, which only the listener-operator writes. `Listener`s that do + /// not exist yet (e.g. around the first reconcile runs) are simply absent. + pub bootstrap_listeners: Vec, } /// Fetches all Kubernetes objects referenced from the [`v1alpha1::KafkaCluster`] spec. @@ -57,9 +99,34 @@ pub async fn dereference( .await .context(GetOpaConfigSnafu)?; + let cluster_name = get_cluster_name(kafka).context(ResolveClusterNameSnafu)?; + let namespace = get_namespace(kafka).context(ResolveNamespaceSnafu)?; + + // Only broker role groups get a bootstrap Listener, so only their names are looked up. + let mut bootstrap_listeners = Vec::new(); + for role_group_name in kafka.spec.brokers.role_groups.keys() { + let role_group_name = + RoleGroupName::from_str(role_group_name).with_context(|_| ParseRoleGroupNameSnafu { + role_group_name: role_group_name.clone(), + })?; + let listener_name = + bootstrap_listener_name(&cluster_name, &KafkaRole::Broker, &role_group_name); + + if let Some(bootstrap_listener) = client + .get_opt::(listener_name.as_ref(), namespace.as_ref()) + .await + .with_context(|_| FetchBootstrapListenerSnafu { + listener_name: listener_name.clone(), + })? + { + bootstrap_listeners.push(bootstrap_listener); + } + } + Ok(DereferencedObjects { authentication_classes, authorization_config, kubernetes_cluster_info: client.kubernetes_cluster_info.clone(), + bootstrap_listeners, }) } diff --git a/rust/operator-binary/src/controller/update_status.rs b/rust/operator-binary/src/controller/update_status.rs new file mode 100644 index 00000000..5c0f0bf5 --- /dev/null +++ b/rust/operator-binary/src/controller/update_status.rs @@ -0,0 +1,55 @@ +//! The update_status step in the KafkaCluster controller. + +use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + client::Client, + status::condition::{ + compute_conditions, operations::ClusterOperationsConditionBuilder, + statefulset::StatefulSetConditionBuilder, + }, +}; +use strum::{EnumDiscriminants, IntoStaticStr}; + +use crate::{ + controller::{Applied, KubernetesResources}, + crd::{KafkaClusterStatus, OPERATOR_NAME, v1alpha1}, +}; + +#[derive(Snafu, Debug, EnumDiscriminants)] +#[strum_discriminants(derive(IntoStaticStr))] +pub enum Error { + #[snafu(display("failed to update status"))] + ApplyStatus { + source: stackable_operator::client::Error, + }, +} + +type Result = std::result::Result; + +/// Computes the cluster status from the applied resources and patches it onto the +/// [`v1alpha1::KafkaCluster`]. Takes [`KubernetesResources`] so the type system +/// proves the status derives from applied resources, not merely built ones. +pub async fn update_status( + client: &Client, + kafka: &v1alpha1::KafkaCluster, + applied: &KubernetesResources, +) -> Result<()> { + let mut ss_cond_builder = StatefulSetConditionBuilder::default(); + for stateful_set in &applied.stateful_sets { + ss_cond_builder.add(stateful_set.clone()); + } + + let cluster_operation_cond_builder = + ClusterOperationsConditionBuilder::new(&kafka.spec.cluster_operation); + + let status = KafkaClusterStatus { + conditions: compute_conditions(kafka, &[&ss_cond_builder, &cluster_operation_cond_builder]), + }; + + client + .apply_patch_status(OPERATOR_NAME, kafka, &status) + .await + .context(ApplyStatusSnafu)?; + + Ok(()) +} diff --git a/rust/operator-binary/src/controller/validate.rs b/rust/operator-binary/src/controller/validate.rs index 29a1c2dc..35405e63 100644 --- a/rust/operator-binary/src/controller/validate.rs +++ b/rust/operator-binary/src/controller/validate.rs @@ -317,6 +317,7 @@ pub fn validate( }, role_configs, role_group_configs, + dereferenced_objects.bootstrap_listeners, )) } From 4c456b2dcdd7877591369b82a22326dea5088198 Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Mon, 10 Aug 2026 09:31:18 +0200 Subject: [PATCH 2/6] changelog --- CHANGELOG.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index ef1f1c00..a2115559 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,10 +13,13 @@ All notable changes to this project will be documented in this file. functions and carry the full set of recommended labels ([#990]). - BREAKING: The `brokers` role is now required by the CRD; a KafkaCluster without it was previously accepted by the API server but failed reconciliation ([#990]). +- The reconciler now applies resources and derives the cluster status in discrete + apply and update_status steps ([#1000]). [#985]: https://github.com/stackabletech/kafka-operator/pull/985 [#990]: https://github.com/stackabletech/kafka-operator/pull/990 [#994]: https://github.com/stackabletech/kafka-operator/pull/994 +[#1000]: https://github.com/stackabletech/kafka-operator/pull/1000 ## [26.7.0] - 2026-07-21 From 452ddcdd0ae495e18fb5c1f45cc82fac0c50bfb5 Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Mon, 10 Aug 2026 10:13:20 +0200 Subject: [PATCH 3/6] updated comments, fixed portname deadlock issue --- .../templates/clusterrole-operator.yaml | 4 +- rust/operator-binary/src/controller.rs | 10 ++- rust/operator-binary/src/controller/apply.rs | 5 ++ .../controller/build/resource/discovery.rs | 65 ++++++++++++------- 4 files changed, 57 insertions(+), 27 deletions(-) diff --git a/deploy/helm/kafka-operator/templates/clusterrole-operator.yaml b/deploy/helm/kafka-operator/templates/clusterrole-operator.yaml index 5e5b475c..656fed50 100644 --- a/deploy/helm/kafka-operator/templates/clusterrole-operator.yaml +++ b/deploy/helm/kafka-operator/templates/clusterrole-operator.yaml @@ -121,7 +121,9 @@ rules: - authenticationclasses verbs: - get - # Listener created per role group. Applied via SSA and tracked for orphan cleanup. + # Listener created per broker role group. Applied via SSA and tracked for orphan cleanup. + # Fetched and watched by the controller, so a reconciliation is triggered once the + # listener-operator writes the ingress addresses (needed for the discovery ConfigMap). - apiGroups: - listeners.stackable.tech resources: diff --git a/rust/operator-binary/src/controller.rs b/rust/operator-binary/src/controller.rs index 259f44aa..b2d781b2 100644 --- a/rust/operator-binary/src/controller.rs +++ b/rust/operator-binary/src/controller.rs @@ -100,9 +100,13 @@ pub struct Applied; /// Every Kubernetes resource produced by the [`build`] step. /// -/// This includes the discovery `ConfigMap` (in [`Self::config_maps`]): it is built from the -/// bootstrap [`Listener`](listener)s as fetched in the dereference step, and is absent while -/// they have no ingress addresses yet. +/// `T` is a marker that indicates if these resources are only [`Prepared`] or already [`Applied`]. +/// The marker is useful e.g. to ensure that the cluster status is updated based on the applied +/// resources. +/// +/// The discovery `ConfigMap` is part of [`Self::config_maps`], but absent while no bootstrap +/// [`Listener`](listener) has a usable ingress address; see +/// [`build::resource::discovery::build_discovery_configmap`]. pub struct KubernetesResources { pub stateful_sets: Vec, pub services: Vec, diff --git a/rust/operator-binary/src/controller/apply.rs b/rust/operator-binary/src/controller/apply.rs index 74370814..75e682cb 100644 --- a/rust/operator-binary/src/controller/apply.rs +++ b/rust/operator-binary/src/controller/apply.rs @@ -65,6 +65,11 @@ impl<'a> Applier<'a> { } } + /// Applies the given Kubernetes resources, deletes resources from earlier reconcile runs + /// that were not applied in this one, and marks the resources as applied. + /// + /// Consumes the applier: a resource applied after the orphan deletion would itself be + /// treated as an orphan and deleted by the next reconcile run. pub async fn apply( mut self, resources: KubernetesResources, diff --git a/rust/operator-binary/src/controller/build/resource/discovery.rs b/rust/operator-binary/src/controller/build/resource/discovery.rs index d19b4664..46fd6034 100644 --- a/rust/operator-binary/src/controller/build/resource/discovery.rs +++ b/rust/operator-binary/src/controller/build/resource/discovery.rs @@ -1,6 +1,6 @@ use std::{num::TryFromIntError, str::FromStr}; -use snafu::{OptionExt, ResultExt, Snafu}; +use snafu::{ResultExt, Snafu}; use stackable_operator::{ builder::{configmap::ConfigMapBuilder, meta::ObjectMetaBuilder}, crd::listener, @@ -15,9 +15,6 @@ use crate::{ #[derive(Snafu, Debug)] pub enum Error { - #[snafu(display("could not find service port with name {}", port_name))] - NoServicePort { port_name: String }, - #[snafu(display("nodePort was out of range"))] InvalidNodePort { source: TryFromIntError }, @@ -32,11 +29,12 @@ pub enum Error { /// /// The bootstrap servers are read from the bootstrap `Listener`s' ingress addresses (carried on /// [`ValidatedCluster::bootstrap_listeners`], fetched in the dereference step), which only the -/// listener-operator writes. Around the first reconcile runs no address exists yet; `Ok(None)` is -/// returned then instead of failing the run -- the `Listener` watch triggers a new run once the -/// addresses are set. In that window a previously tracked discovery `ConfigMap` would be deleted -/// as an orphan and re-created later, but the window only occurs while no address (and therefore -/// no usable `ConfigMap` content) exists at all. +/// listener-operator writes. `Ok(None)` is returned instead of failing the run while no usable +/// address exists: around the first reconcile runs no address exists yet, and after a TLS or +/// Kerberos toggle the stored addresses carry the old port name until the listener-operator has +/// reconciled the new `Listener` spec. The `Listener` watch triggers a new run once the +/// addresses are usable. In that window a previously tracked discovery `ConfigMap` is deleted as +/// an orphan and re-created later. pub fn build_discovery_configmap( validated_cluster: &ValidatedCluster, ) -> Result, Error> { @@ -51,7 +49,8 @@ pub fn build_discovery_configmap( let hosts = listener_hosts(&validated_cluster.bootstrap_listeners, port_name)?; if hosts.is_empty() { tracing::debug!( - "no bootstrap Listener has an ingress address yet, skipping the discovery ConfigMap" + "no bootstrap Listener has an ingress address with the expected client port yet, \ + skipping the discovery ConfigMap" ); return Ok(None); } @@ -101,16 +100,25 @@ fn listener_hosts( .and_then(|s| s.ingress_addresses.as_deref()) }) .flatten() - .map(|addr| { - Ok(( - addr.address.clone(), - addr.ports - .get(port_name) - .copied() - .context(NoServicePortSnafu { port_name })? - .try_into() - .context(InvalidNodePortSnafu)?, - )) + .filter_map(|addr| { + let Some(&port) = addr.ports.get(port_name) else { + // The stored Listener status is stale, e.g. a TLS or Kerberos toggle changed the + // expected port name and the listener-operator has not reconciled the new + // Listener spec yet. Failing the build instead would abort the run before the + // apply step, so the new spec would never reach the listener-operator. + tracing::debug!( + address = addr.address, + port_name, + "skipping ingress address without the expected client port" + ); + return None; + }; + + Some( + u16::try_from(port) + .context(InvalidNodePortSnafu) + .map(|port| (addr.address.clone(), port)), + ) }) .collect::, _>>() } @@ -238,8 +246,14 @@ mod tests { ); } + /// A stored `Listener` whose ingress ports do not (yet) contain the expected client port + /// name is stale, e.g. right after a TLS or Kerberos toggle changed the port name but before + /// the listener-operator has seen the new `Listener` spec. It must be skipped like an + /// address-less `Listener` -- failing the build instead would abort the reconcile run before + /// the apply step, so the updated `Listener` spec would never reach the listener-operator and + /// the stale status would never be refreshed (a deadlock). #[test] - fn address_without_the_client_port_is_an_error() { + fn address_without_the_client_port_is_skipped() { let mut cluster = broker_cluster(); cluster.bootstrap_listeners = vec![bootstrap_listener(Some(vec![ingress_address( "host1", @@ -247,7 +261,12 @@ mod tests { 9093, )]))]; - build_discovery_configmap(&cluster) - .expect_err("an ingress address without the client port must fail the build"); + let discovery_cm = + build_discovery_configmap(&cluster).expect("discovery ConfigMap build should succeed"); + + assert!( + discovery_cm.is_none(), + "a stale ingress address without the client port must be skipped, not fail the build" + ); } } From 84d7e9ba653aa895c1226781fd0e8c56d03f1f22 Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Mon, 10 Aug 2026 10:23:13 +0200 Subject: [PATCH 4/6] fix listener ordering --- .../controller/build/resource/discovery.rs | 43 ++++++++++++++++++- 1 file changed, 41 insertions(+), 2 deletions(-) diff --git a/rust/operator-binary/src/controller/build/resource/discovery.rs b/rust/operator-binary/src/controller/build/resource/discovery.rs index 46fd6034..5c606824 100644 --- a/rust/operator-binary/src/controller/build/resource/discovery.rs +++ b/rust/operator-binary/src/controller/build/resource/discovery.rs @@ -91,7 +91,7 @@ fn listener_hosts( listeners: &[listener::v1alpha1::Listener], port_name: &str, ) -> Result, Error> { - listeners + let mut hosts = listeners .iter() .flat_map(|listener| { listener @@ -120,7 +120,14 @@ fn listener_hosts( .map(|port| (addr.address.clone(), port)), ) }) - .collect::, _>>() + .collect::, _>>()?; + + // The dereference step fetches the Listeners in the iteration order of + // `spec.brokers.roleGroups` -- a `HashMap`, so arbitrary and varying between reconcile runs. + // Sort so that the discovery ConfigMap content does not change while the spec is unchanged. + hosts.sort_unstable(); + + Ok(hosts) } #[cfg(test)] @@ -246,6 +253,38 @@ mod tests { ); } + /// The bootstrap servers must be sorted, not ordered by `bootstrap_listeners`: the + /// dereference step fetches the `Listener`s in the iteration order of + /// `spec.brokers.roleGroups` -- a `HashMap`, so arbitrary and varying between reconcile + /// runs. Without sorting, the discovery ConfigMap content would change between runs with an + /// unchanged spec. + #[test] + fn bootstrap_servers_are_sorted() { + let mut cluster = broker_cluster(); + let port_name = cluster + .cluster_config + .kafka_security + .client_port_name() + .to_owned(); + cluster.bootstrap_listeners = vec![ + bootstrap_listener(Some(vec![ingress_address("host2", &port_name, 31234)])), + bootstrap_listener(Some(vec![ingress_address("host1", &port_name, 9093)])), + ]; + + let discovery_cm = build_discovery_configmap(&cluster) + .expect("discovery ConfigMap build should succeed") + .expect("the listeners have ingress addresses, so a ConfigMap should be built"); + + let data = discovery_cm + .data + .expect("the discovery ConfigMap should carry data"); + assert_eq!( + data.get("KAFKA").map(String::as_str), + Some("host1:9093,host2:31234"), + "the bootstrap servers must be sorted regardless of the Listener fetch order" + ); + } + /// A stored `Listener` whose ingress ports do not (yet) contain the expected client port /// name is stale, e.g. right after a TLS or Kerberos toggle changed the port name but before /// the listener-operator has seen the new `Listener` spec. It must be skipped like an From 787fa2c82d5c6e727e5957d0f6c515be488afb87 Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Mon, 10 Aug 2026 11:05:19 +0200 Subject: [PATCH 5/6] up_date_status consumes applier, preserve existing behaviour by writing out incomplete config map --- rust/operator-binary/src/controller.rs | 64 +++++++- .../src/controller/build/mod.rs | 100 +++++-------- .../controller/build/resource/discovery.rs | 138 +++++------------- .../src/controller/dereference.rs | 22 ++- .../src/controller/update_status.rs | 6 +- 5 files changed, 147 insertions(+), 183 deletions(-) diff --git a/rust/operator-binary/src/controller.rs b/rust/operator-binary/src/controller.rs index b2d781b2..2e3ca3a2 100644 --- a/rust/operator-binary/src/controller.rs +++ b/rust/operator-binary/src/controller.rs @@ -104,9 +104,9 @@ pub struct Applied; /// The marker is useful e.g. to ensure that the cluster status is updated based on the applied /// resources. /// -/// The discovery `ConfigMap` is part of [`Self::config_maps`], but absent while no bootstrap -/// [`Listener`](listener) has a usable ingress address; see -/// [`build::resource::discovery::build_discovery_configmap`]. +/// The discovery `ConfigMap` is part of [`Self::config_maps`]; see +/// [`build::resource::discovery::build_discovery_configmap`] for how its content depends on the +/// bootstrap [`Listener`](listener)s. pub struct KubernetesResources { pub stateful_sets: Vec, pub services: Vec, @@ -550,7 +550,7 @@ pub async fn reconcile_kafka( .context(ApplyResourcesSnafu)?; // update status (client required) - update_status(client, kafka, &applied) + update_status(client, kafka, applied) .await .context(UpdateStatusSnafu)?; @@ -570,9 +570,12 @@ pub fn error_policy( #[cfg(test)] pub(crate) mod test_support { + use std::collections::BTreeMap; + use stackable_operator::{ cli::OperatorEnvironmentOptions, commons::networking::DomainName, + crd::listener, utils::{cluster_info::KubernetesClusterInfo, yaml_from_str_singleton_map}, }; @@ -609,6 +612,59 @@ pub(crate) mod test_support { } } + /// A ZooKeeper-mode cluster with a single `broker` role group and default (TLS) security. + pub fn zookeeper_mode_cluster() -> ValidatedCluster { + let kafka = minimal_kafka( + r#" + apiVersion: kafka.stackable.tech/v1alpha1 + kind: KafkaCluster + metadata: + name: simple-kafka + namespace: default + uid: 12345678-1234-1234-1234-123456789012 + spec: + image: + productVersion: 3.9.2 + clusterConfig: + zookeeperConfigMapName: xyz + brokers: + roleGroups: + default: + replicas: 1 + "#, + ); + validated_cluster(&kafka) + } + + /// A bootstrap `Listener` with the given ingress addresses, as stored in the cluster after + /// the listener-operator has reconciled it. + pub fn bootstrap_listener( + ingress_addresses: Option>, + ) -> listener::v1alpha1::Listener { + listener::v1alpha1::Listener { + metadata: Default::default(), + spec: Default::default(), + status: Some(listener::v1alpha1::ListenerStatus { + service_name: None, + ingress_addresses, + node_ports: None, + }), + } + } + + /// An ingress address exposing a single named port. + pub fn ingress_address( + address: &str, + port_name: &str, + port: i32, + ) -> listener::v1alpha1::ListenerIngress { + listener::v1alpha1::ListenerIngress { + address: address.to_owned(), + address_type: listener::v1alpha1::AddressType::Hostname, + ports: BTreeMap::from([(port_name.to_owned(), port)]), + } + } + /// Runs the real validate step against a minimal (auth/OPA-free) fixture. pub fn validated_cluster(kafka: &v1alpha1::KafkaCluster) -> ValidatedCluster { validate( diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs index 5cdd1544..9bf436bb 100644 --- a/rust/operator-binary/src/controller/build/mod.rs +++ b/rust/operator-binary/src/controller/build/mod.rs @@ -59,9 +59,9 @@ pub enum Error { /// validated by this point, so the only errors are resource-assembly failures. /// /// This includes the discovery `ConfigMap`, built from the bootstrap `Listener`s fetched in the -/// dereference step. It is skipped while no `Listener` has an ingress address yet (only the -/// listener-operator writes them); the `Listener` watch triggers a new reconcile run once the -/// addresses are set. +/// dereference step; see +/// [`build_discovery_configmap`] for how its +/// content depends on their ingress addresses. pub fn build(cluster: &ValidatedCluster) -> Result, Error> { let mut stateful_sets = vec![]; let mut services = vec![]; @@ -145,11 +145,7 @@ pub fn build(cluster: &ValidatedCluster) -> Result } } - if let Some(discovery_cm) = - build_discovery_configmap(cluster).context(DiscoveryConfigMapSnafu)? - { - config_maps.push(discovery_cm); - } + config_maps.push(build_discovery_configmap(cluster).context(DiscoveryConfigMapSnafu)?); Ok(KubernetesResources { stateful_sets, @@ -170,7 +166,10 @@ mod tests { use super::build; use crate::controller::{ ValidatedCluster, - test_support::{minimal_kafka, validated_cluster}, + test_support::{ + bootstrap_listener, ingress_address, minimal_kafka, validated_cluster, + zookeeper_mode_cluster, + }, }; /// Sorted `metadata.name`s of the given resources, for order-independent assertions. @@ -186,7 +185,7 @@ mod tests { /// A KRaft cluster with one `broker` and one `controller` role group, resolved through the real /// validate step (mirroring the other build fixtures), since [`ValidatedCluster`] carries /// several resolved types that are impractical to construct by hand. - fn kraft_cluster() -> ValidatedCluster { + fn kraft_mode_cluster() -> ValidatedCluster { let kafka = minimal_kafka( r#" apiVersion: kafka.stackable.tech/v1alpha1 @@ -213,33 +212,9 @@ mod tests { validated_cluster(&kafka) } - /// A ZooKeeper-mode cluster with a single `broker` role group (no controllers). - fn zookeeper_cluster() -> ValidatedCluster { - let kafka = minimal_kafka( - r#" - apiVersion: kafka.stackable.tech/v1alpha1 - kind: KafkaCluster - metadata: - name: simple-kafka - namespace: default - uid: 12345678-1234-1234-1234-123456789012 - spec: - image: - productVersion: 3.9.2 - clusterConfig: - zookeeperConfigMapName: xyz - brokers: - roleGroups: - default: - replicas: 1 - "#, - ); - validated_cluster(&kafka) - } - #[test] fn build_produces_expected_resource_names() { - let cluster = kraft_cluster(); + let cluster = kraft_mode_cluster(); let resources = build(&cluster).expect("build succeeds"); // One StatefulSet per role group. @@ -250,10 +225,12 @@ mod tests { "simple-kafka-controller-default" ] ); - // One rolegroup ConfigMap per role group. + // One rolegroup ConfigMap per role group, plus the discovery ConfigMap (named after the + // cluster), which is written even while no bootstrap Listener has an address yet. assert_eq!( sorted_names(&resources.config_maps), [ + "simple-kafka", "simple-kafka-broker-default", "simple-kafka-controller-default" ] @@ -289,45 +266,36 @@ mod tests { ); } - /// Once a bootstrap Listener (fetched in the dereference step) carries an ingress address, - /// `build()` emits the discovery ConfigMap, named after the cluster, alongside the rolegroup - /// ConfigMaps. The other tests run without bootstrap Listeners and therefore prove the - /// skip path (no `simple-kafka` entry in their `config_maps` assertions). + /// `build()` threads the bootstrap Listeners (fetched in the dereference step) through to the + /// discovery ConfigMap: once one carries an ingress address, the `KAFKA` entry names it. The + /// other tests run without bootstrap Listeners, where the entry is empty. #[test] - fn build_emits_the_discovery_configmap_once_a_listener_has_an_address() { - use std::collections::BTreeMap; - - use stackable_operator::crd::listener; - - let mut cluster = kraft_cluster(); + fn build_writes_listener_addresses_to_the_discovery_configmap() { + let mut cluster = kraft_mode_cluster(); let port_name = cluster .cluster_config .kafka_security .client_port_name() .to_owned(); - cluster.bootstrap_listeners = vec![listener::v1alpha1::Listener { - metadata: Default::default(), - spec: Default::default(), - status: Some(listener::v1alpha1::ListenerStatus { - service_name: None, - ingress_addresses: Some(vec![listener::v1alpha1::ListenerIngress { - address: "host1".to_owned(), - address_type: listener::v1alpha1::AddressType::Hostname, - ports: BTreeMap::from([(port_name, 9093)]), - }]), - node_ports: None, - }), - }]; + cluster.bootstrap_listeners = vec![bootstrap_listener(Some(vec![ingress_address( + "host1", &port_name, 9093, + )]))]; let resources = build(&cluster).expect("build succeeds"); + let discovery_cm = resources + .config_maps + .iter() + .find(|config_map| config_map.metadata.name.as_deref() == Some("simple-kafka")) + .expect("the discovery ConfigMap should be built"); assert_eq!( - sorted_names(&resources.config_maps), - [ - "simple-kafka", - "simple-kafka-broker-default", - "simple-kafka-controller-default" - ] + discovery_cm + .data + .as_ref() + .expect("the discovery ConfigMap should carry data") + .get("KAFKA") + .map(String::as_str), + Some("host1:9093") ); } @@ -335,7 +303,7 @@ mod tests { /// still producing the broker's bootstrap Listener. #[test] fn build_zookeeper_mode_has_no_controller_resources() { - let cluster = zookeeper_cluster(); + let cluster = zookeeper_mode_cluster(); let resources = build(&cluster).expect("build succeeds"); assert_eq!( diff --git a/rust/operator-binary/src/controller/build/resource/discovery.rs b/rust/operator-binary/src/controller/build/resource/discovery.rs index 5c606824..299598cb 100644 --- a/rust/operator-binary/src/controller/build/resource/discovery.rs +++ b/rust/operator-binary/src/controller/build/resource/discovery.rs @@ -29,15 +29,13 @@ pub enum Error { /// /// The bootstrap servers are read from the bootstrap `Listener`s' ingress addresses (carried on /// [`ValidatedCluster::bootstrap_listeners`], fetched in the dereference step), which only the -/// listener-operator writes. `Ok(None)` is returned instead of failing the run while no usable -/// address exists: around the first reconcile runs no address exists yet, and after a TLS or -/// Kerberos toggle the stored addresses carry the old port name until the listener-operator has -/// reconciled the new `Listener` spec. The `Listener` watch triggers a new run once the -/// addresses are usable. In that window a previously tracked discovery `ConfigMap` is deleted as -/// an orphan and re-created later. -pub fn build_discovery_configmap( - validated_cluster: &ValidatedCluster, -) -> Result, Error> { +/// listener-operator writes. While no usable address exists -- around the first reconcile runs, +/// or after a TLS or Kerberos toggle while the stored addresses still carry the old port name -- +/// the `ConfigMap` is still written, with an empty `KAFKA` value: omitting it instead would let +/// the apply step delete an existing discovery `ConfigMap` as an orphan, breaking consumers that +/// mount it. The `Listener` watch triggers a new run that fills in the value once the addresses +/// are usable. +pub fn build_discovery_configmap(validated_cluster: &ValidatedCluster) -> Result { let kafka_security = &validated_cluster.cluster_config.kafka_security; let port_name = if kafka_security.has_kerberos_enabled() { @@ -50,9 +48,8 @@ pub fn build_discovery_configmap( if hosts.is_empty() { tracing::debug!( "no bootstrap Listener has an ingress address with the expected client port yet, \ - skipping the discovery ConfigMap" + writing an empty KAFKA entry to the discovery ConfigMap" ); - return Ok(None); } // Write a list of bootstrap servers in the format that Kafka clients: @@ -84,7 +81,7 @@ pub fn build_discovery_configmap( .build() .context(BuildConfigMapSnafu)?; - Ok(Some(discovery_cm)) + Ok(discovery_cm) } fn listener_hosts( @@ -132,79 +129,42 @@ fn listener_hosts( #[cfg(test)] mod tests { - use std::collections::BTreeMap; - use stackable_operator::crd::listener; use super::build_discovery_configmap; - use crate::controller::{ - ValidatedCluster, - test_support::{minimal_kafka, validated_cluster}, + use crate::controller::test_support::{ + bootstrap_listener, ingress_address, zookeeper_mode_cluster, }; - /// A ZooKeeper-mode cluster with a single `broker` role group and default (TLS) security. - fn broker_cluster() -> ValidatedCluster { - let kafka = minimal_kafka( - r#" - apiVersion: kafka.stackable.tech/v1alpha1 - kind: KafkaCluster - metadata: - name: simple-kafka - namespace: default - uid: 12345678-1234-1234-1234-123456789012 - spec: - image: - productVersion: 3.9.2 - clusterConfig: - zookeeperConfigMapName: xyz - brokers: - roleGroups: - default: - replicas: 1 - "#, + /// Asserts that the given ConfigMap carries the given `KAFKA` value. + fn assert_kafka_entry( + discovery_cm: &stackable_operator::k8s_openapi::api::core::v1::ConfigMap, + expected: &str, + ) { + assert_eq!( + discovery_cm + .data + .as_ref() + .expect("the discovery ConfigMap should carry data") + .get("KAFKA") + .map(String::as_str), + Some(expected) ); - validated_cluster(&kafka) - } - - fn bootstrap_listener( - ingress_addresses: Option>, - ) -> listener::v1alpha1::Listener { - listener::v1alpha1::Listener { - metadata: Default::default(), - spec: Default::default(), - status: Some(listener::v1alpha1::ListenerStatus { - service_name: None, - ingress_addresses, - node_ports: None, - }), - } - } - - fn ingress_address( - address: &str, - port_name: &str, - port: i32, - ) -> listener::v1alpha1::ListenerIngress { - listener::v1alpha1::ListenerIngress { - address: address.to_owned(), - address_type: listener::v1alpha1::AddressType::Hostname, - ports: BTreeMap::from([(port_name.to_owned(), port)]), - } } #[test] - fn no_bootstrap_listeners_yield_no_configmap() { - let cluster = broker_cluster(); + fn no_bootstrap_listeners_yield_an_empty_kafka_entry() { + let cluster = zookeeper_mode_cluster(); let discovery_cm = build_discovery_configmap(&cluster).expect("discovery ConfigMap build should succeed"); - assert!(discovery_cm.is_none()); + assert_kafka_entry(&discovery_cm, ""); } #[test] - fn addressless_bootstrap_listeners_yield_no_configmap() { - let mut cluster = broker_cluster(); + fn addressless_bootstrap_listeners_yield_an_empty_kafka_entry() { + let mut cluster = zookeeper_mode_cluster(); cluster.bootstrap_listeners = vec![ // Not yet reconciled by the listener-operator at all. listener::v1alpha1::Listener { @@ -218,12 +178,12 @@ mod tests { let discovery_cm = build_discovery_configmap(&cluster).expect("discovery ConfigMap build should succeed"); - assert!(discovery_cm.is_none()); + assert_kafka_entry(&discovery_cm, ""); } #[test] fn listener_addresses_are_written_to_the_configmap() { - let mut cluster = broker_cluster(); + let mut cluster = zookeeper_mode_cluster(); // The fixture keeps the default TLS settings, so the client port is the TLS one. let port_name = cluster .cluster_config @@ -235,22 +195,15 @@ mod tests { bootstrap_listener(Some(vec![ingress_address("host2", &port_name, 31234)])), ]; - let discovery_cm = build_discovery_configmap(&cluster) - .expect("discovery ConfigMap build should succeed") - .expect("the listeners have ingress addresses, so a ConfigMap should be built"); + let discovery_cm = + build_discovery_configmap(&cluster).expect("discovery ConfigMap build should succeed"); assert_eq!( discovery_cm.metadata.name.as_deref(), Some("simple-kafka"), "the discovery ConfigMap must be named after the cluster" ); - let data = discovery_cm - .data - .expect("the discovery ConfigMap should carry data"); - assert_eq!( - data.get("KAFKA").map(String::as_str), - Some("host1:9093,host2:31234") - ); + assert_kafka_entry(&discovery_cm, "host1:9093,host2:31234"); } /// The bootstrap servers must be sorted, not ordered by `bootstrap_listeners`: the @@ -260,7 +213,7 @@ mod tests { /// unchanged spec. #[test] fn bootstrap_servers_are_sorted() { - let mut cluster = broker_cluster(); + let mut cluster = zookeeper_mode_cluster(); let port_name = cluster .cluster_config .kafka_security @@ -271,18 +224,10 @@ mod tests { bootstrap_listener(Some(vec![ingress_address("host1", &port_name, 9093)])), ]; - let discovery_cm = build_discovery_configmap(&cluster) - .expect("discovery ConfigMap build should succeed") - .expect("the listeners have ingress addresses, so a ConfigMap should be built"); + let discovery_cm = + build_discovery_configmap(&cluster).expect("discovery ConfigMap build should succeed"); - let data = discovery_cm - .data - .expect("the discovery ConfigMap should carry data"); - assert_eq!( - data.get("KAFKA").map(String::as_str), - Some("host1:9093,host2:31234"), - "the bootstrap servers must be sorted regardless of the Listener fetch order" - ); + assert_kafka_entry(&discovery_cm, "host1:9093,host2:31234"); } /// A stored `Listener` whose ingress ports do not (yet) contain the expected client port @@ -293,7 +238,7 @@ mod tests { /// the stale status would never be refreshed (a deadlock). #[test] fn address_without_the_client_port_is_skipped() { - let mut cluster = broker_cluster(); + let mut cluster = zookeeper_mode_cluster(); cluster.bootstrap_listeners = vec![bootstrap_listener(Some(vec![ingress_address( "host1", "not-the-client-port", @@ -303,9 +248,6 @@ mod tests { let discovery_cm = build_discovery_configmap(&cluster).expect("discovery ConfigMap build should succeed"); - assert!( - discovery_cm.is_none(), - "a stale ingress address without the client port must be skipped, not fail the build" - ); + assert_kafka_entry(&discovery_cm, ""); } } diff --git a/rust/operator-binary/src/controller/dereference.rs b/rust/operator-binary/src/controller/dereference.rs index ed83395c..605e34c7 100644 --- a/rust/operator-binary/src/controller/dereference.rs +++ b/rust/operator-binary/src/controller/dereference.rs @@ -1,9 +1,15 @@ //! The dereference step in the KafkaCluster controller. //! -//! Fetches all Kubernetes objects referenced by the [`v1alpha1::KafkaCluster`] spec and returns -//! them in [`DereferencedObjects`]. This step only performs I/O; validation of the fetched -//! objects (constraints on which auth class providers are supported, kerberos + TLS -//! compatibility, etc.) happens in the validate step. +//! Fetches the Kubernetes objects the later steps need and returns them in +//! [`DereferencedObjects`]. Most of them are referenced from the [`v1alpha1::KafkaCluster`] +//! spec (e.g. the AuthenticationClasses). The broker role groups' bootstrap `Listener`s are the +//! exception: they are not referenced from the spec but created by this operator itself in a +//! previous reconcile run, and are fetched back because the discovery `ConfigMap` is built from +//! their ingress addresses, which only the listener-operator writes. `Listener`s that do not +//! exist yet (e.g. around the first reconcile runs) are simply absent. +//! +//! Validation of the fetched objects (constraints on which auth class providers are supported, +//! kerberos + TLS compatibility, etc.) happens in the validate step, not here. //! //! `KafkaAuthorization::get_opa_config` is a pure fetch + URL assembly (no validation to peel off) //! and stays here as-is. @@ -64,21 +70,13 @@ pub enum Error { type Result = std::result::Result; -/// Kubernetes objects referenced from the [`v1alpha1::KafkaCluster`] spec, already fetched but -/// not yet validated. pub struct DereferencedObjects { pub authentication_classes: ResolvedAuthenticationClasses, pub authorization_config: Option, pub kubernetes_cluster_info: KubernetesClusterInfo, - /// The broker role groups' bootstrap `Listener`s as currently stored in the cluster. Unlike - /// the other fields they are not referenced from the spec but created by this operator itself - /// in a previous reconcile run. They are fetched because the discovery `ConfigMap` is built - /// from their ingress addresses, which only the listener-operator writes. `Listener`s that do - /// not exist yet (e.g. around the first reconcile runs) are simply absent. pub bootstrap_listeners: Vec, } -/// Fetches all Kubernetes objects referenced from the [`v1alpha1::KafkaCluster`] spec. pub async fn dereference( client: &Client, kafka: &v1alpha1::KafkaCluster, diff --git a/rust/operator-binary/src/controller/update_status.rs b/rust/operator-binary/src/controller/update_status.rs index 5c0f0bf5..376f25eb 100644 --- a/rust/operator-binary/src/controller/update_status.rs +++ b/rust/operator-binary/src/controller/update_status.rs @@ -32,11 +32,11 @@ type Result = std::result::Result; pub async fn update_status( client: &Client, kafka: &v1alpha1::KafkaCluster, - applied: &KubernetesResources, + applied: KubernetesResources, ) -> Result<()> { let mut ss_cond_builder = StatefulSetConditionBuilder::default(); - for stateful_set in &applied.stateful_sets { - ss_cond_builder.add(stateful_set.clone()); + for stateful_set in applied.stateful_sets { + ss_cond_builder.add(stateful_set); } let cluster_operation_cond_builder = From 4fa8f3dfd25ce0d8fd15f7a7260b2053e489784b Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Tue, 11 Aug 2026 13:24:41 +0200 Subject: [PATCH 6/6] add suggested compile-time checks for listener name --- .../src/controller/build/resource/listener.rs | 25 ++++++++++++++++--- 1 file changed, 22 insertions(+), 3 deletions(-) diff --git a/rust/operator-binary/src/controller/build/resource/listener.rs b/rust/operator-binary/src/controller/build/resource/listener.rs index 8648d020..bc625b12 100644 --- a/rust/operator-binary/src/controller/build/resource/listener.rs +++ b/rust/operator-binary/src/controller/build/resource/listener.rs @@ -5,7 +5,7 @@ use stackable_operator::{ crd::listener, v2::{ builder::meta::ownerreference_from_resource, - role_group_utils::ResourceNames, + role_group_utils::{QualifiedRoleGroupName, ResourceNames}, types::{kubernetes::ListenerName, operator::ClusterName}, }, }; @@ -26,14 +26,33 @@ pub fn bootstrap_listener_name( role: &KafkaRole, role_group_name: &RoleGroupName, ) -> ListenerName { + const BOOTSTRAP_SUFFIX: &str = "-bootstrap"; + + // Compile-time checks that `-bootstrap` is a valid ListenerName, 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." + ); + // 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; + let resource_names = ResourceNames { cluster_name: cluster_name.clone(), role_name: role.into(), role_group_name: role_group_name.clone(), }; - ListenerName::from_str(&format!("{}-bootstrap", resource_names.stateful_set_name())) - .expect("the bootstrap listener name is a valid Listener name") + ListenerName::from_str(&format!( + "{qualified_role_group_name}{BOOTSTRAP_SUFFIX}", + qualified_role_group_name = resource_names.qualified_role_group_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