From 848409a6c81cdd378746c2d63fb59716328cdf8e Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Thu, 30 Jul 2026 18:14:23 +0200 Subject: [PATCH] extract apply and update_status steps --- rust/operator-binary/src/controller/apply.rs | 220 ++++++++++++++++++ .../src/controller/build/mod.rs | 7 +- rust/operator-binary/src/controller/mod.rs | 17 +- .../src/controller/update_status.rs | 86 +++++++ rust/operator-binary/src/hdfs_controller.rs | 208 +++-------------- 5 files changed, 353 insertions(+), 185 deletions(-) create mode 100644 rust/operator-binary/src/controller/apply.rs create mode 100644 rust/operator-binary/src/controller/update_status.rs diff --git a/rust/operator-binary/src/controller/apply.rs b/rust/operator-binary/src/controller/apply.rs new file mode 100644 index 00000000..cc019648 --- /dev/null +++ b/rust/operator-binary/src/controller/apply.rs @@ -0,0 +1,220 @@ +//! The apply step in the HdfsCluster controller. + +use std::marker::PhantomData; + +use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + client::Client, + cluster_resources::{ClusterResource, ClusterResourceApplyStrategy, ClusterResources}, + deep_merger::ObjectOverrides, + iter::reverse_if, + k8s_openapi::api::core::v1::ConfigMap, + kube::{ResourceExt, runtime::reflector::ObjectRef}, + status::rollout::check_statefulset_rollout_complete, + v2::cluster_resources::cluster_resources_new, +}; +use strum::{EnumDiscriminants, IntoStaticStr}; + +use crate::{ + controller::{ + Applied, KubernetesResources, Prepared, ValidatedCluster, controller_name, operator_name, + product_name, + }, + crd::{UpgradeState, constants::FIELD_MANAGER_SCOPE}, +}; + +#[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 apply the StatefulSet {name:?}"))] + ApplyRoleGroupStatefulSet { + source: stackable_operator::cluster_resources::Error, + name: String, + }, + + #[snafu(display("cannot create discovery config map {name:?}"))] + ApplyDiscoveryConfigMap { + source: stackable_operator::client::Error, + name: String, + }, + + #[snafu(display("failed to delete orphaned resources"))] + DeleteOrphanedResources { + source: stackable_operator::cluster_resources::Error, + }, +} + +type Result = std::result::Result; + +/// The outcome of the apply step: the applied resources, plus whether every StatefulSet was +/// applied and — during an upgrade or downgrade — fully rolled out. +pub struct AppliedResources { + pub resources: KubernetesResources, + /// `false` while a rolling upgrade or downgrade is still in progress. The role-ordered + /// rollout then stopped at the incomplete StatefulSet, so the later ones were not applied + /// in this run, and the status must keep its upgrade/downgrade state. + pub statefulsets_rolled_out: bool, +} + +/// Applier for the Kubernetes resource specifications produced by this controller. +/// +/// Unlike its siblings in the other operators, this Applier is HDFS-specific: StatefulSets are +/// rolled out in role order during upgrades (reversed for downgrades), each role gated on the +/// previous one's rollout being complete. +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> { + let cluster_resources = cluster_resources_new( + &product_name(), + &operator_name(), + &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. + /// + /// `applied.resources.stateful_sets` contains only the StatefulSets that were actually + /// applied: during an upgrade or downgrade the role-ordered rollout stops at the first + /// StatefulSet whose rollout is incomplete (see [`AppliedResources`]). + pub async fn apply( + mut self, + resources: KubernetesResources, + upgrade_state: Option, + ) -> Result { + // Destructured without `..`, so adding a field to [`KubernetesResources`] fails to + // compile here instead of silently never being applied. + let KubernetesResources { + services, + config_maps, + pod_disruption_budgets, + stateful_sets, + service_accounts, + role_bindings, + status: _, + } = resources; + + // Apply order is: StatefulSets last (a changed mounted ConfigMap/Secret + // must exist first, else Pods restart -- 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 config_maps = self.add_resources(config_maps).await?; + let pod_disruption_budgets = self.add_resources(pod_disruption_budgets).await?; + + // 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"); + } + let mut applied_stateful_sets = vec![]; + let mut statefulsets_rolled_out = true; + for statefulset in reverse_if(downgrading, stateful_sets.into_iter()) { + let name = statefulset.name_any(); + let applied_statefulset = self + .cluster_resources + .add(self.client, statefulset) + .await + .with_context(|_| ApplyRoleGroupStatefulSetSnafu { name })?; + + if upgrade_state.is_some() + && let Err(reason) = check_statefulset_rollout_complete(&applied_statefulset) + { + // Ensure each role is fully upgraded before moving on to the next. + tracing::info!( + rolegroup.statefulset = %ObjectRef::from_obj(&applied_statefulset), + reason = &reason as &dyn std::error::Error, + "rolegroup is still upgrading, waiting..." + ); + applied_stateful_sets.push(applied_statefulset); + statefulsets_rolled_out = false; + break; + } + applied_stateful_sets.push(applied_statefulset); + } + + // During upgrades we do partial deployments; we don't want to garbage collect after + // those since we *will* redeploy (or properly orphan) the remaining resources later. + if statefulsets_rolled_out { + self.cluster_resources + .delete_orphaned_resources(self.client) + .await + .context(DeleteOrphanedResourcesSnafu)?; + } + + Ok(AppliedResources { + resources: KubernetesResources { + stateful_sets: applied_stateful_sets, + services, + config_maps, + pod_disruption_budgets, + service_accounts, + role_bindings, + status: PhantomData, + }, + statefulsets_rolled_out, + }) + } + + async fn add_resources( + &mut self, + resources: Vec, + ) -> Result> { + let mut applied_resources = vec![]; + + for resource in resources { + let applied_resource = self + .cluster_resources + .add(self.client, resource) + .await + .context(ApplyResourceSnafu)?; + applied_resources.push(applied_resource); + } + + Ok(applied_resources) + } +} + +/// Applies the discovery `ConfigMap` directly, outside the [`ClusterResources`] tracking. +/// +/// The discovery CM is linked to the cluster lifecycle via ownerreference. Therefore, it must +/// not be added to the "orphaned" cluster resources: it is applied after +/// [`Applier::apply`], whose orphan deletion must never see it. +pub async fn apply_discovery_config_map(client: &Client, discovery_cm: &ConfigMap) -> Result<()> { + client + .apply_patch(FIELD_MANAGER_SCOPE, discovery_cm, discovery_cm) + .await + .with_context(|_| ApplyDiscoveryConfigMapSnafu { + name: discovery_cm.metadata.name.clone().unwrap_or_default(), + })?; + Ok(()) +} diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs index a19c50d0..f5cb657f 100644 --- a/rust/operator-binary/src/controller/build/mod.rs +++ b/rust/operator-binary/src/controller/build/mod.rs @@ -1,4 +1,4 @@ -use std::collections::HashMap; +use std::{collections::HashMap, marker::PhantomData}; use snafu::{ResultExt, Snafu}; use stackable_operator::{ @@ -13,7 +13,7 @@ use stackable_operator::{ use crate::{ controller::{ - KubernetesResources, ValidatedCluster, + KubernetesResources, Prepared, ValidatedCluster, build::resource::rbac::{build_role_binding, build_service_account}, }, crd::{ @@ -82,7 +82,7 @@ pub enum Error { pub fn build( cluster: &ValidatedCluster, cluster_info: &KubernetesClusterInfo, -) -> Result { +) -> Result, Error> { let mut services = vec![]; let mut config_maps = vec![]; let mut stateful_sets = vec![]; @@ -143,6 +143,7 @@ pub fn build( stateful_sets, service_accounts: vec![build_service_account(cluster)], role_bindings: vec![build_role_binding(cluster)], + status: PhantomData, }) } diff --git a/rust/operator-binary/src/controller/mod.rs b/rust/operator-binary/src/controller/mod.rs index 2f351d7a..1b768d85 100644 --- a/rust/operator-binary/src/controller/mod.rs +++ b/rust/operator-binary/src/controller/mod.rs @@ -1,4 +1,4 @@ -use std::{collections::BTreeMap, str::FromStr}; +use std::{collections::BTreeMap, marker::PhantomData, str::FromStr}; use stackable_operator::{ commons::product_image_selection::ResolvedProductImage, @@ -35,10 +35,18 @@ use crate::{ hdfs_controller::RESOURCE_MANAGER_HDFS_CONTROLLER, }; +pub mod apply; pub mod build; pub mod dereference; +pub mod update_status; pub mod validate; +/// 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 build step. /// /// The resources are flat, unordered collections. The reconcile step re-groups the @@ -46,13 +54,18 @@ pub mod validate; /// 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 { +/// +/// `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 services: Vec, pub config_maps: Vec, pub pod_disruption_budgets: Vec, pub stateful_sets: Vec, pub service_accounts: Vec, pub role_bindings: Vec, + pub status: PhantomData, } /// The [`RoleGroupConfig`] specialised for HDFS: the validated config is the diff --git a/rust/operator-binary/src/controller/update_status.rs b/rust/operator-binary/src/controller/update_status.rs new file mode 100644 index 00000000..c7947712 --- /dev/null +++ b/rust/operator-binary/src/controller/update_status.rs @@ -0,0 +1,86 @@ +//! The update_status step in the HdfsCluster controller. + +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, + controller::{ValidatedCluster, apply::AppliedResources}, + crd::{HdfsClusterStatus, UpgradeState, v1alpha1}, +}; + +#[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 outcome of the apply step and patches it onto the +/// [`v1alpha1::HdfsCluster`]. Takes [`AppliedResources`] so the type system proves the status +/// derives from applied resources — including whether the role-ordered StatefulSet rollout is +/// still in progress — not merely built ones. +pub async fn update_status( + client: &Client, + hdfs: &v1alpha1::HdfsCluster, + cluster: &ValidatedCluster, + applied: &AppliedResources, +) -> Result<()> { + let mut ss_cond_builder = StatefulSetConditionBuilder::default(); + for stateful_set in &applied.resources.stateful_sets { + ss_cond_builder.add(stateful_set.clone()); + } + + let cluster_operation_cond_builder = + ClusterOperationsConditionBuilder::new(&hdfs.spec.cluster_operation); + + let upgrade_state = cluster.status.upgrade_state; + + let status = HdfsClusterStatus { + conditions: compute_conditions(hdfs, &[&ss_cond_builder, &cluster_operation_cond_builder]), + // FIXME: We can't currently leave upgrade mode automatically, since we don't know when an upgrade is finalized + deployed_product_version: Some( + cluster + .status + .deployed_product_version + .clone() + // Keep current version if set, otherwise (on initial deploy) fall back + // to the user's specified version. + .unwrap_or_else(|| cluster.image.product_version.clone()), + ), + upgrade_target_product_version: match upgrade_state { + // User is upgrading, whatever they're upgrading to is (by definition) the target + Some(UpgradeState::Upgrading) => Some(cluster.image.product_version.clone()), + Some(UpgradeState::Downgrading) => { + if applied.statefulsets_rolled_out { + // Downgrade is done, clear + tracing::info!("downgrade deployed, clearing upgrade state"); + None + } else { + // Downgrade is still in progress, preserve the current value + cluster.status.upgrade_target_product_version.clone() + } + } + // Upgrade is complete (if any), clear + None => None, + }, + }; + + client + .apply_patch_status(OPERATOR_NAME, hdfs, &status) + .await + .context(ApplyStatusSnafu)?; + + Ok(()) +} diff --git a/rust/operator-binary/src/hdfs_controller.rs b/rust/operator-binary/src/hdfs_controller.rs index 5b94a220..11a97642 100644 --- a/rust/operator-binary/src/hdfs_controller.rs +++ b/rust/operator-binary/src/hdfs_controller.rs @@ -5,36 +5,27 @@ use stackable_operator::{ cli::OperatorEnvironmentOptions, client::Client, cluster_resources::ClusterResourceApplyStrategy, - iter::reverse_if, kube::{ - Resource, ResourceExt, + Resource, core::{DeserializeGuard, error_boundary}, - runtime::{controller::Action, events::Recorder, reflector::ObjectRef}, + runtime::{controller::Action, events::Recorder}, }, kvp::LabelError, logging::controller::ReconcilerError, shared::time::Duration, - status::{ - condition::{ - compute_conditions, operations::ClusterOperationsConditionBuilder, - statefulset::StatefulSetConditionBuilder, - }, - rollout::check_statefulset_rollout_complete, - }, - v2::cluster_resources::cluster_resources_new, }; use strum::{EnumDiscriminants, IntoEnumIterator, IntoStaticStr}; use crate::{ - OPERATOR_NAME, controller::{ + apply::{self, Applier, apply_discovery_config_map}, build::{ self, resource::discovery::{self, build_discovery_config_map}, }, - controller_name, operator_name, product_name, + update_status::{self, update_status}, }, - crd::{HdfsClusterStatus, HdfsNodeRole, UpgradeState, constants::*, v1alpha1}, + crd::{HdfsNodeRole, v1alpha1}, event::{build_invalid_replica_message, publish_warning_event}, }; @@ -44,10 +35,11 @@ pub const HDFS_CONTROLLER_NAME: &str = "hdfs-controller"; #[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 apply the Kubernetes resources"))] + ApplyResources { source: apply::Error }, + + #[snafu(display("failed to update the cluster status"))] + UpdateStatus { source: update_status::Error }, #[snafu(display("failed to dereference cluster resources"))] Dereference { @@ -59,17 +51,8 @@ pub enum Error { source: crate::controller::validate::Error, }, - #[snafu(display("cannot create role group stateful set {name:?}"))] - ApplyRoleGroupStatefulSet { - source: stackable_operator::cluster_resources::Error, - name: String, - }, - - #[snafu(display("cannot create discovery config map {name:?}"))] - ApplyDiscoveryConfigMap { - source: stackable_operator::client::Error, - name: String, - }, + #[snafu(display("failed to apply the discovery ConfigMap"))] + ApplyDiscoveryConfigMap { source: apply::Error }, #[snafu(display("failed to build Kubernetes resources"))] BuildResources { @@ -82,24 +65,9 @@ pub enum Error { #[snafu(display("cannot build config discovery config map"))] BuildDiscoveryConfigMap { source: discovery::Error }, - #[snafu(display("failed to delete orphaned resources"))] - DeleteOrphanedResources { - source: stackable_operator::cluster_resources::Error, - }, - #[snafu(display("failed to create cluster event"))] FailedToCreateClusterEvent { source: crate::event::Error }, - #[snafu(display("failed to apply PodDisruptionBudget"))] - ApplyPdb { - source: stackable_operator::cluster_resources::Error, - }, - - #[snafu(display("failed to update status"))] - ApplyStatus { - source: stackable_operator::client::Error, - }, - #[snafu(display("failed to build cluster resources label"))] BuildClusterResourcesLabel { source: LabelError }, @@ -147,59 +115,12 @@ pub async fn reconcile_hdfs( ) .context(ValidateSnafu)?; - let mut cluster_resources = cluster_resources_new( - &product_name(), - &operator_name(), - &controller_name(), - &validated_cluster.name, - &validated_cluster.namespace, - &validated_cluster.uid, - ClusterResourceApplyStrategy::from(&hdfs.spec.cluster_operation), - &hdfs.spec.object_overrides, - ); - // 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) .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_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 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)?; - } - - let upgrade_state = validated_cluster.status.upgrade_state; - // Warn about invalid replica counts. This is validation feedback and independent of the // resource application below. for role in HdfsNodeRole::iter() { @@ -219,42 +140,19 @@ pub async fn reconcile_hdfs( } } - let mut ss_cond_builder = StatefulSetConditionBuilder::default(); - let mut deploy_done = true; - - // 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; - } - } - } + let applied = Applier::new( + client, + &validated_cluster, + ClusterResourceApplyStrategy::from(&hdfs.spec.cluster_operation), + &hdfs.spec.object_overrides, + ) + .apply(resources, validated_cluster.status.upgrade_state) + .await + .context(ApplyResourcesSnafu)?; - // Discovery CM will fail to build until the rest of the cluster has been deployed, so do it last - // so that failure won't inhibit the rest of the cluster from booting up. + // Discovery CM will fail to build until the rest of the cluster has been + // deployed, so do it last so that failure won't inhibit the rest of the + // cluster from booting up. let discovery_cm = build_discovery_config_map( &validated_cluster, &client.kubernetes_cluster_info, @@ -267,63 +165,13 @@ pub async fn reconcile_hdfs( ) .context(BuildDiscoveryConfigMapSnafu)?; - // The discovery CM is linked to the cluster lifecycle via ownerreference. - // Therefore, must not be added to the "orphaned" cluster resources - client - .apply_patch(FIELD_MANAGER_SCOPE, &discovery_cm, &discovery_cm) + apply_discovery_config_map(client, &discovery_cm) .await - .with_context(|_| ApplyDiscoveryConfigMapSnafu { - name: discovery_cm.metadata.name.clone().unwrap_or_default(), - })?; - - let cluster_operation_cond_builder = - ClusterOperationsConditionBuilder::new(&hdfs.spec.cluster_operation); - - let status = HdfsClusterStatus { - conditions: compute_conditions(hdfs, &[&ss_cond_builder, &cluster_operation_cond_builder]), - // FIXME: We can't currently leave upgrade mode automatically, since we don't know when an upgrade is finalized - deployed_product_version: Some( - validated_cluster - .status - .deployed_product_version - .clone() - // Keep current version if set, otherwise (on initial deploy) fall back - // to the user's specified version. - .unwrap_or_else(|| validated_cluster.image.product_version.clone()), - ), - upgrade_target_product_version: match upgrade_state { - // User is upgrading, whatever they're upgrading to is (by definition) the target - Some(UpgradeState::Upgrading) => Some(validated_cluster.image.product_version.clone()), - Some(UpgradeState::Downgrading) => { - if deploy_done { - // Downgrade is done, clear - tracing::info!("downgrade deployed, clearing upgrade state"); - None - } else { - // Downgrade is still in progress, preserve the current value - validated_cluster - .status - .upgrade_target_product_version - .clone() - } - } - // Upgrade is complete (if any), clear - None => None, - }, - }; + .context(ApplyDiscoveryConfigMapSnafu)?; - // During upgrades we do partial deployments, we don't want to garbage collect after those - // since we *will* redeploy (or properly orphan) the remaining resources later. - if deploy_done { - cluster_resources - .delete_orphaned_resources(client) - .await - .context(DeleteOrphanedResourcesSnafu)?; - } - client - .apply_patch_status(OPERATOR_NAME, hdfs, &status) + update_status(client, hdfs, &validated_cluster, &applied) .await - .context(ApplyStatusSnafu)?; + .context(UpdateStatusSnafu)?; Ok(Action::await_change()) }