operator: Restart rolling update when Statefulset changes

This PR enhances the detection mechanism for Node Pool rolling upgrades.
Previously, when a Node Pool rolling upgrade was detected, all Pods were
marked with the `ClusterUpdate` Pod Status Condition, and the `restarting`
Cluster Custom Resource Status Node Pool field was marked as `true`.

If a modification to the Cluster Custom Resource occurred during an ongoing
rolling upgrade (e.g., a change to the Redpanda container tag), it could
trigger an update to the StatefulSet resource. In such cases, Pods that had
already been restarted within the affected Node Pool were not restarted again,
resulting in inconsistent state propagation (e.g., mismatched Redpanda
container tags).

With this change, updates to the StatefulSet are explicitly distinguished from
Node Pool rolling upgrades. When a StatefulSet update is required, all
associated Pods are now marked or re-marked with the `ClusterUpdate` Pod Status
Condition to ensure consistent rollout of the updated specification.

Author: RafalKorepta

operator/pkg/resources/statefulset_update.go

@@ -94,15 +94,23 @@ func (r *StatefulSetResource) runUpdate(
 	// Check if we should run update on this specific STS.
 
 	log.V(logger.DebugLevel).Info("Checking that we should update")
-	update, err := r.shouldUpdate(current, modified)
+	stsChanged, nodePoolRestarting, err := r.shouldUpdate(current, modified)
 	if err != nil {
 		return fmt.Errorf("unable to determine the update procedure: %w", err)
 	}
 
-	if !update {
+	if !stsChanged && !nodePoolRestarting {
 		return nil
 	}
 
+	// When a StatefulSet change is detected, mark all Node Pool Pods as requiring a Pod rolling restart
+	if stsChanged {
+		log.V(logger.DebugLevel).Info("Setting ClusterUpdate condition on pods")
+		if err = r.MarkPodsForUpdate(ctx, ClusterUpdateReasonVersion); err != nil {
+			return fmt.Errorf("unable to mark pods for update: %w", err)
+		}
+	}
+
 	// At this point, we have seen a diff and want to update the StatefulSet.
 	npStatus := r.getNodePoolStatus()
 
@@ -612,17 +620,19 @@ func (r *StatefulSetResource) updateStatefulSet(
 	return nil
 }
 
-// shouldUpdate returns true if changes on the CR require update
+// shouldUpdate compares the generated appsv1.StatefulSet from the StatefulSetResource.obj
+// function (referred to as `modified`) with appsv1.StatefulSet retrieved from
+// the Kubernetes API (referred to as `current`). If a difference is detected, the function returns
+// `true` as the first value. Certain fields are ignored because they are not relevant to a Pod
+// rolling upgrade (e.g. `Status`, `VolumeClaimTemplate`, or `Annotations`).
+//
+// If no differences are found, the function returns `false` as first value. The second value
+// depends on the Node Pool Status, indicating whether rolling restart is still in progress.
 func (r *StatefulSetResource) shouldUpdate(
 	current, modified *appsv1.StatefulSet,
-) (bool, error) {
+) (bool, bool, error) {
 	log := r.logger.WithName("shouldUpdate")
 
-	npStatus := r.getNodePoolStatus()
-
-	if npStatus.Restarting || r.pandaCluster.Status.Restarting {
-		return true, nil
-	}
 	prepareResourceForPatch(current, modified)
 	opts := []patch.CalculateOption{
 		patch.IgnoreStatusFields(),
@@ -632,10 +642,16 @@ func (r *StatefulSetResource) shouldUpdate(
 	}
 	patchResult, err := patch.NewPatchMaker(patch.NewAnnotator(redpandaAnnotatorKey), &patch.K8sStrategicMergePatcher{}, &patch.BaseJSONMergePatcher{}).Calculate(current, modified, opts...)
 	if err != nil || patchResult.IsEmpty() {
-		return false, err
+		npStatus := r.getNodePoolStatus()
+
+		if npStatus.Restarting || r.pandaCluster.Status.Restarting {
+			return false, true, nil
+		}
+
+		return false, false, err
 	}
 	log.Info("Detected diff", "patchResult", string(patchResult.Patch))
-	return true, nil
+	return true, false, nil
 }
 
 func (r *StatefulSetResource) getRestartingStatus() bool {

operator/pkg/resources/statefulset_update_test.go

@@ -163,14 +163,36 @@ func TestShouldUpdate_AnnotationChange(t *testing.T) {
 			},
 		},
 	}
-	update, err := ssres.shouldUpdate(sts, stsWithAnnotation)
+	stsChange, _, err := ssres.shouldUpdate(sts, stsWithAnnotation)
 	require.NoError(t, err)
-	require.True(t, update)
+	require.True(t, stsChange)
 
 	// same statefulset with same annotation
-	update, err = ssres.shouldUpdate(stsWithAnnotation, stsWithAnnotation)
+	stsChange, _, err = ssres.shouldUpdate(stsWithAnnotation, stsWithAnnotation)
 	require.NoError(t, err)
-	require.False(t, update)
+	require.False(t, stsChange)
+
+	ssres = StatefulSetResource{
+		nodePool: vectorizedv1alpha1.NodePoolSpecWithDeleted{
+			NodePoolSpec: vectorizedv1alpha1.NodePoolSpec{
+				Name: "default",
+			},
+		},
+		pandaCluster: &vectorizedv1alpha1.Cluster{
+			Status: vectorizedv1alpha1.ClusterStatus{
+				Restarting: false,
+				NodePools: map[string]vectorizedv1alpha1.NodePoolStatus{
+					"default": {
+						Restarting: true,
+					},
+				},
+			},
+		},
+	}
+	// same statefulset with same annotation, but with node pool status restarting
+	_, nodePoolRestarting, err := ssres.shouldUpdate(stsWithAnnotation, stsWithAnnotation)
+	require.NoError(t, err)
+	require.True(t, nodePoolRestarting)
 }
 
 func TestPutInMaintenanceMode(t *testing.T) {

operator/tests/e2e/additional-configuration/00-redpanda-cluster.yaml

@@ -4,7 +4,7 @@ metadata:
   name: additional-configuration
 spec:
   image: "redpandadata/redpanda"
-  version: "v24.2.5"
+  version: "v25.1.11"
   replicas: 1
   resources:
     requests:

operator/tests/e2e/additional-configuration/01-verify-config.yaml

@@ -1,5 +1,5 @@
 apiVersion: kuttl.dev/v1beta1
 kind: TestStep
 commands:
-  - command: ./verify-config.sh -r 10 -b v24.2
+  - command: ./verify-config.sh -r 10 -b v25.1
     namespaced: true

operator/tests/e2e/additional-configuration/03-verify-config-changed.yaml

@@ -1,5 +1,5 @@
 apiVersion: kuttl.dev/v1beta1
 kind: TestStep
 commands:
-  - command: ./verify-config.sh -r 11 -b v24.2
+  - command: ./verify-config.sh -r 11 -b v25.1
     namespaced: true

operator/tests/e2e/additional-configuration/verify-config-dev.sh

@@ -79,6 +79,8 @@ cloud_storage_segment_max_upload_interval_sec: 1800
 default_topic_partitions: 3
 enable_idempotence: true
 enable_rack_awareness: true
+internal_topic_replication_factor: 3
+kafka_nodelete_topics: [_internal_connectors_configs _internal_connectors_offsets _internal_connectors_status _audit __consumer_offsets _redpanda_e2e_probe _schemas]
 log_segment_size: 536870912
 EOF
 )

operator/tests/e2e/additional-configuration/verify-config-v25.1.sh

@@ -0,0 +1,84 @@
+#!/usr/bin/env bash
+
+expected=$(
+  cat <<EOF
+redpanda:
+    data_directory: /var/lib/redpanda/data
+    empty_seed_starts_cluster: false
+    seed_servers:
+        - host:
+            address: additional-configuration-0.additional-configuration.${NAMESPACE}.svc.cluster.local.
+            port: 33145
+    rpc_server:
+        address: 0.0.0.0
+        port: 33145
+    kafka_api:
+        - address: 0.0.0.0
+          port: 9092
+          name: kafka
+          authentication_method: none
+    admin:
+        - address: 0.0.0.0
+          port: 9644
+          name: admin
+    advertised_rpc_api:
+        address: additional-configuration-0.additional-configuration.${NAMESPACE}.svc.cluster.local.
+        port: 33145
+    advertised_kafka_api:
+        - address: additional-configuration-0.additional-configuration.${NAMESPACE}.svc.cluster.local.
+          port: 9092
+          name: kafka
+    developer_mode: true
+    auto_create_topics_enabled: true
+    fetch_reads_debounce_timeout: 10
+    group_initial_rebalance_delay: 0
+    group_topic_partitions: 3
+    log_segment_size_min: 1
+    storage_min_free_bytes: 10485760
+    topic_partitions_per_shard: 1000
+    write_caching_default: "true"
+rpk:
+    overprovisioned: true
+    tune_network: true
+    tune_disk_scheduler: true
+    tune_disk_nomerges: true
+    tune_disk_write_cache: true
+    tune_disk_irq: true
+    tune_cpu: true
+    tune_aio_events: true
+    tune_clocksource: true
+    tune_swappiness: true
+    coredump_dir: /var/lib/redpanda/coredump
+    tune_ballast_file: true
+pandaproxy:
+    pandaproxy_api:
+        - address: 0.0.0.0
+          port: 8082
+          name: proxy
+    advertised_pandaproxy_api:
+        - address: additional-configuration-0.additional-configuration.${NAMESPACE}.svc.cluster.local.
+          port: 8082
+          name: proxy
+pandaproxy_client:
+    brokers:
+        - address: additional-configuration.${NAMESPACE}.svc.cluster.local.
+          port: 9092
+    retries: ${PANDAPROXY_RETRIES}
+schema_registry:
+    schema_registry_api:
+        - address: 0.0.0.0
+          port: 8081
+          name: external
+EOF
+)
+
+expected_bootstrap=$(
+  cat <<EOF
+auto_create_topics_enabled: false
+cloud_storage_segment_max_upload_interval_sec: 1800
+default_topic_partitions: 3
+enable_idempotence: true
+enable_rack_awareness: true
+log_segment_size: 536870912
+EOF
+)