[Merged by Bors] - Implement pause/stop

CHANGELOG.md

@@ -7,19 +7,21 @@ All notable changes to this project will be documented in this file.
 ### Added
 
 - Enabled logging and log aggregation ([#418])
-- Deploy default and support custom affinities ([#436]).
+- Deploy default and support custom affinities ([#436])
 - Added the ability to mount extra volumes for files that may be needed for NiFi processors to work ([#434])
+- Extend cluster resources for status and cluster operation (paused, stopped) ([#447])
 
 ### Changed
 
 - [BREAKING]: Renamed global `config` to `clusterConfig` ([#417])
 - [BREAKING]: Moved `zookeeper_configmap_name` to `clusterConfig` ([#417])
-- Updated operator-rs to 0.33.0  ([#418])
+- `operator-rs` `0.33.0` -> `0.39.0` ([#418], [#447])
 
 [#417]: https://github.com/stackabletech/nifi-operator/pull/417
 [#418]: https://github.com/stackabletech/nifi-operator/pull/418
 [#434]: https://github.com/stackabletech/nifi-operator/pull/434
 [#436]: https://github.com/stackabletech/nifi-operator/pull/436
+[#447]: https://github.com/stackabletech/nifi-operator/pull/447
 
 ## [23.1.0] - 2023-01-23
 

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

@@ -1167,6 +1167,21 @@ spec:
                     - sensitiveProperties
                     - zookeeperConfigMapName
                   type: object
+                clusterOperation:
+                  default:
+                    stopped: false
+                    reconciliationPaused: false
+                  description: Cluster operations like pause reconciliation or cluster stop.
+                  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:
@@ -2934,10 +2949,6 @@ spec:
                   required:
                     - roleGroups
                   type: object
-                stopped:
-                  description: Emergency stop button, if `true` then all pods are stopped without affecting configuration (as setting `replicas` to `0` would)
-                  nullable: true
-                  type: boolean
               required:
                 - clusterConfig
                 - image

docs/modules/nifi/pages/usage_guide/cluster_operations.adoc

@@ -0,0 +1,4 @@
+
+= Cluster Operation
+
+Apache NiFi installations can be configured with different cluster operations like pausing reconciliation or stopping the cluster. See xref:concepts:cluster_operations.adoc[cluster operations] for more details.

docs/modules/nifi/partials/nav.adoc

@@ -12,3 +12,4 @@
 ** xref:nifi:usage_guide/configuration-environment-overrides.adoc[]
 ** xref:nifi:usage_guide/updating.adoc[]
 ** xref:nifi:usage_guide/pod-placement.adoc[]
+** xref:nifi:usage_guide/cluster_operations.adoc[]

rust/crd/Cargo.toml

@@ -15,7 +15,7 @@ serde = "1.0"
 serde_json = "1.0"
 snafu = "0.7"
 rand = "0.8"
-stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "0.36.0" }
+stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "0.39.0" }
 strum = { version = "0.24", features = ["derive"] }
 tracing = "0.1"
 

rust/crd/src/lib.rs

@@ -11,6 +11,7 @@ use stackable_operator::k8s_openapi::api::core::v1::Volume;
 use stackable_operator::{
     commons::{
         affinity::StackableAffinity,
+        cluster_operation::ClusterOperation,
         product_image_selection::ProductImage,
         resources::{
             CpuLimitsFragment, MemoryLimitsFragment, NoRuntimeLimits, NoRuntimeLimitsFragment,
@@ -83,12 +84,12 @@ pub struct NifiSpec {
     pub image: ProductImage,
     /// Global Nifi config for e.g. authentication or sensitive properties
     pub cluster_config: NifiClusterConfig,
+    /// Cluster operations like pause reconciliation or cluster stop.
+    #[serde(default)]
+    pub cluster_operation: ClusterOperation,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     /// Available NiFi roles
     pub nodes: Option<Role<NifiConfigFragment>>,
-    /// Emergency stop button, if `true` then all pods are stopped without affecting configuration (as setting `replicas` to `0` would)
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub stopped: Option<bool>,
 }
 
 #[derive(Clone, Debug, Deserialize, JsonSchema, PartialEq, Serialize)]

rust/operator-binary/Cargo.toml

@@ -20,13 +20,13 @@ semver = "1.0"
 serde = "1.0"
 serde_json = "1.0"
 snafu = "0.7"
-stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "0.36.0" }
+stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "0.39.0" }
 stackable-nifi-crd = { path = "../crd" }
 strum = { version = "0.24", features = ["derive"] }
 tokio = { version = "1.25", features = ["full"] }
 tracing = "0.1"
 
 [build-dependencies]
 built = { version = "0.5", features = ["chrono", "git2"] }
-stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "0.36.0" }
+stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "0.39.0" }
 stackable-nifi-crd = { path = "../crd" }

rust/operator-binary/src/controller.rs

@@ -15,7 +15,7 @@ use stackable_operator::{
         PodSecurityContextBuilder, VolumeBuilder,
     },
     client::Client,
-    cluster_resources::ClusterResources,
+    cluster_resources::{ClusterResourceApplyStrategy, ClusterResources},
     commons::product_image_selection::ResolvedProductImage,
     config::fragment,
     k8s_openapi::{
@@ -308,6 +308,7 @@ pub async fn reconcile_nifi(nifi: Arc<NifiCluster>, ctx: Arc<Ctx>) -> Result<Act
         OPERATOR_NAME,
         CONTROLLER_NAME,
         &nifi.object_ref(&()),
+        ClusterResourceApplyStrategy::from(&nifi.spec.cluster_operation),
     )
     .context(CreateClusterResourcesSnafu)?;
 
@@ -318,7 +319,7 @@ pub async fn reconcile_nifi(nifi: Arc<NifiCluster>, ctx: Arc<Ctx>) -> Result<Act
 
     let node_role_service = build_node_role_service(&nifi, &resolved_product_image)?;
     cluster_resources
-        .add(client, &node_role_service)
+        .add(client, node_role_service)
         .await
         .context(ApplyRoleServiceSnafu)?;
 
@@ -400,19 +401,19 @@ pub async fn reconcile_nifi(nifi: Arc<NifiCluster>, ctx: Arc<Ctx>) -> Result<Act
             )?;
 
             cluster_resources
-                .add(client, &rg_service)
+                .add(client, rg_service)
                 .await
                 .with_context(|_| ApplyRoleGroupServiceSnafu {
                     rolegroup: rolegroup.clone(),
                 })?;
             cluster_resources
-                .add(client, &rg_configmap)
+                .add(client, rg_configmap)
                 .await
                 .with_context(|_| ApplyRoleGroupConfigSnafu {
                     rolegroup: rolegroup.clone(),
                 })?;
             cluster_resources
-                .add(client, &rg_statefulset)
+                .add(client, rg_statefulset)
                 .await
                 .with_context(|_| ApplyRoleGroupStatefulSetSnafu {
                     rolegroup: rolegroup.clone(),
@@ -1034,9 +1035,7 @@ async fn build_node_rolegroup_statefulset(
             .build(),
         spec: Some(StatefulSetSpec {
             pod_management_policy: Some("Parallel".to_string()),
-            replicas: if nifi.spec.stopped.unwrap_or(false)
-                || version_change_state == &VersionChangeState::BeginChange
-            {
+            replicas: if version_change_state == &VersionChangeState::BeginChange {
                 Some(0)
             } else {
                 rolegroup.and_then(|rg| rg.replicas).map(i32::from)

tests/templates/kuttl/cluster_operation/00-assert.yaml.j2

@@ -0,0 +1,10 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+{% if lookup('env', 'VECTOR_AGGREGATOR') %}
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: vector-aggregator-discovery
+{% endif %}

tests/templates/kuttl/cluster_operation/00-install-vector-aggregator-discovery-configmap.yaml.j2

@@ -0,0 +1,9 @@
+{% if lookup('env', 'VECTOR_AGGREGATOR') %}
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: vector-aggregator-discovery
+data:
+  ADDRESS: {{ lookup('env', 'VECTOR_AGGREGATOR') }}
+{% endif %}

tests/templates/kuttl/cluster_operation/10-assert.yaml

@@ -0,0 +1,12 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+timeout: 600
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: test-zk-server-default
+status:
+  readyReplicas: 1
+  replicas: 1

tests/templates/kuttl/cluster_operation/10-install-zk.yaml.j2

@@ -0,0 +1,29 @@
+---
+apiVersion: zookeeper.stackable.tech/v1alpha1
+kind: ZookeeperCluster
+metadata:
+  name: test-zk
+spec:
+  image:
+    productVersion: "{{ test_scenario['values']['zookeeper-latest'].split('-stackable')[0] }}"
+    stackableVersion: "{{ test_scenario['values']['zookeeper-latest'].split('-stackable')[1] }}"
+{% if lookup('env', 'VECTOR_AGGREGATOR') %}
+  clusterConfig:
+    logging:
+      vectorAggregatorConfigMapName: vector-aggregator-discovery
+{% endif %}
+  servers:
+    config:
+      logging:
+        enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }}
+    roleGroups:
+      default:
+        replicas: 1
+---
+apiVersion: zookeeper.stackable.tech/v1alpha1
+kind: ZookeeperZnode
+metadata:
+  name: test-nifi-znode
+spec:
+  clusterRef:
+    name: test-zk

tests/templates/kuttl/cluster_operation/20-assert.yaml

@@ -0,0 +1,12 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+timeout: 1200
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: test-nifi-node-default
+status:
+  readyReplicas: 2
+  replicas: 2

tests/templates/kuttl/cluster_operation/20-install-nifi.yaml.j2

@@ -0,0 +1,42 @@
+---
+apiVersion: v1
+kind: Secret
+metadata:
+  name: nifi-admin-credentials-simple
+stringData:
+  username: admin
+  password: supersecretpassword
+---
+apiVersion: v1
+kind: Secret
+metadata:
+  name: nifi-sensitive-property-key
+stringData:
+  nifiSensitivePropsKey: mYsUp3rS3cr3tk3y
+---
+apiVersion: nifi.stackable.tech/v1alpha1
+kind: NifiCluster
+metadata:
+  name: test-nifi
+spec:
+  image:
+    productVersion: "{{ test_scenario['values']['nifi-latest'].split('-stackable')[0] }}"
+    stackableVersion: "{{ test_scenario['values']['nifi-latest'].split('-stackable')[1] }}"
+  clusterConfig:
+    authentication:
+      method:
+        singleUser:
+          adminCredentialsSecret: nifi-admin-credentials-simple
+    sensitiveProperties:
+      keySecret: nifi-sensitive-property-key
+{% if lookup('env', 'VECTOR_AGGREGATOR') %}
+    vectorAggregatorConfigMapName: vector-aggregator-discovery
+{% endif %}
+    zookeeperConfigMapName: test-nifi-znode
+  nodes:
+    config:
+      logging:
+        enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }}
+    roleGroups:
+      default:
+        replicas: 2

tests/templates/kuttl/cluster_operation/30-assert.yaml

@@ -0,0 +1,11 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+timeout: 1200
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: test-nifi-node-default
+status:
+  replicas: 0

tests/templates/kuttl/cluster_operation/30-stop-nifi.yaml.j2

@@ -0,0 +1,45 @@
+---
+apiVersion: v1
+kind: Secret
+metadata:
+  name: nifi-admin-credentials-simple
+stringData:
+  username: admin
+  password: supersecretpassword
+---
+apiVersion: v1
+kind: Secret
+metadata:
+  name: nifi-sensitive-property-key
+stringData:
+  nifiSensitivePropsKey: mYsUp3rS3cr3tk3y
+---
+apiVersion: nifi.stackable.tech/v1alpha1
+kind: NifiCluster
+metadata:
+  name: test-nifi
+spec:
+  image:
+    productVersion: "{{ test_scenario['values']['nifi-latest'].split('-stackable')[0] }}"
+    stackableVersion: "{{ test_scenario['values']['nifi-latest'].split('-stackable')[1] }}"
+  clusterConfig:
+    authentication:
+      method:
+        singleUser:
+          adminCredentialsSecret: nifi-admin-credentials-simple
+    sensitiveProperties:
+      keySecret: nifi-sensitive-property-key
+{% if lookup('env', 'VECTOR_AGGREGATOR') %}
+    vectorAggregatorConfigMapName: vector-aggregator-discovery
+{% endif %}
+    zookeeperConfigMapName: test-nifi-znode
+  clusterOperation:
+    stopped: true
+    reconciliationPaused: false
+  nodes:
+    config:
+      logging:
+        enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }}
+    roleGroups:
+      default:
+        replicas: 2

tests/templates/kuttl/cluster_operation/40-assert.yaml

@@ -0,0 +1,11 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+timeout: 1200
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: test-nifi-node-default
+status:
+  replicas: 0