filter drpc pvc labels

Signed-off-by: Raghavendra Talur <raghavendra.talur@gmail.com>

controllers/drplacementcontrol_controller.go

@@ -18,6 +18,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -633,6 +634,10 @@ func (r *DRPlacementControlReconciler) Reconcile(ctx context.Context, req ctrl.R
 		return ctrl.Result{}, nil
 	}
 
+	if err, _ := r.ensureHasNoConflictingLabelSelectors(ctx, drpc); err != nil {
+		r.recordFailure(ctx, drpc, placementObj, "Error", err.Error(), logger)
+	}
+
 	drPolicy, err := r.getAndEnsureValidDRPolicy(ctx, drpc, logger)
 	if err != nil {
 		r.recordFailure(ctx, drpc, placementObj, "Error", err.Error(), logger)
@@ -2024,3 +2029,96 @@ func ensureDRPCConditionsInited(conditions *[]metav1.Condition, observedGenerati
 		Message:            message,
 	})
 }
+
+func canMatchSameElements(selector1, selector2 *v1.LabelSelector) bool {
+	// Convert matchLabels to LabelSelectorRequirements for unified processing
+	exprFromLabels1 := labelsToExpressions(selector1.MatchLabels)
+	exprFromLabels2 := labelsToExpressions(selector2.MatchLabels)
+
+	// Consolidate all matchExpressions
+	allExprs1 := append(exprFromLabels1, selector1.MatchExpressions...)
+	allExprs2 := append(exprFromLabels2, selector2.MatchExpressions...)
+
+	// Check if any expression contradicts the other selector
+	return !expressionsContradict(allExprs1, allExprs2) && !expressionsContradict(allExprs2, allExprs1)
+}
+
+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 expressionsContradict(exprs1, exprs2 []v1.LabelSelectorRequirement) bool {
+	for _, expr1 := range exprs1 {
+		for _, expr2 := range exprs2 {
+			if expr1.Key != expr2.Key {
+				continue
+			}
+
+			switch expr1.Operator {
+			case v1.LabelSelectorOpIn:
+				if expr2.Operator == v1.LabelSelectorOpNotIn {
+					for _, val := range expr1.Values {
+						if contains(expr2.Values, val) {
+							return true
+						}
+					}
+				}
+			case v1.LabelSelectorOpNotIn:
+				if expr2.Operator == v1.LabelSelectorOpIn {
+					for _, val := range expr1.Values {
+						if contains(expr2.Values, val) {
+							return true
+						}
+					}
+				}
+			case v1.LabelSelectorOpExists:
+				if expr2.Operator == v1.LabelSelectorOpDoesNotExist {
+					return true
+				}
+			case v1.LabelSelectorOpDoesNotExist:
+				if expr2.Operator == v1.LabelSelectorOpExists {
+					return true
+				}
+			}
+		}
+	}
+
+	return false
+}
+
+func contains(slice []string, item string) bool {
+	for _, a := range slice {
+		if a == item {
+			return true
+		}
+	}
+	return false
+}
+
+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 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
+}