Move upgrade readiness check into utils module

Author: nightkr

rust/operator-binary/src/hdfs_controller.rs

@@ -62,6 +62,7 @@ use crate::{
     },
     product_logging::{extend_role_group_config_map, resolve_vector_aggregator_address},
     security::{self, kerberos, opa::HdfsOpaConfig},
+    utils::statefulset::check_all_replicas_updated,
     OPERATOR_NAME,
 };
 
@@ -412,37 +413,19 @@ pub async fn reconcile_hdfs(hdfs: Arc<HdfsCluster>, ctx: Arc<Ctx>) -> HdfsOperat
                     name: rg_configmap_name,
                 })?;
             let rg_statefulset_name = rg_statefulset.name_any();
-            let mut deployed_rg_statefulset = cluster_resources
+            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 hdfs.is_upgrading() {
-                let status = deployed_rg_statefulset.status.take().unwrap_or_default();
-
-                let current_generation = deployed_rg_statefulset.metadata.generation;
-                let observed_generation = status.observed_generation;
-                if current_generation != observed_generation {
-                    tracing::info!(
-                        object = %ObjectRef::from_obj(&deployed_rg_statefulset),
-                        generation.current = current_generation,
-                        generation.observed = observed_generation,
-                        "rolegroup is still upgrading, waiting... (generation not yet observed by statefulset controller)",
-                    );
-                    deploy_done = false;
-                    break 'roles;
-                }
-
-                let total_replicas = status.replicas;
-                let updated_replicas = status.updated_replicas.unwrap_or(0);
-                if total_replicas != updated_replicas {
+                if let Err(reason) = check_all_replicas_updated(&deployed_rg_statefulset) {
                     tracing::info!(
                         object = %ObjectRef::from_obj(&deployed_rg_statefulset),
-                        replicas.total = total_replicas,
-                        replicas.updated = updated_replicas,
-                        "rolegroup is still upgrading, waiting... (not all replicas are updated)",
+                        reason = &reason as &dyn std::error::Error,
+                        "rolegroup is still upgrading, waiting..."
                     );
                     deploy_done = false;
                     break 'roles;

rust/operator-binary/src/main.rs

@@ -34,6 +34,7 @@ mod hdfs_controller;
 mod operations;
 mod product_logging;
 mod security;
+mod utils;
 
 mod built_info {
     include!(concat!(env!("OUT_DIR"), "/built.rs"));

rust/operator-binary/src/utils/mod.rs

@@ -0,0 +1 @@
+pub mod statefulset;

rust/operator-binary/src/utils/statefulset.rs

@@ -0,0 +1,49 @@
+use std::borrow::Cow;
+
+use snafu::Snafu;
+use stackable_operator::k8s_openapi::api::apps::v1::StatefulSet;
+
+#[derive(Debug, Snafu)]
+#[snafu(module(outdated_statefulset))]
+pub enum OutdatedStatefulSet {
+    #[snafu(display("generation {current_generation:?} not yet observed by statefulset controller, last seen was {observed_generation:?}"))]
+    NotYetObserved {
+        current_generation: Option<i64>,
+        observed_generation: Option<i64>,
+    },
+
+    #[snafu(display("only {updated_replicas} out of {total_replicas} are updated"))]
+    HasOutdatedReplicas {
+        total_replicas: i32,
+        updated_replicas: i32,
+    },
+}
+
+/// Checks whether all ReplicaSet replicas are up-to-date according to `sts.spec`
+pub fn check_all_replicas_updated(sts: &StatefulSet) -> Result<(), OutdatedStatefulSet> {
+    use outdated_statefulset::*;
+
+    let status = sts.status.as_ref().map_or_else(Cow::default, Cow::Borrowed);
+
+    let current_generation = sts.metadata.generation;
+    let observed_generation = status.observed_generation;
+    if current_generation != observed_generation {
+        return NotYetObservedSnafu {
+            current_generation,
+            observed_generation,
+        }
+        .fail();
+    }
+
+    let total_replicas = status.replicas;
+    let updated_replicas = status.updated_replicas.unwrap_or(0);
+    if total_replicas != updated_replicas {
+        return HasOutdatedReplicasSnafu {
+            total_replicas,
+            updated_replicas,
+        }
+        .fail();
+    }
+
+    Ok(())
+}