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 <madhupr007@gmail.com>

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
 

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, 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 == "" {
+ logger.Info("setting owner on PVC annotation", "Name", pvc.Name, "owner", reqOwnerName)
+ 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
+}
+
+// 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
+}

controllers/pvc_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package controllers
 
 import (
+ "context"
  "testing"
 
  replicationv1alpha1 "github.com/csi-addons/volume-replication-operator/api/v1alpha1"
@@ -28,7 +29,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 +104,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 +178,86 @@ 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)
+ }
+
+ 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)
+}

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