Ensure only one VR per PVC

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, 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
+}

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,94 @@ 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.Equal(t, testPVC.ObjectMeta.Annotations[replicationv1alpha1.VolumeReplicationNameAnnotation], vrName)
+ }
+
+ 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, 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")