refactor(scheduler): adopt unified types

cmd/controller-manager/app/controllermanager.go

@@ -39,7 +39,7 @@ const (
 	FederateControllerName                 = "federate"
 	FollowerControllerName                 = "follower"
 	PolicyRCControllerName                 = "policyrc"
-	OverrideControllerName                 = "overridepolicy"
+	OverrideControllerName                 = "override"
 	NamespaceAutoPropagationControllerName = "nsautoprop"
 	StatusControllerName                   = "status"
 	SchedulerName                          = "scheduler"

pkg/controllers/common/constants.go

@@ -140,9 +140,6 @@ const (
 	// TemplateGeneratorMergePatchAnnotation indicates the merge patch document capable of converting
 	// the source object to the template object.
 	TemplateGeneratorMergePatchAnnotation = FederateControllerPrefix + "template-generator-merge-patch"
-
-	PropagationPolicyNameLabel        = DefaultPrefix + "propagation-policy-name"
-	ClusterPropagationPolicyNameLabel = DefaultPrefix + "cluster-propagation-policy-name"
 )
 
 // PropagatedAnnotationKeys and PropagatedLabelKeys are used to store the keys of annotations and labels that are present

pkg/controllers/federate/controller.go

@@ -111,7 +111,6 @@ func NewFederateController(
 	c.eventRecorder = eventsink.NewDefederatingRecorderMux(kubeClient, FederateControllerName, 6)
 	c.worker = worker.NewReconcileWorker[workerKey](
 		FederateControllerName,
-		nil,
 		c.reconcile,
 		worker.RateLimiterOptions{},
 		workerCount,

pkg/controllers/federate/util.go

@@ -270,8 +270,8 @@ var (
 	)
 
 	federatedLabelSet = sets.New[string](
-		common.PropagationPolicyNameLabel,
-		common.ClusterPropagationPolicyNameLabel,
+		scheduler.PropagationPolicyNameLabel,
+		scheduler.ClusterPropagationPolicyNameLabel,
 		override.OverridePolicyNameLabel,
 		override.ClusterOverridePolicyNameLabel,
 	)

pkg/controllers/federatedcluster/clusterstatus.go

@@ -33,6 +33,7 @@
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/client-go/discovery"
 	corev1listers "k8s.io/client-go/listers/core/v1"
+	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/retry"
 	"k8s.io/klog/v2"
 
@@ -68,24 +69,14 @@
 	if !exists {
 		return 0, fmt.Errorf("failed to get cluster client: FederatedInformerManager not yet up-to-date")
 	}
-
 	podLister, podsSynced, exists := c.federatedInformerManager.GetPodLister(cluster.Name)
 	if !exists {
 		return 0, fmt.Errorf("failed to get pod lister: FederatedInformerManager not yet up-to-date")
 	}
-	if !podsSynced() {
-		logger.V(3).Info("Pod informer not synced, will reenqueue")
-		return 100 * time.Millisecond, nil
-	}
-
 	nodeLister, nodesSynced, exists := c.federatedInformerManager.GetNodeLister(cluster.Name)
 	if !exists {
 		return 0, fmt.Errorf("failed to get node lister: FederatedInformerManager not yet up-to-date")
 	}
-	if !nodesSynced() {
-		logger.V(3).Info("Pod informer not synced, will reenqueue")
-		return 100 * time.Millisecond, nil
-	}
 
 	discoveryClient := clusterKubeClient.Discovery()
 
@@ -108,7 +99,14 @@
 
 	// We skip updating cluster resources and api resources if cluster is not ready
 	if readyStatus == corev1.ConditionTrue {
-		if err := updateClusterResources(ctx, &cluster.Status, podLister, nodeLister); err != nil {
+		if err := updateClusterResources(
+			ctx,
+			&cluster.Status,
+			podLister,
+			podsSynced,
+			nodeLister,
+			nodesSynced,
+		); err != nil {
 			logger.Error(err, "Failed to update cluster resources")
 			readyStatus = corev1.ConditionFalse
 			readyReason = ClusterResourceCollectionFailedReason
@@ -174,8 +172,16 @@
 	ctx context.Context,
 	clusterStatus *fedcorev1a1.FederatedClusterStatus,
 	podLister corev1listers.PodLister,
+	podsSynced cache.InformerSynced,
 	nodeLister corev1listers.NodeLister,
+	nodesSynced cache.InformerSynced,
 ) error {
+	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
+	if !cache.WaitForCacheSync(ctx.Done(), podsSynced, nodesSynced) {	
+		return fmt.Errorf("timeout waiting for node and pod informer sync")	
+	}
+
 	nodes, err := nodeLister.List(labels.Everything())
 	if err != nil {
 		return fmt.Errorf("failed to list nodes: %w", err)

pkg/controllers/federatedcluster/controller.go

@@ -96,16 +96,16 @@
 	logger klog.Logger,
 	clusterJoinTimeout time.Duration,
 	workerCount int,
-	fedsystemNamespace string,
+	fedSystemNamespace string,
 ) (*FederatedClusterController, error) {
 	c := &FederatedClusterController{
 		clusterInformer:          clusterInformer,
 		federatedInformerManager: federatedInformerManager,
 		kubeClient:               kubeClient,
 		fedClient:                fedClient,
-		fedSystemNamespace:       fedsystemNamespace,
+		fedSystemNamespace:       fedSystemNamespace,
 		clusterHealthCheckConfig: &ClusterHealthCheckConfig{
-			Period: time.Minute,
+			Period: time.Second*30,
 		},
 		clusterJoinTimeout: clusterJoinTimeout,
 		metrics:            metrics,
@@ -124,7 +124,6 @@
 
 	c.worker = worker.NewReconcileWorker[common.QualifiedName](
 		FederatedClusterControllerName,
-		nil,
 		c.reconcile,
 		worker.RateLimiterOptions{},
 		workerCount,
@@ -133,7 +132,6 @@
 
 	c.statusCollectWorker = worker.NewReconcileWorker[common.QualifiedName](
 		FederatedClusterControllerName,
-		nil,
 		c.collectClusterStatus,
 		worker.RateLimiterOptions{
 			InitialDelay: 50 * time.Millisecond,
@@ -164,7 +162,7 @@
 }
 
 func (c *FederatedClusterController) HasSynced() bool {
-	return c.clusterInformer.Informer().HasSynced()
+	return c.clusterInformer.Informer().HasSynced() && c.federatedInformerManager.HasSynced()
 }
 
 func (c *FederatedClusterController) IsControllerReady() bool {
@@ -292,7 +290,7 @@
 
 	// Trigger initial status collection if successfully joined
 	if joined, alreadyFailed := isClusterJoined(&cluster.Status); joined && !alreadyFailed {
-		c.statusCollectWorker.EnqueueObject(cluster)
+		c.statusCollectWorker.Enqueue(common.NewQualifiedName(cluster))
 	}
 
 	return worker.StatusAllOK
@@ -492,7 +490,7 @@
 
 	for _, cluster := range clusters {
 		if clusterutil.IsClusterJoined(&cluster.Status) {
-			c.statusCollectWorker.EnqueueObject(cluster)
+			c.statusCollectWorker.Enqueue(common.NewQualifiedName(cluster))
 		}
 	}
 }

pkg/controllers/nsautoprop/controller.go

@@ -124,7 +124,6 @@ func NewNamespaceAutoPropagationController(
 
 	c.worker = worker.NewReconcileWorker[common.QualifiedName](
 		NamespaceAutoPropagationControllerName,
-		nil,
 		c.reconcile,
 		worker.RateLimiterOptions{},
 		workerCount,
@@ -136,7 +135,7 @@ func NewNamespaceAutoPropagationController(
 			func(obj *fedcorev1a1.ClusterFederatedObject) {
 				srcMeta, err := obj.Spec.GetTemplateAsUnstructured()
 				if err != nil {
-					logger.Error(
+					c.logger.Error(
 						err,
 						"Failed to get source object's metadata from ClusterFederatedObject",
 						"object",

pkg/controllers/override/overridepolicy_controller.go

@@ -99,7 +99,6 @@ func NewOverridePolicyController(
 	c.eventRecorder = eventsink.NewDefederatingRecorderMux(kubeClient, ControllerName, 4)
 	c.worker = worker.NewReconcileWorker[common.QualifiedName](
 		ControllerName,
-		nil,
 		c.reconcile,
 		worker.RateLimiterOptions{},
 		workerCount,
@@ -235,7 +234,7 @@ func (c *Controller) enqueueFedObjectsUsingPolicy(policy fedcorev1a1.GenericOver
 
 func (c *Controller) reconcileOnClusterChange(cluster *fedcorev1a1.FederatedCluster) {
 	logger := c.logger.WithValues("federated-cluster", cluster.GetName())
-	logger.V(2).Info("observed a cluster change")
+	logger.V(2).Info("Observed a cluster change")
 
 	opRequirement, _ := labels.NewRequirement(OverridePolicyNameLabel, selection.Exists, nil)
 	copRequirement, _ := labels.NewRequirement(ClusterOverridePolicyNameLabel, selection.Exists, nil)
@@ -339,7 +338,7 @@ func (c *Controller) reconcile(ctx context.Context, qualifiedName common.Qualifi
 		return worker.StatusError
 	}
 
-	var overrides util.OverridesMap
+	var overrides overridesMap
 	// Apply overrides from each policy in order
 	for _, policy := range policies {
 		newOverrides, err := parseOverrides(policy, placedClusters)
@@ -358,16 +357,11 @@ func (c *Controller) reconcile(ctx context.Context, qualifiedName common.Qualifi
 		overrides = mergeOverrides(overrides, newOverrides)
 	}
 
-	currentOverrides, err := util.GetOverrides(fedObject, PrefixedControllerName)
-	if err != nil {
-		keyedLogger.Error(err, "Failed to get overrides")
-		return worker.StatusError
-	}
-
+	currentOverrides := fedObject.GetSpec().GetControllerOverrides(PrefixedControllerName)
 	needsUpdate := !equality.Semantic.DeepEqual(overrides, currentOverrides)
 
 	if needsUpdate {
-		err = util.SetOverrides(fedObject, PrefixedControllerName, overrides)
+		err = setOverrides(fedObject, overrides)
 		if err != nil {
 			keyedLogger.Error(err, "Failed to set overrides")
 			return worker.StatusError

pkg/controllers/override/util.go

@@ -26,10 +26,11 @@ import (
 
 	fedcorev1a1 "github.com/kubewharf/kubeadmiral/pkg/apis/core/v1alpha1"
 	fedcorev1a1listers "github.com/kubewharf/kubeadmiral/pkg/client/listers/core/v1alpha1"
-	"github.com/kubewharf/kubeadmiral/pkg/controllers/util"
 	"github.com/kubewharf/kubeadmiral/pkg/util/clusterselector"
 )
 
+type overridesMap map[string]fedcorev1a1.OverridePatches
+
 /*
 lookForMatchedPolicies looks for OverridePolicy and/or ClusterOverridePolicy
 that match the obj in the stores.
@@ -90,8 +91,8 @@ func lookForMatchedPolicies(
 func parseOverrides(
 	policy fedcorev1a1.GenericOverridePolicy,
 	clusters []*fedcorev1a1.FederatedCluster,
-) (util.OverridesMap, error) {
-	overridesMap := make(util.OverridesMap)
+) (overridesMap, error) {
+	overridesMap := make(overridesMap)
 
 	for _, cluster := range clusters {
 		patches := make(fedcorev1a1.OverridePatches, 0)
@@ -130,9 +131,9 @@ func parseOverrides(
 	return overridesMap, nil
 }
 
-func mergeOverrides(dest, src util.OverridesMap) util.OverridesMap {
+func mergeOverrides(dest, src overridesMap) overridesMap {
 	if dest == nil {
-		dest = make(util.OverridesMap)
+		dest = make(overridesMap)
 	}
 
 	for clusterName, srcOverrides := range src {
@@ -229,3 +230,29 @@ func policyJsonPatchOverriderToOverridePatch(
 
 	return overridePatch, nil
 }
+
+func setOverrides(federatedObj fedcorev1a1.GenericFederatedObject, overridesMap overridesMap) error {
+	for clusterName, clusterOverrides := range overridesMap {
+		if len(clusterOverrides) == 0 {
+			delete(overridesMap, clusterName)
+		}
+	}
+
+	if len(overridesMap) == 0 {
+		federatedObj.GetSpec().DeleteControllerOverrides(PrefixedControllerName)
+		return nil
+	}
+
+	overrides := []fedcorev1a1.ClusterReferenceWithPatches{}
+
+	for clusterName, clusterOverrides := range overridesMap {
+		overrides = append(overrides, fedcorev1a1.ClusterReferenceWithPatches{
+			Cluster: clusterName,
+			Patches: clusterOverrides,
+		})
+	}
+
+	federatedObj.GetSpec().SetControllerOverrides(PrefixedControllerName, overrides)
+
+	return nil
+}

pkg/controllers/override/util_test.go

@@ -27,7 +27,6 @@
 	"k8s.io/client-go/tools/cache"
 
 	fedcorev1a1 "github.com/kubewharf/kubeadmiral/pkg/apis/core/v1alpha1"
-	fedtypesv1a1 "github.com/kubewharf/kubeadmiral/pkg/apis/types/v1alpha1"
 	"github.com/kubewharf/kubeadmiral/pkg/controllers/util"
 )
 
@@ -299,7 +298,7 @@
 				}
 			}
 
 			foundPolicies, needsRecheckOnError, err := lookForMatchedPolicies(obj, isNamespaced, overridePolicyStore, clusterOverridePolicyStore)
 			if (err != nil) != testCase.isErrorExpected {
 				t.Fatalf("err = %v, but isErrorExpected = %v", err, testCase.isErrorExpected)
 			}

pkg/controllers/policyrc/controller.go

@@ -32,6 +32,7 @@
 	fedcorev1a1informers "github.com/kubewharf/kubeadmiral/pkg/client/informers/externalversions/core/v1alpha1"
 	"github.com/kubewharf/kubeadmiral/pkg/controllers/common"
 	"github.com/kubewharf/kubeadmiral/pkg/controllers/override"
+	"github.com/kubewharf/kubeadmiral/pkg/controllers/scheduler"
 	"github.com/kubewharf/kubeadmiral/pkg/stats"
 	"github.com/kubewharf/kubeadmiral/pkg/util/eventhandlers"
 	"github.com/kubewharf/kubeadmiral/pkg/util/fedobjectadapters"
@@ -82,23 +83,8 @@
 		logger:                           logger.WithValues("controller", ControllerName),
 	}
 
-	if _, err := c.fedObjectInformer.Informer().AddEventHandler(eventhandlers.NewTriggerOnAllChangesWithTransform(
-		common.NewQualifiedName,
-		c.countWorker.Enqueue,
-	)); err != nil {
-		return nil, err
-	}
-
-	if _, err := c.clusterFedObjectInformer.Informer().AddEventHandler(eventhandlers.NewTriggerOnAllChangesWithTransform(
-		common.NewQualifiedName,
-		c.countWorker.Enqueue,
-	)); err != nil {
-		return nil, err
-	}
-
 	c.countWorker = worker.NewReconcileWorker[common.QualifiedName](
 		"policyrc-controller-count-worker",
-		nil,
 		c.reconcileCount,
 		worker.RateLimiterOptions{},
 		1, // currently only one worker is meaningful due to the global mutex
@@ -107,7 +93,6 @@
 
 	c.persistPpWorker = worker.NewReconcileWorker[common.QualifiedName](
 		"policyrc-controller-persist-worker",
-		nil,
 		func(ctx context.Context, qualifiedName common.QualifiedName) worker.Result {
 			return c.reconcilePersist(
 				ctx,
@@ -125,7 +110,6 @@
 
 	c.persistOpWorker = worker.NewReconcileWorker[common.QualifiedName](
 		"policyrc-controller-persist-worker",
-		nil,
 		func(ctx context.Context, qualifiedName common.QualifiedName) worker.Result {
 			return c.reconcilePersist(
 				ctx,
@@ -141,6 +125,21 @@
 		metrics,
 	)
 
+	if _, err := c.fedObjectInformer.Informer().AddEventHandler(eventhandlers.NewTriggerOnAllChangesWithTransform(
+		common.NewQualifiedName,
+		c.countWorker.Enqueue,
+	)); err != nil {
+		return nil, err
+	}
+
+	if _, err := c.clusterFedObjectInformer.Informer().AddEventHandler(eventhandlers.NewTriggerOnAllChangesWithTransform(
+		common.NewQualifiedName,
+		c.countWorker.Enqueue,
+	)); err != nil {
+		return nil, err
+	}
+
+
 	if _, err := c.propagationPolicyInformer.Informer().AddEventHandler(
 		eventhandlers.NewTriggerOnAllChangesWithTransform(common.NewQualifiedName, c.persistPpWorker.Enqueue),
 	); err != nil {
@@ -236,7 +235,7 @@
 
 	var newPps []PolicyKey
 	if fedObj != nil {
-		newPolicy, newHasPolicy := fedobjectadapters.MatchedPolicyKey(fedObj, fedObj.GetNamespace() != "")
+		newPolicy, newHasPolicy := scheduler.GetMatchedPolicyKey(fedObj)
 		if newHasPolicy {
 			newPps = []PolicyKey{PolicyKey(newPolicy)}
 		}