Ensure only one VR per PVC

VolumeReplication should ensure there
is only one VR per PVC, as otherwise
it can lead to orchestrating the
PVC based on multiple VRs to an
inconsistent state. With this Patch only
one VR will operator on a single PVC.

Signed-off-by: Madhu Rajanna <[email protected]>

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")