From c50096d2ff6b86805a5e85053a360919af8df6b8 Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Wed, 5 Aug 2026 16:24:30 +0200 Subject: [PATCH 01/12] refactor: Build the discovery ConfigMap in the build step The discovery ConfigMap advertises the addresses that the listener operator publishes on the ZooKeeper role Listener, so it used to be built and applied inline in reconcile_zk, from the Listener that had just been applied. That Listener carries no addresses yet on the first reconciliation, so the controller relied on the reconciliation failing and being requeued five seconds later. Follow the pattern the OpenSearch operator already uses: dereference the role Listener, validate its addresses into the ValidatedCluster, and let build() emit an Option. The controller now watches Listeners, so the reconciliation that writes the discovery ConfigMap is triggered as soon as the addresses appear, and it also reruns when they change. The listener address extraction moves into a shared listener_addresses module, because the ZookeeperZnode controller needs it too. The operator ClusterRole gains the watch verb on listeners, which is required for the new watch. --- .../templates/clusterrole-operator.yaml | 6 +- .../operator-binary/src/listener_addresses.rs | 164 +++++++++++++++++ rust/operator-binary/src/main.rs | 9 + rust/operator-binary/src/zk_controller.rs | 73 ++------ .../src/zk_controller/build.rs | 30 +++- .../zk_controller/build/resource/discovery.rs | 165 ++---------------- .../src/zk_controller/dereference.rs | 53 +++++- .../src/zk_controller/validate.rs | 27 ++- rust/operator-binary/src/znode_controller.rs | 24 ++- 9 files changed, 329 insertions(+), 222 deletions(-) create mode 100644 rust/operator-binary/src/listener_addresses.rs diff --git a/deploy/helm/zookeeper-operator/templates/clusterrole-operator.yaml b/deploy/helm/zookeeper-operator/templates/clusterrole-operator.yaml index 651db149..8ed7ed5b 100644 --- a/deploy/helm/zookeeper-operator/templates/clusterrole-operator.yaml +++ b/deploy/helm/zookeeper-operator/templates/clusterrole-operator.yaml @@ -118,8 +118,9 @@ rules: verbs: - create - patch - # Listener created per role group for external access. Applied via SSA and tracked for - # orphan cleanup. + # Listener created per role for external access. Applied via SSA and tracked for + # orphan cleanup. Watched as well, because the discovery ConfigMap advertises the addresses + # that the listener operator publishes on it. - apiGroups: - listeners.stackable.tech resources: @@ -130,6 +131,7 @@ rules: - get - list - patch + - watch # Primary CRD: watched and read during reconciliation. - apiGroups: - {{ include "operator.name" . }}.stackable.tech diff --git a/rust/operator-binary/src/listener_addresses.rs b/rust/operator-binary/src/listener_addresses.rs new file mode 100644 index 00000000..4eaa1e0d --- /dev/null +++ b/rust/operator-binary/src/listener_addresses.rs @@ -0,0 +1,164 @@ +//! Reading client connection addresses from a [`Listener`](listener::v1alpha1::Listener). +//! +//! Shared by both controllers, which turn the dereferenced ZooKeeper role Listener into the +//! addresses advertised by their discovery ConfigMaps. + +use std::{collections::BTreeSet, num::TryFromIntError}; + +use snafu::{ResultExt, Snafu}; +use stackable_operator::{crd::listener, kube::runtime::reflector::ObjectRef}; + +#[derive(Snafu, Debug)] +pub enum Error { + #[snafu(display("{listener} does not have a port with the name {port_name:?}"))] + PortNotFound { + port_name: String, + listener: ObjectRef, + }, + + #[snafu(display("expected an unsigned 16-bit port, got {port_number}"))] + InvalidPort { + source: TryFromIntError, + port_number: i32, + }, +} + +type Result = std::result::Result; + +/// The address and port pairs published by a [`Listener`](listener::v1alpha1::Listener) for a +/// single named port, sorted and deduplicated. +/// +/// An address is a hostname or IP address of a node, a cluster IP or an external load balancer, +/// depending on the Service type behind the Listener. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ListenerAddresses(BTreeSet<(String, u16)>); + +impl ListenerAddresses { + /// Renders the addresses as the comma separated `host1:port1,host2:port2` list that ZooKeeper + /// clients expect. + pub fn to_connection_string(&self) -> String { + self.0 + .iter() + .map(|(address, port)| format!("{address}:{port}")) + .collect::>() + .join(",") + } +} + +/// Reads the addresses that `listener` publishes for `port_name`. +/// +/// Returns `Ok(None)` while the Listener carries no ingress addresses at all, which is the normal +/// state between creating the Listener and the listener operator publishing its addresses. A +/// Listener that does publish addresses, but none for `port_name`, is an error. +// TODO (@NickLarsenNZ): Move this to stackable-operator, so it can be used as +// listener.addresses_for_port(port_name) +pub fn listener_addresses( + listener: &listener::v1alpha1::Listener, + port_name: &str, +) -> Result> { + let Some(ingress_addresses) = listener + .status + .as_ref() + .and_then(|listener_status| listener_status.ingress_addresses.as_ref()) + else { + return Ok(None); + }; + + let address_port_pairs = ingress_addresses + .iter() + // Filter the addresses that have the port we are interested in (they likely all have it though) + .filter_map(|listener_ingress| { + Some(listener_ingress.address.clone()).zip(listener_ingress.ports.get(port_name)) + }) + // Convert the port from i32 to u16 + .map(|(listener_address, &port_number)| { + let port_number: u16 = port_number + .try_into() + .context(InvalidPortSnafu { port_number })?; + Ok((listener_address, port_number)) + }) + .collect::, _>>()?; + + match address_port_pairs.is_empty() { + true => PortNotFoundSnafu { + port_name, + listener, + } + .fail(), + false => Ok(Some(ListenerAddresses(address_port_pairs))), + } +} + +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; + + use stackable_operator::{ + crd::listener::v1alpha1::{ + AddressType, Listener, ListenerIngress, ListenerSpec, ListenerStatus, + }, + k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta, + }; + + use super::*; + use crate::crd::ZOOKEEPER_SERVER_PORT_NAME; + + fn listener(ingress_addresses: Option>) -> Listener { + Listener { + metadata: ObjectMeta { + name: Some("test-listener".to_owned()), + ..ObjectMeta::default() + }, + spec: ListenerSpec::default(), + status: Some(ListenerStatus { + service_name: None, + ingress_addresses, + node_ports: None, + }), + } + } + + fn ingress(port: i32) -> ListenerIngress { + ListenerIngress { + address: "node-0".to_owned(), + address_type: AddressType::Hostname, + ports: BTreeMap::from([(ZOOKEEPER_SERVER_PORT_NAME.to_owned(), port)]), + } + } + + #[test] + fn listener_addresses_returns_host_port_pairs() { + let listener = listener(Some(vec![ingress(2181)])); + let addresses = listener_addresses(&listener, ZOOKEEPER_SERVER_PORT_NAME) + .expect("addresses") + .expect("the listener publishes addresses"); + assert_eq!(addresses.to_connection_string(), "node-0:2181"); + } + + #[test] + fn listener_addresses_without_ingress_is_not_ready_yet() { + assert_eq!( + listener_addresses(&listener(None), ZOOKEEPER_SERVER_PORT_NAME).expect("addresses"), + None + ); + } + + #[test] + fn listener_addresses_missing_port_name_is_error() { + let listener = listener(Some(vec![ingress(2181)])); + assert!(matches!( + listener_addresses(&listener, "does-not-exist"), + Err(Error::PortNotFound { .. }) + )); + } + + #[test] + fn listener_addresses_port_out_of_u16_range_is_error() { + // A port number that does not fit into a u16 must be rejected. + let listener = listener(Some(vec![ingress(70_000)])); + assert!(matches!( + listener_addresses(&listener, ZOOKEEPER_SERVER_PORT_NAME), + Err(Error::InvalidPort { .. }) + )); + } +} diff --git a/rust/operator-binary/src/main.rs b/rust/operator-binary/src/main.rs index 1b9c6067..ab935650 100644 --- a/rust/operator-binary/src/main.rs +++ b/rust/operator-binary/src/main.rs @@ -13,6 +13,7 @@ use futures::{FutureExt, StreamExt, TryFutureExt}; use stackable_operator::{ YamlSchema, cli::{Command, RunArguments}, + crd::listener::v1alpha1::Listener, eos::EndOfSupportChecker, k8s_openapi::api::{ apps::v1::StatefulSet, @@ -40,6 +41,7 @@ use crate::{ }; pub mod crd; +mod listener_addresses; mod webhooks; mod zk_controller; mod znode_controller; @@ -139,6 +141,13 @@ async fn main() -> anyhow::Result<()> { watch_namespace.get_api::>(&client), watcher::Config::default(), ) + // The discovery ConfigMap advertises the addresses that the listener operator + // publishes on the role Listener, so a reconciliation must run once they appear + // or change. + .owns( + watch_namespace.get_api::>(&client), + watcher::Config::default(), + ) .graceful_shutdown_on(sigterm_watcher.handle()) .run( zk_controller::reconcile_zk, diff --git a/rust/operator-binary/src/zk_controller.rs b/rust/operator-binary/src/zk_controller.rs index 07296bae..bff6b670 100644 --- a/rust/operator-binary/src/zk_controller.rs +++ b/rust/operator-binary/src/zk_controller.rs @@ -3,7 +3,7 @@ use std::{hash::Hasher, str::FromStr, sync::Arc}; use const_format::concatcp; use fnv::FnvHasher; -use snafu::{OptionExt, ResultExt, Snafu}; +use snafu::{ResultExt, Snafu}; use stackable_operator::{ cli::OperatorEnvironmentOptions, cluster_resources::ClusterResourceApplyStrategy, @@ -32,10 +32,7 @@ use strum::{EnumDiscriminants, IntoStaticStr}; use crate::{ OPERATOR_NAME, ObjectRef, crd::v1alpha1, - zk_controller::{ - build::resource::discovery, - validate::{operator_name, product_name}, - }, + zk_controller::validate::{operator_name, product_name}, }; pub(crate) mod build; @@ -76,24 +73,6 @@ pub enum Error { source: stackable_operator::cluster_resources::Error, }, - #[snafu(display("object is missing metadata to build owner reference"))] - ObjectMissingMetadataForOwnerRef { - source: stackable_operator::builder::meta::Error, - }, - - #[snafu(display( - "no role Listener was applied; the discovery ConfigMap is derived from the applied role Listener" - ))] - NoRoleListener, - - #[snafu(display("failed to build discovery ConfigMap"))] - BuildDiscoveryConfig { source: discovery::Error }, - - #[snafu(display("failed to apply discovery ConfigMap"))] - ApplyDiscoveryConfig { - source: stackable_operator::cluster_resources::Error, - }, - #[snafu(display("failed to update status"))] ApplyStatus { source: stackable_operator::client::Error, @@ -103,11 +82,6 @@ pub enum Error { DeleteOrphans { source: stackable_operator::cluster_resources::Error, }, - - #[snafu(display("failed to build object meta data"))] - ObjectMeta { - source: stackable_operator::builder::meta::Error, - }, } impl ReconcilerError for Error { @@ -122,25 +96,22 @@ impl ReconcilerError for Error { Error::ValidateCluster { .. } => None, Error::BuildResources { .. } => None, Error::ApplyResource { .. } => None, - Error::ObjectMissingMetadataForOwnerRef { .. } => None, - Error::NoRoleListener => None, - Error::BuildDiscoveryConfig { .. } => None, - Error::ApplyDiscoveryConfig { .. } => None, Error::ApplyStatus { .. } => None, Error::DeleteOrphans { .. } => None, - Error::ObjectMeta { .. } => None, } } } /// Every Kubernetes resource produced by the client-free [`build()`](build::build) step. -/// -/// The discovery `ConfigMap` is deliberately absent — see [`build()`](build::build). pub struct KubernetesResources { pub stateful_sets: Vec, pub services: Vec, pub listeners: Vec, pub config_maps: Vec, + /// The discovery `ConfigMap`, which is only built once the role Listener publishes its + /// addresses (see [`build()`](build::build)). It is kept apart from the role group + /// `config_maps` because the cluster status carries a hash of it. + pub maybe_discovery_config_map: Option, pub pod_disruption_budgets: Vec, pub service_accounts: Vec, pub role_bindings: Vec, @@ -205,17 +176,12 @@ pub async fn reconcile_zk( .context(ApplyResourceSnafu)?; } - // ZooKeeper has a single role Listener; the applied object feeds the discovery ConfigMap. - let mut applied_role_listener: Option = None; for listener in resources.listeners { - applied_role_listener = Some( - cluster_resources - .add(client, listener) - .await - .context(ApplyResourceSnafu)?, - ); + cluster_resources + .add(client, listener) + .await + .context(ApplyResourceSnafu)?; } - let role_listener = applied_role_listener.context(NoRoleListenerSnafu)?; for config_map in resources.config_maps { cluster_resources @@ -247,16 +213,14 @@ pub async fn reconcile_zk( // We don't /need/ stability, but it's still nice to avoid spurious changes where possible. let mut discovery_hash = FnvHasher::with_key(0); - let discovery_cm = - discovery::build_discovery_configmap(&validated_cluster, ZK_CONTROLLER_NAME, role_listener) - .context(BuildDiscoveryConfigSnafu)?; - - let discovery_cm = cluster_resources - .add(client, discovery_cm) - .await - .context(ApplyDiscoveryConfigSnafu)?; - if let Some(generation) = discovery_cm.metadata.resource_version { - discovery_hash.write(generation.as_bytes()) + if let Some(discovery_cm) = resources.maybe_discovery_config_map { + let discovery_cm = cluster_resources + .add(client, discovery_cm) + .await + .context(ApplyResourceSnafu)?; + if let Some(generation) = discovery_cm.metadata.resource_version { + discovery_hash.write(generation.as_bytes()) + } } let cluster_operation_cond_builder = @@ -361,6 +325,7 @@ pub(crate) mod test_support { zk, &DereferencedObjects { authentication_classes: DereferencedAuthenticationClasses::new_for_tests(), + maybe_role_listener: None, }, &operator_environment(), ) diff --git a/rust/operator-binary/src/zk_controller/build.rs b/rust/operator-binary/src/zk_controller/build.rs index f467fe40..0011e5c8 100644 --- a/rust/operator-binary/src/zk_controller/build.rs +++ b/rust/operator-binary/src/zk_controller/build.rs @@ -21,9 +21,9 @@ use stackable_operator::{ use crate::{ crd::ZookeeperRole, zk_controller::{ - KubernetesResources, + KubernetesResources, ZK_CONTROLLER_NAME, build::resource::{ - config_map, + config_map, discovery, listener::build_role_listener, pdb::build_pdb, rbac::{build_role_binding, build_service_account}, @@ -63,6 +63,9 @@ pub enum Error { source: statefulset::Error, rolegroup: RoleGroupName, }, + + #[snafu(display("failed to build the discovery ConfigMap"))] + DiscoveryConfigMap { source: discovery::Error }, } /// Builds every Kubernetes resource for the given validated cluster. @@ -72,9 +75,12 @@ pub enum Error { /// failures only. `cluster_info` is static cluster metadata (not a client call), consumed by the /// role-group ConfigMap builder. /// -/// The discovery `ConfigMap` is deliberately absent: it is built from the *applied* role -/// [`Listener`](stackable_operator::crd::listener::v1alpha1::Listener)'s ingress addresses, so it -/// is assembled in the reconcile step after the Listener has been applied, not here. +/// The discovery `ConfigMap` is only built once the role +/// [`Listener`](stackable_operator::crd::listener::v1alpha1::Listener) publishes ingress addresses. +/// Those are dereferenced and validated into +/// [`ValidatedCluster::discovery_addresses`](ValidatedCluster#structfield.discovery_addresses) +/// before this step runs, so the ConfigMap is absent during the reconciliation that first creates +/// the Listener, and built by the one that the Listener watch triggers afterwards. pub fn build( cluster: &ValidatedCluster, cluster_info: &KubernetesClusterInfo, @@ -127,11 +133,21 @@ pub fn build( let listeners = vec![build_role_listener(cluster, &zk_role)]; + let maybe_discovery_config_map = cluster + .discovery_addresses + .as_ref() + .map(|listener_addresses| { + discovery::build_discovery_configmap(cluster, ZK_CONTROLLER_NAME, listener_addresses) + }) + .transpose() + .context(DiscoveryConfigMapSnafu)?; + Ok(KubernetesResources { stateful_sets, services, listeners, config_maps, + maybe_discovery_config_map, pod_disruption_budgets, service_accounts: vec![build_service_account(cluster)], role_bindings: vec![build_role_binding(cluster)], @@ -214,7 +230,7 @@ mod tests { "simple-zookeeper-server-secondary-metrics", ] ); - // One ConfigMap per role group; the discovery ConfigMap is absent — see `build()`. + // One ConfigMap per role group. assert_eq!( sorted_names(&resources.config_maps), [ @@ -222,6 +238,8 @@ mod tests { "simple-zookeeper-server-secondary", ] ); + // The fixture has no role Listener yet, so the discovery ConfigMap is absent (see `build()`). + assert!(resources.maybe_discovery_config_map.is_none()); // The single role-level Listener for the one ZooKeeper role (`server`). assert_eq!( sorted_names(&resources.listeners), diff --git a/rust/operator-binary/src/zk_controller/build/resource/discovery.rs b/rust/operator-binary/src/zk_controller/build/resource/discovery.rs index d1742478..b0c2bbc6 100644 --- a/rust/operator-binary/src/zk_controller/build/resource/discovery.rs +++ b/rust/operator-binary/src/zk_controller/build/resource/discovery.rs @@ -1,11 +1,10 @@ -use std::{collections::BTreeSet, num::TryFromIntError, str::FromStr}; +use std::str::FromStr; -use snafu::{OptionExt, ResultExt, Snafu}; +use snafu::{ResultExt, Snafu}; use stackable_operator::{ builder::{configmap::ConfigMapBuilder, meta::ObjectMetaBuilder}, - crd::listener, k8s_openapi::api::core::v1::ConfigMap, - kube::{Resource, runtime::reflector::ObjectRef}, + kube::Resource, v2::{ HasName, HasUid, NameIsValidLabelValue, builder::meta::ownerreference_from_resource, @@ -15,7 +14,8 @@ use stackable_operator::{ }; use crate::{ - crd::{ZOOKEEPER_SERVER_PORT_NAME, ZookeeperRole, security::ZookeeperSecurity}, + crd::{ZookeeperRole, security::ZookeeperSecurity}, + listener_addresses::ListenerAddresses, zk_controller::{ build::PLACEHOLDER_DISCOVERY_ROLE_GROUP, validate::{ValidatedCluster, operator_name, product_name}, @@ -30,23 +30,6 @@ pub enum Error { #[snafu(display("chroot path {} was relative (must be absolute)", chroot))] RelativeChroot { chroot: String }, - #[snafu(display("{listener} does not have a port with the name {port_name:?}"))] - PortNotFound { - port_name: String, - listener: ObjectRef, - }, - - #[snafu(display("expected an unsigned 16-bit port, got {port_number}"))] - InvalidPort { - source: TryFromIntError, - port_number: i32, - }, - - #[snafu(display("{listener} has no ingress addresses"))] - NoListenerIngressAddresses { - listener: ObjectRef, - }, - #[snafu(display("failed to build ConfigMap"))] BuildConfigMap { source: stackable_operator::builder::configmap::Error, @@ -61,14 +44,14 @@ pub enum Error { pub fn build_discovery_configmap( validated_cluster: &ValidatedCluster, controller_name: &str, - listener: listener::v1alpha1::Listener, + listener_addresses: &ListenerAddresses, ) -> Result { build_discovery_configmap_for_owner( validated_cluster, &validated_cluster.namespace, controller_name, &validated_cluster.product_version, - listener, + listener_addresses, None, &validated_cluster.cluster_config.zookeeper_security, ) @@ -83,7 +66,7 @@ pub fn build_discovery_configmap( pub fn build_znode_discovery_configmap( validated_znode: &ValidatedZnode, controller_name: &str, - listener: listener::v1alpha1::Listener, + listener_addresses: &ListenerAddresses, chroot: &str, ) -> Result { build_discovery_configmap_for_owner( @@ -91,7 +74,7 @@ pub fn build_znode_discovery_configmap( &validated_znode.namespace, controller_name, &validated_znode.product_version, - listener, + listener_addresses, Some(chroot), &validated_znode.zookeeper_security, ) @@ -108,7 +91,7 @@ fn build_discovery_configmap_for_owner( namespace: impl Into, controller_name: &str, product_version: &ProductVersion, - listener: listener::v1alpha1::Listener, + listener_addresses: &ListenerAddresses, chroot: Option<&str>, zookeeper_security: &ZookeeperSecurity, ) -> Result { @@ -121,16 +104,10 @@ fn build_discovery_configmap_for_owner( .expect("the controller name is a valid label value"); let role_group_name = PLACEHOLDER_DISCOVERY_ROLE_GROUP.clone(); - let listener_addresses = listener_addresses(&listener, ZOOKEEPER_SERVER_PORT_NAME)?; - // Write a connection string of the format that Java ZooKeeper client expects: // "{host1}:{port1},{host2:port2},.../{chroot}" // See https://zookeeper.apache.org/doc/current/apidocs/zookeeper-server/org/apache/zookeeper/ZooKeeper.html#ZooKeeper-java.lang.String-int-org.apache.zookeeper.Watcher- - let listener_addresses = listener_addresses - .into_iter() - .map(|(host, port)| format!("{host}:{port}")) - .collect::>() - .join(","); + let listener_addresses = listener_addresses.to_connection_string(); let mut conn_str = listener_addresses.clone(); if let Some(chroot) = chroot { if !chroot.starts_with('/') { @@ -166,123 +143,3 @@ fn build_discovery_configmap_for_owner( .build() .context(BuildConfigMapSnafu) } - -/// Lists all listener address and port number pairs for a given `port_name` for Pods participating in the [`Listener`][1] -/// -/// This returns pairs of `(Address, Port)`, where address could be a hostname or IP address of a node, clusterIP or external -/// load balancer depending on the Service type. -/// -/// ## Errors -/// -/// An error will be returned if there is no address found for the `port_name`. -/// -/// [1]: listener::v1alpha1::Listener -// TODO (@NickLarsenNZ): Move this to stackable-operator, so it can be used as listener.addresses_for_port(port_name) -fn listener_addresses( - listener: &listener::v1alpha1::Listener, - port_name: &str, -) -> Result + use<>> { - // Get addresses port pairs for addresses that have a port with the name that matches the one we are interested in - let address_port_pairs = listener - .status - .as_ref() - .and_then(|listener_status| listener_status.ingress_addresses.as_ref()) - .context(NoListenerIngressAddressesSnafu { listener })? - .iter() - // Filter the addresses that have the port we are interested in (they likely all have it though) - .filter_map(|listener_ingress| { - Some(listener_ingress.address.clone()).zip(listener_ingress.ports.get(port_name)) - }) - // Convert the port from i32 to u16 - .map(|(listener_address, &port_number)| { - let port_number: u16 = port_number - .try_into() - .context(InvalidPortSnafu { port_number })?; - Ok((listener_address, port_number)) - }) - .collect::, _>>()?; - - // An empty list is considered an error - match address_port_pairs.is_empty() { - true => PortNotFoundSnafu { - port_name, - listener, - } - .fail(), - false => Ok(address_port_pairs), - } -} - -#[cfg(test)] -mod tests { - use std::collections::BTreeMap; - - use stackable_operator::{ - crd::listener::v1alpha1::{ - AddressType, Listener, ListenerIngress, ListenerSpec, ListenerStatus, - }, - k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta, - }; - - use super::*; - - fn listener(ingress_addresses: Option>) -> Listener { - Listener { - metadata: ObjectMeta { - name: Some("test-listener".to_owned()), - ..ObjectMeta::default() - }, - spec: ListenerSpec::default(), - status: Some(ListenerStatus { - service_name: None, - ingress_addresses, - node_ports: None, - }), - } - } - - fn ingress(port: i32) -> ListenerIngress { - ListenerIngress { - address: "node-0".to_owned(), - address_type: AddressType::Hostname, - ports: BTreeMap::from([(ZOOKEEPER_SERVER_PORT_NAME.to_owned(), port)]), - } - } - - #[test] - fn listener_addresses_returns_host_port_pairs() { - let listener = listener(Some(vec![ingress(2181)])); - let pairs: Vec<_> = listener_addresses(&listener, ZOOKEEPER_SERVER_PORT_NAME) - .expect("addresses") - .into_iter() - .collect(); - assert_eq!(pairs, vec![("node-0".to_owned(), 2181u16)]); - } - - #[test] - fn listener_addresses_without_ingress_is_error() { - assert!(matches!( - listener_addresses(&listener(None), ZOOKEEPER_SERVER_PORT_NAME), - Err(Error::NoListenerIngressAddresses { .. }) - )); - } - - #[test] - fn listener_addresses_missing_port_name_is_error() { - let listener = listener(Some(vec![ingress(2181)])); - assert!(matches!( - listener_addresses(&listener, "does-not-exist"), - Err(Error::PortNotFound { .. }) - )); - } - - #[test] - fn listener_addresses_port_out_of_u16_range_is_error() { - // A port number that does not fit into a u16 must be rejected. - let listener = listener(Some(vec![ingress(70_000)])); - assert!(matches!( - listener_addresses(&listener, ZOOKEEPER_SERVER_PORT_NAME), - Err(Error::InvalidPort { .. }) - )); - } -} diff --git a/rust/operator-binary/src/zk_controller/dereference.rs b/rust/operator-binary/src/zk_controller/dereference.rs index 38585d77..ce80dee7 100644 --- a/rust/operator-binary/src/zk_controller/dereference.rs +++ b/rust/operator-binary/src/zk_controller/dereference.rs @@ -6,17 +6,40 @@ //! validate step. use snafu::{ResultExt, Snafu}; -use stackable_operator::client::Client; +use stackable_operator::{ + client::Client, + crd::listener, + v2::{ + controller_utils::{get_cluster_name, get_namespace}, + types::{kubernetes::NamespaceName, operator::ClusterName}, + }, +}; use crate::crd::{ + ZookeeperRole, authentication::{self, DereferencedAuthenticationClasses}, - v1alpha1, + role_listener_name, v1alpha1, }; #[derive(Snafu, Debug)] pub enum Error { #[snafu(display("failed to fetch authentication classes"))] FetchAuthenticationClasses { source: authentication::Error }, + + #[snafu(display("failed to get the cluster name"))] + GetClusterName { + source: stackable_operator::v2::controller_utils::Error, + }, + + #[snafu(display("failed to get the cluster namespace"))] + GetNamespace { + source: stackable_operator::v2::controller_utils::Error, + }, + + #[snafu(display("failed to fetch the role Listener"))] + FetchRoleListener { + source: stackable_operator::client::Error, + }, } type Result = std::result::Result; @@ -25,6 +48,13 @@ type Result = std::result::Result; /// not yet validated. pub struct DereferencedObjects { pub authentication_classes: DereferencedAuthenticationClasses, + + /// The role Listener as created by an earlier reconciliation, if it exists already. + /// + /// The discovery ConfigMap advertises the addresses that the listener operator publishes on + /// this object, so it can only be built once the Listener exists and carries them. The + /// controller watches Listeners, so a reconciliation is triggered as soon as that happens. + pub maybe_role_listener: Option, } /// Fetches all Kubernetes objects referenced from the [`v1alpha1::ZookeeperCluster`] spec. @@ -32,6 +62,9 @@ pub async fn dereference( client: &Client, zk: &v1alpha1::ZookeeperCluster, ) -> Result { + let cluster_name = get_cluster_name(zk).context(GetClusterNameSnafu)?; + let namespace = get_namespace(zk).context(GetNamespaceSnafu)?; + let authentication_classes = DereferencedAuthenticationClasses::fetch_references( client, &zk.spec.cluster_config.authentication, @@ -39,7 +72,23 @@ pub async fn dereference( .await .context(FetchAuthenticationClassesSnafu)?; + let maybe_role_listener = fetch_role_listener(client, &cluster_name, &namespace).await?; + Ok(DereferencedObjects { authentication_classes, + maybe_role_listener, }) } + +async fn fetch_role_listener( + client: &Client, + cluster_name: &ClusterName, + namespace: &NamespaceName, +) -> Result> { + let listener_name = role_listener_name(cluster_name.as_ref(), &ZookeeperRole::Server); + + client + .get_opt(listener_name.as_ref(), namespace.as_ref()) + .await + .context(FetchRoleListenerSnafu) +} diff --git a/rust/operator-binary/src/zk_controller/validate.rs b/rust/operator-binary/src/zk_controller/validate.rs index 93300a46..506dd683 100644 --- a/rust/operator-binary/src/zk_controller/validate.rs +++ b/rust/operator-binary/src/zk_controller/validate.rs @@ -51,11 +51,12 @@ use strum::IntoEnumIterator; use crate::{ crd::{ - APP_NAME, CONTAINER_IMAGE_BASE_NAME, OPERATOR_NAME, ZookeeperRole, ZookeeperServerRoleType, - authentication, + APP_NAME, CONTAINER_IMAGE_BASE_NAME, OPERATOR_NAME, ZOOKEEPER_SERVER_PORT_NAME, + ZookeeperRole, ZookeeperServerRoleType, authentication, security::ZookeeperSecurity, v1alpha1::{self, ZookeeperConfig, ZookeeperConfigOverrides, ZookeeperServerRoleConfig}, }, + listener_addresses::{self, ListenerAddresses, listener_addresses}, zk_controller::{ZK_CONTROLLER_NAME, dereference::DereferencedObjects}, }; @@ -117,6 +118,9 @@ pub enum Error { "the Vector agent is enabled but no Vector aggregator discovery ConfigMap name is set" ))] MissingVectorAggregatorConfigMapName, + + #[snafu(display("failed to read the addresses published by the role Listener"))] + ReadRoleListenerAddresses { source: listener_addresses::Error }, } type Result = std::result::Result; @@ -242,6 +246,12 @@ pub struct ValidatedCluster { /// Object overrides applied to the cluster's resources, carried so the apply step does not reach /// into the raw [`v1alpha1::ZookeeperCluster`]. pub object_overrides: ObjectOverrides, + /// The client addresses published by the role Listener, which the discovery ConfigMap + /// advertises. + /// + /// `None` until the listener operator has published them, in which case the discovery + /// ConfigMap is skipped and built by the reconciliation that the Listener watch triggers. + pub discovery_addresses: Option, } // Placeholder product version used for labels on PVC templates, which cannot be modified once @@ -264,6 +274,7 @@ impl ValidatedCluster { >, cluster_operation: ClusterOperation, object_overrides: ObjectOverrides, + discovery_addresses: Option, ) -> Self { Self { metadata: ObjectMeta { @@ -282,6 +293,7 @@ impl ValidatedCluster { role_group_configs, cluster_operation, object_overrides, + discovery_addresses, } } @@ -508,6 +520,16 @@ pub fn validate( pdb: common.pod_disruption_budget.clone(), }; + // The role Listener does not exist during the very first reconciliation, and carries no + // addresses until the listener operator has published them. + let discovery_addresses = dereferenced_objects + .maybe_role_listener + .as_ref() + .map(|listener| listener_addresses(listener, ZOOKEEPER_SERVER_PORT_NAME)) + .transpose() + .context(ReadRoleListenerAddressesSnafu)? + .flatten(); + Ok(ValidatedCluster::new( name, namespace, @@ -522,6 +544,7 @@ pub fn validate( role_group_configs, zk.spec.cluster_operation.clone(), zk.spec.object_overrides.clone(), + discovery_addresses, )) } diff --git a/rust/operator-binary/src/znode_controller.rs b/rust/operator-binary/src/znode_controller.rs index c0905988..f61f0057 100644 --- a/rust/operator-binary/src/znode_controller.rs +++ b/rust/operator-binary/src/znode_controller.rs @@ -25,7 +25,11 @@ use tracing::{debug, info}; use crate::{ APP_NAME, OPERATOR_NAME, - crd::{ZookeeperRole, role_listener_name, security::ZookeeperSecurity, v1alpha1}, + crd::{ + ZOOKEEPER_SERVER_PORT_NAME, ZookeeperRole, role_listener_name, security::ZookeeperSecurity, + v1alpha1, + }, + listener_addresses::{self, listener_addresses}, zk_controller::build::resource::discovery::{self, build_znode_discovery_configmap}, }; @@ -85,6 +89,14 @@ pub enum Error { znode_path: String, }, + #[snafu(display("failed to read the addresses published by the ZooKeeper role Listener"))] + ReadListenerAddresses { source: listener_addresses::Error }, + + #[snafu(display("{listener} has not published any addresses yet"))] + NoListenerAddresses { + listener: ObjectRef, + }, + #[snafu(display("failed to build discovery information"))] BuildDiscoveryConfigMap { source: discovery::Error }, @@ -150,6 +162,8 @@ impl ReconcilerError for Error { Error::NoZkFqdn { zk } => Some(zk.clone().erase()), Error::EnsureZnode { zk, .. } => Some(zk.clone().erase()), Error::EnsureZnodeMissing { zk, .. } => Some(zk.clone().erase()), + Error::ReadListenerAddresses { .. } => None, + Error::NoListenerAddresses { listener } => Some(listener.clone().erase()), Error::BuildDiscoveryConfigMap { .. } => None, Error::ApplyDiscoveryConfigMap { cm, .. } => Some(cm.clone().erase()), Error::ApplyStatus { .. } => None, @@ -303,10 +317,16 @@ async fn reconcile_apply( zk: ObjectRef::from_obj(&zk), })?; + let listener_addresses = listener_addresses(&listener, ZOOKEEPER_SERVER_PORT_NAME) + .context(ReadListenerAddressesSnafu)? + .with_context(|| NoListenerAddressesSnafu { + listener: ObjectRef::from_obj(&listener), + })?; + let discovery_cm = build_znode_discovery_configmap( validated_znode, ZNODE_CONTROLLER_NAME, - listener, + &listener_addresses, znode_path, ) .context(BuildDiscoveryConfigMapSnafu)?; From ce3b38545b10ae8468afb41815e04aa6c9aa8bc3 Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Wed, 5 Aug 2026 16:32:29 +0200 Subject: [PATCH 02/12] refactor: Extract the apply step into an Applier reconcile_zk built the ClusterResources itself and applied every collection with its own inline loop, so the driver carried the apply order, the orphan deletion and the resource specific error variants. Extract all of that into an Applier, following the airflow and hbase operators. KubernetesResources is now marked as either Prepared or Applied, which makes it impossible to derive the cluster status from resources that were only built. apply() destructures the resource set exhaustively, so a new field fails to compile here instead of silently never being applied. Unlike the sibling operators, the apply module is declared in zk_controller.rs itself, so the Applier is imported without `self` to avoid a name collision with the module declaration. --- rust/operator-binary/src/zk_controller.rs | 128 +++++----------- .../src/zk_controller/apply.rs | 141 ++++++++++++++++++ .../src/zk_controller/build.rs | 7 +- 3 files changed, 181 insertions(+), 95 deletions(-) create mode 100644 rust/operator-binary/src/zk_controller/apply.rs diff --git a/rust/operator-binary/src/zk_controller.rs b/rust/operator-binary/src/zk_controller.rs index bff6b670..a2da1c2a 100644 --- a/rust/operator-binary/src/zk_controller.rs +++ b/rust/operator-binary/src/zk_controller.rs @@ -1,5 +1,5 @@ //! Ensures that `Pod`s are configured and running for each [`v1alpha1::ZookeeperCluster`] -use std::{hash::Hasher, str::FromStr, sync::Arc}; +use std::{hash::Hasher, marker::PhantomData, sync::Arc}; use const_format::concatcp; use fnv::FnvHasher; @@ -25,16 +25,16 @@ use stackable_operator::{ compute_conditions, operations::ClusterOperationsConditionBuilder, statefulset::StatefulSetConditionBuilder, }, - v2::{cluster_resources::cluster_resources_new, types::operator::ControllerName}, }; use strum::{EnumDiscriminants, IntoStaticStr}; use crate::{ OPERATOR_NAME, ObjectRef, crd::v1alpha1, - zk_controller::validate::{operator_name, product_name}, + zk_controller::apply::Applier, }; +pub(crate) mod apply; pub(crate) mod build; mod dereference; pub(crate) mod validate; @@ -68,20 +68,13 @@ 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 apply the Kubernetes resources"))] + ApplyResources { source: apply::Error }, #[snafu(display("failed to update status"))] ApplyStatus { source: stackable_operator::client::Error, }, - - #[snafu(display("failed to delete orphaned resources"))] - DeleteOrphans { - source: stackable_operator::cluster_resources::Error, - }, } impl ReconcilerError for Error { @@ -95,15 +88,24 @@ impl ReconcilerError for Error { Error::Dereference { .. } => None, Error::ValidateCluster { .. } => None, Error::BuildResources { .. } => None, - Error::ApplyResource { .. } => None, + Error::ApplyResources { .. } => None, Error::ApplyStatus { .. } => None, - Error::DeleteOrphans { .. } => None, } } } +/// 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 client-free [`build()`](build::build) step. -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. +pub struct KubernetesResources { pub stateful_sets: Vec, pub services: Vec, pub listeners: Vec, @@ -115,6 +117,7 @@ pub struct KubernetesResources { pub pod_disruption_budgets: Vec, pub service_accounts: Vec, pub role_bindings: Vec, + pub status: PhantomData, } pub async fn reconcile_zk( @@ -138,89 +141,34 @@ pub async fn reconcile_zk( validate::validate(zk, &dereferenced_objects, &ctx.operator_environment) .context(ValidateClusterSnafu)?; - // Names are derived from compile-time constants. - let mut cluster_resources = cluster_resources_new( - &product_name(), - &operator_name(), - &ControllerName::from_str(ZK_CONTROLLER_NAME) - .expect("ZK_CONTROLLER_NAME should be a valid controller name"), - &validated_cluster.name, - &validated_cluster.namespace, - &validated_cluster.uid, - ClusterResourceApplyStrategy::from(&validated_cluster.cluster_operation), - &validated_cluster.object_overrides, - ); - + // build (no client required) let resources = build::build(&validated_cluster, &client.kubernetes_cluster_info) .context(BuildResourcesSnafu)?; - let mut ss_cond_builder = StatefulSetConditionBuilder::default(); - - 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)?; - } - - for listener in resources.listeners { - cluster_resources - .add(client, 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)?; - } + // apply (client required) + let applied = Applier::new( + client, + &validated_cluster, + ClusterResourceApplyStrategy::from(&validated_cluster.cluster_operation), + &validated_cluster.object_overrides, + ) + .apply(resources) + .await + .context(ApplyResourcesSnafu)?; - // Note: The StatefulSet needs to be applied after all ConfigMaps and Secrets it mounts - // to prevent unnecessary Pod restarts. - // See https://github.com/stackabletech/commons-operator/issues/111 for details. - for statefulset in resources.stateful_sets { - ss_cond_builder.add( - cluster_resources - .add(client, statefulset) - .await - .context(ApplyResourceSnafu)?, - ); + let mut ss_cond_builder = StatefulSetConditionBuilder::default(); + for stateful_set in applied.stateful_sets { + ss_cond_builder.add(stateful_set); } // std's SipHasher is deprecated, and DefaultHasher is unstable across Rust releases. // We don't /need/ stability, but it's still nice to avoid spurious changes where possible. let mut discovery_hash = FnvHasher::with_key(0); - if let Some(discovery_cm) = resources.maybe_discovery_config_map { - let discovery_cm = cluster_resources - .add(client, discovery_cm) - .await - .context(ApplyResourceSnafu)?; - if let Some(generation) = discovery_cm.metadata.resource_version { - discovery_hash.write(generation.as_bytes()) - } + if let Some(discovery_cm) = applied.maybe_discovery_config_map + && let Some(generation) = discovery_cm.metadata.resource_version + { + discovery_hash.write(generation.as_bytes()) } let cluster_operation_cond_builder = @@ -233,10 +181,6 @@ pub async fn reconcile_zk( conditions: compute_conditions(zk, &[&ss_cond_builder, &cluster_operation_cond_builder]), }; - cluster_resources - .delete_orphaned_resources(client) - .await - .context(DeleteOrphansSnafu)?; client .apply_patch_status(OPERATOR_NAME, zk, &status) .await diff --git a/rust/operator-binary/src/zk_controller/apply.rs b/rust/operator-binary/src/zk_controller/apply.rs new file mode 100644 index 00000000..94d34d0c --- /dev/null +++ b/rust/operator-binary/src/zk_controller/apply.rs @@ -0,0 +1,141 @@ +//! The apply step in the ZookeeperCluster controller. + +use std::{marker::PhantomData, str::FromStr}; + +use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + client::Client, + cluster_resources::{ClusterResource, ClusterResourceApplyStrategy, ClusterResources}, + deep_merger::ObjectOverrides, + v2::{cluster_resources::cluster_resources_new, types::operator::ControllerName}, +}; +use strum::{EnumDiscriminants, IntoStaticStr}; + +use crate::zk_controller::{ + Applied, KubernetesResources, Prepared, ZK_CONTROLLER_NAME, + validate::{ValidatedCluster, 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> { + // Names are derived from compile-time constants. + let cluster_resources = cluster_resources_new( + &product_name(), + &operator_name(), + &ControllerName::from_str(ZK_CONTROLLER_NAME) + .expect("ZK_CONTROLLER_NAME should be a valid controller name"), + &cluster.name, + &cluster.namespace, + &cluster.uid, + apply_strategy, + object_overrides, + ); + + Applier { + client, + cluster_resources, + } + } + + /// Applies the given Kubernetes resources and marks them as applied. + 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, + maybe_discovery_config_map, + 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, see 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 maybe_discovery_config_map = match maybe_discovery_config_map { + Some(config_map) => Some(self.add_resource(config_map).await?), + None => None, + }; + 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, + maybe_discovery_config_map, + 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 { + applied_resources.push(self.add_resource(resource).await?); + } + + Ok(applied_resources) + } + + async fn add_resource(&mut self, resource: T) -> Result { + self.cluster_resources + .add(self.client, resource) + .await + .context(ApplyResourceSnafu) + } +} diff --git a/rust/operator-binary/src/zk_controller/build.rs b/rust/operator-binary/src/zk_controller/build.rs index 0011e5c8..7fcc5235 100644 --- a/rust/operator-binary/src/zk_controller/build.rs +++ b/rust/operator-binary/src/zk_controller/build.rs @@ -9,7 +9,7 @@ //! remaining submodules ([`command`], [`graceful_shutdown`], [`jvm`], //! [`properties`]) produce fragments that those resource builders assemble. -use std::str::FromStr; +use std::{marker::PhantomData, str::FromStr}; use snafu::{ResultExt, Snafu}; use stackable_operator::{ @@ -21,7 +21,7 @@ use stackable_operator::{ use crate::{ crd::ZookeeperRole, zk_controller::{ - KubernetesResources, ZK_CONTROLLER_NAME, + KubernetesResources, Prepared, ZK_CONTROLLER_NAME, build::resource::{ config_map, discovery, listener::build_role_listener, @@ -84,7 +84,7 @@ pub enum Error { pub fn build( cluster: &ValidatedCluster, cluster_info: &KubernetesClusterInfo, -) -> Result { +) -> Result, Error> { let mut stateful_sets = vec![]; let mut services = vec![]; let mut config_maps = vec![]; @@ -151,6 +151,7 @@ pub fn build( pod_disruption_budgets, service_accounts: vec![build_service_account(cluster)], role_bindings: vec![build_role_binding(cluster)], + status: PhantomData, }) } From fd484b5827383fa1a5d9acd55f24a23fdc1b1905 Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Wed, 5 Aug 2026 16:51:14 +0200 Subject: [PATCH 03/12] refactor: Extract the update status step The cluster conditions and the discovery hash were computed inline at the end of reconcile_zk, from resources that happened to be in scope. Move both into an update_status step, following the airflow and hbase operators. It takes KubernetesResources, so the type system proves the status is derived from resources that were actually applied rather than merely built. The discovery hash, which the sibling operators do not have, becomes a private helper next to it. reconcile_zk is now the dereference, validate, build, apply and update_status pipeline and nothing else. --- rust/operator-binary/src/zk_controller.rs | 53 +++-------- .../src/zk_controller/update_status.rs | 89 +++++++++++++++++++ 2 files changed, 102 insertions(+), 40 deletions(-) create mode 100644 rust/operator-binary/src/zk_controller/update_status.rs diff --git a/rust/operator-binary/src/zk_controller.rs b/rust/operator-binary/src/zk_controller.rs index a2da1c2a..9a116ad5 100644 --- a/rust/operator-binary/src/zk_controller.rs +++ b/rust/operator-binary/src/zk_controller.rs @@ -1,8 +1,11 @@ //! Ensures that `Pod`s are configured and running for each [`v1alpha1::ZookeeperCluster`] -use std::{hash::Hasher, marker::PhantomData, sync::Arc}; +//! +//! This is the controller driver: it runs the +//! `dereference -> validate -> build -> apply -> update_status` pipeline, with each step living +//! in its own submodule. +use std::{marker::PhantomData, sync::Arc}; use const_format::concatcp; -use fnv::FnvHasher; use snafu::{ResultExt, Snafu}; use stackable_operator::{ cli::OperatorEnvironmentOptions, @@ -21,22 +24,19 @@ use stackable_operator::{ }, logging::controller::ReconcilerError, shared::time::Duration, - status::condition::{ - compute_conditions, operations::ClusterOperationsConditionBuilder, - statefulset::StatefulSetConditionBuilder, - }, }; use strum::{EnumDiscriminants, IntoStaticStr}; use crate::{ OPERATOR_NAME, ObjectRef, crd::v1alpha1, - zk_controller::apply::Applier, + zk_controller::{apply::Applier, update_status::update_status}, }; pub(crate) mod apply; pub(crate) mod build; mod dereference; +mod update_status; pub(crate) mod validate; pub const ZK_CONTROLLER_NAME: &str = "zookeepercluster"; @@ -71,10 +71,8 @@ pub enum Error { #[snafu(display("failed to apply the Kubernetes resources"))] ApplyResources { source: apply::Error }, - #[snafu(display("failed to update status"))] - ApplyStatus { - source: stackable_operator::client::Error, - }, + #[snafu(display("failed to update the cluster status"))] + UpdateStatus { source: update_status::Error }, } impl ReconcilerError for Error { @@ -89,7 +87,7 @@ impl ReconcilerError for Error { Error::ValidateCluster { .. } => None, Error::BuildResources { .. } => None, Error::ApplyResources { .. } => None, - Error::ApplyStatus { .. } => None, + Error::UpdateStatus { .. } => None, } } } @@ -156,35 +154,10 @@ pub async fn reconcile_zk( .await .context(ApplyResourcesSnafu)?; - let mut ss_cond_builder = StatefulSetConditionBuilder::default(); - for stateful_set in applied.stateful_sets { - ss_cond_builder.add(stateful_set); - } - - // std's SipHasher is deprecated, and DefaultHasher is unstable across Rust releases. - // We don't /need/ stability, but it's still nice to avoid spurious changes where possible. - let mut discovery_hash = FnvHasher::with_key(0); - - if let Some(discovery_cm) = applied.maybe_discovery_config_map - && let Some(generation) = discovery_cm.metadata.resource_version - { - discovery_hash.write(generation.as_bytes()) - } - - let cluster_operation_cond_builder = - ClusterOperationsConditionBuilder::new(&zk.spec.cluster_operation); - - let status = v1alpha1::ZookeeperClusterStatus { - // Serialize as a string to discourage users from trying to parse the value, - // and to keep things flexible if we end up changing the hasher at some point. - discovery_hash: Some(discovery_hash.finish().to_string()), - conditions: compute_conditions(zk, &[&ss_cond_builder, &cluster_operation_cond_builder]), - }; - - client - .apply_patch_status(OPERATOR_NAME, zk, &status) + // update_status (client required) + update_status(client, zk, &applied) .await - .context(ApplyStatusSnafu)?; + .context(UpdateStatusSnafu)?; Ok(controller::Action::await_change()) } diff --git a/rust/operator-binary/src/zk_controller/update_status.rs b/rust/operator-binary/src/zk_controller/update_status.rs new file mode 100644 index 00000000..f45e1b21 --- /dev/null +++ b/rust/operator-binary/src/zk_controller/update_status.rs @@ -0,0 +1,89 @@ +//! The update_status step in the ZookeeperCluster controller. + +use std::hash::Hasher; + +use fnv::FnvHasher; +use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + client::Client, + status::condition::{ + compute_conditions, operations::ClusterOperationsConditionBuilder, + statefulset::StatefulSetConditionBuilder, + }, +}; +use strum::{EnumDiscriminants, IntoStaticStr}; + +use crate::{ + OPERATOR_NAME, + crd::v1alpha1, + zk_controller::{Applied, KubernetesResources}, +}; + +#[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::ZookeeperCluster`]. +/// +/// Takes [`KubernetesResources`] so the type system proves that the status derives from +/// applied resources, not merely built ones. +pub async fn update_status( + client: &Client, + zk: &v1alpha1::ZookeeperCluster, + applied: &KubernetesResources, +) -> Result<()> { + let mut stateful_set_condition_builder = StatefulSetConditionBuilder::default(); + for stateful_set in &applied.stateful_sets { + stateful_set_condition_builder.add(stateful_set.clone()); + } + + let cluster_operation_cond_builder = + ClusterOperationsConditionBuilder::new(&zk.spec.cluster_operation); + + let status = v1alpha1::ZookeeperClusterStatus { + discovery_hash: Some(discovery_hash(applied)), + conditions: compute_conditions( + zk, + &[ + &stateful_set_condition_builder, + &cluster_operation_cond_builder, + ], + ), + }; + + client + .apply_patch_status(OPERATOR_NAME, zk, &status) + .await + .context(ApplyStatusSnafu)?; + + Ok(()) +} + +/// Hashes the resource version of the applied discovery ConfigMap, so that clients can tell when +/// the published connection details changed. +/// +/// The hash covers nothing while the discovery ConfigMap is absent, which is the case until the +/// role Listener publishes its addresses. +fn discovery_hash(applied: &KubernetesResources) -> String { + // std's SipHasher is deprecated, and DefaultHasher is unstable across Rust releases. + // We don't /need/ stability, but it's still nice to avoid spurious changes where possible. + let mut discovery_hash = FnvHasher::with_key(0); + + if let Some(discovery_config_map) = &applied.maybe_discovery_config_map + && let Some(resource_version) = &discovery_config_map.metadata.resource_version + { + discovery_hash.write(resource_version.as_bytes()) + } + + // Serialize as a string to discourage users from trying to parse the value, + // and to keep things flexible if we end up changing the hasher at some point. + discovery_hash.finish().to_string() +} From 7c6ae4ed619801a2723b72bae4cc47afe6a5da11 Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Wed, 5 Aug 2026 16:58:52 +0200 Subject: [PATCH 04/12] refactor: Extract build and apply steps in the znode controller reconcile_apply created the ClusterResources, talked to ZooKeeper, fetched the role Listener, built the discovery ConfigMap and applied it, all inline, so the ZookeeperZnode controller was the only one left without a pipeline. Give it the same dereference, validate, build and apply structure as the ZookeeperCluster controller. The Listener fetch moves into the dereference step, where a missing Listener stays a non error so it can never block finalizer removal, and its addresses become a validated field on ValidatedZnode. Creating the znode inside the ZooKeeper ensemble lives in the apply step as a free function, next to the Applier, because it is a client side effect that the client free build() cannot perform. The discovery ConfigMap builders move to a shared discovery module, so the znode controller no longer reaches into the cluster controller's build tree. Unlike the cluster controller, the resources carry no Prepared or Applied marker: the ZookeeperZnode has no cluster conditions, so there is no status step the marker could protect. --- .../build/resource => }/discovery.rs | 16 +- rust/operator-binary/src/main.rs | 1 + .../src/zk_controller/build.rs | 7 +- .../src/zk_controller/build/resource/mod.rs | 1 - rust/operator-binary/src/znode_controller.rs | 150 +++++------------- .../src/znode_controller/apply.rs | 131 +++++++++++++++ .../src/znode_controller/build.rs | 33 ++++ .../src/znode_controller/dereference.rs | 53 ++++++- .../src/znode_controller/validate.rs | 30 +++- 9 files changed, 294 insertions(+), 128 deletions(-) rename rust/operator-binary/src/{zk_controller/build/resource => }/discovery.rs (88%) create mode 100644 rust/operator-binary/src/znode_controller/apply.rs create mode 100644 rust/operator-binary/src/znode_controller/build.rs diff --git a/rust/operator-binary/src/zk_controller/build/resource/discovery.rs b/rust/operator-binary/src/discovery.rs similarity index 88% rename from rust/operator-binary/src/zk_controller/build/resource/discovery.rs rename to rust/operator-binary/src/discovery.rs index b0c2bbc6..a8e1c0a6 100644 --- a/rust/operator-binary/src/zk_controller/build/resource/discovery.rs +++ b/rust/operator-binary/src/discovery.rs @@ -1,3 +1,8 @@ +//! Builders for the discovery ConfigMaps, which advertise how to connect to a ZooKeeper ensemble. +//! +//! Shared by the build steps of both controllers: the ZookeeperCluster controller publishes the +//! whole ensemble, the ZookeeperZnode controller publishes the same ensemble narrowed to a chroot. + use std::str::FromStr; use snafu::{ResultExt, Snafu}; @@ -9,20 +14,21 @@ use stackable_operator::{ HasName, HasUid, NameIsValidLabelValue, builder::meta::ownerreference_from_resource, kvp::label::recommended_labels, - types::operator::{ControllerName, ProductVersion}, + types::operator::{ControllerName, ProductVersion, RoleGroupName}, }, }; use crate::{ crd::{ZookeeperRole, security::ZookeeperSecurity}, listener_addresses::ListenerAddresses, - zk_controller::{ - build::PLACEHOLDER_DISCOVERY_ROLE_GROUP, - validate::{ValidatedCluster, operator_name, product_name}, - }, + zk_controller::validate::{ValidatedCluster, operator_name, product_name}, znode_controller::validate::ValidatedZnode, }; +// Placeholder role-group name used for the recommended labels of the role-level discovery +// `ConfigMap` (which is not tied to a single role group). +stackable_operator::constant!(PLACEHOLDER_DISCOVERY_ROLE_GROUP: RoleGroupName = "discovery"); + type Result = std::result::Result; #[derive(Snafu, Debug)] diff --git a/rust/operator-binary/src/main.rs b/rust/operator-binary/src/main.rs index ab935650..c8babfeb 100644 --- a/rust/operator-binary/src/main.rs +++ b/rust/operator-binary/src/main.rs @@ -41,6 +41,7 @@ use crate::{ }; pub mod crd; +mod discovery; mod listener_addresses; mod webhooks; mod zk_controller; diff --git a/rust/operator-binary/src/zk_controller/build.rs b/rust/operator-binary/src/zk_controller/build.rs index 7fcc5235..e2af711b 100644 --- a/rust/operator-binary/src/zk_controller/build.rs +++ b/rust/operator-binary/src/zk_controller/build.rs @@ -20,10 +20,11 @@ use stackable_operator::{ use crate::{ crd::ZookeeperRole, + discovery, zk_controller::{ KubernetesResources, Prepared, ZK_CONTROLLER_NAME, build::resource::{ - config_map, discovery, + config_map, listener::build_role_listener, pdb::build_pdb, rbac::{build_role_binding, build_service_account}, @@ -36,10 +37,6 @@ use crate::{ }, }; -// Placeholder role-group name used for the recommended labels of the role-level discovery -// `ConfigMap` (which is not tied to a single role group). -stackable_operator::constant!(pub(crate) PLACEHOLDER_DISCOVERY_ROLE_GROUP: RoleGroupName = "discovery"); - // Placeholder role-group name used for the recommended labels of the role-level `Listener` // (which is not tied to a single role group). stackable_operator::constant!(pub(crate) NONE_ROLE_GROUP_NAME: RoleGroupName = "none"); diff --git a/rust/operator-binary/src/zk_controller/build/resource/mod.rs b/rust/operator-binary/src/zk_controller/build/resource/mod.rs index 7f57f617..6b846018 100644 --- a/rust/operator-binary/src/zk_controller/build/resource/mod.rs +++ b/rust/operator-binary/src/zk_controller/build/resource/mod.rs @@ -2,7 +2,6 @@ //! into complete Kubernetes resources. pub mod config_map; -pub mod discovery; pub mod listener; pub mod pdb; pub mod rbac; diff --git a/rust/operator-binary/src/znode_controller.rs b/rust/operator-binary/src/znode_controller.rs index f61f0057..831487c5 100644 --- a/rust/operator-binary/src/znode_controller.rs +++ b/rust/operator-binary/src/znode_controller.rs @@ -1,17 +1,20 @@ //! Reconciles state for ZooKeeper znodes between Kubernetes [`v1alpha1::ZookeeperZnode`] objects and the ZooKeeper cluster //! //! See [`v1alpha1::ZookeeperZnode`] for more details. +//! +//! This is the controller driver: it runs the `dereference -> validate -> build -> apply` +//! pipeline, with each step living in its own submodule. There is no update_status step, because +//! the only status the ZookeeperZnode carries (the znode path) is written before the finalizer +//! runs. use std::{borrow::Cow, convert::Infallible, sync::Arc}; use const_format::concatcp; use snafu::{OptionExt, ResultExt, Snafu}; use stackable_operator::{ cli::OperatorEnvironmentOptions, - cluster_resources::{ClusterResourceApplyStrategy, ClusterResources}, - crd::listener, + cluster_resources::ClusterResourceApplyStrategy, k8s_openapi::api::core::v1::ConfigMap, kube::{ - Resource, ResourceExt, api::ObjectMeta, core::{DeserializeGuard, DynamicObject, error_boundary}, runtime::{controller, finalizer, reflector::ObjectRef}, @@ -24,15 +27,13 @@ use strum::{EnumDiscriminants, IntoStaticStr}; use tracing::{debug, info}; use crate::{ - APP_NAME, OPERATOR_NAME, - crd::{ - ZOOKEEPER_SERVER_PORT_NAME, ZookeeperRole, role_listener_name, security::ZookeeperSecurity, - v1alpha1, - }, - listener_addresses::{self, listener_addresses}, - zk_controller::build::resource::discovery::{self, build_znode_discovery_configmap}, + OPERATOR_NAME, + crd::{security::ZookeeperSecurity, v1alpha1}, + znode_controller::apply::{Applier, ensure_znode_exists}, }; +pub(crate) mod apply; +pub(crate) mod build; mod dereference; pub(crate) mod validate; @@ -64,24 +65,11 @@ pub enum Error { ))] ObjectMissingMetadata, - #[snafu(display("could not find server role service for {zk:?}"))] - FindZkSvc { - source: stackable_operator::client::Error, - zk: ObjectRef, - }, - #[snafu(display("failed to calculate FQDN for {zk:?}"))] NoZkFqdn { zk: ObjectRef, }, - #[snafu(display("failed to ensure that ZNode {znode_path:?} exists in {zk:?}"))] - EnsureZnode { - source: znode_mgmt::Error, - zk: ObjectRef, - znode_path: String, - }, - #[snafu(display("failed to ensure that ZNode {znode_path:?} is missing from {zk:?}"))] EnsureZnodeMissing { source: znode_mgmt::Error, @@ -89,22 +77,11 @@ pub enum Error { znode_path: String, }, - #[snafu(display("failed to read the addresses published by the ZooKeeper role Listener"))] - ReadListenerAddresses { source: listener_addresses::Error }, - - #[snafu(display("{listener} has not published any addresses yet"))] - NoListenerAddresses { - listener: ObjectRef, - }, - - #[snafu(display("failed to build discovery information"))] - BuildDiscoveryConfigMap { source: discovery::Error }, + #[snafu(display("failed to build the Kubernetes resources"))] + BuildResources { source: build::Error }, - #[snafu(display("failed to save discovery information to {cm:?}"))] - ApplyDiscoveryConfigMap { - source: stackable_operator::cluster_resources::Error, - cm: ObjectRef, - }, + #[snafu(display("failed to apply the Kubernetes resources"))] + ApplyResources { source: apply::Error }, #[snafu(display("failed to update status"))] ApplyStatus { @@ -115,14 +92,6 @@ pub enum Error { Finalizer { source: finalizer::Error, }, - - #[snafu(display("failed to delete orphaned resources"))] - DeleteOrphans { - source: stackable_operator::cluster_resources::Error, - }, - - #[snafu(display("object has no namespace"))] - ObjectHasNoNamespace, } type Result = std::result::Result; @@ -158,22 +127,24 @@ impl ReconcilerError for Error { Error::Dereference { .. } => None, Error::ValidateCluster { .. } => None, Error::ObjectMissingMetadata => None, - Error::FindZkSvc { zk, .. } => Some(zk.clone().erase()), Error::NoZkFqdn { zk } => Some(zk.clone().erase()), - Error::EnsureZnode { zk, .. } => Some(zk.clone().erase()), Error::EnsureZnodeMissing { zk, .. } => Some(zk.clone().erase()), - Error::ReadListenerAddresses { .. } => None, - Error::NoListenerAddresses { listener } => Some(listener.clone().erase()), - Error::BuildDiscoveryConfigMap { .. } => None, - Error::ApplyDiscoveryConfigMap { cm, .. } => Some(cm.clone().erase()), + Error::BuildResources { .. } => None, + Error::ApplyResources { .. } => None, Error::ApplyStatus { .. } => None, Error::Finalizer { .. } => None, - Error::DeleteOrphans { .. } => None, - Error::ObjectHasNoNamespace => None, } } } +/// Every Kubernetes resource produced by the client-free [`build()`](build::build) step. +/// +/// The znode path inside the ZooKeeper ensemble is not a Kubernetes object, so it is absent here +/// and created by the apply step instead. +pub struct KubernetesResources { + pub discovery_config_maps: Vec, +} + pub async fn reconcile_znode( znode: Arc>, ctx: Arc, @@ -274,23 +245,8 @@ async fn reconcile_apply( zk: v1alpha1::ZookeeperCluster, znode_path: &str, ) -> Result { - // Infallible: `ValidatedZnode`'s object reference always contains name, namespace and uid - // (set unconditionally during the validate step), which is all `ClusterResources::new` - // requires. - let mut cluster_resources = ClusterResources::new( - APP_NAME, - OPERATOR_NAME, - ZNODE_CONTROLLER_NAME, - &validated_znode.object_ref(&()), - ClusterResourceApplyStrategy::from(&validated_znode.cluster_operation), - &validated_znode.object_overrides, - ) - .expect( - "ClusterResources should be created because the ValidatedZnode's object reference \ - always contains name, namespace and uid", - ); - - znode_mgmt::ensure_znode_exists( + // The znode must exist in the ZooKeeper ensemble before the discovery ConfigMap advertises it. + ensure_znode_exists( &zk_mgmt_addr( &zk, &validated_znode.zookeeper_security, @@ -299,48 +255,22 @@ async fn reconcile_apply( znode_path, ) .await - .with_context(|_| EnsureZnodeSnafu { - zk: ObjectRef::from_obj(&zk), - znode_path, - })?; + .context(ApplyResourcesSnafu)?; - let listener = client - .get::( - role_listener_name(&zk.name_any(), &ZookeeperRole::Server).as_ref(), - zk.metadata - .namespace - .as_deref() - .context(ObjectHasNoNamespaceSnafu)?, - ) - .await - .context(FindZkSvcSnafu { - zk: ObjectRef::from_obj(&zk), - })?; - - let listener_addresses = listener_addresses(&listener, ZOOKEEPER_SERVER_PORT_NAME) - .context(ReadListenerAddressesSnafu)? - .with_context(|| NoListenerAddressesSnafu { - listener: ObjectRef::from_obj(&listener), - })?; - - let discovery_cm = build_znode_discovery_configmap( + // build (no client required) + let resources = build::build(validated_znode, znode_path).context(BuildResourcesSnafu)?; + + // apply (client required) + Applier::new( + client, validated_znode, - ZNODE_CONTROLLER_NAME, - &listener_addresses, - znode_path, + ClusterResourceApplyStrategy::from(&validated_znode.cluster_operation), + &validated_znode.object_overrides, ) - .context(BuildDiscoveryConfigMapSnafu)?; - - let obj_ref = ObjectRef::from_obj(&discovery_cm); - cluster_resources - .add(client, discovery_cm) - .await - .with_context(|_| ApplyDiscoveryConfigMapSnafu { cm: obj_ref })?; - - cluster_resources - .delete_orphaned_resources(client) - .await - .context(DeleteOrphansSnafu)?; + .apply(resources) + .await + .context(ApplyResourcesSnafu)?; + Ok(controller::Action::await_change()) } diff --git a/rust/operator-binary/src/znode_controller/apply.rs b/rust/operator-binary/src/znode_controller/apply.rs new file mode 100644 index 00000000..c96106e0 --- /dev/null +++ b/rust/operator-binary/src/znode_controller/apply.rs @@ -0,0 +1,131 @@ +//! The apply step in the ZookeeperZnode controller. + +use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + client::Client, + cluster_resources::{ClusterResource, ClusterResourceApplyStrategy, ClusterResources}, + deep_merger::ObjectOverrides, + kube::Resource, +}; + +use crate::{ + APP_NAME, OPERATOR_NAME, + znode_controller::{ + KubernetesResources, ZNODE_CONTROLLER_NAME, validate::ValidatedZnode, znode_mgmt, + }, +}; + +#[derive(Snafu, Debug)] +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, + }, + + #[snafu(display("failed to ensure that ZNode {znode_path:?} exists in {zk_mgmt_addr}"))] + EnsureZnode { + source: znode_mgmt::Error, + zk_mgmt_addr: String, + znode_path: String, + }, +} + +type Result = std::result::Result; + +/// Applier for the Kubernetes resource specifications produced by this controller. +/// +/// Unlike the ZookeeperCluster controller's applier, the resources are not marked as prepared or +/// applied: the ZookeeperZnode has no cluster conditions, so there is no status step that could +/// derive the status from resources that were never applied. +pub struct Applier<'a> { + client: &'a Client, + cluster_resources: ClusterResources<'a>, +} + +impl<'a> Applier<'a> { + pub fn new( + client: &'a Client, + znode: &ValidatedZnode, + apply_strategy: ClusterResourceApplyStrategy, + object_overrides: &'a ObjectOverrides, + ) -> Applier<'a> { + // Infallible: `ValidatedZnode`'s object reference always contains name, namespace and uid + // (set unconditionally during the validate step), which is all `ClusterResources::new` + // requires. + let cluster_resources = ClusterResources::new( + APP_NAME, + OPERATOR_NAME, + ZNODE_CONTROLLER_NAME, + &znode.object_ref(&()), + apply_strategy, + object_overrides, + ) + .expect( + "ClusterResources should be created because the ValidatedZnode's object reference \ + always contains name, namespace and uid", + ); + + Applier { + client, + cluster_resources, + } + } + + /// Applies the given Kubernetes 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 { + discovery_config_maps, + } = resources; + + let discovery_config_maps = self.add_resources(discovery_config_maps).await?; + + self.cluster_resources + .delete_orphaned_resources(self.client) + .await + .context(DeleteOrphanedResourcesSnafu)?; + + Ok(KubernetesResources { + discovery_config_maps, + }) + } + + async fn add_resources( + &mut self, + resources: Vec, + ) -> Result> { + let mut applied_resources = vec![]; + + for resource in resources { + applied_resources.push( + self.cluster_resources + .add(self.client, resource) + .await + .context(ApplyResourceSnafu)?, + ); + } + + Ok(applied_resources) + } +} + +/// Ensures that the znode exists in the ZooKeeper ensemble reachable at `zk_mgmt_addr`. +/// +/// The znode is a path inside ZooKeeper rather than a Kubernetes object, so it cannot be part of +/// the client-free `build()` step, and it is not tracked in +/// [`ClusterResources`](stackable_operator::cluster_resources::ClusterResources) either. It must +/// exist before the discovery ConfigMap advertises it to clients. +pub async fn ensure_znode_exists(zk_mgmt_addr: &str, znode_path: &str) -> Result<()> { + znode_mgmt::ensure_znode_exists(zk_mgmt_addr, znode_path) + .await + .with_context(|_| EnsureZnodeSnafu { + zk_mgmt_addr, + znode_path, + }) +} diff --git a/rust/operator-binary/src/znode_controller/build.rs b/rust/operator-binary/src/znode_controller/build.rs new file mode 100644 index 00000000..509b4adb --- /dev/null +++ b/rust/operator-binary/src/znode_controller/build.rs @@ -0,0 +1,33 @@ +//! The build step in the ZookeeperZnode controller. + +use snafu::{ResultExt, Snafu}; + +use crate::{ + discovery::{self, build_znode_discovery_configmap}, + znode_controller::{KubernetesResources, ZNODE_CONTROLLER_NAME, validate::ValidatedZnode}, +}; + +#[derive(Snafu, Debug)] +pub enum Error { + #[snafu(display("failed to build the discovery ConfigMap"))] + DiscoveryConfigMap { source: discovery::Error }, +} + +/// Builds every Kubernetes resource for the given validated znode. +/// +/// Does not need a Kubernetes client: the referenced cluster and the addresses published by its +/// role Listener are already dereferenced and validated by this point. The znode itself (a path +/// inside the ZooKeeper ensemble, not a Kubernetes object) is created by the apply step. +pub fn build(znode: &ValidatedZnode, znode_path: &str) -> Result { + let discovery_config_map = build_znode_discovery_configmap( + znode, + ZNODE_CONTROLLER_NAME, + &znode.discovery_addresses, + znode_path, + ) + .context(DiscoveryConfigMapSnafu)?; + + Ok(KubernetesResources { + discovery_config_maps: vec![discovery_config_map], + }) +} diff --git a/rust/operator-binary/src/znode_controller/dereference.rs b/rust/operator-binary/src/znode_controller/dereference.rs index d29ee402..1d0a4055 100644 --- a/rust/operator-binary/src/znode_controller/dereference.rs +++ b/rust/operator-binary/src/znode_controller/dereference.rs @@ -1,19 +1,21 @@ //! The dereference step in the ZookeeperZnode controller. //! //! Fetches the parent [`v1alpha1::ZookeeperCluster`] referenced by the znode's -//! `spec.clusterRef`, plus the [`DereferencedAuthenticationClasses`] of that cluster. Both Apply -//! and Cleanup paths in `reconcile_znode` share this output. Synchronous validation of the -//! fetched objects happens in the validate step. +//! `spec.clusterRef`, plus the [`DereferencedAuthenticationClasses`] and the role Listener of that +//! cluster. Both Apply and Cleanup paths in `reconcile_znode` share this output. Synchronous +//! validation of the fetched objects happens in the validate step. -use snafu::{ResultExt, Snafu}; +use snafu::{OptionExt, ResultExt, Snafu}; use stackable_operator::{ client::Client, - kube::{self, runtime::reflector::ObjectRef}, + crd::listener, + kube::{self, ResourceExt, runtime::reflector::ObjectRef}, }; use crate::crd::{ + ZookeeperRole, authentication::{self, DereferencedAuthenticationClasses}, - v1alpha1, + role_listener_name, v1alpha1, }; #[derive(Snafu, Debug)] @@ -35,6 +37,17 @@ pub enum Error { #[snafu(display("failed to fetch authentication classes"))] FetchAuthenticationClasses { source: authentication::Error }, + + #[snafu(display("{zk} has no namespace"))] + ZkHasNoNamespace { + zk: ObjectRef, + }, + + #[snafu(display("failed to fetch the role Listener of {zk}"))] + FetchRoleListener { + source: stackable_operator::client::Error, + zk: ObjectRef, + }, } type Result = std::result::Result; @@ -43,6 +56,13 @@ type Result = std::result::Result; pub struct DereferencedObjects { pub zk: v1alpha1::ZookeeperCluster, pub authentication_classes: DereferencedAuthenticationClasses, + + /// The role Listener of the referenced cluster, if it exists already. + /// + /// The znode's discovery ConfigMap advertises the addresses that the listener operator + /// publishes on it. The Cleanup path does not need it, so a missing Listener is not an error + /// here and never blocks finalizer removal. + pub maybe_role_listener: Option, } /// Fetches all Kubernetes objects referenced from the [`v1alpha1::ZookeeperZnode`] spec. @@ -59,12 +79,33 @@ pub async fn dereference( .await .context(FetchAuthenticationClassesSnafu)?; + let maybe_role_listener = fetch_role_listener(client, &zk).await?; + Ok(DereferencedObjects { zk, authentication_classes, + maybe_role_listener, }) } +async fn fetch_role_listener( + client: &Client, + zk: &v1alpha1::ZookeeperCluster, +) -> Result> { + let zk_ref = ObjectRef::from_obj(zk); + let namespace = zk + .metadata + .namespace + .as_deref() + .with_context(|| ZkHasNoNamespaceSnafu { zk: zk_ref.clone() })?; + let listener_name = role_listener_name(&zk.name_any(), &ZookeeperRole::Server); + + client + .get_opt(listener_name.as_ref(), namespace) + .await + .with_context(|_| FetchRoleListenerSnafu { zk: zk_ref }) +} + async fn find_zk_of_znode( client: &Client, znode: &v1alpha1::ZookeeperZnode, diff --git a/rust/operator-binary/src/znode_controller/validate.rs b/rust/operator-binary/src/znode_controller/validate.rs index 899b6cdb..570721c2 100644 --- a/rust/operator-binary/src/znode_controller/validate.rs +++ b/rust/operator-binary/src/znode_controller/validate.rs @@ -24,7 +24,11 @@ use stackable_operator::{ }; use crate::{ - crd::{CONTAINER_IMAGE_BASE_NAME, authentication, security::ZookeeperSecurity, v1alpha1}, + crd::{ + CONTAINER_IMAGE_BASE_NAME, ZOOKEEPER_SERVER_PORT_NAME, authentication, + security::ZookeeperSecurity, v1alpha1, + }, + listener_addresses::{self, ListenerAddresses, listener_addresses}, znode_controller::dereference::DereferencedObjects, }; @@ -62,6 +66,14 @@ pub enum Error { source: stackable_operator::v2::macros::attributed_string_type::Error, product_version: String, }, + + #[snafu(display("failed to read the addresses published by the ZooKeeper role Listener"))] + ReadRoleListenerAddresses { source: listener_addresses::Error }, + + #[snafu(display( + "the ZooKeeper role Listener does not exist yet, or has not published any addresses yet" + ))] + NoRoleListenerAddresses, } type Result = std::result::Result; @@ -92,6 +104,12 @@ pub struct ValidatedZnode { /// Object overrides applied to the znode's resources, carried so the apply step does not reach /// into the raw [`v1alpha1::ZookeeperZnode`]. pub object_overrides: ObjectOverrides, + /// The client addresses published by the referenced cluster's role Listener, which the znode's + /// discovery ConfigMap advertises. + /// + /// Unlike the cluster controller, the znode controller cannot produce anything without them, + /// so validation fails while they are missing and the reconciliation is retried. + pub discovery_addresses: ListenerAddresses, } impl HasName for ValidatedZnode { @@ -184,6 +202,15 @@ pub fn validate( } })?; + let discovery_addresses = dereferenced_objects + .maybe_role_listener + .as_ref() + .map(|listener| listener_addresses(listener, ZOOKEEPER_SERVER_PORT_NAME)) + .transpose() + .context(ReadRoleListenerAddressesSnafu)? + .flatten() + .context(NoRoleListenerAddressesSnafu)?; + Ok(ValidatedZnode { metadata: ObjectMeta { name: Some(name.clone()), @@ -198,5 +225,6 @@ pub fn validate( zookeeper_security, cluster_operation: dereferenced_objects.zk.spec.cluster_operation.clone(), object_overrides: znode.spec.object_overrides.clone(), + discovery_addresses, }) } From f2f2b0a53a8d970557a39a2c102210bc56489dc1 Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Wed, 5 Aug 2026 17:15:55 +0200 Subject: [PATCH 05/12] chore: adapted changelog --- CHANGELOG.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2845c927..5bf76d8c 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 ([#1060]). - BREAKING: The `servers` role is now required by the CRD. Previously a ZookeeperCluster without it was accepted by the API server but failed reconciliation ([#1060]). +- The reconciler now applies resources and derives the cluster status in discrete + apply and update_status steps for the `zk_controller` and `znode_controller` ([#1069]). [#1053]: https://github.com/stackabletech/zookeeper-operator/pull/1053 [#1060]: https://github.com/stackabletech/zookeeper-operator/pull/1060 [#1063]: https://github.com/stackabletech/zookeeper-operator/pull/1063 +[#1069]: https://github.com/stackabletech/zookeeper-operator/pull/1069 ## [26.7.0] - 2026-07-21 From 88043d38be39764666d11b9f6e816181ec601253 Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Wed, 12 Aug 2026 12:25:58 +0200 Subject: [PATCH 06/12] docs: fix rustdoc warnings in discovery and znode apply --- rust/operator-binary/src/discovery.rs | 4 ++-- rust/operator-binary/src/znode_controller/apply.rs | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/rust/operator-binary/src/discovery.rs b/rust/operator-binary/src/discovery.rs index a8e1c0a6..93403ee2 100644 --- a/rust/operator-binary/src/discovery.rs +++ b/rust/operator-binary/src/discovery.rs @@ -86,8 +86,8 @@ pub fn build_znode_discovery_configmap( ) } -/// Build a discovery [`ConfigMap`] containing ZooKeeper connection details from a -/// [`listener::v1alpha1::Listener`]. +/// Build a discovery [`ConfigMap`] containing ZooKeeper connection details from the +/// [`ListenerAddresses`] published by the role Listener. /// /// `owner` owns the ConfigMap (the [`ZookeeperCluster`](crate::crd::v1alpha1::ZookeeperCluster) for the cluster /// controller, or the [`ZookeeperZnode`](crate::crd::v1alpha1::ZookeeperZnode) for the znode controller) and diff --git a/rust/operator-binary/src/znode_controller/apply.rs b/rust/operator-binary/src/znode_controller/apply.rs index c96106e0..86c9d69d 100644 --- a/rust/operator-binary/src/znode_controller/apply.rs +++ b/rust/operator-binary/src/znode_controller/apply.rs @@ -119,7 +119,7 @@ impl<'a> Applier<'a> { /// /// The znode is a path inside ZooKeeper rather than a Kubernetes object, so it cannot be part of /// the client-free `build()` step, and it is not tracked in -/// [`ClusterResources`](stackable_operator::cluster_resources::ClusterResources) either. It must +/// [`ClusterResources`] either. It must /// exist before the discovery ConfigMap advertises it to clients. pub async fn ensure_znode_exists(zk_mgmt_addr: &str, znode_path: &str) -> Result<()> { znode_mgmt::ensure_znode_exists(zk_mgmt_addr, znode_path) From 17bc91455da7a647a4d9cef3b811e6443dbeff1d Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Wed, 12 Aug 2026 12:35:17 +0200 Subject: [PATCH 07/12] fix: Always build the discovery ConfigMap Skipping the discovery ConfigMap while the role Listener publishes no addresses let the apply step delete the already published one as an orphan, breaking consumers that mount it. Before the pipeline extraction, the missing addresses aborted the run before delete_orphaned_resources() ran, so this only surfaced now. Follow the kafka-operator pattern instead: always write the ConfigMap, with an empty ZOOKEEPER value while no address is known. The Listener watch triggers a new run that fills the value in. --- rust/operator-binary/src/discovery.rs | 9 +++++ .../operator-binary/src/listener_addresses.rs | 5 ++- rust/operator-binary/src/zk_controller.rs | 7 ++-- .../src/zk_controller/apply.rs | 9 ++--- .../src/zk_controller/build.rs | 36 ++++++++++--------- .../src/zk_controller/update_status.rs | 13 ++++--- .../src/zk_controller/validate.rs | 16 +++++---- 7 files changed, 54 insertions(+), 41 deletions(-) diff --git a/rust/operator-binary/src/discovery.rs b/rust/operator-binary/src/discovery.rs index 93403ee2..d19708e6 100644 --- a/rust/operator-binary/src/discovery.rs +++ b/rust/operator-binary/src/discovery.rs @@ -47,6 +47,15 @@ pub enum Error { /// /// The ConfigMap is owned by, and placed in the namespace of, the cluster. The image and security /// settings are taken from the [`ValidatedCluster`] rather than being passed in separately. +/// +/// The connection details are read from the addresses published by the role Listener (carried on +/// [`ValidatedCluster::discovery_addresses`](ValidatedCluster#structfield.discovery_addresses), +/// fetched in the dereference step), which only the listener operator writes. While no address +/// exists around the first reconciliations, which create the Listener in the first place, the +/// ConfigMap is still written, with an empty `ZOOKEEPER` 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 +/// published. pub fn build_discovery_configmap( validated_cluster: &ValidatedCluster, controller_name: &str, diff --git a/rust/operator-binary/src/listener_addresses.rs b/rust/operator-binary/src/listener_addresses.rs index 4eaa1e0d..2c12a0e3 100644 --- a/rust/operator-binary/src/listener_addresses.rs +++ b/rust/operator-binary/src/listener_addresses.rs @@ -30,7 +30,10 @@ type Result = std::result::Result; /// /// An address is a hostname or IP address of a node, a cluster IP or an external load balancer, /// depending on the Service type behind the Listener. -#[derive(Clone, Debug, PartialEq, Eq)] +/// +/// The [`Default`] value is empty, which renders as an empty connection string, see +/// [`build_discovery_configmap`](crate::discovery::build_discovery_configmap). +#[derive(Clone, Debug, Default, PartialEq, Eq)] pub struct ListenerAddresses(BTreeSet<(String, u16)>); impl ListenerAddresses { diff --git a/rust/operator-binary/src/zk_controller.rs b/rust/operator-binary/src/zk_controller.rs index 9a116ad5..f1f29801 100644 --- a/rust/operator-binary/src/zk_controller.rs +++ b/rust/operator-binary/src/zk_controller.rs @@ -108,10 +108,9 @@ pub struct KubernetesResources { pub services: Vec, pub listeners: Vec, pub config_maps: Vec, - /// The discovery `ConfigMap`, which is only built once the role Listener publishes its - /// addresses (see [`build()`](build::build)). It is kept apart from the role group - /// `config_maps` because the cluster status carries a hash of it. - pub maybe_discovery_config_map: Option, + /// The discovery `ConfigMap`, kept apart from the role group `config_maps` because the cluster + /// status carries a hash of it. + pub discovery_config_map: ConfigMap, pub pod_disruption_budgets: Vec, pub service_accounts: Vec, pub role_bindings: Vec, diff --git a/rust/operator-binary/src/zk_controller/apply.rs b/rust/operator-binary/src/zk_controller/apply.rs index 94d34d0c..9169e09a 100644 --- a/rust/operator-binary/src/zk_controller/apply.rs +++ b/rust/operator-binary/src/zk_controller/apply.rs @@ -79,7 +79,7 @@ impl<'a> Applier<'a> { services, listeners, config_maps, - maybe_discovery_config_map, + discovery_config_map, pod_disruption_budgets, service_accounts, role_bindings, @@ -94,10 +94,7 @@ impl<'a> Applier<'a> { let services = self.add_resources(services).await?; let listeners = self.add_resources(listeners).await?; let config_maps = self.add_resources(config_maps).await?; - let maybe_discovery_config_map = match maybe_discovery_config_map { - Some(config_map) => Some(self.add_resource(config_map).await?), - None => None, - }; + let discovery_config_map = self.add_resource(discovery_config_map).await?; let pod_disruption_budgets = self.add_resources(pod_disruption_budgets).await?; let stateful_sets = self.add_resources(stateful_sets).await?; @@ -111,7 +108,7 @@ impl<'a> Applier<'a> { services, listeners, config_maps, - maybe_discovery_config_map, + discovery_config_map, pod_disruption_budgets, service_accounts, role_bindings, diff --git a/rust/operator-binary/src/zk_controller/build.rs b/rust/operator-binary/src/zk_controller/build.rs index e2af711b..b762fc4d 100644 --- a/rust/operator-binary/src/zk_controller/build.rs +++ b/rust/operator-binary/src/zk_controller/build.rs @@ -72,12 +72,12 @@ pub enum Error { /// failures only. `cluster_info` is static cluster metadata (not a client call), consumed by the /// role-group ConfigMap builder. /// -/// The discovery `ConfigMap` is only built once the role -/// [`Listener`](stackable_operator::crd::listener::v1alpha1::Listener) publishes ingress addresses. -/// Those are dereferenced and validated into -/// [`ValidatedCluster::discovery_addresses`](ValidatedCluster#structfield.discovery_addresses) -/// before this step runs, so the ConfigMap is absent during the reconciliation that first creates -/// the Listener, and built by the one that the Listener watch triggers afterwards. +/// This includes the discovery `ConfigMap`, built from the addresses published by the role +/// [`Listener`](stackable_operator::crd::listener::v1alpha1::Listener) that were dereferenced and +/// validated into +/// [`ValidatedCluster::discovery_addresses`](ValidatedCluster#structfield.discovery_addresses); see +/// [`build_discovery_configmap`](discovery::build_discovery_configmap) for how its content depends +/// on them. pub fn build( cluster: &ValidatedCluster, cluster_info: &KubernetesClusterInfo, @@ -130,21 +130,19 @@ pub fn build( let listeners = vec![build_role_listener(cluster, &zk_role)]; - let maybe_discovery_config_map = cluster - .discovery_addresses - .as_ref() - .map(|listener_addresses| { - discovery::build_discovery_configmap(cluster, ZK_CONTROLLER_NAME, listener_addresses) - }) - .transpose() - .context(DiscoveryConfigMapSnafu)?; + let discovery_config_map = discovery::build_discovery_configmap( + cluster, + ZK_CONTROLLER_NAME, + &cluster.discovery_addresses, + ) + .context(DiscoveryConfigMapSnafu)?; Ok(KubernetesResources { stateful_sets, services, listeners, config_maps, - maybe_discovery_config_map, + discovery_config_map, pod_disruption_budgets, service_accounts: vec![build_service_account(cluster)], role_bindings: vec![build_role_binding(cluster)], @@ -236,8 +234,12 @@ mod tests { "simple-zookeeper-server-secondary", ] ); - // The fixture has no role Listener yet, so the discovery ConfigMap is absent (see `build()`). - assert!(resources.maybe_discovery_config_map.is_none()); + // The discovery ConfigMap is named after the cluster, and written even though the fixture + // has no role Listener publishing addresses yet (see `build()`). + assert_eq!( + resources.discovery_config_map.meta().name.as_deref(), + Some("simple-zookeeper") + ); // The single role-level Listener for the one ZooKeeper role (`server`). assert_eq!( sorted_names(&resources.listeners), diff --git a/rust/operator-binary/src/zk_controller/update_status.rs b/rust/operator-binary/src/zk_controller/update_status.rs index f45e1b21..69fa963e 100644 --- a/rust/operator-binary/src/zk_controller/update_status.rs +++ b/rust/operator-binary/src/zk_controller/update_status.rs @@ -6,6 +6,7 @@ use fnv::FnvHasher; use snafu::{ResultExt, Snafu}; use stackable_operator::{ client::Client, + k8s_openapi::api::core::v1::ConfigMap, status::condition::{ compute_conditions, operations::ClusterOperationsConditionBuilder, statefulset::StatefulSetConditionBuilder, @@ -49,7 +50,7 @@ pub async fn update_status( ClusterOperationsConditionBuilder::new(&zk.spec.cluster_operation); let status = v1alpha1::ZookeeperClusterStatus { - discovery_hash: Some(discovery_hash(applied)), + discovery_hash: Some(discovery_hash(&applied.discovery_config_map)), conditions: compute_conditions( zk, &[ @@ -70,16 +71,14 @@ pub async fn update_status( /// Hashes the resource version of the applied discovery ConfigMap, so that clients can tell when /// the published connection details changed. /// -/// The hash covers nothing while the discovery ConfigMap is absent, which is the case until the -/// role Listener publishes its addresses. -fn discovery_hash(applied: &KubernetesResources) -> String { +/// A ConfigMap that was never applied carries no resource version, in which case the hash covers +/// nothing. +fn discovery_hash(discovery_config_map: &ConfigMap) -> String { // std's SipHasher is deprecated, and DefaultHasher is unstable across Rust releases. // We don't /need/ stability, but it's still nice to avoid spurious changes where possible. let mut discovery_hash = FnvHasher::with_key(0); - if let Some(discovery_config_map) = &applied.maybe_discovery_config_map - && let Some(resource_version) = &discovery_config_map.metadata.resource_version - { + if let Some(resource_version) = &discovery_config_map.metadata.resource_version { discovery_hash.write(resource_version.as_bytes()) } diff --git a/rust/operator-binary/src/zk_controller/validate.rs b/rust/operator-binary/src/zk_controller/validate.rs index 506dd683..2eea33b3 100644 --- a/rust/operator-binary/src/zk_controller/validate.rs +++ b/rust/operator-binary/src/zk_controller/validate.rs @@ -249,9 +249,10 @@ pub struct ValidatedCluster { /// The client addresses published by the role Listener, which the discovery ConfigMap /// advertises. /// - /// `None` until the listener operator has published them, in which case the discovery - /// ConfigMap is skipped and built by the reconciliation that the Listener watch triggers. - pub discovery_addresses: Option, + /// Empty until the listener operator has published them, in which case the discovery ConfigMap + /// advertises an empty connection string until the reconciliation that the Listener watch + /// triggers fills it in. + pub discovery_addresses: ListenerAddresses, } // Placeholder product version used for labels on PVC templates, which cannot be modified once @@ -274,7 +275,7 @@ impl ValidatedCluster { >, cluster_operation: ClusterOperation, object_overrides: ObjectOverrides, - discovery_addresses: Option, + discovery_addresses: ListenerAddresses, ) -> Self { Self { metadata: ObjectMeta { @@ -521,14 +522,17 @@ pub fn validate( }; // The role Listener does not exist during the very first reconciliation, and carries no - // addresses until the listener operator has published them. + // addresses until the listener operator has published them. Both cases resolve to no addresses + // rather than an error, so that the discovery ConfigMap is still built (see + // [`build_discovery_configmap`](crate::discovery::build_discovery_configmap)). let discovery_addresses = dereferenced_objects .maybe_role_listener .as_ref() .map(|listener| listener_addresses(listener, ZOOKEEPER_SERVER_PORT_NAME)) .transpose() .context(ReadRoleListenerAddressesSnafu)? - .flatten(); + .flatten() + .unwrap_or_default(); Ok(ValidatedCluster::new( name, From 27f9e181ceeaaed38c02ea05367e83377ec0989f Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Wed, 12 Aug 2026 12:40:27 +0200 Subject: [PATCH 08/12] test: Cover the znode validate step's role Listener errors The znode controller had no tests at all. Add fixtures for it and cover both ways the role Listener can block validation: addresses published under an unexpected port name (ReadRoleListenerAddresses), and no addresses at all, either because the Listener does not exist yet or because the listener operator has not published its status (NoRoleListenerAddresses). The Listener fixtures move into a shared test_support module in listener_addresses, so the cluster controller's tests can use them too. --- .../operator-binary/src/listener_addresses.rs | 41 ++++++--- rust/operator-binary/src/zk_controller.rs | 2 +- rust/operator-binary/src/znode_controller.rs | 87 +++++++++++++++++++ .../src/znode_controller/validate.rs | 76 ++++++++++++++++ 4 files changed, 193 insertions(+), 13 deletions(-) diff --git a/rust/operator-binary/src/listener_addresses.rs b/rust/operator-binary/src/listener_addresses.rs index 2c12a0e3..fff6c18b 100644 --- a/rust/operator-binary/src/listener_addresses.rs +++ b/rust/operator-binary/src/listener_addresses.rs @@ -92,8 +92,9 @@ pub fn listener_addresses( } } +/// Shared helpers for building role [`Listener`](listener::v1alpha1::Listener) fixtures. #[cfg(test)] -mod tests { +pub(crate) mod test_support { use std::collections::BTreeMap; use stackable_operator::{ @@ -103,10 +104,9 @@ mod tests { k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta, }; - use super::*; - use crate::crd::ZOOKEEPER_SERVER_PORT_NAME; - - fn listener(ingress_addresses: Option>) -> Listener { + /// A role Listener publishing `ingress_addresses`. `None` is a Listener that the listener + /// operator has not written a status for yet. + pub fn role_listener(ingress_addresses: Option>) -> Listener { Listener { metadata: ObjectMeta { name: Some("test-listener".to_owned()), @@ -121,17 +121,33 @@ mod tests { } } - fn ingress(port: i32) -> ListenerIngress { + /// A single ingress address of a role Listener, publishing `port` under `port_name`. + pub fn ingress_address(address: &str, port_name: &str, port: i32) -> ListenerIngress { ListenerIngress { - address: "node-0".to_owned(), + address: address.to_owned(), address_type: AddressType::Hostname, - ports: BTreeMap::from([(ZOOKEEPER_SERVER_PORT_NAME.to_owned(), port)]), + ports: BTreeMap::from([(port_name.to_owned(), port)]), } } +} + +#[cfg(test)] +mod tests { + use stackable_operator::crd::listener::v1alpha1::ListenerIngress; + + use super::{ + test_support::{ingress_address, role_listener}, + *, + }; + use crate::crd::ZOOKEEPER_SERVER_PORT_NAME; + + fn ingress(port: i32) -> ListenerIngress { + ingress_address("node-0", ZOOKEEPER_SERVER_PORT_NAME, port) + } #[test] fn listener_addresses_returns_host_port_pairs() { - let listener = listener(Some(vec![ingress(2181)])); + let listener = role_listener(Some(vec![ingress(2181)])); let addresses = listener_addresses(&listener, ZOOKEEPER_SERVER_PORT_NAME) .expect("addresses") .expect("the listener publishes addresses"); @@ -141,14 +157,15 @@ mod tests { #[test] fn listener_addresses_without_ingress_is_not_ready_yet() { assert_eq!( - listener_addresses(&listener(None), ZOOKEEPER_SERVER_PORT_NAME).expect("addresses"), + listener_addresses(&role_listener(None), ZOOKEEPER_SERVER_PORT_NAME) + .expect("addresses"), None ); } #[test] fn listener_addresses_missing_port_name_is_error() { - let listener = listener(Some(vec![ingress(2181)])); + let listener = role_listener(Some(vec![ingress(2181)])); assert!(matches!( listener_addresses(&listener, "does-not-exist"), Err(Error::PortNotFound { .. }) @@ -158,7 +175,7 @@ mod tests { #[test] fn listener_addresses_port_out_of_u16_range_is_error() { // A port number that does not fit into a u16 must be rejected. - let listener = listener(Some(vec![ingress(70_000)])); + let listener = role_listener(Some(vec![ingress(70_000)])); assert!(matches!( listener_addresses(&listener, ZOOKEEPER_SERVER_PORT_NAME), Err(Error::InvalidPort { .. }) diff --git a/rust/operator-binary/src/zk_controller.rs b/rust/operator-binary/src/zk_controller.rs index f1f29801..cb984e52 100644 --- a/rust/operator-binary/src/zk_controller.rs +++ b/rust/operator-binary/src/zk_controller.rs @@ -224,7 +224,7 @@ pub(crate) mod test_support { } } - fn operator_environment() -> OperatorEnvironmentOptions { + pub fn operator_environment() -> OperatorEnvironmentOptions { OperatorEnvironmentOptions { operator_namespace: "stackable-operators".to_owned(), operator_service_name: "zookeeper-operator".to_owned(), diff --git a/rust/operator-binary/src/znode_controller.rs b/rust/operator-binary/src/znode_controller.rs index 831487c5..85172ebd 100644 --- a/rust/operator-binary/src/znode_controller.rs +++ b/rust/operator-binary/src/znode_controller.rs @@ -330,6 +330,93 @@ pub fn error_policy( controller::Action::requeue(*Duration::from_secs(5)) } +/// Shared helpers for building validated test znodes from minimal YAML fixtures. +#[cfg(test)] +pub(crate) mod test_support { + use stackable_operator::crd::listener; + + use crate::{ + crd::{authentication::DereferencedAuthenticationClasses, v1alpha1}, + zk_controller::test_support::{minimal_zk, operator_environment}, + znode_controller::{ + dereference::DereferencedObjects, + validate::{ValidatedZnode, validate}, + }, + }; + + /// Parses a minimal `ZookeeperZnode` test fixture, defaulting `namespace`/`uid` so the validate + /// step can build a [`ValidatedZnode`]. + pub fn minimal_znode(yaml: &str) -> v1alpha1::ZookeeperZnode { + let mut znode: v1alpha1::ZookeeperZnode = + serde_yaml::from_str(yaml).expect("invalid test ZookeeperZnode YAML"); + znode + .metadata + .namespace + .get_or_insert_with(|| "default".to_owned()); + znode + .metadata + .uid + .get_or_insert_with(|| "e5dbf9c2-d8b0-4c1e-9f4a-1d2e3f4a5b6c".to_owned()); + znode + } + + /// The `ZookeeperCluster` that the znode fixtures reference. The znode validate step reads the + /// image, security settings and cluster operation from it. + pub fn referenced_zk() -> v1alpha1::ZookeeperCluster { + minimal_zk( + r#" + apiVersion: zookeeper.stackable.tech/v1alpha1 + kind: ZookeeperCluster + metadata: + name: simple-zookeeper + spec: + image: + productVersion: "3.9.5" + servers: + roleGroups: + default: + replicas: 3 + "#, + ) + } + + /// Runs the real validate step against a minimal (auth-free) fixture and the referenced + /// cluster's role Listener, returning the result so tests can assert on validation errors. + pub fn try_validate( + znode: &v1alpha1::ZookeeperZnode, + maybe_role_listener: Option, + ) -> Result { + validate( + znode, + &DereferencedObjects { + zk: referenced_zk(), + authentication_classes: DereferencedAuthenticationClasses::new_for_tests(), + maybe_role_listener, + }, + &operator_environment(), + ) + } + + /// Runs the real validate step against a minimal (auth-free) fixture whose role Listener + /// publishes `node-0:2181`. + pub fn validated_znode(znode: &v1alpha1::ZookeeperZnode) -> ValidatedZnode { + use crate::{ + crd::ZOOKEEPER_SERVER_PORT_NAME, + listener_addresses::test_support::{ingress_address, role_listener}, + }; + + try_validate( + znode, + Some(role_listener(Some(vec![ingress_address( + "node-0", + ZOOKEEPER_SERVER_PORT_NAME, + 2181, + )]))), + ) + .expect("validate should succeed for the test fixture") + } +} + mod znode_mgmt { use std::{collections::VecDeque, net::SocketAddr}; diff --git a/rust/operator-binary/src/znode_controller/validate.rs b/rust/operator-binary/src/znode_controller/validate.rs index 570721c2..6b4f8aea 100644 --- a/rust/operator-binary/src/znode_controller/validate.rs +++ b/rust/operator-binary/src/znode_controller/validate.rs @@ -228,3 +228,79 @@ pub fn validate( discovery_addresses, }) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + listener_addresses::test_support::{ingress_address, role_listener}, + zk_controller::test_support::app_version_label, + znode_controller::test_support::{minimal_znode, try_validate, validated_znode}, + }; + + const ZNODE_YAML: &str = r#" + apiVersion: zookeeper.stackable.tech/v1alpha1 + kind: ZookeeperZnode + metadata: + name: simple-znode + spec: + clusterRef: + name: simple-zookeeper + "#; + + /// Locks the values the validate step derives from the znode itself and from the referenced + /// cluster, including the addresses that the znode's discovery ConfigMap advertises. + #[test] + fn validate_ok_derives_expected_values() { + let validated = validated_znode(&minimal_znode(ZNODE_YAML)); + + assert_eq!(validated.name, "simple-znode"); + assert_eq!(validated.namespace.to_string(), "default"); + assert_eq!( + validated.uid.to_string(), + "e5dbf9c2-d8b0-4c1e-9f4a-1d2e3f4a5b6c" + ); + // The product version comes from the referenced cluster, not from the znode. + assert_eq!( + validated.product_version.to_string(), + app_version_label("3.9.5") + ); + assert!(validated.zookeeper_security.tls_enabled()); + assert_eq!( + validated.discovery_addresses.to_connection_string(), + "node-0:2181" + ); + } + + /// The znode's discovery ConfigMap is the only resource this controller produces, so a role + /// Listener that publishes addresses the znode cannot use must fail validation rather than + /// advertise nothing. + #[test] + fn role_listener_without_the_expected_port_fails_validation() { + let listener = role_listener(Some(vec![ingress_address("node-0", "not-the-zk-port", 2181)])); + + assert!(matches!( + try_validate(&minimal_znode(ZNODE_YAML), Some(listener)), + Err(Error::ReadRoleListenerAddresses { .. }) + )); + } + + /// The znode controller runs on its own schedule, so it can observe the referenced cluster + /// before the cluster controller has created the role Listener at all. + #[test] + fn missing_role_listener_fails_validation() { + assert!(matches!( + try_validate(&minimal_znode(ZNODE_YAML), None), + Err(Error::NoRoleListenerAddresses) + )); + } + + /// The Listener exists, but the listener operator has not published its addresses yet. + #[test] + fn role_listener_without_addresses_fails_validation() { + assert!(matches!( + try_validate(&minimal_znode(ZNODE_YAML), Some(role_listener(None))), + Err(Error::NoRoleListenerAddresses) + )); + } +} From b26a4ec3bd5b82560a6fd683ad211d419de47b7c Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Wed, 12 Aug 2026 12:45:52 +0200 Subject: [PATCH 09/12] test: Cover the role Listener addresses in the cluster validate step Add a try_validate variant that takes a role Listener, and cover the three outcomes: addresses published under an unexpected port name fail with ReadRoleListenerAddresses, published addresses reach the validated cluster sorted, and a missing or unpublished Listener validates to no addresses rather than an error, which is what lets the build step still write the discovery ConfigMap. --- rust/operator-binary/src/zk_controller.rs | 17 +++-- .../src/zk_controller/validate.rs | 69 ++++++++++++++++++- 2 files changed, 80 insertions(+), 6 deletions(-) diff --git a/rust/operator-binary/src/zk_controller.rs b/rust/operator-binary/src/zk_controller.rs index cb984e52..f246b8d5 100644 --- a/rust/operator-binary/src/zk_controller.rs +++ b/rust/operator-binary/src/zk_controller.rs @@ -178,7 +178,7 @@ pub fn error_policy( #[cfg(test)] pub(crate) mod test_support { use stackable_operator::{ - cli::OperatorEnvironmentOptions, commons::networking::DomainName, + cli::OperatorEnvironmentOptions, commons::networking::DomainName, crd::listener, utils::cluster_info::KubernetesClusterInfo, }; @@ -232,16 +232,25 @@ pub(crate) mod test_support { } } - /// Runs the real validate step against a minimal (auth-free) fixture, returning the result so - /// tests can assert on validation errors. + /// Runs the real validate step against a minimal (auth-free) fixture whose role Listener does + /// not exist yet, returning the result so tests can assert on validation errors. pub fn try_validate( zk: &v1alpha1::ZookeeperCluster, + ) -> Result { + try_validate_with_role_listener(zk, None) + } + + /// Runs the real validate step against a minimal (auth-free) fixture and the given role + /// Listener, returning the result so tests can assert on validation errors. + pub fn try_validate_with_role_listener( + zk: &v1alpha1::ZookeeperCluster, + maybe_role_listener: Option, ) -> Result { validate( zk, &DereferencedObjects { authentication_classes: DereferencedAuthenticationClasses::new_for_tests(), - maybe_role_listener: None, + maybe_role_listener, }, &operator_environment(), ) diff --git a/rust/operator-binary/src/zk_controller/validate.rs b/rust/operator-binary/src/zk_controller/validate.rs index 2eea33b3..b423090c 100644 --- a/rust/operator-binary/src/zk_controller/validate.rs +++ b/rust/operator-binary/src/zk_controller/validate.rs @@ -604,10 +604,28 @@ mod tests { use stackable_operator::k8s_openapi::apimachinery::pkg::api::resource::Quantity; use super::*; - use crate::zk_controller::test_support::{ - app_version_label, minimal_zk, try_validate, validated_cluster, + use crate::{ + listener_addresses::test_support::{ingress_address, role_listener}, + zk_controller::test_support::{ + app_version_label, minimal_zk, try_validate, try_validate_with_role_listener, + validated_cluster, + }, }; + const MINIMAL_ZK_YAML: &str = r#" + apiVersion: zookeeper.stackable.tech/v1alpha1 + kind: ZookeeperCluster + metadata: + name: simple-zookeeper + spec: + image: + productVersion: "3.9.5" + servers: + roleGroups: + default: + replicas: 3 + "#; + /// Locks every value the validate step itself derives from the minimal fixture — so a /// validation regression fails here, with a validate-shaped message, instead of surfacing as /// a confusing build-test failure downstream. @@ -825,4 +843,51 @@ mod tests { let _: RoleName = role.into(); } } + + /// The `zk` port is a constant, so a role Listener that publishes addresses without it is a + /// fault rather than a transient state, and must fail validation. + #[test] + fn role_listener_without_the_expected_port_fails_validation() { + let zk = minimal_zk(MINIMAL_ZK_YAML); + let listener = role_listener(Some(vec![ingress_address("node-0", "not-the-zk-port", 2181)])); + + assert!(matches!( + try_validate_with_role_listener(&zk, Some(listener)), + Err(Error::ReadRoleListenerAddresses { .. }) + )); + } + + /// The published addresses reach the discovery ConfigMap through `discovery_addresses`. + #[test] + fn role_listener_addresses_are_carried_into_the_validated_cluster() { + let zk = minimal_zk(MINIMAL_ZK_YAML); + let listener = role_listener(Some(vec![ + ingress_address("node-1", ZOOKEEPER_SERVER_PORT_NAME, 2181), + ingress_address("node-0", ZOOKEEPER_SERVER_PORT_NAME, 2181), + ])); + + let validated = try_validate_with_role_listener(&zk, Some(listener)) + .expect("validate should succeed for the test fixture"); + + assert_eq!( + validated.discovery_addresses.to_connection_string(), + "node-0:2181,node-1:2181" + ); + } + + /// Unlike the znode controller, missing addresses are not an error here: the reconciliation + /// that creates the role Listener in the first place has to get as far as the apply step. + #[test] + fn missing_role_listener_addresses_validate_to_no_addresses() { + let zk = minimal_zk(MINIMAL_ZK_YAML); + + // The Listener does not exist yet. + let validated = validated_cluster(&zk); + assert_eq!(validated.discovery_addresses, ListenerAddresses::default()); + + // The Listener exists, but the listener operator has not published its status yet. + let validated = try_validate_with_role_listener(&zk, Some(role_listener(None))) + .expect("validate should succeed for the test fixture"); + assert_eq!(validated.discovery_addresses, ListenerAddresses::default()); + } } From 5df6ab22726a920d5b708385c3357e92d0ca6d9b Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Wed, 12 Aug 2026 12:46:10 +0200 Subject: [PATCH 10/12] test: Cover the discovery hash in the update status step --- .../src/zk_controller/update_status.rs | 32 +++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/rust/operator-binary/src/zk_controller/update_status.rs b/rust/operator-binary/src/zk_controller/update_status.rs index 69fa963e..455d34a8 100644 --- a/rust/operator-binary/src/zk_controller/update_status.rs +++ b/rust/operator-binary/src/zk_controller/update_status.rs @@ -86,3 +86,35 @@ fn discovery_hash(discovery_config_map: &ConfigMap) -> String { // and to keep things flexible if we end up changing the hasher at some point. discovery_hash.finish().to_string() } + +#[cfg(test)] +mod tests { + use stackable_operator::kube::api::ObjectMeta; + + use super::{ConfigMap, discovery_hash}; + + fn discovery_config_map(resource_version: Option<&str>) -> ConfigMap { + ConfigMap { + metadata: ObjectMeta { + name: Some("simple-zookeeper".to_owned()), + resource_version: resource_version.map(ToOwned::to_owned), + ..ObjectMeta::default() + }, + ..ConfigMap::default() + } + } + + /// The hash exists so that clients can tell when the published connection details changed, so + /// it must follow the discovery ConfigMap's resource version. + #[test] + fn discovery_hash_tracks_the_resource_version() { + let hash = discovery_hash(&discovery_config_map(Some("42"))); + + assert_ne!(hash, discovery_hash(&discovery_config_map(Some("43")))); + assert_eq!(hash, discovery_hash(&discovery_config_map(Some("42")))); + + // A ConfigMap that was never applied carries no resource version, in which case the hash + // covers nothing and stays at the hasher's initial state. + assert_ne!(hash, discovery_hash(&discovery_config_map(None))); + } +} From 13effc97cd203903e6b302e88b94db6a50e1f827 Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Wed, 12 Aug 2026 12:49:45 +0200 Subject: [PATCH 11/12] test: Cover the discovery ConfigMap builders Assert the ZOOKEEPER, ZOOKEEPER_HOSTS, ZOOKEEPER_CLIENT_PORT and ZOOKEEPER_CHROOT content for both the cluster and the znode variant, that only ZOOKEEPER carries the chroot, and that a relative chroot is rejected. The empty-address case is covered too, pinning that the ConfigMap is written even while the role Listener publishes nothing. The Listener fixtures now publish the secure client port, which is what the role Listener serves while the fixture keeps TLS enabled. --- rust/operator-binary/src/discovery.rs | 165 ++++++++++++++++++ .../src/zk_controller/validate.rs | 6 +- rust/operator-binary/src/znode_controller.rs | 5 +- .../src/znode_controller/validate.rs | 8 +- 4 files changed, 179 insertions(+), 5 deletions(-) diff --git a/rust/operator-binary/src/discovery.rs b/rust/operator-binary/src/discovery.rs index d19708e6..9b3a7d8b 100644 --- a/rust/operator-binary/src/discovery.rs +++ b/rust/operator-binary/src/discovery.rs @@ -158,3 +158,168 @@ fn build_discovery_configmap_for_owner( .build() .context(BuildConfigMapSnafu) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + crd::ZOOKEEPER_SERVER_PORT_NAME, + listener_addresses::{ + listener_addresses, + test_support::{ingress_address, role_listener}, + }, + zk_controller::{ + ZK_CONTROLLER_NAME, + test_support::{minimal_zk, try_validate_with_role_listener}, + }, + znode_controller::{ + ZNODE_CONTROLLER_NAME, + test_support::{minimal_znode, validated_znode}, + }, + }; + + const ZK_YAML: &str = r#" + apiVersion: zookeeper.stackable.tech/v1alpha1 + kind: ZookeeperCluster + metadata: + name: simple-zookeeper + spec: + image: + productVersion: "3.9.5" + servers: + roleGroups: + default: + replicas: 3 + "#; + + const ZNODE_YAML: &str = r#" + apiVersion: zookeeper.stackable.tech/v1alpha1 + kind: ZookeeperZnode + metadata: + name: simple-znode + spec: + clusterRef: + name: simple-zookeeper + "#; + + /// The znode path that the znode controller derives from the fixture's UID. + const ZNODE_PATH: &str = "/znode-e5dbf9c2-d8b0-4c1e-9f4a-1d2e3f4a5b6c"; + + /// The value of `key` in the given discovery ConfigMap. + fn data(config_map: &ConfigMap, key: &str) -> String { + config_map + .data + .as_ref() + .expect("the discovery ConfigMap should carry data") + .get(key) + .unwrap_or_else(|| panic!("the discovery ConfigMap should carry {key}")) + .clone() + } + + /// Addresses published under the ZooKeeper server port name, built through the real reader so + /// the fixtures cannot drift from what the validate step produces. + fn published_addresses(addresses: &[(&str, u16)]) -> ListenerAddresses { + let listener = role_listener(Some( + addresses + .iter() + .map(|(address, port)| { + ingress_address(address, ZOOKEEPER_SERVER_PORT_NAME, i32::from(*port)) + }) + .collect(), + )); + + listener_addresses(&listener, ZOOKEEPER_SERVER_PORT_NAME) + .expect("the fixture publishes addresses under the server port name") + .expect("the fixture publishes addresses") + } + + /// The cluster controller advertises the whole ensemble, rooted at `/`. The fixture keeps TLS + /// enabled, so the client port is the secure one. + #[test] + fn cluster_discovery_config_map_advertises_the_published_addresses() { + let cluster = try_validate_with_role_listener(&minimal_zk(ZK_YAML), None) + .expect("validate should succeed for the test fixture"); + let addresses = published_addresses(&[("node-0", 2282), ("node-1", 2282)]); + + let config_map = + build_discovery_configmap(&cluster, ZK_CONTROLLER_NAME, &addresses).expect("build"); + + assert_eq!( + config_map.metadata.name.as_deref(), + Some("simple-zookeeper") + ); + assert_eq!(config_map.metadata.namespace.as_deref(), Some("default")); + assert_eq!(data(&config_map, "ZOOKEEPER"), "node-0:2282,node-1:2282"); + assert_eq!( + data(&config_map, "ZOOKEEPER_HOSTS"), + "node-0:2282,node-1:2282" + ); + assert_eq!(data(&config_map, "ZOOKEEPER_CLIENT_PORT"), "2282"); + assert_eq!(data(&config_map, "ZOOKEEPER_CHROOT"), "/"); + } + + /// While the role Listener publishes no addresses the ConfigMap is still written, with an + /// empty connection string, so that the apply step does not delete the published one as an + /// orphan. + #[test] + fn cluster_discovery_config_map_without_addresses_is_still_written() { + let cluster = try_validate_with_role_listener(&minimal_zk(ZK_YAML), None) + .expect("validate should succeed for the test fixture"); + + let config_map = + build_discovery_configmap(&cluster, ZK_CONTROLLER_NAME, &ListenerAddresses::default()) + .expect("build"); + + assert_eq!( + config_map.metadata.name.as_deref(), + Some("simple-zookeeper") + ); + assert_eq!(data(&config_map, "ZOOKEEPER"), ""); + assert_eq!(data(&config_map, "ZOOKEEPER_HOSTS"), ""); + // The port and chroot do not depend on the addresses, so they stay populated. + assert_eq!(data(&config_map, "ZOOKEEPER_CLIENT_PORT"), "2282"); + assert_eq!(data(&config_map, "ZOOKEEPER_CHROOT"), "/"); + } + + /// The znode controller advertises the same ensemble, narrowed to the znode's chroot. Only + /// `ZOOKEEPER` carries the chroot, because some clients cannot parse the merged format. + #[test] + fn znode_discovery_config_map_narrows_the_ensemble_to_the_chroot() { + let znode = validated_znode(&minimal_znode(ZNODE_YAML)); + + let config_map = build_znode_discovery_configmap( + &znode, + ZNODE_CONTROLLER_NAME, + &znode.discovery_addresses, + ZNODE_PATH, + ) + .expect("build"); + + // The ConfigMap is named after the znode, not after the referenced cluster. + assert_eq!(config_map.metadata.name.as_deref(), Some("simple-znode")); + assert_eq!(config_map.metadata.namespace.as_deref(), Some("default")); + assert_eq!( + data(&config_map, "ZOOKEEPER"), + format!("node-0:2282{ZNODE_PATH}") + ); + assert_eq!(data(&config_map, "ZOOKEEPER_HOSTS"), "node-0:2282"); + assert_eq!(data(&config_map, "ZOOKEEPER_CLIENT_PORT"), "2282"); + assert_eq!(data(&config_map, "ZOOKEEPER_CHROOT"), ZNODE_PATH); + } + + /// A relative chroot would silently produce a connection string pointing at the ensemble root. + #[test] + fn relative_chroot_is_rejected() { + let znode = validated_znode(&minimal_znode(ZNODE_YAML)); + + assert!(matches!( + build_znode_discovery_configmap( + &znode, + ZNODE_CONTROLLER_NAME, + &znode.discovery_addresses, + "znode-without-a-leading-slash", + ), + Err(Error::RelativeChroot { .. }) + )); + } +} diff --git a/rust/operator-binary/src/zk_controller/validate.rs b/rust/operator-binary/src/zk_controller/validate.rs index b423090c..79e16f01 100644 --- a/rust/operator-binary/src/zk_controller/validate.rs +++ b/rust/operator-binary/src/zk_controller/validate.rs @@ -849,7 +849,11 @@ mod tests { #[test] fn role_listener_without_the_expected_port_fails_validation() { let zk = minimal_zk(MINIMAL_ZK_YAML); - let listener = role_listener(Some(vec![ingress_address("node-0", "not-the-zk-port", 2181)])); + let listener = role_listener(Some(vec![ingress_address( + "node-0", + "not-the-zk-port", + 2181, + )])); assert!(matches!( try_validate_with_role_listener(&zk, Some(listener)), diff --git a/rust/operator-binary/src/znode_controller.rs b/rust/operator-binary/src/znode_controller.rs index 85172ebd..f9014701 100644 --- a/rust/operator-binary/src/znode_controller.rs +++ b/rust/operator-binary/src/znode_controller.rs @@ -398,7 +398,8 @@ pub(crate) mod test_support { } /// Runs the real validate step against a minimal (auth-free) fixture whose role Listener - /// publishes `node-0:2181`. + /// publishes `node-0:2282`, the secure client port that the referenced cluster serves because + /// its fixture keeps TLS enabled. pub fn validated_znode(znode: &v1alpha1::ZookeeperZnode) -> ValidatedZnode { use crate::{ crd::ZOOKEEPER_SERVER_PORT_NAME, @@ -410,7 +411,7 @@ pub(crate) mod test_support { Some(role_listener(Some(vec![ingress_address( "node-0", ZOOKEEPER_SERVER_PORT_NAME, - 2181, + 2282, )]))), ) .expect("validate should succeed for the test fixture") diff --git a/rust/operator-binary/src/znode_controller/validate.rs b/rust/operator-binary/src/znode_controller/validate.rs index 6b4f8aea..862f3c90 100644 --- a/rust/operator-binary/src/znode_controller/validate.rs +++ b/rust/operator-binary/src/znode_controller/validate.rs @@ -268,7 +268,7 @@ mod tests { assert!(validated.zookeeper_security.tls_enabled()); assert_eq!( validated.discovery_addresses.to_connection_string(), - "node-0:2181" + "node-0:2282" ); } @@ -277,7 +277,11 @@ mod tests { /// advertise nothing. #[test] fn role_listener_without_the_expected_port_fails_validation() { - let listener = role_listener(Some(vec![ingress_address("node-0", "not-the-zk-port", 2181)])); + let listener = role_listener(Some(vec![ingress_address( + "node-0", + "not-the-zk-port", + 2181, + )])); assert!(matches!( try_validate(&minimal_znode(ZNODE_YAML), Some(listener)), From 20ad91ab6d4214651e07f3f2b0a347ad8d60c2b2 Mon Sep 17 00:00:00 2001 From: maltesander Date: Wed, 12 Aug 2026 14:36:14 +0200 Subject: [PATCH 12/12] Update CHANGELOG.md Co-authored-by: Andrew Kenworthy <1712947+adwk67@users.noreply.github.com> --- CHANGELOG.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 42d07c26..4683c192 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,8 @@ All notable changes to this project will be documented in this file. Previously a ZookeeperCluster without it was accepted by the API server but failed reconciliation ([#1060]). - The reconciler now applies resources and derives the cluster status in discrete apply and update_status steps for the `zk_controller` and `znode_controller` ([#1069]). +- The discovery ConfigMap is now always written, with empty `ZOOKEEPER` and `ZOOKEEPER_HOSTS` while + the listener publishes no addresses ([#1069]). - All product containers now run with `securityContext.runAsNonRoot` set to `true` to improve security ([#1070]). ### Fixed