implement connect server status tracking

Author: razvan

Cargo.lock

@@ -35,5 +35,5 @@ indoc = "2"
 
 # TODO: revert before merging
 [patch."https://github.com/stackabletech/operator-rs.git"]
-stackable-operator = { git = "https://github.com/stackabletech//operator-rs.git", branch = "main" }
-# stackable-operator = { path = "../operator-rs/crates/stackable-operator" }
+# stackable-operator = { git = "https://github.com/stackabletech//operator-rs.git", branch = "main" }
+stackable-operator = { path = "../operator-rs/crates/stackable-operator" }

Cargo.nix

@@ -2150,10 +2150,58 @@ spec:
               required:
                 - image
               type: object
+            status:
+              nullable: true
+              properties:
+                conditions:
+                  default: []
+                  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
+              type: object
           required:
             - spec
           title: SparkConnectServer
           type: object
       served: true
       storage: true
-      subresources: {}
+      subresources:
+        status: {}

Cargo.toml

@@ -113,6 +113,7 @@ rules:
       - spark.stackable.tech
     resources:
       - sparkapplications/status
+      - sparkconnectservers/status
     verbs:
       - patch
   - apiGroups:

crate-hashes.json

@@ -44,6 +44,10 @@ use stackable_operator::{
         },
     },
     role_utils::{JavaCommonConfig, JvmArgumentOverrides, RoleGroupRef},
+    status::condition::{
+        compute_conditions, deployment::DeploymentConditionBuilder,
+        operations::ClusterOperationsConditionBuilder,
+    },
     time::Duration,
 };
 use strum::{Display, EnumDiscriminants, IntoStaticStr};
@@ -53,6 +57,7 @@ use super::crd::{
     CONNECT_CONTROLLER_NAME, CONNECT_GRPC_PORT, CONNECT_SERVER_ROLE_NAME, CONNECT_UI_PORT,
 };
 use crate::{
+    connect::crd::SparkConnectServerStatus,
     crd::constants::{
         APP_NAME, JVM_SECURITY_PROPERTIES_FILE, LOG4J2_CONFIG_FILE, MAX_SPARK_LOG_FILES_SIZE,
         METRICS_PORT, OPERATOR_NAME, SPARK_DEFAULTS_FILE_NAME, SPARK_IMAGE_BASE_NAME, SPARK_UID,
@@ -69,6 +74,12 @@ const DUMMY_SPARK_CONNECT_GROUP_NAME: &str = "default";
 #[strum_discriminants(derive(IntoStaticStr))]
 #[allow(clippy::enum_variant_names)]
 pub enum Error {
+    #[snafu(display("failed to update status of spark connect server {name}"))]
+    ApplyStatus {
+        source: stackable_operator::client::Error,
+        name: String,
+    },
+
     #[snafu(display("failed to merge jvm argument overrides"))]
     MergeJvmArgumentOverrides {
         source: stackable_operator::role_utils::Error,
@@ -311,16 +322,35 @@ pub async fn reconcile(
         &config_map,
         args,
     )?;
-    cluster_resources
-        .add(client, deployment)
-        .await
-        .context(ApplyDeploymentSnafu)?;
+
+    let mut ss_cond_builder = DeploymentConditionBuilder::default();
+
+    ss_cond_builder.add(
+        cluster_resources
+            .add(client, deployment)
+            .await
+            .context(ApplyDeploymentSnafu)?,
+    );
 
     cluster_resources
         .delete_orphaned_resources(client)
         .await
         .context(DeleteOrphanedResourcesSnafu)?;
 
+    let cluster_operation_cond_builder =
+        ClusterOperationsConditionBuilder::new(&scs.spec.cluster_operation);
+
+    let status = SparkConnectServerStatus {
+        conditions: compute_conditions(scs, &[&ss_cond_builder, &cluster_operation_cond_builder]),
+    };
+
+    client
+        .apply_patch_status(OPERATOR_NAME, scs, &status)
+        .await
+        .context(ApplyStatusSnafu {
+            name: scs.name_any(),
+        })?;
+
     Ok(Action::await_change())
 }
 

deploy/helm/spark-k8s-operator/crds/crds.yaml

@@ -19,6 +19,7 @@ use stackable_operator::{
     product_logging::{self, spec::Logging},
     role_utils::{CommonConfiguration, JavaCommonConfig},
     schemars::{self, JsonSchema},
+    status::condition::{ClusterCondition, HasStatusCondition},
     time::Duration,
 };
 use stackable_versioned::versioned;
@@ -67,6 +68,7 @@ pub mod versioned {
         kind = "SparkConnectServer",
         plural = "sparkconnectservers",
         shortname = "sparkconnect",
+        status = "SparkConnectServerStatus",
         namespaced,
         crates(
             kube_core = "stackable_operator::kube::core",
@@ -245,3 +247,19 @@ impl v1alpha1::SparkConnectServer {
         .context(FragmentValidationFailureSnafu)
     }
 }
+
+#[derive(Clone, Debug, Default, Deserialize, JsonSchema, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct SparkConnectServerStatus {
+    #[serde(default)]
+    pub conditions: Vec<ClusterCondition>,
+}
+
+impl HasStatusCondition for v1alpha1::SparkConnectServer {
+    fn conditions(&self) -> Vec<ClusterCondition> {
+        match &self.status {
+            Some(status) => status.conditions.clone(),
+            None => vec![],
+        }
+    }
+}