diff --git a/CHANGELOG.md b/CHANGELOG.md index 769160f1..4e3b1a1a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,8 @@ All notable changes to this project will be documented in this file. - 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]). - All product containers now run with `securityContext.runAsNonRoot` set to `true` to improve security ([#998]). +- The reconciler now applies resources and derives the cluster status in discrete + apply and update_status steps ([#1000]). ### Fixed @@ -25,6 +27,7 @@ All notable changes to this project will be documented in this file. [#990]: https://github.com/stackabletech/kafka-operator/pull/990 [#994]: https://github.com/stackabletech/kafka-operator/pull/994 [#998]: https://github.com/stackabletech/kafka-operator/pull/998 +[#1000]: https://github.com/stackabletech/kafka-operator/pull/1000 ## [26.7.0] - 2026-07-21 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 5b3cea97..2e3ca3a2 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,22 @@ 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 { +/// `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`]; 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, pub listeners: Vec, @@ -104,6 +115,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 +143,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 +161,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 +183,7 @@ impl ValidatedCluster { cluster_config, role_configs, role_group_configs, + bootstrap_listeners, } } @@ -303,12 +322,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 +463,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 +478,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 +495,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 +526,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 +534,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()) } @@ -663,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}, }; @@ -702,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( @@ -710,6 +673,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..75e682cb --- /dev/null +++ b/rust/operator-binary/src/controller/apply.rs @@ -0,0 +1,135 @@ +//! 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, + } + } + + /// 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, + ) -> 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..9bf436bb 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; 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![]; let mut listeners = vec![]; @@ -138,6 +145,8 @@ pub fn build(cluster: &ValidatedCluster) -> Result { } } + config_maps.push(build_discovery_configmap(cluster).context(DiscoveryConfigMapSnafu)?); + Ok(KubernetesResources { stateful_sets, services, @@ -146,6 +155,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, }) } @@ -156,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. @@ -172,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 @@ -199,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. @@ -236,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" ] @@ -275,11 +266,44 @@ mod tests { ); } + /// `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_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![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!( + discovery_cm + .data + .as_ref() + .expect("the discovery ConfigMap should carry data") + .get("KAFKA") + .map(String::as_str), + Some("host1:9093") + ); + } + /// ZooKeeper mode has no `controller` role, so `build()` emits no controller resources while /// 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 de25fd3a..299598cb 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 }, @@ -29,10 +26,16 @@ pub enum Error { /// Build a discovery [`ConfigMap`] containing information about how to connect to a certain /// `v1alpha1::KafkaCluster`. -pub fn build_discovery_configmap( - validated_cluster: &ValidatedCluster, - listeners: &[listener::v1alpha1::Listener], -) -> Result { +/// +/// 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. 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() { @@ -41,14 +44,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 with the expected client port yet, \ + writing an empty KAFKA entry to the discovery ConfigMap" + ); + } + // 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,14 +79,16 @@ pub fn build_discovery_configmap( ) .add_data("KAFKA", bootstrap_servers) .build() - .context(BuildConfigMapSnafu) + .context(BuildConfigMapSnafu)?; + + Ok(discovery_cm) } fn listener_hosts( listeners: &[listener::v1alpha1::Listener], port_name: &str, -) -> Result + use<>, Error> { - listeners +) -> Result, Error> { + let mut hosts = listeners .iter() .flat_map(|listener| { listener @@ -84,16 +97,157 @@ 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::, _>>() + .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)] +mod tests { + use stackable_operator::crd::listener; + + use super::build_discovery_configmap; + use crate::controller::test_support::{ + bootstrap_listener, ingress_address, zookeeper_mode_cluster, + }; + + /// 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) + ); + } + + #[test] + 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_kafka_entry(&discovery_cm, ""); + } + + #[test] + 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 { + 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_kafka_entry(&discovery_cm, ""); + } + + #[test] + fn listener_addresses_are_written_to_the_configmap() { + 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 + .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"); + + assert_eq!( + discovery_cm.metadata.name.as_deref(), + Some("simple-kafka"), + "the discovery ConfigMap must be named after the cluster" + ); + assert_kafka_entry(&discovery_cm, "host1:9093,host2:31234"); + } + + /// 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 = zookeeper_mode_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"); + + assert_kafka_entry(&discovery_cm, "host1:9093,host2:31234"); + } + + /// 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_skipped() { + let mut cluster = zookeeper_mode_cluster(); + cluster.bootstrap_listeners = vec![bootstrap_listener(Some(vec![ingress_address( + "host1", + "not-the-client-port", + 9093, + )]))]; + + let discovery_cm = + build_discovery_configmap(&cluster).expect("discovery ConfigMap build should succeed"); + + assert_kafka_entry(&discovery_cm, ""); + } } 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..605e34c7 100644 --- a/rust/operator-binary/src/controller/dereference.rs +++ b/rust/operator-binary/src/controller/dereference.rs @@ -1,20 +1,40 @@ //! 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. +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,19 +44,39 @@ 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; -/// 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, + pub bootstrap_listeners: Vec, } -/// Fetches all Kubernetes objects referenced from the [`v1alpha1::KafkaCluster`] spec. pub async fn dereference( client: &Client, kafka: &v1alpha1::KafkaCluster, @@ -57,9 +97,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..376f25eb --- /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); + } + + 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, )) }