From c50096d2ff6b86805a5e85053a360919af8df6b8 Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Wed, 5 Aug 2026 16:24:30 +0200 Subject: [PATCH 1/5] 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 2/5] 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 3/5] 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 4/5] 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 5/5] 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