Update only affected replicated policies

Signed-off-by: Yi Rae Kim <[email protected]>

Author: yiraeChristineKim

controllers/common/common.go

@@ -245,6 +245,74 @@ func FullNameForPolicy(plc *policiesv1.Policy) string {
 	return plc.GetNamespace() + "." + plc.GetName()
 }
 
+// GetRepPoliciesInPlacementBinding returns a list of the RepPolicies that are either direct subjects of
+// the given PlacementBinding, or are in PolicySets that are subjects of the PlacementBinding.
+// The list items are guaranteed to be unique (for example if a policy is in multiple sets).
+func GetRepPoliciesInPlacementBinding(
+	ctx context.Context, c client.Client, pb *policiesv1.PlacementBinding,
+) []reconcile.Request {
+	result := make([]reconcile.Request, 0)
+
+	decisions, err := GetDecisions(c, pb) //nolint:contextcheck
+	if err != nil {
+		return result
+	}
+
+	for _, subject := range pb.Subjects {
+		if subject.APIGroup != policiesv1.SchemeGroupVersion.Group {
+			continue
+		}
+
+		switch subject.Kind {
+		case policiesv1.Kind:
+			for _, pd := range decisions {
+				result = append(result, reconcile.Request{NamespacedName: types.NamespacedName{
+					Name:      pb.GetNamespace() + "." + subject.Name,
+					Namespace: pd.ClusterName,
+				}})
+			}
+
+		case policiesv1.PolicySetKind:
+			setNN := types.NamespacedName{
+				Name:      subject.Name,
+				Namespace: pb.GetNamespace(),
+			}
+
+			policySet := policiesv1beta1.PolicySet{}
+			if err := c.Get(ctx, setNN, &policySet); err != nil {
+				continue
+			}
+
+			for _, plc := range policySet.Spec.Policies {
+				for _, pd := range decisions {
+					result = append(result, reconcile.Request{NamespacedName: types.NamespacedName{
+						Name:      pb.GetNamespace() + "." + string(plc),
+						Namespace: pd.ClusterName,
+					}})
+				}
+			}
+		}
+	}
+
+	var unique []reconcile.Request
+
+	table := map[reconcile.Request]bool{}
+	// Remove duplicated policies
+	if len(result) != 0 {
+		for _, r := range result {
+			if !table[r] {
+				table[r] = true
+
+				unique = append(unique, r)
+			}
+		}
+
+		result = unique
+	}
+
+	return result
+}
+
 // TypeConverter is a helper function to converter type struct a to b
 func TypeConverter(a, b interface{}) error {
 	js, err := json.Marshal(a)
@@ -254,3 +322,26 @@ func TypeConverter(a, b interface{}) error {
 
 	return json.Unmarshal(js, b)
 }
+
+// Select objects that are deleted or created
+func GetAffectedOjbs[T comparable](oldObjs []T, newObjs []T) []T {
+	table := make(map[T]int)
+
+	for _, oldObj := range oldObjs {
+		table[oldObj] = 1
+	}
+
+	for _, newObj := range newObjs {
+		table[newObj]++
+	}
+
+	result := []T{}
+
+	for key, val := range table {
+		if val == 1 {
+			result = append(result, key)
+		}
+	}
+
+	return result
+}

controllers/common/common_test.go

@@ -1,6 +1,13 @@
 package common
 
-import "testing"
+import (
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"k8s.io/apimachinery/pkg/types"
+	clusterv1beta1 "open-cluster-management.io/api/cluster/v1beta1"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+)
 
 func TestParseRootPolicyLabel(t *testing.T) {
 	tests := map[string]struct {
@@ -28,3 +35,43 @@ func TestParseRootPolicyLabel(t *testing.T) {
 		})
 	}
 }
+
+func TestGetAffectedOjbsWithDecision(t *testing.T) {
+	newOjbs := []clusterv1beta1.ClusterDecision{
+		{ClusterName: "managed1", Reason: "test11"},
+		{ClusterName: "managed2", Reason: "test11"},
+	}
+	oldObjs := []clusterv1beta1.ClusterDecision{
+		{ClusterName: "managed1", Reason: "test11"},
+		{ClusterName: "managed3", Reason: "test11"},
+	}
+	expectedResult := []clusterv1beta1.ClusterDecision{
+		{ClusterName: "managed2", Reason: "test11"},
+		{ClusterName: "managed3", Reason: "test11"},
+	}
+
+	result := GetAffectedOjbs(newOjbs, oldObjs)
+	if !cmp.Equal(result, expectedResult) {
+		t.Fatalf(`GetAffectedOjbs test failed expected: %+v but result is %+v`, expectedResult, result)
+	}
+}
+
+func TestGetAffectedOjbsWithRequest(t *testing.T) {
+	newOjbs := []reconcile.Request{
+		{NamespacedName: types.NamespacedName{Namespace: "test1", Name: "test1"}},
+		{NamespacedName: types.NamespacedName{Namespace: "test2", Name: "test2"}},
+	}
+	oldOjbs := []reconcile.Request{
+		{NamespacedName: types.NamespacedName{Namespace: "test1", Name: "test1"}},
+		{NamespacedName: types.NamespacedName{Namespace: "test3", Name: "test3"}},
+	}
+	expectedResult := []reconcile.Request{
+		{NamespacedName: types.NamespacedName{Namespace: "test2", Name: "test2"}},
+		{NamespacedName: types.NamespacedName{Namespace: "test3", Name: "test3"}},
+	}
+
+	result := GetAffectedOjbs(newOjbs, oldOjbs)
+	if !cmp.Equal(result, expectedResult) {
+		t.Fatalf(`GetAffectedOjbs test failed expected: %+v but result is %+v`, expectedResult, result)
+	}
+}

controllers/propagator/replicatedpolicy_controller.go

@@ -17,6 +17,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	policiesv1 "open-cluster-management.io/governance-policy-propagator/api/v1"
+	policyv1beta1 "open-cluster-management.io/governance-policy-propagator/api/v1beta1"
 	"open-cluster-management.io/governance-policy-propagator/controllers/common"
 )
 
@@ -126,6 +127,22 @@ func (r *ReplicatedPolicyReconciler) Reconcile(ctx context.Context, request ctrl
 		return reconcile.Result{}, nil
 	}
 
+	// err = r.rootStatusUpdate(rootPolicy)
+	// if err != nil {
+	// 	return reconcile.Result{}, err
+	// }
+	// Handle replicated policy, which is related to policyset
+	isHandled, err := r.handlePolicySet(ctx, rootPolicy, replicatedPolicy)
+	if err == nil && isHandled {
+		return reconcile.Result{}, nil
+	}
+
+	if err != nil && !isHandled {
+		log.Error(err, "Failed to remove the replicated policy for PolicySet, requeueing")
+
+		return reconcile.Result{}, err
+	}
+
 	if rootPolicy.Spec.Disabled {
 		if replicatedExists {
 			if err := r.cleanUpReplicated(ctx, replicatedPolicy); err != nil {
@@ -511,3 +528,61 @@ func (r *ReplicatedPolicyReconciler) isSingleClusterInDecisions(
 
 	return false, nil
 }
+
+// Check this replicated policy related to policyset and
+// Check policyset existing and delete policies deteched from policyset
+func (r *ReplicatedPolicyReconciler) handlePolicySet(
+	ctx context.Context, rootPlc *policiesv1.Policy, replicatedPolicy *policiesv1.Policy,
+) (isHandled bool, err error) {
+	// Find PlacementBinding
+	pbList := &policiesv1.PlacementBindingList{}
+
+	err = r.List(ctx, pbList, &client.ListOptions{Namespace: rootPlc.GetNamespace()})
+	if err != nil {
+		return false, err
+	}
+
+	for _, pb := range pbList.Items {
+		for _, sub := range pb.Subjects {
+			policySet := &policyv1beta1.PolicySet{}
+			if sub.Kind == "PolicySet" && policyv1beta1.GroupVersion.Group == sub.APIGroup {
+				if err := r.Get(ctx, types.NamespacedName{
+					Namespace: pb.Namespace,
+					Name:      sub.Name,
+				}, policySet); err != nil {
+					isCorrectPolicyset := false
+
+					for _, plc := range rootPlc.Status.Placement {
+						if plc.PolicySet == sub.Name {
+							isCorrectPolicyset = true
+						}
+					}
+
+					if !isCorrectPolicyset {
+						log.V(2).Info("Incorrect PolicySet", "policySetName", sub.Name)
+
+						return false, nil
+					}
+
+					if k8serrors.IsNotFound(err) {
+						log.Info("PolicySet deleted so delete related replicated policy")
+
+						if err := r.cleanUpReplicated(ctx, replicatedPolicy); err != nil {
+							if !k8serrors.IsNotFound(err) {
+								return false, err
+							}
+						}
+
+						return true, nil
+					}
+
+					log.Error(err, "Error to get policySet")
+
+					return false, err
+				}
+			}
+		}
+	}
+
+	return false, nil
+}

controllers/propagator/replicatedpolicy_setup.go

@@ -2,20 +2,26 @@ package propagator
 
 import (
 	"sync"
+	"time"
 
 	"k8s.io/apimachinery/pkg/api/equality"
+	clusterv1beta1 "open-cluster-management.io/api/cluster/v1beta1"
+	appsv1 "open-cluster-management.io/multicloud-operators-subscription/pkg/apis/apps/placementrule/v1"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	policiesv1 "open-cluster-management.io/governance-policy-propagator/api/v1"
 	"open-cluster-management.io/governance-policy-propagator/controllers/common"
 )
 
+var emptyRequest = []reconcile.Request{}
+
 func (r *ReplicatedPolicyReconciler) SetupWithManager(
 	mgr ctrl.Manager,
 	maxConcurrentReconciles uint,
@@ -32,6 +38,30 @@ func (r *ReplicatedPolicyReconciler) SetupWithManager(
 		WatchesRawSource(dependenciesSource, &handler.EnqueueRequestForObject{}).
 		WatchesRawSource(updateSrc, &handler.EnqueueRequestForObject{}).
 		WatchesRawSource(templateSrc, &handler.EnqueueRequestForObject{}).
+		Watches(
+			&clusterv1beta1.PlacementDecision{},
+			EnqueueReqsFromMapFuncByDecision(mgr.GetClient()),
+		).
+		Watches(
+			&policiesv1.PlacementBinding{},
+			EnqueueReqsFromMapFuncByBinding(mgr.GetClient()),
+		).
+		Watches(
+			&appsv1.PlacementRule{},
+			EnqueueReqsFromMapFuncByRule(mgr.GetClient()),
+		).
+		WatchesRawSource(
+			dynWatcherSrc,
+			// The dependency-watcher could create an event before the same sort of watch in the
+			// controller-runtime triggers an update in the cache. This tries to ensure the cache is
+			// updated before the reconcile is triggered.
+			&delayGeneric{
+				EventHandler: &handler.EnqueueRequestForObject{},
+				delay:        time.Second * 3,
+			}).
+		WatchesRawSource(
+			updateSrc,
+			&handler.EnqueueRequestForObject{}).
 		Complete(r)
 }