Fixes issue when multiple VRGs conflict with the PVCs being protected

internal/controller/util/nslock.go

@@ -0,0 +1,45 @@
+// SPDX-FileCopyrightText: The RamenDR authors
+// SPDX-License-Identifier: Apache-2.0
+
+package util
+
+import (
+	"sync"
+)
+
+// NamespaceLock implements atomic operation for namespace. It will have the namespace
+// having multiple vrgs in which VRGs are being processed.
+type NamespaceLock struct {
+	namespace string
+	mux       sync.Mutex
+}
+
+// NewNamespaceLock returns new NamespaceLock
+func NewNamespaceLock() *NamespaceLock {
+	return &NamespaceLock{}
+}
+
+// TryToAcquireLock tries to acquire the lock for processing VRG in a namespace having
+// multiple VRGs and returns true if successful.
+// If processing has already begun in the namespace, returns false.
+func (nl *NamespaceLock) TryToAcquireLock(namespace string) bool {
+	nl.mux.Lock()
+	defer nl.mux.Unlock()
+
+	if nl.namespace == namespace {
+		return false
+	}
+
+	if nl.namespace == "" {
+		nl.namespace = namespace
+	}
+
+	return true
+}
+
+// Release removes lock on the namespace
+func (nl *NamespaceLock) Release(namespace string) {
+	nl.mux.Lock()
+	defer nl.mux.Unlock()
+	nl.namespace = ""
+}

internal/controller/util/nslock_test.go

@@ -0,0 +1,18 @@
+// SPDX-FileCopyrightText: The RamenDR authors
+// SPDX-License-Identifier: Apache-2.0
+
+package util_test
+
+import (
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"github.com/ramendr/ramen/internal/controller/util"
+)
+
+var _ = Describe("Testing Locks", func() {
+	nsLock := util.NewNamespaceLock()
+	Expect(nsLock.TryToAcquireLock("test")).To(BeTrue())
+	Expect(nsLock.TryToAcquireLock("test")).To(BeFalse())
+	nsLock.Release("test")
+	Expect(nsLock.TryToAcquireLock("test")).To(BeTrue())
+})

internal/controller/volumereplicationgroup_controller.go

@@ -55,6 +55,7 @@ type VolumeReplicationGroupReconciler struct {
 	kubeObjects         kubeobjects.RequestsManager
 	RateLimiter         *workqueue.RateLimiter
 	veleroCRsAreWatched bool
+	locks               *rmnutil.NamespaceLock
 }
 
 // SetupWithManager sets up the controller with the Manager.
@@ -115,6 +116,8 @@ func (r *VolumeReplicationGroupReconciler) SetupWithManager(
 		r.Log.Info("Kube object protection disabled; don't watch kube objects requests")
 	}
 
+	r.locks = rmnutil.NewNamespaceLock()
+
 	return ctrlBuilder.Complete(r)
 }
 
@@ -437,6 +440,12 @@ func (r *VolumeReplicationGroupReconciler) Reconcile(ctx context.Context, req ct
 				"Please install velero/oadp and restart the operator", v.instance.Namespace, v.instance.Name)
 	}
 
+	err = v.vrgParallelProcessingCheck(adminNamespaceVRG)
+
+	if err != nil {
+		return ctrl.Result{Requeue: true}, err
+	}
+
 	v.volSyncHandler = volsync.NewVSHandler(ctx, r.Client, log, v.instance,
 		v.instance.Spec.Async, cephFSCSIDriverNameOrDefault(v.ramenConfig),
 		volSyncDestinationCopyMethodOrDefault(v.ramenConfig), adminNamespaceVRG)
@@ -1626,3 +1635,32 @@ func (r *VolumeReplicationGroupReconciler) addKubeObjectsOwnsAndWatches(ctrlBuil
 
 	return ctrlBuilder
 }
+
+func (v *VRGInstance) vrgParallelProcessingCheck(adminNamespaceVRG bool) error {
+	ns := v.instance.Namespace
+
+	if !adminNamespaceVRG {
+		vrgList := &ramendrv1alpha1.VolumeReplicationGroupList{}
+		listOps := &client.ListOptions{
+			Namespace: ns,
+		}
+
+		err := v.reconciler.APIReader.List(context.Background(), vrgList, listOps)
+		if err != nil {
+			v.log.Error(err, "Unable to list the VRGs in the", " namespace ", ns)
+
+			return err
+		}
+
+		// if the number of vrgs in the ns is more than 1, lock is needed.
+		if len(vrgList.Items) > 1 {
+			if isLockAcquired := v.reconciler.locks.TryToAcquireLock(ns); !isLockAcquired {
+				// Acquiring lock failed, VRG reconcile should be requeued
+				return err
+			}
+			defer v.reconciler.locks.Release(ns)
+		}
+	}
+
+	return nil
+}