Don't allow two DRPCs with conflicting selectors in the same namespace

controllers/drpc_namespace_conflict_test.go

@@ -0,0 +1,63 @@
+// SPDX-FileCopyrightText: The RamenDR authors
+// SPDX-License-Identifier: Apache-2.0
+
+package controllers_test
+
+import (
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"k8s.io/apimachinery/pkg/types"
+
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	rmn "github.com/ramendr/ramen/api/v1alpha1"
+	plrv1 "github.com/stolostron/multicloud-operators-placementrule/pkg/apis/apps/v1"
+)
+
+var _ = Describe("DRPlacementControl Reconciler New", func() {
+	Specify("DRClusters", func() {
+		populateDRClusters()
+	})
+	Context("DRPlacementControl Reconciler Async DR using PlacementRule (Subscription)", func() {
+		var userPlacementRule *plrv1.PlacementRule
+		var drpc *rmn.DRPlacementControl
+		When("An Application is deployed for the first time", func() {
+			It("Should deploy to East1ManagedCluster", func() {
+				By("Initial Deployment")
+				var placementObj client.Object
+				placementObj, drpc = InitialDeploymentAsync(
+					DRPCNamespaceName, UserPlacementRuleName, East1ManagedCluster, UsePlacementRule)
+				testLogger.Info("dprc rtalur logging", "drpc", drpc)
+				userPlacementRule = placementObj.(*plrv1.PlacementRule)
+				Expect(userPlacementRule).NotTo(BeNil())
+				verifyInitialDRPCDeployment(userPlacementRule, East1ManagedCluster)
+				verifyDRPCOwnedByPlacement(userPlacementRule, getLatestDRPC())
+				drpc2 := createDRPC(drpc.Spec.PlacementRef.Name, "overlappingdrpc", drpc.Namespace, drpc.Spec.DRPolicyRef.Name)
+				testLogger.Info("dprc rtalur logging", "drpc", drpc2)
+				drpc2, err := getLatestDRPCByNamespacedName(types.NamespacedName{Namespace: drpc2.Namespace, Name: drpc2.Name})
+				Expect(err).NotTo(HaveOccurred())
+				testLogger.Info("dprc rtalur logging", "drpc", drpc2)
+			})
+		})
+		//When("Deleting user PlacementRule", func() {
+		//	It("Should cleanup DRPC", func() {
+		//		// ----------------------------- DELETE DRPC from PRIMARY --------------------------------------
+		//		By("\n\n*** DELETE User PlacementRule ***\n\n")
+		//		deleteUserPlacementRule()
+		//	})
+		//})
+
+		//When("Deleting DRPC", func() {
+		//	It("Should delete VRG and NS MWs and MCVs from Primary (East1ManagedCluster)", func() {
+		//		// ----------------------------- DELETE DRPC from PRIMARY --------------------------------------
+		//		By("\n\n*** DELETE DRPC ***\n\n")
+		//		Expect(getManifestWorkCount(East1ManagedCluster)).Should(Equal(3)) // DRCluster + VRG MW
+		//		deleteDRPC()
+		//		waitForCompletion("deleted")
+		//		Expect(getManifestWorkCount(East1ManagedCluster)).Should(Equal(2))       // DRCluster + NS MW only
+		//		Expect(getManagedClusterViewCount(East1ManagedCluster)).Should(Equal(0)) // NS + VRG MCV
+		//		deleteNamespaceMWsFromAllClusters(DRPCNamespaceName)
+		//	})
+		//})
+	})
+})

controllers/drplacementcontrol_controller.go

@@ -723,6 +723,25 @@
 		return ctrl.Result{}, nil
 	}
 
+	if err, otherDRPC := r.ensureHasNoConflictingLabelSelectors(ctx, drpc); err != nil {
+		// check which drpc has the finalizer
+		if controllerutil.ContainsFinalizer(drpc, DRPCFinalizer) {
+			if controllerutil.ContainsFinalizer(otherDRPC, DRPCFinalizer) {
+				logger.Error(err, "Two valid DRPC found with label selectors that conflict",
+					"this DRPC", drpc, "other DRPC", otherDRPC)
+				r.recordFailure(ctx, drpc, placementObj, "Error: two DRPCs have the same pvc label selector", err.Error(), logger)
+			} else {
+				// otherDRPC has no finalizer, so we will assume this drpc is the correct one
+				// log only in debug cases
+				logger.V(1).Info("Conflicting DRPC has been created", "Error", err)
+			}
+		} else {
+			logger.Error(err, "Existing DRPC found with label selectors that conflict with label selectors of this drpc",
+				"this DRPC", drpc, "existing drpc", otherDRPC)
+			r.recordFailure(ctx, drpc, placementObj, "Error: two DRPCs have the same pvc label selector", err.Error(), logger)
+		}
+	}
+
 	drPolicy, err := r.getAndEnsureValidDRPolicy(ctx, drpc, logger)
 	if err != nil {
 		r.recordFailure(ctx, drpc, placementObj, "Error", err.Error(), logger)
@@ -2422,3 +2441,27 @@
 
 	return AllowFailover, msg, nil
 }
+
+func (r *DRPlacementControlReconciler) ensureHasNoConflictingLabelSelectors(ctx context.Context, drpc *rmn.DRPlacementControl) (error, *rmn.DRPlacementControl) {
+	drpcList := &rmn.DRPlacementControlList{}
+	err := r.APIReader.List(ctx, drpcList)
+	if err != nil {
+		return fmt.Errorf("failed to list DRPC objects %w", err), nil
+	}
+
+	for _, otherDRPC := range drpcList.Items {
+		if otherDRPC.Name == drpc.Name {
+			continue
+		}
+
+		if otherDRPC.Namespace != drpc.Namespace {
+			continue
+		}
+
+		if canMatchSameElements(&drpc.Spec.PVCSelector, &otherDRPC.Spec.PVCSelector) {
+			return fmt.Errorf("DRPC %s has conflicting PVCSelector with DRPC %s", drpc.Name, otherDRPC.Name), &otherDRPC
+		}
+	}
+
+	return nil, nil
+}

controllers/drplacementcontrol_controller_test.go

@@ -749,13 +749,19 @@ func setDRPCSpecExpectationTo(namespace, preferredCluster, failoverCluster strin
 	}, timeout, interval).Should(BeTrue(), "failed to update DRPC DR action on time")
 }
 
+func getLatestDRPCByNamespacedName(key types.NamespacedName) (*rmn.DRPlacementControl, error) {
+	latestDRPC := &rmn.DRPlacementControl{}
+	err := apiReader.Get(context.TODO(), key, latestDRPC)
+
+	return latestDRPC, err
+}
+
 func getLatestDRPC(namespace string) *rmn.DRPlacementControl {
 	drpcLookupKey := types.NamespacedName{
 		Name:      DRPCCommonName,
 		Namespace: namespace,
 	}
-	latestDRPC := &rmn.DRPlacementControl{}
-	err := apiReader.Get(context.TODO(), drpcLookupKey, latestDRPC)
+	latestDRPC, err := getLatestDRPCByNamespacedName(drpcLookupKey)
 	Expect(err).NotTo(HaveOccurred())
 
 	return latestDRPC

controllers/label_selector.go

@@ -0,0 +1,140 @@
+// SPDX-FileCopyrightText: The RamenDR authors
+// SPDX-License-Identifier: Apache-2.0
+
+package controllers
+
+import (
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// Relationship type between two value sets.
+type relationship int
+
+const (
+	overlap relationship = iota
+	disjoint
+	subset
+	superset
+	equal
+)
+
+func determineRelationship(a, b []string) relationship {
+	setA := make(map[string]bool)
+	setB := make(map[string]bool)
+
+	for _, v := range a {
+		setA[v] = true
+	}
+	for _, v := range b {
+		setB[v] = true
+	}
+
+	// Check if A is a subset or equal to B
+	isSubset := true
+	for v := range setA {
+		if !setB[v] {
+			isSubset = false
+			break
+		}
+	}
+
+	// Check if B is a subset or equal to A
+	isSuperset := true
+	for v := range setB {
+		if !setA[v] {
+			isSuperset = false
+			break
+		}
+	}
+
+	if isSubset && isSuperset {
+		return equal
+	}
+	if isSubset {
+		return subset
+	}
+	if isSuperset {
+		return superset
+	}
+
+	for v := range setA {
+		if setB[v] {
+			return overlap
+		}
+	}
+
+	return disjoint
+}
+
+func isConflicting(op1, op2 v1.LabelSelectorOperator, rel relationship) bool {
+	switch op1 {
+	case v1.LabelSelectorOpIn:
+		switch op2 {
+		case v1.LabelSelectorOpIn:
+			return rel == disjoint
+		case v1.LabelSelectorOpNotIn:
+			return rel == subset || rel == equal
+		case v1.LabelSelectorOpDoesNotExist:
+			return rel != disjoint
+		}
+	case v1.LabelSelectorOpNotIn:
+		switch op2 {
+		case v1.LabelSelectorOpIn:
+			return rel == superset || rel == equal
+		case v1.LabelSelectorOpDoesNotExist:
+			return rel != disjoint
+		}
+	case v1.LabelSelectorOpExists:
+		switch op2 {
+		case v1.LabelSelectorOpNotIn, v1.LabelSelectorOpDoesNotExist:
+			return true
+		}
+	case v1.LabelSelectorOpDoesNotExist:
+		switch op2 {
+		case v1.LabelSelectorOpIn, v1.LabelSelectorOpNotIn, v1.LabelSelectorOpExists:
+			return true
+		}
+	}
+	return false
+}
+
+func canMatchSameElements(selector1, selector2 *v1.LabelSelector) bool {
+	// Convert matchLabels to LabelSelectorRequirements for unified processing
+	allExprs1 := labelsToExpressions(selector1.MatchLabels)
+	allExprs2 := labelsToExpressions(selector2.MatchLabels)
+
+	// Consolidate all matchExpressions
+	allExprs1 = append(allExprs1, selector1.MatchExpressions...)
+	allExprs2 = append(allExprs2, selector2.MatchExpressions...)
+
+	// Check if any expression contradicts the other selector
+	return !selectorsWillNeverMatch(allExprs1, allExprs2)
+}
+
+func labelsToExpressions(matchLabels map[string]string) []v1.LabelSelectorRequirement {
+	var exprs []v1.LabelSelectorRequirement
+	for k, v := range matchLabels {
+		exprs = append(exprs, v1.LabelSelectorRequirement{
+			Key:      k,
+			Operator: v1.LabelSelectorOpIn,
+			Values:   []string{v},
+		})
+	}
+	return exprs
+}
+
+func selectorsWillNeverMatch(exprs1, exprs2 []v1.LabelSelectorRequirement) bool {
+	for _, e1 := range exprs1 {
+		for _, e2 := range exprs2 {
+			if e1.Key != e2.Key {
+				continue
+			}
+
+			relationship := determineRelationship(e1.Values, e2.Values)
+			if isConflicting(e1.Operator, e2.Operator, relationship) {
+				return true
+			}
+		}
+	}
+	return false
+}

controllers/label_selector_internal_test.go

@@ -0,0 +1,117 @@
+// SPDX-FileCopyrightText: The RamenDR authors
+// SPDX-License-Identifier: Apache-2.0
+
+package controllers
+
+import (
+	"testing"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+//nolint:funlen
+func TestCanMatchSameElements(t *testing.T) {
+	tests := []struct {
+		name        string
+		selectorA   *v1.LabelSelector
+		selectorB   *v1.LabelSelector
+		shouldMatch bool
+	}{
+		{
+			name: "Test with two keys where one can match",
+			selectorA: &v1.LabelSelector{
+				MatchExpressions: []v1.LabelSelectorRequirement{
+					{Key: "tier", Operator: v1.LabelSelectorOpIn, Values: []string{"frontend"}},
+					{Key: "env", Operator: v1.LabelSelectorOpIn, Values: []string{"dev", "staging"}},
+				},
+			},
+			selectorB: &v1.LabelSelector{
+				MatchExpressions: []v1.LabelSelectorRequirement{
+					{Key: "tier", Operator: v1.LabelSelectorOpIn, Values: []string{"frontend", "backend"}},
+					{Key: "env", Operator: v1.LabelSelectorOpNotIn, Values: []string{"prod"}},
+				},
+			},
+			shouldMatch: true,
+		},
+		{
+			name: "Test with one k=v and one k exists",
+			selectorA: &v1.LabelSelector{
+				MatchExpressions: []v1.LabelSelectorRequirement{
+					{Key: "tier", Operator: v1.LabelSelectorOpIn, Values: []string{"frontend"}},
+				},
+			},
+			selectorB: &v1.LabelSelector{
+				MatchExpressions: []v1.LabelSelectorRequirement{
+					{Key: "tier", Operator: v1.LabelSelectorOpExists},
+				},
+			},
+			shouldMatch: true,
+		},
+		{
+			name: "Test with k={v1} and k={v2}",
+			selectorA: &v1.LabelSelector{
+				MatchExpressions: []v1.LabelSelectorRequirement{
+					{Key: "tier", Operator: v1.LabelSelectorOpIn, Values: []string{"cache"}},
+				},
+			},
+			selectorB: &v1.LabelSelector{
+				MatchExpressions: []v1.LabelSelectorRequirement{
+					{Key: "tier", Operator: v1.LabelSelectorOpIn, Values: []string{"backup"}},
+				},
+			},
+			shouldMatch: false,
+		},
+		{
+			name: "Test with k1={v1},k2!={v3} and k={v2},k2!={v3}",
+			selectorA: &v1.LabelSelector{
+				MatchExpressions: []v1.LabelSelectorRequirement{
+					{Key: "tier", Operator: v1.LabelSelectorOpIn, Values: []string{"cache"}},
+					{Key: "env", Operator: v1.LabelSelectorOpNotIn, Values: []string{"dev"}},
+				},
+			},
+			selectorB: &v1.LabelSelector{
+				MatchExpressions: []v1.LabelSelectorRequirement{
+					{Key: "tier", Operator: v1.LabelSelectorOpIn, Values: []string{"backup"}},
+					{Key: "env", Operator: v1.LabelSelectorOpNotIn, Values: []string{"dev"}},
+				},
+			},
+			shouldMatch: false,
+		},
+		{
+			name: "Test with cache and not cache",
+			selectorA: &v1.LabelSelector{
+				MatchExpressions: []v1.LabelSelectorRequirement{
+					{Key: "tier", Operator: v1.LabelSelectorOpIn, Values: []string{"cache"}},
+				},
+			},
+			selectorB: &v1.LabelSelector{
+				MatchExpressions: []v1.LabelSelectorRequirement{
+					{Key: "tier", Operator: v1.LabelSelectorOpNotIn, Values: []string{"cache"}},
+				},
+			},
+			shouldMatch: false,
+		},
+		{
+			name:      "Test with empty selector",
+			selectorA: &v1.LabelSelector{},
+			selectorB: &v1.LabelSelector{
+				MatchExpressions: []v1.LabelSelectorRequirement{
+					{Key: "tier", Operator: v1.LabelSelectorOpIn, Values: []string{"backup"}},
+				},
+			},
+			shouldMatch: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if match := canMatchSameElements(tt.selectorA, tt.selectorB); match != tt.shouldMatch {
+				if tt.shouldMatch {
+					t.Fatalf("Expected selectors to potentially match the same elements, but they were determined not to.")
+				} else {
+					t.Fatalf("Expected selectors to not match the same elements, but they were determined to.")
+				}
+			}
+		})
+	}
+}