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 is based on multiple VRs to an inconsistent state. With this Patch only one VR will operate on a single PVC. Signed-off-by: Madhu Rajanna <[email protected]>
csi-addons · Aug 25, 2022 · 4316fcc · 4316fcc
1 parent b77dac3
commit 4316fcc
Show file tree

Hide file tree

Showing 4 changed files with 169 additions and 2 deletions.
diff --git a/api/v1alpha1/volumereplication_types.go b/api/v1alpha1/volumereplication_types.go
@@ -36,6 +36,10 @@ const (
 	Resync ReplicationState = "resync"
 )
 
+const (
+	VolumeReplicationNameAnnotation = "replication.storage.openshift.io/volume-replication-name"
+)
+
 // State captures the latest state of the replication operation.
 type State string
 

diff --git a/controllers/pvc.go b/controllers/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/volume-replication-operator/api/v1alpha1"
 )
 
 // getPVCDataSource get pvc, pv object from the request.
@@ -56,3 +58,52 @@ 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, reqOwnerName string, pvc *corev1.PersistentVolumeClaim) error {
+	if pvc.ObjectMeta.Annotations == nil {
+		pvc.ObjectMeta.Annotations = map[string]string{}
+	}
+
+	currentOwnerName := pvc.ObjectMeta.Annotations[replicationv1alpha1.VolumeReplicationNameAnnotation]
+	if currentOwnerName == "" {
+		logger.Info("setting owner on PVC annotation", "Name", pvc.Name, "owner", reqOwnerName)
+		pvc.ObjectMeta.Annotations[replicationv1alpha1.VolumeReplicationNameAnnotation] = reqOwnerName
+		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 currentOwnerName != reqOwnerName {
+		logger.Info("cannot change the owner of PVC",
+			"PVC name", pvc.Name,
+			"current owner", currentOwnerName,
+			"requested owner", reqOwnerName)
+
+		return fmt.Errorf("PVC %q not owned by VolumeReplication %q",
+			pvc.Name, reqOwnerName)
+	}
+
+	return nil
+}
+
+// removeOwnerFromPVCAnnotation removes the VolumeReplication owner from the PVC annotations.
+func (r *VolumeReplicationReconciler) removeOwnerFromPVCAnnotation(ctx context.Context, logger logr.Logger, pvc *corev1.PersistentVolumeClaim) error {
+	if _, ok := pvc.ObjectMeta.Annotations[replicationv1alpha1.VolumeReplicationNameAnnotation]; ok {
+		logger.Info("removing annotation from PersistentVolumeClaim object", "Annotation", replicationv1alpha1.VolumeReplicationNameAnnotation)
+		delete(pvc.ObjectMeta.Annotations, replicationv1alpha1.VolumeReplicationNameAnnotation)
+		if err := r.Client.Update(ctx, pvc); err != nil {
+			return fmt.Errorf("failed to remove annotation %q from PersistentVolumeClaim "+
+				"%q %w",
+				replicationv1alpha1.VolumeReplicationNameAnnotation, pvc.Name, err)
+		}
+	}
+
+	return nil
+}
diff --git a/controllers/pvc_test.go b/controllers/pvc_test.go
@@ -17,6 +17,8 @@ limitations under the License.
 package controllers
 
 import (
+	"context"
+	"fmt"
 	"testing"
 
 	replicationv1alpha1 "github.com/csi-addons/volume-replication-operator/api/v1alpha1"
@@ -28,7 +30,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
-	logf "sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/log"
 )
 
 const (
@@ -103,7 +105,7 @@ func createFakeVolumeReplicationReconciler(t *testing.T, obj ...runtime.Object)
 	return VolumeReplicationReconciler{
 		Client:       client,
 		Scheme:       scheme,
-		Log:          logf.Log.WithName("controller_volumereplication_test"),
+		Log:          log.Log.WithName("controller_volumereplication_test"),
 		DriverConfig: &config.DriverConfig{DriverName: "test-driver"},
 	}
 }
@@ -177,3 +179,98 @@ 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)
+
+		ctx := context.TODO()
+
+		reconciler := createFakeVolumeReplicationReconciler(t, testPVC, volumeReplication)
+		err := reconciler.annotatePVCWithOwner(ctx, log.FromContext(context.TODO()), vrName, testPVC)
+		if tc.errorExpected {
+			assert.Error(t, err)
+		} else {
+			assert.NoError(t, err)
+		}
+
+		pvcNamespacedName := types.NamespacedName{
+			Name:      testPVC.Name,
+			Namespace: testPVC.Namespace,
+		}
+
+		// check annotation is added
+		err = reconciler.Get(ctx, pvcNamespacedName, testPVC)
+		assert.NoError(t, err)
+
+		assert.Contains(t, testPVC.ObjectMeta.Annotations,
+			replicationv1alpha1.VolumeReplicationNameAnnotation,
+			fmt.Sprintf("missing annotation %q, existing annotations are %v",
+				replicationv1alpha1.VolumeReplicationNameAnnotation,
+				testPVC.ObjectMeta.Annotations))
+
+		err = reconciler.removeOwnerFromPVCAnnotation(context.TODO(), log.FromContext(context.TODO()), testPVC)
+		assert.NoError(t, err)
+
+		// try calling delete again, it should not fail
+		err = reconciler.removeOwnerFromPVCAnnotation(context.TODO(), log.FromContext(context.TODO()), testPVC)
+		assert.NoError(t, err)
+	}
+
+	// try removeOwnerFromPVCAnnotation for empty map
+	pvc := &corev1.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "pvc-name",
+			Namespace: mockNamespace,
+		},
+	}
+	volumeReplication := &replicationv1alpha1.VolumeReplication{}
+	reconciler := createFakeVolumeReplicationReconciler(t, pvc, volumeReplication)
+	err := reconciler.removeOwnerFromPVCAnnotation(context.TODO(), log.FromContext(context.TODO()), pvc)
+	assert.NoError(t, err)
+}
diff --git a/controllers/volumereplication_controller.go b/controllers/volumereplication_controller.go
@@ -188,6 +188,14 @@ func (r *VolumeReplicationReconciler) Reconcile(ctx context.Context, req ctrl.Re
 
 			return reconcile.Result{}, err
 		}
+
+		err = r.annotatePVCWithOwner(ctx, logger, req.Name, pvc)
+		if err != nil {
+			logger.Error(err, "Failed to annotate PVC owner")
+
+			return ctrl.Result{}, err
+		}
+
 		if err = r.addFinalizerToPVC(logger, pvc); err != nil {
 			logger.Error(err, "Failed to add PersistentVolumeClaim finalizer")
 
@@ -201,6 +209,13 @@ func (r *VolumeReplicationReconciler) Reconcile(ctx context.Context, req ctrl.Re
 
 				return ctrl.Result{}, err
 			}
+
+			if err = r.removeOwnerFromPVCAnnotation(ctx, logger, pvc); err != nil {
+				logger.Error(err, "Failed to remove VolumeReplication annotation")
+
+				return reconcile.Result{}, err
+			}
+
 			if err = r.removeFinalizerFromPVC(logger, pvc); err != nil {
 				logger.Error(err, "Failed to remove PersistentVolumeClaim finalizer")