Perform downgrades in order

Author: nightkr

Cargo.lock

@@ -21,7 +21,7 @@ serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 serde_yaml = "0.9"
 snafu = "0.8"
-stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "stackable-operator-0.70.0" }
+stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "stackable-operator-0.73.0" }
 product-config = { git = "https://github.com/stackabletech/product-config.git", tag = "0.7.0" }
 strum = { version = "0.26", features = ["derive"] }
 tokio = { version = "1.38", features = ["full"] }
@@ -31,4 +31,4 @@ tracing-futures = { version = "0.2", features = ["futures-03"] }
 [patch."https://github.com/stackabletech/operator-rs.git"]
 #stackable-operator = { path = "../operator-rs/crates/stackable-operator" }
 #stackable-operator = { git = "https://github.com/stackabletech//operator-rs.git", branch = "main" }
-stackable-operator = { git = "https://github.com/stackabletech//operator-rs.git", branch = "revert/lazylock" }
+stackable-operator = { git = "https://github.com/stackabletech//operator-rs.git", branch = "feature/iter-reverse-if" }

Cargo.nix

@@ -21,8 +21,6 @@ $ kubectl patch hdfs/simple-hdfs --patch '{"spec": {"image": {"productVersion":
 hdfscluster.hdfs.stackable.tech/simple-hdfs patched
 ----
 
-WARNING: The Stackable Operator for HDFS will _not_ currently roll downgrades in order, they will incur some downtime.
-
 == Preparing HDFS
 
 HDFS must be configured to initiate the upgrade process. To do this, put the cluster into upgrade mode by running the following commands in an HDFS superuser environment

Cargo.toml

@@ -804,13 +804,22 @@ impl HdfsCluster {
         Ok(result)
     }
 
-    pub fn is_upgrading(&self) -> bool {
-        self.status
-            .as_ref()
-            .and_then(|status| status.deployed_product_version.as_deref())
-            .map_or(false, |deployed_version| {
-                deployed_version != self.spec.image.product_version()
-            })
+    pub fn upgrade_state(&self) -> Option<UpgradeState> {
+        let status = self.status.as_ref()?;
+        let requested_version = self.spec.image.product_version();
+
+        if requested_version != status.deployed_product_version.as_deref()? {
+            // If we're requesting a different version than what is deployed, assume that we're upgrading.
+            // Could also be a downgrade to an older version, but we don't support downgrades after upgrade finalization.
+            Some(UpgradeState::Upgrading)
+        } else if requested_version != status.upgrading_product_version.as_deref()? {
+            // If we're requesting the old version mid-upgrade, assume that we're downgrading.
+            // We only support downgrading to the exact previous version.
+            Some(UpgradeState::Downgrading)
+        } else {
+            // All three versions match, upgrade was completed without clearing `upgrading_product_version`.
+            None
+        }
     }
 
     pub fn authentication_config(&self) -> Option<&AuthenticationConfig> {
@@ -966,6 +975,14 @@ impl HdfsPodRef {
     }
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum UpgradeState {
+    /// The cluster is currently being upgraded to a new version.
+    Upgrading,
+    /// The cluster is currently being downgraded to the previous version.
+    Downgrading,
+}
+
 #[derive(
     Clone,
     Debug,
@@ -1334,7 +1351,13 @@ pub struct HdfsClusterStatus {
     #[serde(default)]
     pub conditions: Vec<ClusterCondition>,
 
+    /// The product version that the HDFS cluster is currently running.
+    ///
+    /// During upgrades, this field contains the *old* version.
     pub deployed_product_version: Option<String>,
+
+    /// The product version that is currently being upgraded to, otherwise null.
+    pub upgrading_product_version: Option<String>,
 }
 
 impl HasStatusCondition for HdfsCluster {

crate-hashes.json

@@ -12,6 +12,7 @@
 use crate::DATANODE_ROOT_DATA_DIR_PREFIX;
 use crate::JVM_SECURITY_PROPERTIES_FILE;
 use crate::LOG4J_PROPERTIES;
+use stackable_hdfs_crd::UpgradeState;
 use stackable_operator::utils::COMMON_BASH_TRAP_FUNCTIONS;
 use std::{collections::BTreeMap, str::FromStr};
 
@@ -548,7 +549,9 @@ impl ContainerConfig {
             args.push_str(&Self::export_kerberos_real_env_var_command());
         }
 
-        let upgrade_args = if hdfs.is_upgrading() && *role == HdfsRole::NameNode {
+        let upgrade_args = if hdfs.upgrade_state() == Some(UpgradeState::Upgrading)
+            && *role == HdfsRole::NameNode
+        {
             "-rollingUpgrade started"
         } else {
             ""

docs/modules/hdfs/pages/usage-guide/upgrading.adoc

@@ -21,6 +21,7 @@ use stackable_operator::{
         product_image_selection::ResolvedProductImage,
         rbac::{build_rbac_resources, service_account_name},
     },
+    iter::reverse_if,
     k8s_openapi::{
         api::{
             apps::v1::{StatefulSet, StatefulSetSpec},
@@ -50,7 +51,7 @@ use stackable_operator::{
 use strum::{EnumDiscriminants, IntoEnumIterator, IntoStaticStr};
 
 use stackable_hdfs_crd::{
-    constants::*, AnyNodeConfig, HdfsCluster, HdfsClusterStatus, HdfsPodRef, HdfsRole,
+    constants::*, AnyNodeConfig, HdfsCluster, HdfsClusterStatus, HdfsPodRef, HdfsRole, UpgradeState,
 };
 
 use crate::{
@@ -325,10 +326,23 @@ pub async fn reconcile_hdfs(hdfs: Arc<HdfsCluster>, ctx: Arc<Ctx>) -> HdfsOperat
     let dfs_replication = hdfs.spec.cluster_config.dfs_replication;
     let mut ss_cond_builder = StatefulSetConditionBuilder::default();
 
+    let upgrade_state = hdfs.upgrade_state();
     let mut deploy_done = true;
 
-    // Roles must be deployed in order during rolling upgrades
-    'roles: for role in HdfsRole::iter() {
+    // 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,
+        },
+        HdfsRole::iter(),
+    );
+    'roles: for role in roles {
         let role_name: &str = role.into();
         let Some(group_config) = validated_config.get(role_name) else {
             tracing::debug!(?role, "role has no configuration, skipping");
@@ -422,12 +436,12 @@ pub async fn reconcile_hdfs(hdfs: Arc<HdfsCluster>, ctx: Arc<Ctx>) -> HdfsOperat
                     name: rg_statefulset_name,
                 })?;
             ss_cond_builder.add(deployed_rg_statefulset.clone());
-            if hdfs.is_upgrading() {
+            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!(
-                        object = %ObjectRef::from_obj(&deployed_rg_statefulset),
+                        rolegroup.statefulset = %ObjectRef::from_obj(&deployed_rg_statefulset),
                         reason = &reason as &dyn std::error::Error,
                         "rolegroup is still upgrading, waiting..."
                     );
@@ -482,17 +496,30 @@ pub async fn reconcile_hdfs(hdfs: Arc<HdfsCluster>, ctx: Arc<Ctx>) -> HdfsOperat
         deployed_product_version: Some(
             hdfs.status
                 .as_ref()
+                // Keep current version if set
                 .and_then(|status| status.deployed_product_version.as_deref())
+                // Otherwise (on initial deploy) fall back to user's specified version
                 .unwrap_or(hdfs.spec.image.product_version())
                 .to_string(),
         ),
-        // deployed_product_version: if deploy_done {
-        //     Some(hdfs.spec.image.product_version().to_string())
-        // } else {
-        //     hdfs.status
-        //         .as_ref()
-        //         .and_then(|status| status.deployed_product_version.clone())
-        // },
+        upgrading_product_version: match upgrade_state {
+            // User is upgrading, whatever they're upgrading to is (by definition) the target
+            Some(UpgradeState::Upgrading) => Some(hdfs.spec.image.product_version().to_string()),
+            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
+                    hdfs.status
+                        .as_ref()
+                        .and_then(|status| status.upgrading_product_version.clone())
+                }
+            }
+            // Upgrade is complete (if any), clear
+            None => None,
+        },
     };
 
     // During upgrades we do partial deployments, we don't want to garbage collect after those