From b16b748723a59acc975519a2392db7b3ebd5898a Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Sat, 8 Aug 2026 09:51:15 +0200 Subject: [PATCH] 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, )) }