implement cluster operation

Author: razvan

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

@@ -1895,6 +1895,21 @@ spec:
                         - external-stable
                       type: string
                   type: object
+                clusterOperation:
+                  default:
+                    reconciliationPaused: false
+                    stopped: false
+                  description: '[Cluster operations](https://docs.stackable.tech/home/nightly/concepts/operations/cluster_operations) properties, allow stopping the product instance as well as pausing reconciliation.'
+                  properties:
+                    reconciliationPaused:
+                      default: false
+                      description: Flag to stop cluster reconciliation by the operator. This means that all changes in the custom resource spec are ignored until this flag is set to false or removed. The operator will however still watch the deployed resources at the time and update the custom resource status field. If applied at the same time with `stopped`, `reconciliationPaused` will take precedence over `stopped` and stop the reconciliation immediately.
+                      type: boolean
+                    stopped:
+                      default: false
+                      description: Flag to stop the cluster. This means all deployed resources (e.g. Services, StatefulSets, ConfigMaps) are kept but all deployed Pods (e.g. replicas from a StatefulSet) are scaled to 0 and therefore stopped and removed. If applied at the same time with `reconciliationPaused`, the latter will pause reconciliation and `stopped` will take no effect until `reconciliationPaused` is set to false or removed.
+                      type: boolean
+                  type: object
                 image:
                   anyOf:
                     - required:

rust/operator-binary/src/connect/controller.rs

@@ -233,7 +233,7 @@ pub async fn reconcile(
         OPERATOR_NAME,
         CONNECT_CONTROLLER_NAME,
         &scs.object_ref(&()),
-        ClusterResourceApplyStrategy::Default,
+        ClusterResourceApplyStrategy::from(&scs.spec.cluster_operation),
     )
     .context(CreateClusterResourcesSnafu)?;
 

rust/operator-binary/src/connect/crd.rs

@@ -3,6 +3,7 @@ use serde::{Deserialize, Serialize};
 use snafu::{ResultExt, Snafu};
 use stackable_operator::{
     commons::{
+        cluster_operation::ClusterOperation,
         product_image_selection::ProductImage,
         resources::{
             CpuLimitsFragment, MemoryLimitsFragment, NoRuntimeLimits, NoRuntimeLimitsFragment,
@@ -82,6 +83,10 @@ pub mod versioned {
         #[serde(default)]
         pub cluster_config: v1alpha1::SparkConnectServerClusterConfig,
 
+        // no doc string - See ClusterOperation struct
+        #[serde(default)]
+        pub cluster_operation: ClusterOperation,
+
         /// Name of the Vector aggregator discovery ConfigMap.
         /// It must contain the key `ADDRESS` with the address of the Vector aggregator.
         #[serde(skip_serializing_if = "Option::is_none")]