Rolling HDFS upgrade

Cargo.toml

@@ -21,13 +21,13 @@ 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"] }
 tracing = "0.1"
 tracing-futures = { version = "0.2", features = ["futures-03"] }
 
-#[patch."https://github.com/stackabletech/operator-rs.git"]
+[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 = "main" }

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

@@ -0,0 +1,107 @@
+= Upgrading HDFS
+
+IMPORTANT: HDFS upgrades are experimental, and details may change at any time
+
+HDFS currently requires a manual process to upgrade. This guide will take you through an example case, upgrading an example cluster (from our xref:getting_started/index.adoc[Getting Started] guide) from HDFS 3.3.6 to 3.4.0.
+
+== Preparing for the worst
+
+Upgrades can fail, and it is important to prepare for when that happens. Apache HDFS supports https://hadoop.apache.org/docs/r3.4.0/hadoop-project-dist/hadoop-hdfs/HdfsRollingUpgrade.html#Downgrade_and_Rollback[two ways to revert an upgrade]:
+
+Rollback:: Reverts all user data to the pre-upgrade state. Requires taking the cluster offline.
+Downgrade:: Downgrades the HDFS software but preserves all changes made by users. Can be performed as a rolling change, keeping the cluster online.
+
+The Stackable Operator for HDFS supports downgrading but not rollbacks.
+
+In order to downgrade, revert the `.spec.image.productVersion` field, and then proceed to xref:#finalize[finalizing] once the cluster is downgraded:
+
+[source,shell]
+----
+$ kubectl patch hdfs/simple-hdfs --patch '{"spec": {"image": {"productVersion": "3.3.6"}}}' --type=merge
+hdfscluster.hdfs.stackable.tech/simple-hdfs patched
+----
+
+== 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
+(either a client configured with a superuser account, or from inside a NameNode pod):
+
+// This could be automated by the operator, but dfsadmin does not have good machine-readable output.
+// It *can* be queried over JMX, but we're not so lucky for finalization.
+
+[source,shell]
+----
+$ hdfs dfsadmin -rollingUpgrade prepare
+PREPARE rolling upgrade ...
+Preparing for upgrade. Data is being saved for rollback.
+Run "dfsadmin -rollingUpgrade query" to check the status
+for proceeding with rolling upgrade
+  Block Pool ID: BP-841432641-10.244.0.29-1722612757853
+     Start Time: Fri Aug 02 15:49:12 GMT 2024 (=1722613752341)
+  Finalize Time: <NOT FINALIZED>
+
+$ # Then run query until the HDFS is ready to proceed
+$ hdfs dfsadmin -rollingUpgrade query
+QUERY rolling upgrade ...
+Preparing for upgrade. Data is being saved for rollback.
+Run "dfsadmin -rollingUpgrade query" to check the status
+for proceeding with rolling upgrade
+  Block Pool ID: BP-841432641-10.244.0.29-1722612757853
+     Start Time: Fri Aug 02 15:49:12 GMT 2024 (=1722613752341)
+  Finalize Time: <NOT FINALIZED>
+
+$ # It is safe to proceed when the output indicates so, like this:
+$ hdfs dfsadmin -rollingUpgrade query
+QUERY rolling upgrade ...
+Proceed with rolling upgrade:
+  Block Pool ID: BP-841432641-10.244.0.29-1722612757853
+     Start Time: Fri Aug 02 15:49:12 GMT 2024 (=1722613752341)
+  Finalize Time: <NOT FINALIZED>
+----
+
+== Starting the upgrade
+
+Once HDFS is ready to upgrade, the HdfsCluster can be updated with the new product version:
+
+[source,shell]
+----
+$ kubectl patch hdfs/simple-hdfs --patch '{"spec": {"image": {"productVersion": "3.4.0"}}}' --type=merge
+hdfscluster.hdfs.stackable.tech/simple-hdfs patched
+----
+
+Then wait until all pods have restarted, are in the Ready state, and running the new HDFS version.
+
+NOTE: This will automatically enable the NameNodes' compatibility mode, allowing them to start despite the fsImage version mismatch.
+
+NOTE: Services will be upgraded in order: JournalNodes, then NameNodes, then DataNodes.
+
+[#finalize]
+== Finalizing the upgrade
+
+Once all HDFS pods are running the new version, the HDFS upgrade can be finalized (from the HDFS superuser environment as described in the preparation step):
+
+[source,shell]
+----
+$ hdfs dfsadmin -rollingUpgrade finalize
+FINALIZE rolling upgrade ...
+Rolling upgrade is finalized.
+  Block Pool ID: BP-841432641-10.244.0.29-1722612757853
+     Start Time: Fri Aug 02 15:49:12 GMT 2024 (=1722613752341)
+  Finalize Time: Fri Aug 02 15:58:39 GMT 2024 (=1722614319854)
+----
+
+// We can't safely automate this, because finalize is asynchronous and doesn't tell us whether all NameNodes have even received the request to finalize.
+
+WARNING: Please ensure that all NameNodes are running and available before proceeding. NameNodes that have not finalized yet will crash on launch when taken out of compatibility mode.
+
+Finally, mark the cluster as upgraded:
+
+[source,shell]
+----
+$ kubectl patch hdfs/simple-hdfs --subresource=status --patch '{"status": {"deployedProductVersion": "3.4.0"}}' --type=merge
+hdfscluster.hdfs.stackable.tech/simple-hdfs patched
+----
+
+NOTE: `deployedProductVersion` is located in the _status_ subresource, which will not be modified by most graphical editors, and `kubectl` requires the `--subresource=status` flag.
+
+The NameNodes will then be restarted a final time, taking them out of compatibility mode.

docs/modules/hdfs/partials/nav.adoc

@@ -10,6 +10,7 @@
 ** xref:hdfs:usage-guide/logging-log-aggregation.adoc[]
 ** xref:hdfs:usage-guide/monitoring.adoc[]
 ** xref:hdfs:usage-guide/configuration-environment-overrides.adoc[]
+** xref:hdfs:usage-guide/upgrading.adoc[]
 ** xref:hdfs:usage-guide/operations/index.adoc[]
 *** xref:hdfs:usage-guide/operations/cluster-operations.adoc[]
 *** xref:hdfs:usage-guide/operations/pod-placement.adoc[]

rust/crd/src/lib.rs

@@ -41,7 +41,7 @@ use stackable_operator::{
     status::condition::{ClusterCondition, HasStatusCondition},
     time::Duration,
 };
-use strum::{Display, EnumIter, EnumString};
+use strum::{Display, EnumIter, EnumString, IntoStaticStr};
 
 use crate::{
     affinity::get_affinity,
@@ -312,27 +312,29 @@ impl AnyNodeConfig {
 
 #[derive(
     Clone,
+    Copy,
     Debug,
     Deserialize,
     Display,
     EnumIter,
     EnumString,
+    IntoStaticStr,
     Eq,
     Hash,
     JsonSchema,
     PartialEq,
     Serialize,
 )]
 pub enum HdfsRole {
+    #[serde(rename = "journalnode")]
+    #[strum(serialize = "journalnode")]
+    JournalNode,
     #[serde(rename = "namenode")]
     #[strum(serialize = "namenode")]
     NameNode,
     #[serde(rename = "datanode")]
     #[strum(serialize = "datanode")]
     DataNode,
-    #[serde(rename = "journalnode")]
-    #[strum(serialize = "journalnode")]
-    JournalNode,
 }
 
 impl HdfsRole {
@@ -802,6 +804,24 @@ impl HdfsCluster {
         Ok(result)
     }
 
+    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.upgrade_target_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> {
         self.spec.cluster_config.authentication.as_ref()
     }
@@ -955,6 +975,15 @@ 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,
@@ -1322,6 +1351,14 @@ impl Configuration for JournalNodeConfigFragment {
 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 upgrade_target_product_version: Option<String>,
 }
 
 impl HasStatusCondition for HdfsCluster {

rust/operator-binary/src/container.rs

@@ -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};
 
@@ -212,7 +213,7 @@ impl ContainerConfig {
         labels: &Labels,
     ) -> Result<(), Error> {
         // HDFS main container
-        let main_container_config = Self::from(role.clone());
+        let main_container_config = Self::from(*role);
         pb.add_volumes(main_container_config.volumes(merged_config, object_name, labels)?);
         pb.add_container(main_container_config.main_container(
             hdfs,
@@ -548,6 +549,14 @@ impl ContainerConfig {
             args.push_str(&Self::export_kerberos_real_env_var_command());
         }
 
+        let upgrade_args = if hdfs.upgrade_state() == Some(UpgradeState::Upgrading)
+            && *role == HdfsRole::NameNode
+        {
+            "-rollingUpgrade started"
+        } else {
+            ""
+        };
+
         match self {
             ContainerConfig::Hdfs { role, .. } => {
                 args.push_str(&self.copy_log4j_properties_cmd(
@@ -566,7 +575,7 @@ if [[ -d {LISTENER_VOLUME_DIR} ]]; then
         export $(basename $i | tr a-z- A-Z_)_PORT="$(cat $i)"
     done
 fi
-{hadoop_home}/bin/hdfs {role} &
+{hadoop_home}/bin/hdfs {role} {upgrade_args} &
 wait_for_termination $!
 {create_vector_shutdown_file_command}
 "#,
@@ -1317,7 +1326,7 @@ impl From<HdfsRole> for ContainerConfig {
     fn from(role: HdfsRole) -> Self {
         match role {
             HdfsRole::NameNode => Self::Hdfs {
-                role: role.clone(),
+                role,
                 container_name: role.to_string(),
                 volume_mounts: ContainerVolumeDirs::from(role),
                 ipc_port_name: SERVICE_PORT_NAME_RPC,
@@ -1327,7 +1336,7 @@ impl From<HdfsRole> for ContainerConfig {
                 metrics_port: DEFAULT_NAME_NODE_METRICS_PORT,
             },
             HdfsRole::DataNode => Self::Hdfs {
-                role: role.clone(),
+                role,
                 container_name: role.to_string(),
                 volume_mounts: ContainerVolumeDirs::from(role),
                 ipc_port_name: SERVICE_PORT_NAME_IPC,
@@ -1337,7 +1346,7 @@ impl From<HdfsRole> for ContainerConfig {
                 metrics_port: DEFAULT_DATA_NODE_METRICS_PORT,
             },
             HdfsRole::JournalNode => Self::Hdfs {
-                role: role.clone(),
+                role,
                 container_name: role.to_string(),
                 volume_mounts: ContainerVolumeDirs::from(role),
                 ipc_port_name: SERVICE_PORT_NAME_RPC,