test

Signed-off-by: Ryan Zhang <[email protected]>

Author: Ryan Zhang

pkg/scheduler/scheduler.go

@@ -103,7 +103,7 @@ func (s *Scheduler) scheduleOnce(ctx context.Context, worker int) {
 	// Retrieve the next item (name of a placement) from the work queue.
 	//
 	// Note that this will block if no item is available.
-	placementName, closed := s.queue.NextPlacementKey()
+	placementKey, closed := s.queue.NextPlacementKey()
 	if closed {
 		// End the run immediately if the work queue has been closed.
 		klog.InfoS("Work queue has been closed")
@@ -115,48 +115,47 @@ func (s *Scheduler) scheduleOnce(ctx context.Context, worker int) {
 		//
 		// Note that this will happen even if an error occurs. Should the key get requeued by Add()
 		// during the call, it will be added to the queue after this call returns.
-		s.queue.Done(placementName)
+		s.queue.Done(placementKey)
 	}()
 
 	// keep track of the number of active scheduling loop
 	metrics.SchedulerActiveWorkers.WithLabelValues().Add(1)
 	defer metrics.SchedulerActiveWorkers.WithLabelValues().Add(-1)
 
 	startTime := time.Now()
-	placementRef := klog.KRef("", string(placementName))
-	klog.V(2).InfoS("Schedule once started", "placement", placementRef, "worker", worker)
+	klog.V(2).InfoS("Schedule once started", "placement", placementKey, "worker", worker)
 	defer func() {
 		// Note that the time spent on pulling keys from the work queue (and the time spent on waiting
 		// for a key to arrive) is not counted here, as we cannot reliably distinguish between
 		// system processing latencies and actual duration of placement absence.
 		latency := time.Since(startTime).Milliseconds()
-		klog.V(2).InfoS("Schedule once completed", "placement", placementRef, "latency", latency, "worker", worker)
+		klog.V(2).InfoS("Schedule once completed", "placement", placementKey, "latency", latency, "worker", worker)
 	}()
 
 	// Retrieve the placement object (either ClusterResourcePlacement or ResourcePlacement).
-	placement, err := controller.FetchPlacementFromKey(ctx, s.client, placementName)
+	placement, err := controller.FetchPlacementFromKey(ctx, s.client, placementKey)
 	if err != nil {
 		if apiErrors.IsNotFound(err) {
 			// The placement has been gone before the scheduler gets a chance to
 			// process it; normally this would not happen as sources would not enqueue any placement that
 			// has been marked for deletion but does not have the scheduler cleanup finalizer to
 			// the work queue. Such placements needs no further processing any way though, as the absence
 			// of the cleanup finalizer implies that bindings derived from the placement are no longer present.
-			klog.ErrorS(err, "placement is already deleted", "placement", placementRef)
+			klog.ErrorS(err, "placement is already deleted", "placement", placementKey)
 			return
 		}
 		if errors.Is(err, controller.ErrUnexpectedBehavior) {
 			// The placement is in an unexpected state; this is a scheduler-side error, and
 			// Note that this is a scheduler-side error, so it does not return an error to the caller.
 			// Raise an alert for it.
-			klog.ErrorS(err, "Placement is in an unexpected state", "placement", placementRef)
+			klog.ErrorS(err, "Placement is in an unexpected state", "placement", placementKey)
 			return
 		}
 		// Wrap the error for metrics; this method does not return an error.
-		klog.ErrorS(controller.NewAPIServerError(true, err), "Failed to get placement", "placement", placementRef)
+		klog.ErrorS(controller.NewAPIServerError(true, err), "Failed to get placement", "placement", placementKey)
 
 		// Requeue for later processing.
-		s.queue.AddRateLimited(placementName)
+		s.queue.AddRateLimited(placementKey)
 		return
 	}
 
@@ -165,21 +164,21 @@ func (s *Scheduler) scheduleOnce(ctx context.Context, worker int) {
 		// Use SchedulerCRPCleanupFinalizer consistently for all placement types
 		if controllerutil.ContainsFinalizer(placement, fleetv1beta1.SchedulerCleanupFinalizer) {
 			if err := s.cleanUpAllBindingsFor(ctx, placement); err != nil {
-				klog.ErrorS(err, "Failed to clean up all bindings for placement", "placement", placementRef)
+				klog.ErrorS(err, "Failed to clean up all bindings for placement", "placement", placementKey)
 				if errors.Is(err, controller.ErrUnexpectedBehavior) {
 					// The placement is in an unexpected state; this is a scheduler-side error, and
 					return
 				}
 				// Requeue for later processing.
-				s.queue.AddRateLimited(placementName)
+				s.queue.AddRateLimited(placementKey)
 				return
 			}
 		}
 		// The placement has been marked for deletion but no longer has the scheduler cleanup finalizer; no
 		// additional handling is needed.
 
 		// Untrack the key from the rate limiter.
-		s.queue.Forget(placementName)
+		s.queue.Forget(placementKey)
 		return
 	}
 
@@ -188,20 +187,20 @@ func (s *Scheduler) scheduleOnce(ctx context.Context, worker int) {
 	// Verify that it has an active policy snapshot.
 	latestPolicySnapshot, err := s.lookupLatestPolicySnapshot(ctx, placement)
 	if err != nil {
-		klog.ErrorS(err, "Failed to lookup latest policy snapshot", "placement", placementRef)
+		klog.ErrorS(err, "Failed to lookup latest policy snapshot", "placement", placementKey)
 		// No requeue is needed; the scheduler will be triggered again when an active policy
 		// snapshot is created.
 
 		// Untrack the key for quicker reprocessing.
-		s.queue.Forget(placementName)
+		s.queue.Forget(placementKey)
 		return
 	}
 
 	// Add the scheduler cleanup finalizer to the placement (if it does not have one yet).
 	if err := s.addSchedulerCleanUpFinalizer(ctx, placement); err != nil {
-		klog.ErrorS(err, "Failed to add scheduler cleanup finalizer", "placement", placementRef)
+		klog.ErrorS(err, "Failed to add scheduler cleanup finalizer", "placement", placementKey)
 		// Requeue for later processing.
-		s.queue.AddRateLimited(placementName)
+		s.queue.AddRateLimited(placementKey)
 		return
 	}
 
@@ -216,26 +215,26 @@ func (s *Scheduler) scheduleOnce(ctx context.Context, worker int) {
 			// The placement is in an unexpected state; this is a scheduler-side error, and
 			// Note that this is a scheduler-side error, so it does not return an error to the caller.
 			// Raise an alert for it.
-			klog.ErrorS(err, "Placement is in an unexpected state", "placement", placementRef)
+			klog.ErrorS(err, "Placement is in an unexpected state", "placement", placementKey)
 			observeSchedulingCycleMetrics(cycleStartTime, true, false)
 			return
 		}
 		// Requeue for later processing.
-		klog.ErrorS(err, "Failed to run scheduling cycle", "placement", placementRef)
-		s.queue.AddRateLimited(placementName)
+		klog.ErrorS(err, "Failed to run scheduling cycle", "placement", placementKey)
+		s.queue.AddRateLimited(placementKey)
 		observeSchedulingCycleMetrics(cycleStartTime, true, true)
 		return
 	}
 
 	// Requeue if the scheduling cycle suggests so.
 	if res.Requeue {
 		if res.RequeueAfter > 0 {
-			s.queue.AddAfter(placementName, res.RequeueAfter)
+			s.queue.AddAfter(placementKey, res.RequeueAfter)
 			observeSchedulingCycleMetrics(cycleStartTime, false, true)
 			return
 		}
 		// Untrack the key from the rate limiter.
-		s.queue.Forget(placementName)
+		s.queue.Forget(placementKey)
 		// Requeue for later processing.
 		//
 		// Note that the key is added directly to the queue without having to wait for any rate limiter's
@@ -244,11 +243,11 @@ func (s *Scheduler) scheduleOnce(ctx context.Context, worker int) {
 		// one cycle (e.g., a plugin sets up a per-cycle batch limit, and consequently the scheduler must
 		// finish the scheduling in multiple cycles); in such cases, rate limiter should not add
 		// any delay to the requeues.
-		s.queue.Add(placementName)
+		s.queue.Add(placementKey)
 		observeSchedulingCycleMetrics(cycleStartTime, false, true)
 	} else {
 		// no more failure, the following queue don't need to be rate limited
-		s.queue.Forget(placementName)
+		s.queue.Forget(placementKey)
 		observeSchedulingCycleMetrics(cycleStartTime, false, false)
 	}
 }

pkg/scheduler/watchers/clusterresourceplacement/watcher.go

@@ -21,14 +21,15 @@ package clusterresourceplacement
 import (
 	"context"
 	"fmt"
-	"time"
 
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog/v2"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/event"
-	"sigs.k8s.io/controller-runtime/pkg/predicate"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	fleetv1beta1 "github.com/kubefleet-dev/kubefleet/apis/placement/v1beta1"
 	"github.com/kubefleet-dev/kubefleet/pkg/scheduler/queue"
@@ -45,65 +46,47 @@ type Reconciler struct {
 
 // Reconcile reconciles the CRP.
 func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
-	crpRef := klog.KRef("", req.Name)
-	startTime := time.Now()
-	klog.V(2).InfoS("Scheduler source reconciliation starts", "clusterResourcePlacement", crpRef)
-	defer func() {
-		latency := time.Since(startTime).Milliseconds()
-		klog.V(2).InfoS("Scheduler source reconciliation ends", "clusterResourcePlacement", crpRef, "latency", latency)
-	}()
-
-	// Retrieve the CRP.
-	crp := &fleetv1beta1.ClusterResourcePlacement{}
-	if err := r.Client.Get(ctx, req.NamespacedName, crp); err != nil {
-		klog.ErrorS(err, "Failed to get cluster resource placement", "clusterResourcePlacement", crpRef)
-		return ctrl.Result{}, controller.NewAPIServerError(true, client.IgnoreNotFound(err))
-	}
-
-	// Check if the CRP has been deleted and has the scheduler finalizer.
-	if crp.DeletionTimestamp != nil && controllerutil.ContainsFinalizer(crp, fleetv1beta1.SchedulerCleanupFinalizer) {
-		// The CRP has been deleted and still has the scheduler finalizer;
-		// enqueue it for the scheduler to process.
-		r.SchedulerWorkQueue.AddRateLimited(queue.PlacementKey(crp.Name))
-	}
+	r.SchedulerWorkQueue.AddRateLimited(controller.GetPlacementKeyFromRequest(req))
 
 	// No action is needed for the scheduler to take in other cases.
 	return ctrl.Result{}, nil
 }
 
 // SetupWithManager sets up the controller with the manager.
 func (r *Reconciler) SetupWithManager(mgr ctrl.Manager) error {
-	customPredicate := predicate.Funcs{
-		CreateFunc: func(e event.CreateEvent) bool {
-			// Ignore creation events.
-			return false
-		},
-		DeleteFunc: func(e event.DeleteEvent) bool {
-			// Ignore deletion events (events emitted when the object is actually removed
-			// from storage).
-			return false
-		},
-		UpdateFunc: func(e event.UpdateEvent) bool {
-			// Check if the update event is valid.
-			if e.ObjectOld == nil || e.ObjectNew == nil {
-				err := controller.NewUnexpectedBehaviorError(fmt.Errorf("update event is invalid"))
-				klog.ErrorS(err, "Failed to process update event")
-				return false
-			}
-
-			// Check if the deletion timestamp has been set.
-			oldDeletionTimestamp := e.ObjectOld.GetDeletionTimestamp()
-			newDeletionTimestamp := e.ObjectNew.GetDeletionTimestamp()
-			if oldDeletionTimestamp == nil && newDeletionTimestamp != nil {
-				return true
-			}
-
-			return false
-		},
+	// Create a new controller for the CRP watcher.
+	// This controller watches for deletions of ClusterResourcePlacement and ResourcePlacement objects.
+	updateFunc := func(ctx context.Context, e event.UpdateEvent, q workqueue.TypedRateLimitingInterface[reconcile.Request]) {
+		klog.V(2).InfoS("Handling a placement update event", "placement", klog.KObj(e.ObjectNew))
+		// Check if the update event is valid.
+		if e.ObjectOld == nil || e.ObjectNew == nil {
+			err := controller.NewUnexpectedBehaviorError(fmt.Errorf("update event is invalid"))
+			klog.ErrorS(err, "Failed to process update event")
+		}
+
+		// Check if the deletion timestamp has been set.
+		oldDeletionTimestamp := e.ObjectOld.GetDeletionTimestamp()
+		newDeletionTimestamp := e.ObjectNew.GetDeletionTimestamp()
+		if oldDeletionTimestamp == nil && newDeletionTimestamp != nil {
+			q.Add(reconcile.Request{
+				NamespacedName: types.NamespacedName{Name: e.ObjectNew.GetName(), Namespace: e.ObjectNew.GetNamespace()},
+			})
+			klog.V(2).InfoS("Enqueued placement for scheduler processing", "placement", klog.KObj(e.ObjectNew))
+		}
 	}
 
-	return ctrl.NewControllerManagedBy(mgr).Named("clusterresourceplacement-scheduler-watcher").
-		For(&fleetv1beta1.ClusterResourcePlacement{}).
-		WithEventFilter(customPredicate).
+	deleteFunc := func(ctx context.Context, e event.DeleteEvent, q workqueue.TypedRateLimitingInterface[reconcile.Request]) {
+		klog.V(2).InfoS("Handling a placement delete event", "placement", klog.KObj(e.Object))
+		q.Add(reconcile.Request{
+			NamespacedName: types.NamespacedName{Name: e.Object.GetName(), Namespace: e.Object.GetNamespace()},
+		})
+	}
+	handler := handler.Funcs{
+		DeleteFunc: deleteFunc,
+		UpdateFunc: updateFunc,
+	}
+	return ctrl.NewControllerManagedBy(mgr).Named("placement-scheduler-watcher").
+		Watches(&fleetv1beta1.ClusterResourcePlacement{}, handler).
+		Watches(&fleetv1beta1.ResourcePlacement{}, handler).
 		Complete(r)
 }

pkg/utils/controller/placement_resolver.go

@@ -22,6 +22,7 @@ import (
 	"strings"
 
 	"k8s.io/apimachinery/pkg/types"
+	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	fleetv1beta1 "github.com/kubefleet-dev/kubefleet/apis/placement/v1beta1"
@@ -76,6 +77,17 @@ func GetPlacementKeyFromObj(obj fleetv1beta1.PlacementObj) queue.PlacementKey {
 	}
 }
 
+// GetPlacementKeyFromObj generates a PlacementKey from a placement object.
+func GetPlacementKeyFromRequest(req ctrl.Request) queue.PlacementKey {
+	if req.Namespace == "" {
+		// Cluster-scoped placement
+		return queue.PlacementKey(req.Name)
+	} else {
+		// Namespaced placement
+		return queue.PlacementKey(req.Namespace + namespaceSeparator + req.Name)
+	}
+}
+
 // ExtractNamespaceNameFromKey resolves a PlacementKey to a (namespace, name) tuple of the placement object.
 func ExtractNamespaceNameFromKey(placementKey queue.PlacementKey) (string, string, error) {
 	keyStr := string(placementKey)