fixup storage tracking

Author: lmiccini

apis/bases/rabbitmq.openstack.org_rabbitmqs.yaml

@@ -1969,6 +1969,14 @@ spec:
                   the opentack-operator in the top-level CR (e.g. the ContainerImage)
                 format: int64
                 type: integer
+              pvsBeingDeleted:
+                description: |-
+                  PVsBeingDeleted - list of PV names that are expected to be deleted during storage wipe
+                  Tracked to avoid scanning all PVs in the cluster during cleanup verification
+                items:
+                  type: string
+                type: array
+                x-kubernetes-list-type: set
               queueType:
                 description: QueueType - store whether default ha-all policy is present
                   or not

apis/rabbitmq/v1beta1/rabbitmq_types.go

@@ -151,6 +151,11 @@ type RabbitMqStatus struct {
 	// StorageWipeStartedAt - timestamp when storage wipe process started
 	// Used to implement timeout protection against stuck PV/PVC deletions
 	StorageWipeStartedAt *metav1.Time `json:"storageWipeStartedAt,omitempty"`
+
+	// PVsBeingDeleted - list of PV names that are expected to be deleted during storage wipe
+	// Tracked to avoid scanning all PVs in the cluster during cleanup verification
+	// +listType=set
+	PVsBeingDeleted []string `json:"pvsBeingDeleted,omitempty"`
 }
 
 //+kubebuilder:object:root=true

apis/rabbitmq/v1beta1/zz_generated.deepcopy.go

@@ -1969,6 +1969,14 @@ spec:
                   the opentack-operator in the top-level CR (e.g. the ContainerImage)
                 format: int64
                 type: integer
+              pvsBeingDeleted:
+                description: |-
+                  PVsBeingDeleted - list of PV names that are expected to be deleted during storage wipe
+                  Tracked to avoid scanning all PVs in the cluster during cleanup verification
+                items:
+                  type: string
+                type: array
+                x-kubernetes-list-type: set
               queueType:
                 description: QueueType - store whether default ha-all policy is present
                   or not

config/crd/bases/rabbitmq.openstack.org_rabbitmqs.yaml

@@ -598,6 +598,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (result ct
 			CurrentQueueType:     string(instance.Status.QueueType),
 			Reason:               wipeReason,
 			StorageWipeStartedAt: storageWipeStartedAt,
+			PVsBeingDeleted:      instance.Status.PVsBeingDeleted,
 			DeleteCluster: func(ctx context.Context) error {
 				err := rmqCluster.Delete(ctx, helper)
 				if err != nil && !k8s_errors.IsNotFound(err) {
@@ -618,6 +619,11 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (result ct
 			return result, err
 		}
 
+		// Update status with PV tracking information (if changed during this reconcile)
+		if len(wipeParams.PVsBeingDeleted) > 0 {
+			instance.Status.PVsBeingDeleted = wipeParams.PVsBeingDeleted
+		}
+
 		// If result has Requeue set, we're still in progress
 		if result.Requeue || result.RequeueAfter > 0 {
 			return result, nil
@@ -628,9 +634,10 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (result ct
 			if instance.Annotations != nil {
 				if targetVersion, hasTarget := instance.Annotations[rabbitmqv1beta1.AnnotationTargetVersion]; hasTarget && targetVersion != "" {
 					instance.Status.CurrentVersion = targetVersion
-					// Clear the upgrade phase and timestamp
+					// Clear the upgrade phase, timestamp, and PV tracking
 					instance.Status.UpgradePhase = ""
 					instance.Status.StorageWipeStartedAt = nil
+					instance.Status.PVsBeingDeleted = nil
 
 					// If queue type changed during upgrade, update Status.QueueType to prevent
 					// triggering another wipe for "queue type migration"
@@ -654,6 +661,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (result ct
 			// Queue migration complete - update Status.QueueType
 			instance.Status.UpgradePhase = ""
 			instance.Status.StorageWipeStartedAt = nil
+			instance.Status.PVsBeingDeleted = nil
 			if instance.Spec.QueueType != nil {
 				switch *instance.Spec.QueueType {
 				case rabbitmqv1beta1.QueueTypeQuorum:

internal/controller/rabbitmq/rabbitmq_controller.go

@@ -75,7 +75,6 @@ import (
 	"context"
 	"fmt"
 	"sort"
-	"strings"
 	"time"
 
 	"github.com/go-logr/logr"
@@ -184,33 +183,102 @@ func removeProtectionFinalizer(pvc *corev1.PersistentVolumeClaim, log logr.Logge
 	return false
 }
 
-// verifyPVCleanupComplete checks if all PVs associated with the RabbitMQ instance
-// have been deleted. Returns true if cleanup is complete, false if still in progress.
+// removePVProtectionFinalizer removes the kubernetes.io/pv-protection finalizer from a PV.
+// Returns true if the finalizer was found and removed.
+//
+// Other finalizers (CSI drivers, storage backend) are preserved.
+//
+// The kubernetes.io/pv-protection finalizer prevents PV deletion while bound to a PVC.
+// When a PVC is deleted, the PV goes to Released state, but sometimes the finalizer
+// isn't automatically removed by Kubernetes. We remove it to allow the Delete reclaim
+// policy to take effect.
+func removePVProtectionFinalizer(pv *corev1.PersistentVolume, log logr.Logger) bool {
+	originalCount := len(pv.Finalizers)
+	if originalCount == 0 {
+		return false
+	}
+
+	// Filter out only the kubernetes.io/pv-protection finalizer
+	var filtered []string
+	for _, f := range pv.Finalizers {
+		if f != "kubernetes.io/pv-protection" {
+			filtered = append(filtered, f)
+		}
+	}
+
+	if len(filtered) < originalCount {
+		log.Info("Removing PV protection finalizer to allow deletion during upgrade",
+			"pvName", pv.Name,
+			"removedFinalizer", "kubernetes.io/pv-protection",
+			"remainingFinalizers", filtered)
+		pv.Finalizers = filtered
+		return true
+	}
+
+	return false
+}
+
+// verifyPVCleanupComplete checks if all PVs from the provided list have been deleted.
+// Returns true if cleanup is complete, false if still in progress.
 //
 // Verifies clean storage before recreating the cluster.
 //
 // Implements timeout protection (30 minutes) for PV deletion.
-func (r *Reconciler) verifyPVCleanupComplete(ctx context.Context, namespace, instanceName string, storageWipeStartedAt *time.Time, log logr.Logger) (bool, error) {
-	pvList := &corev1.PersistentVolumeList{}
-	if err := r.List(ctx, pvList); err != nil {
-		return false, fmt.Errorf("failed to list PVs during upgrade verification: %w. "+
-			"List PVs: kubectl get pv -A", err)
+//
+// pvNames is the list of PV names that were captured during PVC deletion.
+// This optimization avoids scanning all PVs in the cluster.
+func (r *Reconciler) verifyPVCleanupComplete(ctx context.Context, pvNames []string, storageWipeStartedAt *time.Time, log logr.Logger) (bool, error) {
+	// If no PVs were tracked, cleanup is complete
+	if len(pvNames) == 0 {
+		log.Info("No PVs to clean up")
+		return true, nil
 	}
 
-	// Check if any PVs are still referencing our PVCs
-	pvcPrefix := fmt.Sprintf("persistence-%s-", instanceName)
+	// Check which PVs still exist and remove protection finalizers if needed
 	var stuckPVs []string
-	for _, pv := range pvList.Items {
-		if pv.Spec.ClaimRef != nil &&
-			pv.Spec.ClaimRef.Namespace == namespace &&
-			strings.HasPrefix(pv.Spec.ClaimRef.Name, pvcPrefix) {
-
-			log.Info("Waiting for PV cleanup to complete",
-				"pvName", pv.Name,
-				"phase", pv.Status.Phase,
-				"pvcName", pv.Spec.ClaimRef.Name)
-			stuckPVs = append(stuckPVs, pv.Name)
+	for _, pvName := range pvNames {
+		pv := &corev1.PersistentVolume{}
+		err := r.Get(ctx, types.NamespacedName{Name: pvName}, pv)
+
+		// If PV doesn't exist, it's been cleaned up (expected)
+		if k8s_errors.IsNotFound(err) {
+			continue
+		}
+
+		// If we can't get the PV for other reasons, include it to be safe
+		if err != nil {
+			log.V(1).Info("Unable to get PV, including in cleanup check",
+				"pvName", pvName,
+				"error", err)
+			stuckPVs = append(stuckPVs, pvName)
+			continue
 		}
+
+		// If PV is in Released phase (PVC deleted but PV still exists),
+		// remove kubernetes.io/pv-protection finalizer to allow deletion
+		if pv.Status.Phase == corev1.VolumeReleased {
+			if removePVProtectionFinalizer(pv, log) {
+				if err := r.Update(ctx, pv); err != nil {
+					log.Error(err, "Failed to remove pv-protection finalizer from PV",
+						"pvName", pvName)
+					// Don't fail - just log and continue, will retry next reconcile
+				} else {
+					log.Info("Removed pv-protection finalizer from Released PV",
+						"pvName", pvName)
+				}
+			}
+		}
+
+		// PV still exists - track it
+		pvcName := ""
+		if pv.Spec.ClaimRef != nil {
+			pvcName = pv.Spec.ClaimRef.Name
+		}
+		log.Info("Waiting for PV cleanup to complete",
+			"pvName", pvName,
+			"phase", pv.Status.Phase,
+			"pvcName", pvcName)
+		stuckPVs = append(stuckPVs, pvName)
 	}
 
 	if len(stuckPVs) > 0 {
@@ -250,10 +318,10 @@ func (r *Reconciler) verifyPVCleanupComplete(ctx context.Context, namespace, ins
 // 1. PV deletion (due to Delete reclaim policy)
 // 2. Storage backend cleanup
 //
-// Returns true if PVCs were deleted.
-func (r *Reconciler) deletePVCsForUpgrade(ctx context.Context, pvcList *corev1.PersistentVolumeClaimList, log logr.Logger) (bool, error) {
+// Returns true if PVCs were deleted and the list of PV names that should be deleted.
+func (r *Reconciler) deletePVCsForUpgrade(ctx context.Context, pvcList *corev1.PersistentVolumeClaimList, log logr.Logger) (bool, []string, error) {
 	if len(pvcList.Items) == 0 {
-		return false, nil
+		return false, nil, nil
 	}
 
 	// Sort PVCs by name to ensure deterministic deletion order
@@ -264,14 +332,17 @@ func (r *Reconciler) deletePVCsForUpgrade(ctx context.Context, pvcList *corev1.P
 
 	log.Info("Deleting PVCs in deterministic order", "pvcCount", len(pvcList.Items))
 
+	// Track which PVs should be deleted (for efficient cleanup verification)
+	pvNames := make([]string, 0, len(pvcList.Items))
+
 	for i := range pvcList.Items {
 		pvc := &pvcList.Items[i]
 
 		// Remove kubernetes.io/pvc-protection finalizer to allow deletion
 		// Other finalizers are preserved for CSI driver cleanup
 		if removeProtectionFinalizer(pvc, log) {
 			if err := r.Update(ctx, pvc); err != nil {
-				return false, fmt.Errorf("failed to remove finalizer from PVC %s: %w. "+
+				return false, nil, fmt.Errorf("failed to remove finalizer from PVC %s: %w. "+
 					"Check PVC: kubectl get pvc %s -n %s -o yaml",
 					pvc.Name, err, pvc.Name, pvc.Namespace)
 			}
@@ -280,17 +351,22 @@ func (r *Reconciler) deletePVCsForUpgrade(ctx context.Context, pvcList *corev1.P
 		// Delete the PVC
 		// The Delete reclaim policy on the PV triggers automatic storage cleanup
 		if err := r.Delete(ctx, pvc); err != nil && !k8s_errors.IsNotFound(err) {
-			return false, fmt.Errorf("failed to delete PVC %s: %w. "+
+			return false, nil, fmt.Errorf("failed to delete PVC %s: %w. "+
 				"Check PVC: kubectl describe pvc %s -n %s",
 				pvc.Name, err, pvc.Name, pvc.Namespace)
 		}
 
 		log.Info("Deleted PVC - storage will be automatically wiped by Delete reclaim policy",
 			"pvcName", pvc.Name,
 			"pvName", pvc.Spec.VolumeName)
+
+		// Track the PV for cleanup verification
+		if pvc.Spec.VolumeName != "" {
+			pvNames = append(pvNames, pvc.Spec.VolumeName)
+		}
 	}
 
-	return true, nil
+	return true, pvNames, nil
 }
 
 // waitForPodsTermination checks if all pods associated with the RabbitMQ instance
@@ -340,6 +416,8 @@ type StorageWipeParams struct {
 	Reason string
 	// Timestamp when storage wipe started (for timeout tracking)
 	StorageWipeStartedAt *time.Time
+	// PV names being deleted (populated during PVC deletion for efficient verification)
+	PVsBeingDeleted []string
 	// Function to delete the RabbitMQCluster
 	DeleteCluster func(ctx context.Context) error
 	// Function to delete the ha-all mirrored policy (optional)
@@ -429,20 +507,24 @@ func (r *Reconciler) performStorageWipe(
 
 		// Step 4b: Delete all PVCs
 		// The Delete reclaim policy will trigger automatic storage cleanup
-		pvcDeleted, err := r.deletePVCsForUpgrade(ctx, pvcList, log)
+		pvcDeleted, pvNames, err := r.deletePVCsForUpgrade(ctx, pvcList, log)
 		if err != nil {
 			log.Error(err, "Failed to delete PVCs during storage wipe")
 			return ctrl.Result{}, err
 		}
 
 		if pvcDeleted {
-			log.Info("Deleted all PVCs, waiting for PV cleanup to complete", "pvcCount", len(pvcList.Items))
+			// Store the PV names for efficient cleanup verification
+			params.PVsBeingDeleted = pvNames
+			log.Info("Deleted all PVCs, waiting for PV cleanup to complete",
+				"pvcCount", len(pvcList.Items),
+				"pvCount", len(pvNames))
 			return ctrl.Result{RequeueAfter: UpgradeCheckInterval}, nil
 		}
 	}
 
 	// Step 5: Verify all PVs are completely cleaned up
-	cleanupComplete, err := r.verifyPVCleanupComplete(ctx, params.Namespace, params.InstanceName, params.StorageWipeStartedAt, log)
+	cleanupComplete, err := r.verifyPVCleanupComplete(ctx, params.PVsBeingDeleted, params.StorageWipeStartedAt, log)
 	if err != nil {
 		log.Error(err, "Failed to verify PV cleanup")
 		return ctrl.Result{}, fmt.Errorf("PV cleanup verification failed: %w. "+