Fix CG PVC selection issue due to storageId mismatch on failover

Recently, we started using storageId as part of the label to determine
whether a PVC belongs to a consistency group. While the initial deployment
and synchronization from primary to secondary clusters work correctly,
failover or relocation results in a different storageId on the new primary.

This mismatch causes issues when setting up the source and destination again.
This fix ensures that PVC selection accounts for storageId differences
to maintain correct CG membership during failover and relocation.

Signed-off-by: Benamar Mekhissi <[email protected]>

api/v1alpha1/replicationgroupdestination_types.go

@@ -51,6 +51,7 @@ type ReplicationGroupDestinationStatus struct {
 // +kubebuilder:printcolumn:name="Last sync",type="string",format="date-time",JSONPath=`.status.lastSyncTime`
 // +kubebuilder:printcolumn:name="Duration",type="string",JSONPath=`.status.lastSyncDuration`
 // +kubebuilder:printcolumn:name="Last sync start",type="string",format="date-time",JSONPath=`.status.lastSyncStartTime`
+// +kubebuilder:resource:shortName=rgd
 
 // ReplicationGroupDestination is the Schema for the replicationgroupdestinations API
 type ReplicationGroupDestination struct {

api/v1alpha1/replicationgroupsource_types.go

@@ -72,6 +72,7 @@ type ReplicationGroupSourceStatus struct {
 // +kubebuilder:printcolumn:name="Next sync",type="string",format="date-time",JSONPath=`.status.nextSyncTime`
 // +kubebuilder:printcolumn:name="Source",type="string",JSONPath=`.spec.volumeGroupSnapshotSource`
 // +kubebuilder:printcolumn:name="Last sync start",type="string",format="date-time",JSONPath=`.status.lastSyncStartTime`
+// +kubebuilder:resource:shortName=rgs
 
 // ReplicationGroupSource is the Schema for the replicationgroupsources API
 type ReplicationGroupSource struct {

config/crd/bases/ramendr.openshift.io_replicationgroupdestinations.yaml

@@ -11,6 +11,8 @@ spec:
     kind: ReplicationGroupDestination
     listKind: ReplicationGroupDestinationList
     plural: replicationgroupdestinations
+    shortNames:
+    - rgd
     singular: replicationgroupdestination
   scope: Namespaced
   versions:

config/crd/bases/ramendr.openshift.io_replicationgroupsources.yaml

@@ -11,6 +11,8 @@ spec:
     kind: ReplicationGroupSource
     listKind: ReplicationGroupSourceList
     plural: replicationgroupsources
+    shortNames:
+    - rgs
     singular: replicationgroupsource
   scope: Namespaced
   versions:

internal/controller/cephfscg/cghandler.go

@@ -245,7 +245,6 @@ func (c *cgHandler) CreateOrUpdateReplicationGroupSource(
 
 		return nil, false, err
 	}
-
 	//
 	// For final sync only - check status to make sure the final sync is complete
 	// and also run cleanup (removes PVC we just ran the final sync from)

internal/controller/volumereplicationgroup_controller.go

@@ -743,37 +743,44 @@ func (v *VRGInstance) labelPVCsForCG() error {
 }
 
 func (v *VRGInstance) addConsistencyGroupLabel(pvc *corev1.PersistentVolumeClaim) error {
-	scName := pvc.Spec.StorageClassName
+	cgLabelVal, err := v.getCGLabelValue(pvc.Spec.StorageClassName, pvc.GetName(), pvc.GetNamespace())
+	if err != nil {
+		return err
+	}
 
+	// Add a CG label to indicate that this PVC belongs to a consistency group.
+	return util.NewResourceUpdater(pvc).
+		AddLabel(ConsistencyGroupLabel, cgLabelVal).
+		Update(v.ctx, v.reconciler.Client)
+}
+
+func (v *VRGInstance) getCGLabelValue(scName *string, pvcName, pvcNamespace string) (string, error) {
 	if scName == nil || *scName == "" {
-		return fmt.Errorf("missing storage class name for PVC %s/%s", pvc.GetNamespace(), pvc.GetName())
+		return "", fmt.Errorf("missing storage class name for PVC %s/%s", pvcNamespace, pvcName)
 	}
 
 	storageClass := &storagev1.StorageClass{}
 	if err := v.reconciler.Get(v.ctx, types.NamespacedName{Name: *scName}, storageClass); err != nil {
 		v.log.Info(fmt.Sprintf("Failed to get the storageclass %s", *scName))
 
-		return fmt.Errorf("failed to get the storageclass with name %s (%w)", *scName, err)
+		return "", fmt.Errorf("failed to get the storageclass with name %s (%w)", *scName, err)
 	}
 
 	storageID, ok := storageClass.GetLabels()[StorageIDLabel]
 	if !ok {
-		v.log.Info("Missing storageID for PVC %s/%s", pvc.GetNamespace(), pvc.GetName())
+		v.log.Info("Missing storageID for PVC %s/%s", pvcNamespace, pvcName)
 
-		return fmt.Errorf("missing storageID for PVC %s/%s", pvc.GetNamespace(), pvc.GetName())
+		return "", fmt.Errorf("missing storageID for PVC %s/%s", pvcNamespace, pvcName)
 	}
 
 	// FIXME: a temporary workaround for issue DFBUGS-1209
 	// Remove this block once DFBUGS-1209 is fixed
-	storageID = "cephfs-" + storageID
+	cgLabelVal := "cephfs-" + storageID
 	if storageClass.Provisioner != DefaultCephFSCSIDriverName {
-		storageID = "rbd-" + storageID
+		cgLabelVal = "rbd-" + storageID
 	}
 
-	// Add label for PVC, showing that this PVC is part of consistency group
-	return util.NewResourceUpdater(pvc).
-		AddLabel(ConsistencyGroupLabel, storageID).
-		Update(v.ctx, v.reconciler.Client)
+	return cgLabelVal, nil
 }
 
 func (v *VRGInstance) updateReplicationClassList() error {

internal/controller/vrg_volsync.go

@@ -37,15 +37,20 @@ func (v *VRGInstance) restorePVsAndPVCsForVolSync() (int, error) {
 		// as this would result in incorrect information.
 		rdSpec.ProtectedPVC.Conditions = nil
 
-		cg, ok := rdSpec.ProtectedPVC.Labels[ConsistencyGroupLabel]
+		cgLabelVal, ok := rdSpec.ProtectedPVC.Labels[ConsistencyGroupLabel]
 		if ok && util.IsCGEnabled(v.instance.Annotations) {
-			v.log.Info("rdSpec has CG label", "Labels", rdSpec.ProtectedPVC.Labels)
-			cephfsCGHandler := cephfscg.NewVSCGHandler(
-				v.ctx, v.reconciler.Client, v.instance,
-				&metav1.LabelSelector{MatchLabels: map[string]string{ConsistencyGroupLabel: cg}},
-				v.volSyncHandler, cg, v.log,
-			)
-			err = cephfsCGHandler.EnsurePVCfromRGD(rdSpec, failoverAction)
+			v.log.Info("The CG label from the primary cluster found in RDSpec", "Label", cgLabelVal)
+			// Get the CG label value for this cluster
+			cgLabelVal, err = v.getCGLabelValue(rdSpec.ProtectedPVC.StorageClassName,
+				rdSpec.ProtectedPVC.Name, rdSpec.ProtectedPVC.Namespace)
+			if err == nil {
+				cephfsCGHandler := cephfscg.NewVSCGHandler(
+					v.ctx, v.reconciler.Client, v.instance,
+					&metav1.LabelSelector{MatchLabels: map[string]string{ConsistencyGroupLabel: cgLabelVal}},
+					v.volSyncHandler, cgLabelVal, v.log,
+				)
+				err = cephfsCGHandler.EnsurePVCfromRGD(rdSpec, failoverAction)
+			}
 		} else {
 			// Create a PVC from snapshot or for direct copy
 			err = v.volSyncHandler.EnsurePVCfromRD(rdSpec, failoverAction)
@@ -263,98 +268,133 @@ func (v *VRGInstance) reconcileVolSyncAsSecondary() bool {
 	return v.reconcileRDSpecForDeletionOrReplication()
 }
 
-//nolint:gocognit,funlen,cyclop,nestif
 func (v *VRGInstance) reconcileRDSpecForDeletionOrReplication() bool {
-	requeue := false
-	rdinCGs := []ramendrv1alpha1.VolSyncReplicationDestinationSpec{}
+	rdSpecsUsingCG, requeue, err := v.reconcileCGMembership()
+	if err != nil {
+		v.log.Error(err, "Failed to reconcile CG for deletion or replication")
+
+		return requeue
+	}
 
-	// TODO: Set the workload status in CG code path later
 	for _, rdSpec := range v.instance.Spec.VolSync.RDSpec {
-		cg, ok := rdSpec.ProtectedPVC.Labels[ConsistencyGroupLabel]
-		if ok && util.IsCGEnabled(v.instance.Annotations) {
-			v.log.Info("rdSpec has CG label", "Labels", rdSpec.ProtectedPVC.Labels)
-			cephfsCGHandler := cephfscg.NewVSCGHandler(
-				v.ctx, v.reconciler.Client, v.instance,
-				&metav1.LabelSelector{MatchLabels: map[string]string{ConsistencyGroupLabel: cg}},
-				v.volSyncHandler, cg, v.log,
-			)
-
-			rdinCG, err := cephfsCGHandler.GetRDInCG()
-			if err != nil {
-				v.log.Error(err, "Failed to get RD in CG")
+		v.log.Info("Reconcile RD as Secondary", "RDSpec", rdSpec.ProtectedPVC.Name)
 
-				requeue = true
+		key := fmt.Sprintf("%s-%s", rdSpec.ProtectedPVC.Namespace, rdSpec.ProtectedPVC.Name)
 
-				return requeue
-			}
+		_, ok := rdSpecsUsingCG[key]
+		if ok {
+			v.log.Info("Skip Reconcile RD as Secondary as it's in a consistency group", "RDSpec", rdSpec.ProtectedPVC.Name)
 
-			if len(rdinCG) > 0 {
-				v.log.Info("Create ReplicationGroupDestination with RDSpecs", "RDSpecs", rdinCG)
+			continue
+		}
 
-				replicationGroupDestination, err := cephfsCGHandler.CreateOrUpdateReplicationGroupDestination(
-					v.instance.Name, v.instance.Namespace, rdinCG,
-				)
-				if err != nil {
-					v.log.Error(err, "Failed to create ReplicationGroupDestination")
+		rd, err := v.volSyncHandler.ReconcileRD(rdSpec)
+		if err != nil {
+			v.log.Error(err, "Failed to reconcile VolSync Replication Destination")
 
-					requeue = true
+			requeue = true
+
+			break
+		}
+
+		if rd == nil {
+			v.log.Info(fmt.Sprintf("ReconcileRD - ReplicationDestination for %s is not ready. We'll retry...",
+				rdSpec.ProtectedPVC.Name))
 
-					return requeue
-				}
+			requeue = true
+		}
+	}
 
-				ready, err := util.IsReplicationGroupDestinationReady(v.ctx, v.reconciler.Client, replicationGroupDestination)
-				if err != nil {
-					v.log.Error(err, "Failed to check if ReplicationGroupDestination if ready")
+	if !requeue {
+		v.log.Info("Successfully reconciled VolSync as Secondary")
+	}
 
-					requeue = true
+	return requeue
+}
 
-					return requeue
-				}
+func (v *VRGInstance) reconcileCGMembership() (map[string]struct{}, bool, error) {
+	groups := map[string][]ramendrv1alpha1.VolSyncReplicationDestinationSpec{}
 
-				if !ready {
-					v.log.Info(fmt.Sprintf("ReplicationGroupDestination for %s is not ready. We'll retry...",
-						replicationGroupDestination.Name))
+	rdSpecsUsingCG := make(map[string]struct{})
 
-					requeue = true
-				}
+	for _, rdSpec := range v.instance.Spec.VolSync.RDSpec {
+		cgLabelVal, ok := rdSpec.ProtectedPVC.Labels[ConsistencyGroupLabel]
+		if ok && util.IsCGEnabled(v.instance.Annotations) {
+			v.log.Info("RDSpec contains the CG label from the primary cluster", "Label", cgLabelVal)
+			// Get the CG label value for this cluster
+			cgLabelVal, err := v.getCGLabelValue(rdSpec.ProtectedPVC.StorageClassName,
+				rdSpec.ProtectedPVC.Name, rdSpec.ProtectedPVC.Namespace)
+			if err != nil {
+				v.log.Error(err, "Failed to get cgLabelVal")
 
-				rdinCGs = append(rdinCGs, rdinCG...)
+				return rdSpecsUsingCG, true, err
 			}
+
+			key := fmt.Sprintf("%s-%s", rdSpec.ProtectedPVC.Namespace, rdSpec.ProtectedPVC.Name)
+			rdSpecsUsingCG[key] = struct{}{}
+
+			groups[cgLabelVal] = append(groups[cgLabelVal], rdSpec)
 		}
 	}
 
-	for _, rdSpec := range v.instance.Spec.VolSync.RDSpec {
-		v.log.Info("Reconcile RD as Secondary", "RDSpec", rdSpec)
+	requeue, err := v.createOrUpdateReplicationDestinations(groups)
 
-		if util.IsRDExist(rdSpec, rdinCGs) {
-			v.log.Info("Skip Reconcile RD as Secondary as it's in a consistency group",
-				"RDSpec", rdSpec, "RDInCGs", rdinCGs)
+	return rdSpecsUsingCG, requeue, err
+}
 
-			continue
+func (v *VRGInstance) createOrUpdateReplicationDestinations(
+	groups map[string][]ramendrv1alpha1.VolSyncReplicationDestinationSpec,
+) (bool, error) {
+	requeue := false
+
+	for groupKey := range groups {
+		cephfsCGHandler := cephfscg.NewVSCGHandler(
+			v.ctx, v.reconciler.Client, v.instance,
+			&metav1.LabelSelector{MatchLabels: map[string]string{ConsistencyGroupLabel: groupKey}},
+			v.volSyncHandler, groupKey, v.log,
+		)
+
+		v.log.Info("Create ReplicationGroupDestination with RDSpecs", "RDSpecs", v.getRDSpecGroupName(groups[groupKey]))
+
+		replicationGroupDestination, err := cephfsCGHandler.CreateOrUpdateReplicationGroupDestination(
+			v.instance.Name, v.instance.Namespace, groups[groupKey],
+		)
+		if err != nil {
+			v.log.Error(err, "Failed to create ReplicationGroupDestination")
+
+			requeue = true
+
+			return requeue, err
 		}
 
-		rd, err := v.volSyncHandler.ReconcileRD(rdSpec)
+		ready, err := util.IsReplicationGroupDestinationReady(v.ctx, v.reconciler.Client, replicationGroupDestination)
 		if err != nil {
-			v.log.Error(err, "Failed to reconcile VolSync Replication Destination")
+			v.log.Error(err, "Failed to check if ReplicationGroupDestination if ready")
 
 			requeue = true
 
-			break
+			return requeue, err
 		}
 
-		if rd == nil {
-			v.log.Info(fmt.Sprintf("ReconcileRD - ReplicationDestination for %s is not ready. We'll retry...",
-				rdSpec.ProtectedPVC.Name))
+		if !ready {
+			v.log.Info(fmt.Sprintf("ReplicationGroupDestination for %s is not ready. We'll retry...",
+				replicationGroupDestination.Name))
 
 			requeue = true
 		}
 	}
 
-	if !requeue {
-		v.log.Info("Successfully reconciled VolSync as Secondary")
+	return requeue, nil
+}
+
+func (v *VRGInstance) getRDSpecGroupName(rdSpecs []ramendrv1alpha1.VolSyncReplicationDestinationSpec) string {
+	names := make([]string, 0, len(rdSpecs))
+
+	for _, rdSpec := range rdSpecs {
+		names = append(names, rdSpec.ProtectedPVC.Name)
 	}
 
-	return requeue
+	return strings.Join(names, ",")
 }
 
 func (v *VRGInstance) aggregateVolSyncDataReadyCondition() *metav1.Condition {