Added status field support. (#349)

Part of : stackabletech/issues#343
🟢 CI https://ci.stackable.tech/view/02%20Operator%20Tests%20(custom)/job/superset-operator-it-custom/30/

Author: razvan

CHANGELOG.md

@@ -7,6 +7,7 @@
 - Log aggregation added ([#326]).
 - Deploy default and support custom affinities ([#337]).
 - Extend cluster resources for status and cluster operation (paused, stopped) ([#348])
+- Cluster status conditions ([#349])
 
 ### Changed
 
@@ -26,6 +27,7 @@
 [#337]: https://github.com/stackabletech/superset-operator/pull/337
 [#344]: https://github.com/stackabletech/superset-operator/pull/344
 [#348]: https://github.com/stackabletech/superset-operator/pull/348
+[#349]: https://github.com/stackabletech/superset-operator/pull/349
 [#350]: https://github.com/stackabletech/superset-operator/pull/350
 
 ## [23.1.0] - 2023-01-23

deploy/helm/superset-operator/crds/crds.yaml

@@ -1517,6 +1517,51 @@ spec:
               type: object
             status:
               nullable: true
+              properties:
+                conditions:
+                  items:
+                    properties:
+                      lastTransitionTime:
+                        description: Last time the condition transitioned from one status to another.
+                        format: date-time
+                        nullable: true
+                        type: string
+                      lastUpdateTime:
+                        description: The last time this condition was updated.
+                        format: date-time
+                        nullable: true
+                        type: string
+                      message:
+                        description: A human readable message indicating details about the transition.
+                        nullable: true
+                        type: string
+                      reason:
+                        description: The reason for the condition's last transition.
+                        nullable: true
+                        type: string
+                      status:
+                        description: Status of the condition, one of True, False, Unknown.
+                        enum:
+                          - 'True'
+                          - 'False'
+                          - Unknown
+                        type: string
+                      type:
+                        description: Type of deployment condition.
+                        enum:
+                          - Available
+                          - Degraded
+                          - Progressing
+                          - ReconciliationPaused
+                          - Stopped
+                        type: string
+                    required:
+                      - status
+                      - type
+                    type: object
+                  type: array
+              required:
+                - conditions
               type: object
           required:
             - spec

deploy/helm/superset-operator/templates/roles.yaml

@@ -46,6 +46,7 @@ rules:
     resources:
       - statefulsets
     verbs:
+      - get
       - create
       - delete
       - list

rust/crd/src/lib.rs

@@ -10,6 +10,7 @@ use stackable_operator::commons::cluster_operation::ClusterOperation;
 use stackable_operator::commons::product_image_selection::ProductImage;
 use stackable_operator::kube::ResourceExt;
 use stackable_operator::role_utils::RoleGroup;
+use stackable_operator::status::condition::{ClusterCondition, HasStatusCondition};
 use stackable_operator::{
     commons::resources::{
         CpuLimitsFragment, MemoryLimitsFragment, NoRuntimeLimits, NoRuntimeLimitsFragment,
@@ -431,7 +432,18 @@ impl Configuration for SupersetConfigFragment {
 
 #[derive(Clone, Debug, Default, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
 #[serde(rename_all = "camelCase")]
-pub struct SupersetClusterStatus {}
+pub struct SupersetClusterStatus {
+    pub conditions: Vec<ClusterCondition>,
+}
+
+impl HasStatusCondition for SupersetCluster {
+    fn conditions(&self) -> Vec<ClusterCondition> {
+        match &self.status {
+            Some(status) => status.conditions.clone(),
+            None => vec![],
+        }
+    }
+}
 
 impl SupersetCluster {
     /// The name of the role-level load-balanced Kubernetes `Service`

rust/operator-binary/src/superset_controller.rs

@@ -39,7 +39,14 @@ use stackable_operator::{
     product_config_utils::{transform_all_roles_to_config, validate_all_roles_and_groups_config},
     product_logging::{self, spec::Logging},
     role_utils::RoleGroupRef,
+    status::condition::{
+        compute_conditions, operations::ClusterOperationsConditionBuilder,
+        statefulset::StatefulSetConditionBuilder, ClusterCondition, ClusterConditionSet,
+        ClusterConditionStatus, ClusterConditionType, ConditionBuilder,
+    },
 };
+use stackable_superset_crd::supersetdb::SupersetDBStatus;
+use stackable_superset_crd::SupersetClusterStatus;
 use stackable_superset_crd::{
     supersetdb::{SupersetDB, SupersetDBStatusCondition},
     Container, SupersetCluster, SupersetConfig, SupersetConfigOptions, SupersetRole, APP_NAME,
@@ -52,7 +59,6 @@ use std::{
     time::Duration,
 };
 use strum::{EnumDiscriminants, IntoStaticStr};
-use tracing::log::debug;
 
 pub const SUPERSET_CONTROLLER_NAME: &str = "supersetcluster";
 pub const DOCKER_IMAGE_BASE_NAME: &str = "superset";
@@ -172,6 +178,10 @@ pub enum Error {
         source: crate::product_logging::Error,
         cm_name: String,
     },
+    #[snafu(display("failed to update status"))]
+    ApplyStatus {
+        source: stackable_operator::error::Error,
+    },
 }
 
 type Result<T, E = Error> = std::result::Result<T, E>;
@@ -189,43 +199,17 @@ pub async fn reconcile_superset(superset: Arc<SupersetCluster>, ctx: Arc<Ctx>) -
     let resolved_product_image: ResolvedProductImage =
         superset.spec.image.resolve(DOCKER_IMAGE_BASE_NAME);
 
-    // Ensure DB Schema exists
-    let superset_db = SupersetDB::for_superset(&superset, &resolved_product_image)
-        .context(CreateSupersetObjectSnafu)?;
-    client
-        .apply_patch(SUPERSET_CONTROLLER_NAME, &superset_db, &superset_db)
-        .await
-        .context(ApplySupersetDBSnafu)?;
+    let cluster_operation_cond_builder =
+        ClusterOperationsConditionBuilder::new(&superset.spec.cluster_operation);
 
-    let superset_db = client
-        .get::<SupersetDB>(
-            &superset.name_unchecked(),
-            superset
-                .namespace()
-                .as_deref()
-                .context(ObjectHasNoNamespaceSnafu)?,
-        )
-        .await
-        .context(SupersetDBRetrievalSnafu)?;
-
-    if let Some(ref status) = superset_db.status {
-        match status.condition {
-            SupersetDBStatusCondition::Pending | SupersetDBStatusCondition::Initializing => {
-                debug!(
-                    "Waiting for SupersetDB initialization to complete, not starting Superset yet"
-                );
-                return Ok(Action::await_change());
-            }
-            SupersetDBStatusCondition::Failed => {
-                return SupersetDBFailedSnafu {
-                    superset_db: ObjectRef::from_obj(&superset_db),
-                }
-                .fail();
-            }
-            SupersetDBStatusCondition::Ready => (), // Continue starting Superset
-        }
-    } else {
-        debug!("Waiting for SupersetDBStatus to be reported, not starting Superset yet");
+    if wait_for_db_and_update_status(
+        client,
+        &superset,
+        &resolved_product_image,
+        &cluster_operation_cond_builder,
+    )
+    .await?
+    {
         return Ok(Action::await_change());
     }
 
@@ -300,6 +284,8 @@ pub async fn reconcile_superset(superset: Arc<SupersetCluster>, ctx: Arc<Ctx>) -
         None => None,
     };
 
+    let mut ss_cond_builder = StatefulSetConditionBuilder::default();
+
     for (rolegroup_name, rolegroup_config) in role_node_config.iter() {
         let rolegroup = superset.node_rolegroup_ref(rolegroup_name);
 
@@ -338,19 +324,32 @@ pub async fn reconcile_superset(superset: Arc<SupersetCluster>, ctx: Arc<Ctx>) -
             .with_context(|_| ApplyRoleGroupConfigSnafu {
                 rolegroup: rolegroup.clone(),
             })?;
-        cluster_resources
-            .add(client, rg_statefulset)
-            .await
-            .with_context(|_| ApplyRoleGroupStatefulSetSnafu {
-                rolegroup: rolegroup.clone(),
-            })?;
+        ss_cond_builder.add(
+            cluster_resources
+                .add(client, rg_statefulset.clone())
+                .await
+                .with_context(|_| ApplyRoleGroupStatefulSetSnafu {
+                    rolegroup: rolegroup.clone(),
+                })?,
+        );
     }
 
     cluster_resources
         .delete_orphaned_resources(client)
         .await
         .context(DeleteOrphanedResourcesSnafu)?;
 
+    let status = SupersetClusterStatus {
+        conditions: compute_conditions(
+            superset.as_ref(),
+            &[&ss_cond_builder, &cluster_operation_cond_builder],
+        ),
+    };
+    client
+        .apply_patch_status(OPERATOR_NAME, &*superset, &status)
+        .await
+        .context(ApplyStatusSnafu)?;
+
     Ok(Action::await_change())
 }
 
@@ -714,3 +713,117 @@ fn add_authentication_volumes_and_volume_mounts(
 pub fn error_policy(_obj: Arc<SupersetCluster>, _error: &Error, _ctx: Arc<Ctx>) -> Action {
     Action::requeue(Duration::from_secs(5))
 }
+
+/// Return true if the controller should wait for the DB to be set up.
+///
+/// As a side-effect, the Superset cluster status is updated as long as the controller waits
+/// for the DB to come up.
+///
+/// Having the DB set up by a Job managed by a different controller has it's own
+/// set of problems as described here: <https://github.com/stackabletech/superset-operator/issues/351>.
+/// The Airflow operator uses the same pattern as implemented here for setting up the DB.
+///
+/// When the ticket above is implemented, this function will most likely be removed completely.
+async fn wait_for_db_and_update_status(
+    client: &stackable_operator::client::Client,
+    superset: &SupersetCluster,
+    resolved_product_image: &ResolvedProductImage,
+    cluster_operation_condition_builder: &ClusterOperationsConditionBuilder<'_>,
+) -> Result<bool> {
+    // Ensure DB Schema exists
+    let superset_db = SupersetDB::for_superset(superset, resolved_product_image)
+        .context(CreateSupersetObjectSnafu)?;
+    client
+        .apply_patch(SUPERSET_CONTROLLER_NAME, &superset_db, &superset_db)
+        .await
+        .context(ApplySupersetDBSnafu)?;
+
+    let superset_db = client
+        .get::<SupersetDB>(
+            &superset.name_unchecked(),
+            superset
+                .namespace()
+                .as_deref()
+                .context(ObjectHasNoNamespaceSnafu)?,
+        )
+        .await
+        .context(SupersetDBRetrievalSnafu)?;
+
+    tracing::debug!("{}", format!("Checking status: {:#?}", superset_db.status));
+
+    // Update the Superset cluster status, only if the controller needs to wait.
+    // This avoids updating the status twice per reconcile call. when the DB
+    // has a ready condition.
+    let db_cond_builder = DbConditionBuilder(superset_db.status);
+    if bool::from(&db_cond_builder) {
+        let status = SupersetClusterStatus {
+            conditions: compute_conditions(
+                superset,
+                &[&db_cond_builder, cluster_operation_condition_builder],
+            ),
+        };
+
+        client
+            .apply_patch_status(OPERATOR_NAME, superset, &status)
+            .await
+            .context(ApplyStatusSnafu)?;
+    }
+
+    Ok(bool::from(&db_cond_builder))
+}
+
+struct DbConditionBuilder(Option<SupersetDBStatus>);
+impl ConditionBuilder for DbConditionBuilder {
+    fn build_conditions(&self) -> ClusterConditionSet {
+        let (status, message) = if let Some(ref status) = self.0 {
+            match status.condition {
+                SupersetDBStatusCondition::Pending | SupersetDBStatusCondition::Initializing => (
+                    ClusterConditionStatus::False,
+                    "Waiting for SupersetDB initialization to complete",
+                ),
+                SupersetDBStatusCondition::Failed => (
+                    ClusterConditionStatus::False,
+                    "Superset database initialization failed.",
+                ),
+                SupersetDBStatusCondition::Ready => (
+                    ClusterConditionStatus::True,
+                    "Superset database initialization ready.",
+                ),
+            }
+        } else {
+            (
+                ClusterConditionStatus::Unknown,
+                "Waiting for Superset database initialization to start.",
+            )
+        };
+
+        let cond = ClusterCondition {
+            reason: None,
+            message: Some(String::from(message)),
+            status,
+            type_: ClusterConditionType::Available,
+            last_transition_time: None,
+            last_update_time: None,
+        };
+
+        vec![cond].into()
+    }
+}
+
+/// Evaluates to true if the DB is not ready yet (the controller needs to wait).
+/// Otherwise false.
+impl From<&DbConditionBuilder> for bool {
+    fn from(cond_builder: &DbConditionBuilder) -> bool {
+        if let Some(ref status) = cond_builder.0 {
+            match status.condition {
+                SupersetDBStatusCondition::Pending | SupersetDBStatusCondition::Initializing => {
+                    true
+                }
+                SupersetDBStatusCondition::Failed => true,
+                SupersetDBStatusCondition::Ready => false,
+            }
+        } else {
+            true
+        }
+    }
+}

tests/templates/kuttl/cluster-operation/02-assert.yaml

@@ -4,6 +4,8 @@ kind: TestAssert
 metadata:
   name: install-superset
 timeout: 300
+commands:
+  - script: kubectl -n $NAMESPACE wait --for=condition=available=true supersetclusters.superset.stackable.tech/superset --timeout 301s
 ---
 apiVersion: apps/v1
 kind: StatefulSet

tests/templates/kuttl/cluster-operation/03-assert.yaml

@@ -4,6 +4,8 @@ kind: TestAssert
 metadata:
   name: install-superset
 timeout: 300
+commands:
+  - script: kubectl -n $NAMESPACE wait --for=condition=reconciliationPaused=true supersetclusters.superset.stackable.tech/superset --timeout 301s
 ---
 apiVersion: apps/v1
 kind: StatefulSet

tests/templates/kuttl/cluster-operation/04-assert.yaml

@@ -4,6 +4,8 @@ kind: TestAssert
 metadata:
   name: install-superset
 timeout: 300
+commands:
+  - script: kubectl -n $NAMESPACE wait --for=condition=stopped=true supersetclusters.superset.stackable.tech/superset --timeout 301s
 ---
 apiVersion: apps/v1
 kind: StatefulSet

tests/templates/kuttl/cluster-operation/05-assert.yaml

@@ -4,6 +4,8 @@ kind: TestAssert
 metadata:
   name: install-superset
 timeout: 300
+commands:
+  - script: kubectl -n $NAMESPACE wait --for=condition=available=true supersetclusters.superset.stackable.tech/superset --timeout 301s
 ---
 apiVersion: apps/v1
 kind: StatefulSet