fix #648: add more comments in prepareForDeletion and better error ms…

…g in statefulSetUID

controllers/reconcile_finalizer.go

@@ -41,7 +41,7 @@ func (r *RabbitmqClusterReconciler) removeFinalizer(ctx context.Context, rabbitm
 func (r *RabbitmqClusterReconciler) prepareForDeletion(ctx context.Context, rabbitmqCluster *rabbitmqv1beta1.RabbitmqCluster) error {
 	if controllerutil.ContainsFinalizer(rabbitmqCluster, deletionFinalizer) {
 		if err := clientretry.RetryOnConflict(clientretry.DefaultRetry, func() error {
-			uid, err := r.statefulSetUID(ctx, rabbitmqCluster);
+			uid, err := r.statefulSetUID(ctx, rabbitmqCluster)
 			if err != nil {
 				return err
 			}
@@ -58,6 +58,8 @@ func (r *RabbitmqClusterReconciler) prepareForDeletion(ctx context.Context, rabb
 			// Delete StatefulSet immediately after changing pod labels to minimize risk of them respawning.
 			// There is a window where the StatefulSet could respawn Pods without the deletion label in this order.
 			// But we can't delete it before because the DownwardAPI doesn't update once a Pod enters Terminating.
+			// Addressing #648: if both rabbitmqCluster and the statefulSet returned by r.Get() are stale (and match each other),
+			// setting the stale statefulSet's uid in the precondition can avoid mis-deleting any currently running statefulSet sharing the same name.
 			if err := r.Client.Delete(ctx, sts, &client.DeleteOptions{Preconditions: &metav1.Preconditions{UID: &uid}}); client.IgnoreNotFound(err) != nil {
 				return fmt.Errorf("cannot delete StatefulSet: %s", err.Error())
 			}

controllers/utils.go

@@ -7,9 +7,9 @@ import (
 	rabbitmqv1beta1 "github.com/rabbitmq/cluster-operator/api/v1beta1"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/util/retry"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -67,16 +67,21 @@ func (r *RabbitmqClusterReconciler) statefulSet(ctx context.Context, rmq *rabbit
 	return sts, nil
 }
 
+// statefulSetUID only returns the UID successfully when the controller reference uid matches the rmq uid
 func (r *RabbitmqClusterReconciler) statefulSetUID(ctx context.Context, rmq *rabbitmqv1beta1.RabbitmqCluster) (types.UID, error) {
-	uid := types.UID("")
-	if sts, err := r.statefulSet(ctx, rmq); err == nil {
-		if ref := metav1.GetControllerOf(sts); ref != nil {
-			if string(rmq.GetUID()) == string(ref.UID) {
-				return sts.UID, nil
-			}
-		}
+	var err error
+	var sts *appsv1.StatefulSet
+	var ref *metav1.OwnerReference
+	if sts, err = r.statefulSet(ctx, rmq); err != nil {
+		return "", fmt.Errorf("failed to get statefulSet: %s", err.Error())
+	}
+	if ref = metav1.GetControllerOf(sts); ref == nil {
+		return "", errors.New("failed to get controller reference for statefulSet")
+	}
+	if string(rmq.GetUID()) != string(ref.UID) {
+		return "", errors.New("statefulSet not owner by current RabbitmqCluster")
 	}
-	return uid, fmt.Errorf("failed to get the uid of the statefulset owned by the current rabbitmqCluster")
+	return sts.UID, nil
 }
 
 func (r *RabbitmqClusterReconciler) configMap(ctx context.Context, rmq *rabbitmqv1beta1.RabbitmqCluster, name string) (*corev1.ConfigMap, error) {