Ensure only one VR per PVC

apis/replication.storage/v1alpha1/volumereplication_types.go

@@ -21,6 +21,10 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
+const (
+	VolumeReplicationNameAnnotation = "replication.storage.openshift.io/volume-replication-name"
+)
+
 // ReplicationState represents the replication operations to be performed on the volume.
 // +kubebuilder:validation:Enum=primary;secondary;resync
 type ReplicationState string

controllers/replication.storage/pvc.go

@@ -24,6 +24,8 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/types"
+
+	replicationv1alpha1 "github.com/csi-addons/kubernetes-csi-addons/apis/replication.storage/v1alpha1"
 )
 
 // getPVCDataSource get pvc, pv object from the request.
@@ -56,3 +58,36 @@ func (r VolumeReplicationReconciler) getPVCDataSource(logger logr.Logger, req ty
 
 	return pvc, pv, nil
 }
+
+// annotatePVCWithOwner will add the VolumeReplication details to the PVC annotations.
+func (r *VolumeReplicationReconciler) annotatePVCWithOwner(ctx context.Context, logger logr.Logger, req types.NamespacedName, pvc *corev1.PersistentVolumeClaim) error {
+	if pvc.ObjectMeta.Annotations == nil {
+		pvc.ObjectMeta.Annotations = map[string]string{}
+	}
+
+	ownerName := pvc.ObjectMeta.Annotations[replicationv1alpha1.VolumeReplicationNameAnnotation]
+	if ownerName == "" {
+		pvc.ObjectMeta.Annotations[replicationv1alpha1.VolumeReplicationNameAnnotation] = req.Name
+		err := r.Update(ctx, pvc)
+		if err != nil {
+			logger.Error(err, "Failed to update PVC annotation", "Name", pvc.Name)
+
+			return fmt.Errorf("failed to update PVC %q annotation for VolumeReplication: %w",
+				pvc.Name, err)
+		}
+
+		return nil
+	}
+
+	if ownerName != req.Name {
+		logger.Info("cannot change the owner of PVC",
+			"PVC name", pvc.Name,
+			"current owner", ownerName,
+			"requested owner", req.Name)
+
+		return fmt.Errorf("PVC %q not owned by VolumeReplication %q",
+			pvc.Name, req.Name)
+	}
+
+	return nil
+}

controllers/replication.storage/pvc_test.go

@@ -21,7 +21,6 @@ import (
 	"testing"
 
 	replicationv1alpha1 "github.com/csi-addons/kubernetes-csi-addons/apis/replication.storage/v1alpha1"
-
 	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -175,3 +174,67 @@ func TestGetVolumeHandle(t *testing.T) {
 		}
 	}
 }
+
+func TestVolumeReplicationReconciler_annotatePVCWithOwner(t *testing.T) {
+	t.Parallel()
+	vrName := "test-vr"
+
+	testcases := []struct {
+		name          string
+		pvc           *corev1.PersistentVolumeClaim
+		errorExpected bool
+	}{
+		{
+			name:          "case 1: no VR is owning the PVC",
+			pvc:           mockPersistentVolumeClaim,
+			errorExpected: false,
+		},
+		{
+			name: "case 2: pvc is already owned by same VR",
+			pvc: &corev1.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "pvc-name",
+					Namespace: mockNamespace,
+					Annotations: map[string]string{
+						replicationv1alpha1.VolumeReplicationNameAnnotation: vrName,
+					},
+				},
+			},
+			errorExpected: false,
+		},
+		{
+			name: "case 2: pvc is owned by different VR",
+			pvc: &corev1.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "pvc-name",
+					Namespace: mockNamespace,
+					Annotations: map[string]string{
+						replicationv1alpha1.VolumeReplicationNameAnnotation: "test-vr-1",
+					},
+				},
+			},
+			errorExpected: true,
+		},
+	}
+
+	for _, tc := range testcases {
+		volumeReplication := &replicationv1alpha1.VolumeReplication{}
+		mockVolumeReplicationObj.DeepCopyInto(volumeReplication)
+
+		testPVC := &corev1.PersistentVolumeClaim{}
+		tc.pvc.DeepCopyInto(testPVC)
+
+		namespacedName := types.NamespacedName{
+			Name:      vrName,
+			Namespace: mockNamespace,
+		}
+
+		reconciler := createFakeVolumeReplicationReconciler(t, testPVC, volumeReplication)
+		err := reconciler.annotatePVCWithOwner(context.TODO(), log.FromContext(context.TODO()), namespacedName, testPVC)
+		if tc.errorExpected {
+			assert.Error(t, err)
+		} else {
+			assert.NoError(t, err)
+		}
+	}
+}

controllers/replication.storage/volumereplication_controller.go

@@ -172,6 +172,12 @@ func (r *VolumeReplicationReconciler) Reconcile(ctx context.Context, req ctrl.Re
 		logger.Info("Replication handle", "ReplicationHandleName", replicationHandle)
 	}
 
+	err = r.annotatePVCWithOwner(ctx, logger, nameSpacedName, pvc)
+	if err != nil {
+		logger.Error(err, "Failed to annotate PVC owner")
+		return ctrl.Result{}, err
+	}
+
 	replicationClient, err := r.getReplicationClient(vrcObj.Spec.Provisioner)
 	if err != nil {
 		logger.Error(err, "Failed to get ReplicationClient")