From e812aba9b5d3bf96c82ad619160303b64fcac7c3 Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Fri, 10 Jul 2026 13:35:43 +0200 Subject: [PATCH 1/3] refactor: Decouple StatefulSet builder from applied ServiceAccount --- .../src/controller/build/resource/statefulset.rs | 11 ++++------- rust/operator-binary/src/hdfs_controller.rs | 2 +- 2 files changed, 5 insertions(+), 8 deletions(-) diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs index 5f6ac00b..6b7b32f2 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -5,13 +5,10 @@ use stackable_operator::{ builder::pod::{PodBuilder, security::PodSecurityContextBuilder}, k8s_openapi::{ DeepMerge, - api::{ - apps::v1::{StatefulSet, StatefulSetSpec}, - core::v1::ServiceAccount, - }, + api::apps::v1::{StatefulSet, StatefulSetSpec}, apimachinery::pkg::apis::meta::v1::LabelSelector, }, - kube::{ResourceExt, api::ObjectMeta}, + kube::api::ObjectMeta, kvp::{LabelError, Labels}, utils::cluster_info::KubernetesClusterInfo, v2::types::operator::RoleGroupName, @@ -50,7 +47,7 @@ pub(crate) fn build_rolegroup_statefulset( role: &HdfsNodeRole, role_group_name: &RoleGroupName, rolegroup_config: &ValidatedRoleGroupConfig, - service_account: &ServiceAccount, + service_account_name: &str, ) -> Result { tracing::info!("Setting up StatefulSet for role {role} role group {role_group_name}"); @@ -72,7 +69,7 @@ pub(crate) fn build_rolegroup_statefulset( pb.metadata(pb_metadata) .image_pull_secrets_from_product_image(image) .affinity(&merged_config.affinity) - .service_account_name(service_account.name_any()) + .service_account_name(service_account_name) .security_context(PodSecurityContextBuilder::new().fs_group(1000).build()); // Adds all containers and volumes to the pod builder diff --git a/rust/operator-binary/src/hdfs_controller.rs b/rust/operator-binary/src/hdfs_controller.rs index 1b073fed..604ce24e 100644 --- a/rust/operator-binary/src/hdfs_controller.rs +++ b/rust/operator-binary/src/hdfs_controller.rs @@ -269,7 +269,7 @@ pub async fn reconcile_hdfs( &role, rolegroup_name, validated_cluster_rg_config, - &rbac_sa, + &rbac_sa.name_any(), ) .context(BuildRoleGroupStatefulSetSnafu)?; From 8af09ecf5571196a1205aebd2908b7e86624df6a Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Fri, 10 Jul 2026 14:00:36 +0200 Subject: [PATCH 2/3] refactor: Introduce build aggregator --- .../src/controller/build/mod.rs | 161 ++++++++++++++- rust/operator-binary/src/controller/mod.rs | 19 ++ rust/operator-binary/src/hdfs_controller.rs | 188 +++++++----------- 3 files changed, 252 insertions(+), 116 deletions(-) diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs index 9b4f1138..9b18e795 100644 --- a/rust/operator-binary/src/controller/build/mod.rs +++ b/rust/operator-binary/src/controller/build/mod.rs @@ -1,8 +1,10 @@ use std::{collections::HashMap, str::FromStr}; +use snafu::{ResultExt, Snafu}; use stackable_operator::{ builder::meta::ObjectMetaBuilder, kvp::{LabelError, Labels}, + utils::cluster_info::KubernetesClusterInfo, v2::{ builder::meta::ownerreference_from_resource, types::{common::Port, kubernetes::ServiceName, operator::RoleGroupName}, @@ -11,7 +13,7 @@ use stackable_operator::{ use crate::{ build_recommended_labels, - controller::ValidatedCluster, + controller::{KubernetesResources, ValidatedCluster}, crd::{ HdfsNodeRole, HdfsPodRef, constants::{ @@ -41,6 +43,111 @@ pub mod opa; pub mod properties; pub mod resource; +#[derive(Snafu, Debug)] +pub enum Error { + #[snafu(display("failed to build Service for role {role} role group {role_group}"))] + Service { + source: resource::service::Error, + role: HdfsNodeRole, + role_group: RoleGroupName, + }, + + #[snafu(display("failed to build ConfigMap for role {role} role group {role_group}"))] + ConfigMap { + source: resource::config_map::Error, + role: HdfsNodeRole, + role_group: RoleGroupName, + }, + + #[snafu(display("failed to build StatefulSet for role {role} role group {role_group}"))] + StatefulSet { + source: resource::statefulset::Error, + role: HdfsNodeRole, + role_group: RoleGroupName, + }, +} + +/// Builds every Kubernetes resource for the given validated cluster. +/// +/// Does not need a Kubernetes client: every external reference is already dereferenced and +/// validated by this point, so the errors returned here are resource-assembly failures only. +/// `cluster_info` carries static cluster information resolved at operator startup (e.g. the +/// cluster domain used to build Kerberos principals), not a live client. +/// +/// `service_account_name` is the name of the RBAC `ServiceAccount` the role-group Pods run under. +/// The RBAC resources are built and applied separately in the reconcile step. +/// +/// The resources are returned as flat, unordered collections. The reconcile step re-groups the +/// StatefulSets by role to preserve HDFS's ordered, rollout-gated deployment during upgrades. The +/// discovery `ConfigMap` is deliberately not built here: it needs a live client to resolve +/// listener addresses and is therefore handled in the reconcile step. +pub fn build( + cluster: &ValidatedCluster, + cluster_info: &KubernetesClusterInfo, + service_account_name: &str, +) -> Result { + let mut services = vec![]; + let mut config_maps = vec![]; + let mut stateful_sets = vec![]; + let mut pod_disruption_budgets = vec![]; + + for (role, role_group_configs) in &cluster.role_groups { + for (role_group_name, rg_config) in role_group_configs { + services.push( + resource::service::rolegroup_headless_service(cluster, role, role_group_name) + .context(ServiceSnafu { + role: *role, + role_group: role_group_name.clone(), + })?, + ); + services.push( + resource::service::rolegroup_metrics_service(cluster, role, role_group_name) + .context(ServiceSnafu { + role: *role, + role_group: role_group_name.clone(), + })?, + ); + config_maps.push( + resource::config_map::build_rolegroup_config_map( + cluster, + cluster_info, + role, + role_group_name, + ) + .context(ConfigMapSnafu { + role: *role, + role_group: role_group_name.clone(), + })?, + ); + stateful_sets.push( + resource::statefulset::build_rolegroup_statefulset( + cluster, + cluster_info, + role, + role_group_name, + rg_config, + service_account_name, + ) + .context(StatefulSetSnafu { + role: *role, + role_group: role_group_name.clone(), + })?, + ); + } + + if let Some(pdb) = resource::pdb::build_pdb(cluster, role) { + pod_disruption_budgets.push(pdb); + } + } + + Ok(KubernetesResources { + services, + config_maps, + pod_disruption_budgets, + stateful_sets, + }) +} + /// Builds the [`HdfsPodRef`]s expected for every pod of the given `role`, across all /// of its role groups. /// @@ -263,3 +370,55 @@ fn role_data_ports(role: &HdfsNodeRole, https_enabled: bool) -> Vec<(String, Por ], } } + +#[cfg(test)] +mod tests { + use stackable_operator::kube::Resource; + + use super::build; + use crate::controller::build::properties::test_support::{cluster_info, validated_cluster}; + + /// The sorted `metadata.name`s of a resource collection. + fn sorted_names(resources: &[impl Resource]) -> Vec { + let mut names: Vec = resources + .iter() + .filter_map(|resource| resource.meta().name.clone()) + .collect(); + names.sort(); + names + } + + /// The aggregator emits, for the minimal three-role cluster (one `default` role group each): + /// one StatefulSet and one ConfigMap per role group, one headless plus one metrics Service per + /// role group, and one default PDB per role. + #[test] + fn build_produces_expected_resource_names() { + let cluster = validated_cluster(); + let resources = + build(&cluster, &cluster_info(), "hdfs-serviceaccount").expect("build succeeds"); + + assert_eq!( + sorted_names(&resources.stateful_sets), + [ + "hdfs-datanode-default", + "hdfs-journalnode-default", + "hdfs-namenode-default", + ] + ); + // One headless and one metrics Service per role group. + assert_eq!(resources.services.len(), 6); + assert_eq!( + sorted_names(&resources.config_maps), + [ + "hdfs-datanode-default", + "hdfs-journalnode-default", + "hdfs-namenode-default", + ] + ); + // A default PDB per role. + assert_eq!( + sorted_names(&resources.pod_disruption_budgets), + ["hdfs-datanode", "hdfs-journalnode", "hdfs-namenode"] + ); + } +} diff --git a/rust/operator-binary/src/controller/mod.rs b/rust/operator-binary/src/controller/mod.rs index 3e4fea84..800d6157 100644 --- a/rust/operator-binary/src/controller/mod.rs +++ b/rust/operator-binary/src/controller/mod.rs @@ -2,6 +2,11 @@ use std::{collections::BTreeMap, str::FromStr}; use stackable_operator::{ commons::product_image_selection::ResolvedProductImage, + k8s_openapi::api::{ + apps::v1::StatefulSet, + core::v1::{ConfigMap, Service}, + policy::v1::PodDisruptionBudget, + }, kube::{Resource, api::ObjectMeta}, v2::{ HasName, HasUid, NameIsValidLabelValue, @@ -30,6 +35,20 @@ pub mod build; pub mod dereference; pub mod validate; +/// Every Kubernetes resource produced by the build step. +/// +/// The resources are flat, unordered collections. The reconcile step re-groups the +/// StatefulSets by role to preserve HDFS's ordered, rollout-gated deployment during +/// upgrades. The discovery `ConfigMap` is not part of this set: it depends on a live +/// Kubernetes client (to resolve listener addresses) and is therefore built and applied +/// separately in the reconcile step. +pub struct KubernetesResources { + pub services: Vec, + pub config_maps: Vec, + pub pod_disruption_budgets: Vec, + pub stateful_sets: Vec, +} + /// The [`RoleGroupConfig`] specialised for HDFS: the validated config is the /// per-role [`AnyNodeConfig`], pub type ValidatedRoleGroupConfig = RoleGroupConfig< diff --git a/rust/operator-binary/src/hdfs_controller.rs b/rust/operator-binary/src/hdfs_controller.rs index 604ce24e..2df9ac2e 100644 --- a/rust/operator-binary/src/hdfs_controller.rs +++ b/rust/operator-binary/src/hdfs_controller.rs @@ -31,12 +31,7 @@ use crate::{ controller::{ build::{ self, - resource::{ - discovery::{self, build_discovery_config_map}, - pdb::build_pdb, - service::{self, rolegroup_headless_service, rolegroup_metrics_service}, - statefulset::{self, build_rolegroup_statefulset}, - }, + resource::discovery::{self, build_discovery_config_map}, }, controller_name, operator_name, product_name, }, @@ -84,9 +79,9 @@ pub enum Error { name: String, }, - #[snafu(display("failed to build the role group ConfigMap"))] - BuildRoleGroupConfigMap { - source: crate::controller::build::resource::config_map::Error, + #[snafu(display("failed to build Kubernetes resources"))] + BuildResources { + source: crate::controller::build::Error, }, #[snafu(display("cannot collect discovery configuration"))] @@ -131,16 +126,10 @@ pub enum Error { #[snafu(display("failed to build cluster resources label"))] BuildClusterResourcesLabel { source: LabelError }, - #[snafu(display("failed to build the role group StatefulSet"))] - BuildRoleGroupStatefulSet { source: statefulset::Error }, - #[snafu(display("HdfsCluster object is invalid"))] InvalidHdfsCluster { source: error_boundary::InvalidObject, }, - - #[snafu(display("failed to build service"))] - BuildService { source: service::Error }, } impl ReconcilerError for Error { @@ -211,30 +200,48 @@ pub async fn reconcile_hdfs( .await .context(ApplyRoleBindingSnafu)?; - let mut ss_cond_builder = StatefulSetConditionBuilder::default(); + // Build every (non-discovery) Kubernetes resource up front. This step needs no client: all + // external references are already dereferenced and validated. The ServiceAccount name is + // deterministic on the built RBAC object, so the build does not depend on the applied one. + let resources = build::build( + &validated_cluster, + &client.kubernetes_cluster_info, + &rbac_sa.name_any(), + ) + .context(BuildResourcesSnafu)?; + + // Apply Services, ConfigMaps and PodDisruptionBudgets first. The StatefulSets are applied + // afterwards so that every ConfigMap a Pod mounts already exists, which prevents unnecessary + // Pod restarts. See https://github.com/stackabletech/commons-operator/issues/111 for details. + for service in resources.services { + let name = service.name_any(); + cluster_resources + .add(client, service) + .await + .with_context(|_| ApplyRoleGroupServiceSnafu { name })?; + } + for config_map in resources.config_maps { + let name = config_map.name_any(); + cluster_resources + .add(client, config_map) + .await + .with_context(|_| ApplyRoleGroupConfigMapSnafu { name })?; + } + for pdb in resources.pod_disruption_budgets { + cluster_resources + .add(client, pdb) + .await + .context(ApplyPdbSnafu)?; + } let upgrade_state = validated_cluster.status.upgrade_state; - let mut deploy_done = true; - // Roles must be deployed in order during rolling upgrades, - // namenode version must be >= datanode version (and so on). - let roles = reverse_if( - match upgrade_state { - // Downgrades have the opposite version relationship, so they need to be rolled out in reverse order. - Some(UpgradeState::Downgrading) => { - tracing::info!("HdfsCluster is being downgraded, deploying in reverse order"); - true - } - _ => false, - }, - HdfsNodeRole::iter(), - ); - 'roles: for role in roles { - let Some(group_config) = validated_cluster.role_groups.get(&role) else { - tracing::debug!(?role, "role has no configuration, skipping"); + // Warn about invalid replica counts. This is validation feedback and independent of the + // resource application below. + for role in HdfsNodeRole::iter() { + if !validated_cluster.role_groups.contains_key(&role) { continue; - }; - + } if let Some(message) = build_invalid_replica_message(&validated_cluster, &role) { publish_warning_event( &ctx, @@ -246,88 +253,39 @@ pub async fn reconcile_hdfs( .await .context(FailedToCreateClusterEventSnafu)?; } + } - for (rolegroup_name, validated_cluster_rg_config) in group_config.iter() { - let rg_service = rolegroup_headless_service(&validated_cluster, &role, rolegroup_name) - .context(BuildServiceSnafu)?; - let rg_metrics_service = - rolegroup_metrics_service(&validated_cluster, &role, rolegroup_name) - .context(BuildServiceSnafu)?; - - let rg_configmap = - crate::controller::build::resource::config_map::build_rolegroup_config_map( - &validated_cluster, - &client.kubernetes_cluster_info, - &role, - rolegroup_name, - ) - .context(BuildRoleGroupConfigMapSnafu)?; - - let rg_statefulset = build_rolegroup_statefulset( - &validated_cluster, - &client.kubernetes_cluster_info, - &role, - rolegroup_name, - validated_cluster_rg_config, - &rbac_sa.name_any(), - ) - .context(BuildRoleGroupStatefulSetSnafu)?; - - let rg_service_name = rg_service.name_any(); - let rg_metrics_service_name = rg_metrics_service.name_any(); - - cluster_resources - .add(client, rg_service) - .await - .with_context(|_| ApplyRoleGroupServiceSnafu { - name: rg_service_name, - })?; - cluster_resources - .add(client, rg_metrics_service) - .await - .with_context(|_| ApplyRoleGroupServiceSnafu { - name: rg_metrics_service_name, - })?; - let rg_configmap_name = rg_configmap.name_any(); - cluster_resources - .add(client, rg_configmap.clone()) - .await - .with_context(|_| ApplyRoleGroupConfigMapSnafu { - name: rg_configmap_name, - })?; - - // 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. - let rg_statefulset_name = rg_statefulset.name_any(); - let deployed_rg_statefulset = cluster_resources - .add(client, rg_statefulset.clone()) - .await - .with_context(|_| ApplyRoleGroupStatefulSetSnafu { - name: rg_statefulset_name, - })?; - ss_cond_builder.add(deployed_rg_statefulset.clone()); - - if upgrade_state.is_some() { - // When upgrading, ensure that each role is upgraded before moving on to the next as recommended by - // https://hadoop.apache.org/docs/r3.4.0/hadoop-project-dist/hadoop-hdfs/HdfsRollingUpgrade.html#Upgrading_Non-Federated_Clusters - if let Err(reason) = check_statefulset_rollout_complete(&deployed_rg_statefulset) { - tracing::info!( - rolegroup.statefulset = %ObjectRef::from_obj(&deployed_rg_statefulset), - reason = &reason as &dyn std::error::Error, - "rolegroup is still upgrading, waiting..." - ); - deploy_done = false; - break 'roles; - } - } - } + let mut ss_cond_builder = StatefulSetConditionBuilder::default(); + let mut deploy_done = true; - if let Some(pdb) = build_pdb(&validated_cluster, &role) { - cluster_resources - .add(client, pdb) - .await - .context(ApplyPdbSnafu)?; + // StatefulSets must be rolled out in role order during upgrades (a namenode's version must be + // >= the datanodes', and so on), with each role finishing its rollout before the next starts. + // https://hadoop.apache.org/docs/r3.4.0/hadoop-project-dist/hadoop-hdfs/HdfsRollingUpgrade.html#Upgrading_Non-Federated_Clusters + // The build output is already ordered by role, so it is applied as-is; downgrades have the + // opposite version relationship and are therefore rolled out in reverse. + let downgrading = matches!(upgrade_state, Some(UpgradeState::Downgrading)); + if downgrading { + tracing::info!("HdfsCluster is being downgraded, deploying in reverse order"); + } + for statefulset in reverse_if(downgrading, resources.stateful_sets.iter()) { + let name = statefulset.name_any(); + let deployed_statefulset = cluster_resources + .add(client, statefulset.clone()) + .await + .with_context(|_| ApplyRoleGroupStatefulSetSnafu { name })?; + ss_cond_builder.add(deployed_statefulset.clone()); + + if upgrade_state.is_some() { + // Ensure each role is fully upgraded before moving on to the next. + if let Err(reason) = check_statefulset_rollout_complete(&deployed_statefulset) { + tracing::info!( + rolegroup.statefulset = %ObjectRef::from_obj(&deployed_statefulset), + reason = &reason as &dyn std::error::Error, + "rolegroup is still upgrading, waiting..." + ); + deploy_done = false; + break; + } } } From 41058ed73c64e06e0c3ac364956fc4ba36dc82eb Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Tue, 21 Jul 2026 16:57:23 +0200 Subject: [PATCH 3/3] changelog --- CHANGELOG.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index aaebdb82..0d84fb15 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,13 @@ All notable changes to this project will be documented in this file. ## [Unreleased] +### Changed + +- Internal operator refactoring: introduce a build() step in the reconciler that + assembles all relevant Kubernetes resources before anything is applied ([#801]). + +[#801]: https://github.com/stackabletech/hdfs-operator/pull/801 + ## [26.7.0] - 2026-07-21 ## [26.7.0-rc1] - 2026-07-16