refactor: Extract build and apply steps in the zk- and znode-controller

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
 

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

rust/operator-binary/src/discovery.rs

@@ -0,0 +1,151 @@
+//! 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};
+use stackable_operator::{
+    builder::{configmap::ConfigMapBuilder, meta::ObjectMetaBuilder},
+    k8s_openapi::api::core::v1::ConfigMap,
+    kube::Resource,
+    v2::{
+        HasName, HasUid, NameIsValidLabelValue,
+        builder::meta::ownerreference_from_resource,
+        kvp::label::recommended_labels,
+        types::operator::{ControllerName, ProductVersion, RoleGroupName},
+    },
+};
+
+use crate::{
+    crd::{ZookeeperRole, security::ZookeeperSecurity},
+    listener_addresses::ListenerAddresses,
+    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<T, E = Error> = std::result::Result<T, E>;
+
+#[derive(Snafu, Debug)]
+pub enum Error {
+    #[snafu(display("chroot path {} was relative (must be absolute)", chroot))]
+    RelativeChroot { chroot: String },
+
+    #[snafu(display("failed to build ConfigMap"))]
+    BuildConfigMap {
+        source: stackable_operator::builder::configmap::Error,
+    },
+}
+
+/// Build the discovery [`ConfigMap`] for the cluster controller from the
+/// [`ValidatedCluster`].
+///
+/// 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.
+pub fn build_discovery_configmap(
+    validated_cluster: &ValidatedCluster,
+    controller_name: &str,
+    listener_addresses: &ListenerAddresses,
+) -> Result<ConfigMap> {
+    build_discovery_configmap_for_owner(
+        validated_cluster,
+        &validated_cluster.namespace,
+        controller_name,
+        &validated_cluster.product_version,
+        listener_addresses,
+        None,
+        &validated_cluster.cluster_config.zookeeper_security,
+    )
+}
+
+/// Build the discovery [`ConfigMap`] for the znode controller.
+///
+/// The ConfigMap is owned by, and placed in the namespace of, the
+/// [`ValidatedZnode`]. The product version and `zookeeper_security` originate from the referenced
+/// cluster (via the validated znode), while `chroot` isolates the znode within the shared ZooKeeper
+/// ensemble.
+pub fn build_znode_discovery_configmap(
+    validated_znode: &ValidatedZnode,
+    controller_name: &str,
+    listener_addresses: &ListenerAddresses,
+    chroot: &str,
+) -> Result<ConfigMap> {
+    build_discovery_configmap_for_owner(
+        validated_znode,
+        &validated_znode.namespace,
+        controller_name,
+        &validated_znode.product_version,
+        listener_addresses,
+        Some(chroot),
+        &validated_znode.zookeeper_security,
+    )
+}
+
+/// Build a discovery [`ConfigMap`] containing ZooKeeper connection details from a
+/// [`listener::v1alpha1::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
+/// `namespace` is where the ConfigMap is placed.
+fn build_discovery_configmap_for_owner(
+    owner: &(impl Resource<DynamicType = ()> + HasName + HasUid + NameIsValidLabelValue),
+    namespace: impl Into<String>,
+    controller_name: &str,
+    product_version: &ProductVersion,
+    listener_addresses: &ListenerAddresses,
+    chroot: Option<&str>,
+    zookeeper_security: &ZookeeperSecurity,
+) -> Result<ConfigMap> {
+    let name = owner.to_name();
+
+    // The discovery ConfigMap is a role-level resource of the `server` role, conventionally
+    // labelled with the `discovery` role group. The controller name differs between the cluster and
+    // znode controllers, so it is passed in and validated into the type-safe newtype here.
+    let controller_name = ControllerName::from_str(controller_name)
+        .expect("the controller name is a valid label value");
+    let role_group_name = PLACEHOLDER_DISCOVERY_ROLE_GROUP.clone();
+
+    // 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.to_connection_string();
+    let mut conn_str = listener_addresses.clone();
+    if let Some(chroot) = chroot {
+        if !chroot.starts_with('/') {
+            return RelativeChrootSnafu { chroot }.fail();
+        }
+        conn_str.push_str(chroot);
+    }
+    ConfigMapBuilder::new()
+        .metadata(
+            ObjectMetaBuilder::new()
+                .name(name)
+                .namespace(namespace)
+                .ownerreference(ownerreference_from_resource(owner, None, Some(true)))
+                .with_labels(recommended_labels(
+                    owner,
+                    &product_name(),
+                    product_version,
+                    &operator_name(),
+                    &controller_name,
+                    &ZookeeperRole::Server.into(),
+                    &role_group_name,
+                ))
+                .build(),
+        )
+        .add_data("ZOOKEEPER", conn_str)
+        // Some clients don't support ZooKeeper's merged `hosts/chroot` format, so export them separately for these clients
+        .add_data("ZOOKEEPER_HOSTS", listener_addresses)
+        .add_data(
+            "ZOOKEEPER_CLIENT_PORT",
+            zookeeper_security.client_port().to_string(),
+        )
+        .add_data("ZOOKEEPER_CHROOT", chroot.unwrap_or("/"))
+        .build()
+        .context(BuildConfigMapSnafu)
+}

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<listener::v1alpha1::Listener>,
+    },
+
+    #[snafu(display("expected an unsigned 16-bit port, got {port_number}"))]
+    InvalidPort {
+        source: TryFromIntError,
+        port_number: i32,
+    },
+}
+
+type Result<T, E = Error> = std::result::Result<T, E>;
+
+/// 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::<Vec<_>>()
+            .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<Option<ListenerAddresses>> {
+    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::<Result<BTreeSet<_>, _>>()?;
+
+    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<Vec<ListenerIngress>>) -> 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 { .. })
+        ));
+    }
+}

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,8 @@ use crate::{
 };
 
 pub mod crd;
+mod discovery;
+mod listener_addresses;
 mod webhooks;
 mod zk_controller;
 mod znode_controller;
@@ -139,6 +142,13 @@ async fn main() -> anyhow::Result<()> {
                     watch_namespace.get_api::<DeserializeGuard<ConfigMap>>(&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::<DeserializeGuard<Listener>>(&client),
+                    watcher::Config::default(),
+                )
                 .graceful_shutdown_on(sigterm_watcher.handle())
                 .run(
                     zk_controller::reconcile_zk,