refactor: Extract apply and update status steps

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<T, E = Error> = std::result::Result<T, E>;
+
+/// 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<Applied>,
+    /// `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<Prepared>,
+        upgrade_state: Option<UpgradeState>,
+    ) -> Result<AppliedResources> {
+        // 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<T: ClusterResource + Sync>(
+        &mut self,
+        resources: Vec<T>,
+    ) -> Result<Vec<T>> {
+        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(())
+}

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<KubernetesResources, Error> {
+) -> Result<KubernetesResources<Prepared>, 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,
     })
 }
 

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,24 +35,37 @@ 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
 /// 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 {
+///
+/// `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<T> {
     pub services: Vec<Service>,
     pub config_maps: Vec<ConfigMap>,
     pub pod_disruption_budgets: Vec<PodDisruptionBudget>,
     pub stateful_sets: Vec<StatefulSet>,
     pub service_accounts: Vec<ServiceAccount>,
     pub role_bindings: Vec<RoleBinding>,
+    pub status: PhantomData<T>,
 }
 
 /// The [`RoleGroupConfig`] specialised for HDFS: the validated config is the

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<T, E = Error> = std::result::Result<T, E>;
+
+/// 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(())
+}