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,22 @@ 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
 	}
 
+	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()
 
@@ -615,13 +622,13 @@ func (r *StatefulSetResource) updateStatefulSet(
 // shouldUpdate returns true if changes on the CR require update
 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
+		return false, true, nil
 	}
 	prepareResourceForPatch(current, modified)
 	opts := []patch.CalculateOption{
@@ -632,10 +639,10 @@ 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
+		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) {