From bdb6aff12cdc75f012b8c085a3c01f8eea39bb71 Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Fri, 10 Jul 2026 17:54:57 +0200 Subject: [PATCH] refactor: Introduce build aggregator --- rust/operator-binary/src/zk_controller.rs | 214 +++++++----------- .../src/zk_controller/build.rs | 191 +++++++++++++++- 2 files changed, 268 insertions(+), 137 deletions(-) diff --git a/rust/operator-binary/src/zk_controller.rs b/rust/operator-binary/src/zk_controller.rs index 67100563..4f81ffa6 100644 --- a/rust/operator-binary/src/zk_controller.rs +++ b/rust/operator-binary/src/zk_controller.rs @@ -3,11 +3,17 @@ use std::{hash::Hasher, str::FromStr, sync::Arc}; use const_format::concatcp; use fnv::FnvHasher; -use snafu::{ResultExt, Snafu}; +use snafu::{OptionExt, ResultExt, Snafu}; use stackable_operator::{ cli::OperatorEnvironmentOptions, cluster_resources::ClusterResourceApplyStrategy, commons::rbac::build_rbac_resources, + crd::listener::v1alpha1::Listener, + k8s_openapi::api::{ + apps::v1::StatefulSet, + core::v1::{ConfigMap, Service}, + policy::v1::PodDisruptionBudget, + }, kube::{ api::DynamicObject, core::{DeserializeGuard, error_boundary}, @@ -20,26 +26,15 @@ use stackable_operator::{ compute_conditions, operations::ClusterOperationsConditionBuilder, statefulset::StatefulSetConditionBuilder, }, - v2::{ - cluster_resources::cluster_resources_new, - types::operator::{ControllerName, RoleGroupName}, - }, + v2::{cluster_resources::cluster_resources_new, types::operator::ControllerName}, }; use strum::{EnumDiscriminants, IntoStaticStr}; use crate::{ APP_NAME, OPERATOR_NAME, ObjectRef, - crd::{ZookeeperRole, v1alpha1}, + crd::v1alpha1, zk_controller::{ - build::resource::{ - config_map, discovery, - listener::build_role_listener, - pdb::build_pdb, - service::{ - build_server_rolegroup_headless_service, build_server_rolegroup_metrics_service, - }, - statefulset::build_server_rolegroup_statefulset, - }, + build::resource::discovery, validate::{operator_name, product_name}, }, }; @@ -62,7 +57,6 @@ type Result = std::result::Result; #[derive(Snafu, Debug, EnumDiscriminants)] #[strum_discriminants(derive(IntoStaticStr))] -#[allow(clippy::enum_variant_names)] pub enum Error { #[snafu(display("ZookeeperCluster object is invalid"))] InvalidZookeeperCluster { @@ -81,34 +75,12 @@ pub enum Error { #[snafu(display("internal operator failure"))] InternalOperatorFailure { source: crate::crd::Error }, - #[snafu(display("failed to apply Service for role group {rolegroup}"))] - ApplyRoleGroupService { - source: stackable_operator::cluster_resources::Error, - rolegroup: RoleGroupName, - }, - - #[snafu(display("failed to build ConfigMap for role group {rolegroup}"))] - BuildRoleGroupConfigMap { - source: config_map::Error, - rolegroup: RoleGroupName, - }, + #[snafu(display("failed to build the Kubernetes resources"))] + BuildResources { source: build::Error }, - #[snafu(display("failed to apply ConfigMap for role group {rolegroup}"))] - ApplyRoleGroupConfig { + #[snafu(display("failed to apply Kubernetes resource"))] + ApplyResource { source: stackable_operator::cluster_resources::Error, - rolegroup: RoleGroupName, - }, - - #[snafu(display("failed to build StatefulSet for role group {rolegroup}"))] - BuildRoleGroupStatefulSet { - source: build::resource::statefulset::Error, - rolegroup: RoleGroupName, - }, - - #[snafu(display("failed to apply StatefulSet for role group {rolegroup}"))] - ApplyRoleGroupStatefulSet { - source: stackable_operator::cluster_resources::Error, - rolegroup: RoleGroupName, }, #[snafu(display("object is missing metadata to build owner reference"))] @@ -116,6 +88,11 @@ pub enum Error { 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 }, @@ -149,11 +126,6 @@ pub enum Error { source: stackable_operator::cluster_resources::Error, }, - #[snafu(display("failed to apply PodDisruptionBudget"))] - ApplyPdb { - source: stackable_operator::cluster_resources::Error, - }, - #[snafu(display("failed to build label"))] BuildLabel { source: LabelError }, @@ -161,11 +133,6 @@ pub enum Error { ObjectMeta { source: stackable_operator::builder::meta::Error, }, - - #[snafu(display("failed to apply group listener"))] - ApplyGroupListener { - source: stackable_operator::cluster_resources::Error, - }, } impl ReconcilerError for Error { @@ -180,12 +147,10 @@ impl ReconcilerError for Error { Error::ValidateCluster { .. } => None, Error::CrdValidationFailure { .. } => None, Error::InternalOperatorFailure { .. } => None, - Error::ApplyRoleGroupService { .. } => None, - Error::BuildRoleGroupConfigMap { .. } => None, - Error::ApplyRoleGroupConfig { .. } => None, - Error::BuildRoleGroupStatefulSet { .. } => None, - Error::ApplyRoleGroupStatefulSet { .. } => None, + Error::BuildResources { .. } => None, + Error::ApplyResource { .. } => None, Error::ObjectMissingMetadataForOwnerRef { .. } => None, + Error::NoRoleListener => None, Error::BuildDiscoveryConfig { .. } => None, Error::ApplyDiscoveryConfig { .. } => None, Error::ApplyStatus { .. } => None, @@ -193,14 +158,23 @@ impl ReconcilerError for Error { Error::ApplyRoleBinding { .. } => None, Error::BuildRbacResources { .. } => None, Error::DeleteOrphans { .. } => None, - Error::ApplyPdb { .. } => None, Error::BuildLabel { .. } => None, Error::ObjectMeta { .. } => None, - Error::ApplyGroupListener { .. } => 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, + pub pod_disruption_budgets: Vec, +} + pub async fn reconcile_zk( zk: Arc>, ctx: Arc, @@ -245,7 +219,7 @@ pub async fn reconcile_zk( .context(BuildRbacResourcesSnafu)?; cluster_resources - .add(client, rbac_sa.clone()) + .add(client, rbac_sa) .await .context(ApplyServiceAccountSnafu)?; @@ -254,98 +228,63 @@ pub async fn reconcile_zk( .await .context(ApplyRoleBindingSnafu)?; - let mut ss_cond_builder = StatefulSetConditionBuilder::default(); + let resources = build::build(&validated_cluster, &client.kubernetes_cluster_info) + .context(BuildResourcesSnafu)?; - let zk_role = ZookeeperRole::Server; - let server_role_group_configs = validated_cluster - .role_group_configs - .get(&zk_role) - .into_iter() - .flatten(); - for (rolegroup_name, rolegroup_config) in server_role_group_configs { - // Resource naming, labels and owner references are derived from the `ValidatedCluster` and - // the type-safe `RoleGroupName`. - let rg_headless_service = - build_server_rolegroup_headless_service(&validated_cluster, rolegroup_name); - let rg_metrics_service = build_server_rolegroup_metrics_service( - &validated_cluster, - rolegroup_name, - rolegroup_config, - ); - let rg_configmap = config_map::build_server_rolegroup_config_map( - &validated_cluster, - &client.kubernetes_cluster_info, - rolegroup_name, - rolegroup_config, - ) - .context(BuildRoleGroupConfigMapSnafu { - rolegroup: rolegroup_name.clone(), - })?; - let rg_statefulset = build_server_rolegroup_statefulset( - &validated_cluster, - rolegroup_name, - rolegroup_config, - ) - .with_context(|_| BuildRoleGroupStatefulSetSnafu { - rolegroup: rolegroup_name.clone(), - })?; + let mut ss_cond_builder = StatefulSetConditionBuilder::default(); + for service in resources.services { cluster_resources - .add(client, rg_headless_service) - .await - .with_context(|_| ApplyRoleGroupServiceSnafu { - rolegroup: rolegroup_name.clone(), - })?; - cluster_resources - .add(client, rg_metrics_service) + .add(client, service) .await - .with_context(|_| ApplyRoleGroupServiceSnafu { - rolegroup: rolegroup_name.clone(), - })?; - cluster_resources - .add(client, rg_configmap) - .await - .with_context(|_| ApplyRoleGroupConfigSnafu { - rolegroup: rolegroup_name.clone(), - })?; + .context(ApplyResourceSnafu)?; + } - // 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. - ss_cond_builder.add( + // 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, rg_statefulset) + .add(client, listener) .await - .with_context(|_| ApplyRoleGroupStatefulSetSnafu { - rolegroup: rolegroup_name.clone(), - })?, + .context(ApplyResourceSnafu)?, ); } + let role_listener = applied_role_listener.context(NoRoleListenerSnafu)?; - if let Some(role_config) = &validated_cluster.role_config - && let Some(pdb) = build_pdb(&role_config.pdb, &validated_cluster, &zk_role) - { + 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(ApplyPdbSnafu)?; + .context(ApplyResourceSnafu)?; } - let listener = build_role_listener(&validated_cluster, &zk_role); - let applied_listener = cluster_resources - .add(client, listener) - .await - .context(ApplyGroupListenerSnafu)?; + // 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)?, + ); + } // 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); - let discovery_cm = discovery::build_discovery_configmap( - &validated_cluster, - ZK_CONTROLLER_NAME, - applied_listener, - ) - .context(BuildDiscoveryConfigSnafu)?; + + 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) @@ -462,8 +401,13 @@ mod tests { k8s_openapi::api::core::v1::ConfigMap, v2::types::operator::RoleGroupName, }; - use super::*; - use crate::zk_controller::test_support::{cluster_info, minimal_zk, validated_cluster}; + use crate::{ + crd::ZookeeperRole, + zk_controller::{ + build::resource::config_map, + test_support::{cluster_info, minimal_zk, validated_cluster}, + }, + }; #[test] fn test_default_config() { diff --git a/rust/operator-binary/src/zk_controller/build.rs b/rust/operator-binary/src/zk_controller/build.rs index 71f3ba30..1c6dd520 100644 --- a/rust/operator-binary/src/zk_controller/build.rs +++ b/rust/operator-binary/src/zk_controller/build.rs @@ -1,6 +1,6 @@ //! Build steps for the ZookeeperCluster controller. //! -//! Each submodule turns the [`ValidatedCluster`](super::validate::ValidatedCluster) +//! Each submodule turns the [`ValidatedCluster`] //! into a Kubernetes resource. The [`v1alpha1::ZookeeperCluster`](crate::crd::v1alpha1::ZookeeperCluster) //! is only used for the owner reference and object metadata, never for //! configuration. @@ -11,7 +11,28 @@ use std::str::FromStr; -use stackable_operator::v2::types::operator::{ProductVersion, RoleGroupName}; +use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + utils::cluster_info::KubernetesClusterInfo, + v2::types::operator::{ProductVersion, RoleGroupName}, +}; + +use crate::{ + crd::ZookeeperRole, + zk_controller::{ + KubernetesResources, + build::resource::{ + config_map, + listener::build_role_listener, + pdb::build_pdb, + service::{ + build_server_rolegroup_headless_service, build_server_rolegroup_metrics_service, + }, + statefulset::{self, build_server_rolegroup_statefulset}, + }, + validate::ValidatedCluster, + }, +}; // Placeholder role-group name used for the recommended labels of the role-level discovery // `ConfigMap` (which is not tied to a single role group). @@ -30,3 +51,169 @@ pub mod graceful_shutdown; pub mod jvm; pub mod properties; pub mod resource; + +#[derive(Snafu, Debug)] +pub enum Error { + #[snafu(display("failed to build ConfigMap for role group {rolegroup}"))] + ConfigMap { + source: config_map::Error, + rolegroup: RoleGroupName, + }, + + #[snafu(display("failed to build StatefulSet for role group {rolegroup}"))] + StatefulSet { + source: statefulset::Error, + rolegroup: RoleGroupName, + }, +} + +/// Builds every Kubernetes resource for the given validated cluster. +/// +/// Does not need a Kubernetes client: every reference to another Kubernetes resource is already +/// dereferenced and validated by this point, so the errors returned here are resource-assembly +/// 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. +pub fn build( + cluster: &ValidatedCluster, + cluster_info: &KubernetesClusterInfo, +) -> Result { + let mut stateful_sets = vec![]; + let mut services = vec![]; + let mut config_maps = vec![]; + let mut pod_disruption_budgets = vec![]; + + let zk_role = ZookeeperRole::Server; + let server_role_group_configs = cluster + .role_group_configs + .get(&zk_role) + .into_iter() + .flatten(); + for (rolegroup_name, rolegroup_config) in server_role_group_configs { + // Resource naming, labels and owner references are derived from the `ValidatedCluster` and + // the type-safe `RoleGroupName`. + services.push(build_server_rolegroup_headless_service( + cluster, + rolegroup_name, + )); + services.push(build_server_rolegroup_metrics_service( + cluster, + rolegroup_name, + rolegroup_config, + )); + config_maps.push( + config_map::build_server_rolegroup_config_map( + cluster, + cluster_info, + rolegroup_name, + rolegroup_config, + ) + .context(ConfigMapSnafu { + rolegroup: rolegroup_name.clone(), + })?, + ); + stateful_sets.push( + build_server_rolegroup_statefulset(cluster, rolegroup_name, rolegroup_config) + .with_context(|_| StatefulSetSnafu { + rolegroup: rolegroup_name.clone(), + })?, + ); + } + + if let Some(role_config) = &cluster.role_config + && let Some(pdb) = build_pdb(&role_config.pdb, cluster, &zk_role) + { + pod_disruption_budgets.push(pdb); + } + + let listeners = vec![build_role_listener(cluster, &zk_role)]; + + Ok(KubernetesResources { + stateful_sets, + services, + listeners, + config_maps, + pod_disruption_budgets, + }) +} + +#[cfg(test)] +mod tests { + use stackable_operator::kube::Resource; + + use super::build; + use crate::zk_controller::test_support::{cluster_info, minimal_zk, validated_cluster}; + + /// Collects the `.metadata.name`s of the given resources, sorted for stable comparison. + fn sorted_names(resources: &[impl Resource]) -> Vec<&str> { + let mut names: Vec<&str> = resources + .iter() + .filter_map(|resource| resource.meta().name.as_deref()) + .collect(); + names.sort(); + names + } + + #[test] + fn build_produces_expected_resource_names() { + let zookeeper_yaml = r#" + apiVersion: zookeeper.stackable.tech/v1alpha1 + kind: ZookeeperCluster + metadata: + name: simple-zookeeper + spec: + image: + productVersion: "3.9.5" + servers: + roleGroups: + default: + replicas: 3 + secondary: + replicas: 1 + "#; + let zookeeper = minimal_zk(zookeeper_yaml); + let cluster = validated_cluster(&zookeeper); + + let resources = build(&cluster, &cluster_info()).expect("build succeeds"); + + // One StatefulSet per role group. + assert_eq!( + sorted_names(&resources.stateful_sets), + [ + "simple-zookeeper-server-default", + "simple-zookeeper-server-secondary", + ] + ); + // One headless and one metrics Service per role group. + assert_eq!( + sorted_names(&resources.services), + [ + "simple-zookeeper-server-default-headless", + "simple-zookeeper-server-default-metrics", + "simple-zookeeper-server-secondary-headless", + "simple-zookeeper-server-secondary-metrics", + ] + ); + // One ConfigMap per role group; the discovery ConfigMap is absent — see `build()`. + assert_eq!( + sorted_names(&resources.config_maps), + [ + "simple-zookeeper-server-default", + "simple-zookeeper-server-secondary", + ] + ); + // The single role-level Listener for the one ZooKeeper role (`server`). + assert_eq!( + sorted_names(&resources.listeners), + ["simple-zookeeper-server"] + ); + // The default PodDisruptionBudget for the `server` role. + assert_eq!( + sorted_names(&resources.pod_disruption_budgets), + ["simple-zookeeper-server"] + ); + } +}