diff --git a/.changes/unreleased/changed-20260805-165940.yaml b/.changes/unreleased/changed-20260805-165940.yaml new file mode 100644 index 000000000..a70ab0382 --- /dev/null +++ b/.changes/unreleased/changed-20260805-165940.yaml @@ -0,0 +1,3 @@ +kind: Changed +body: |- + Avoid duplicate binder reconciliation events diff --git a/pkg/binder/controllers/bindrequest_controller.go b/pkg/binder/controllers/bindrequest_controller.go index d94026852..9baad6b55 100644 --- a/pkg/binder/controllers/bindrequest_controller.go +++ b/pkg/binder/controllers/bindrequest_controller.go @@ -106,7 +106,7 @@ func (r *BindRequestReconciler) Reconcile(ctx context.Context, req ctrl.Request) return result, nil } - if bindRequest.Status.Phase == schedulingv1alpha2.BindRequestPhaseSucceeded { + if !isActionableBindRequest(bindRequest) { return result, nil } @@ -176,7 +176,7 @@ func (r *BindRequestReconciler) Reconcile(ctx context.Context, req ctrl.Request) // SetupWithManager sets up the controller with the Manager. func (r *BindRequestReconciler) SetupWithManager(mgr ctrl.Manager) error { return ctrl.NewControllerManagedBy(mgr). - For(&schedulingv1alpha2.BindRequest{}). + Named("bindrequest"). Watches(&schedulingv1alpha2.BindRequest{}, r.eventHandlers()). WithOptions(controller.Options{ MaxConcurrentReconciles: r.params.MaxConcurrentReconciles, @@ -192,10 +192,16 @@ func (r *BindRequestReconciler) SetupWithManager(mgr ctrl.Manager) error { func (r *BindRequestReconciler) eventHandlers() handler.Funcs { return handler.Funcs{ CreateFunc: func(ctx context.Context, event event.CreateEvent, wq workqueue.TypedRateLimitingInterface[reconcile.Request]) { + if !isActionableBindRequestObject(event.Object) { + return + } h := handler.EnqueueRequestForObject{} h.Create(ctx, event, wq) }, UpdateFunc: func(ctx context.Context, event event.UpdateEvent, wq workqueue.TypedRateLimitingInterface[reconcile.Request]) { + if !shouldEnqueueBindRequestUpdate(event.ObjectOld, event.ObjectNew) { + return + } h := handler.EnqueueRequestForObject{} h.Update(ctx, event, wq) }, @@ -205,6 +211,41 @@ func (r *BindRequestReconciler) eventHandlers() handler.Funcs { } } +func isActionableBindRequestObject(object client.Object) bool { + bindRequest, ok := object.(*schedulingv1alpha2.BindRequest) + return ok && isActionableBindRequest(bindRequest) +} + +func isActionableBindRequest(bindRequest *schedulingv1alpha2.BindRequest) bool { + if bindRequest == nil || bindRequest.DeletionTimestamp != nil { + return false + } + + switch bindRequest.Status.Phase { + case schedulingv1alpha2.BindRequestPhaseSucceeded: + return false + case schedulingv1alpha2.BindRequestPhaseFailed: + return bindRequest.Spec.BackoffLimit != nil && + bindRequest.Status.FailedAttempts < *bindRequest.Spec.BackoffLimit + default: + return true + } +} + +func shouldEnqueueBindRequestUpdate(oldObject, newObject client.Object) bool { + newBindRequest, ok := newObject.(*schedulingv1alpha2.BindRequest) + if !ok || !isActionableBindRequest(newBindRequest) { + return false + } + + oldBindRequest, ok := oldObject.(*schedulingv1alpha2.BindRequest) + if !ok { + return true + } + + return oldBindRequest.Generation != newBindRequest.Generation +} + func (r *BindRequestReconciler) deleteHandler(ctx context.Context, event event.TypedDeleteEvent[client.Object], _ workqueue.TypedRateLimitingInterface[reconcile.Request]) { logger := log.FromContext(ctx) @@ -231,10 +272,12 @@ func (r *BindRequestReconciler) UpdateStatus( originalBindRequest := &schedulingv1alpha2.BindRequest{} bindRequest.DeepCopyInto(originalBindRequest) + shouldRetry := false if err != nil { if bindRequest.Spec.BackoffLimit != nil && *bindRequest.Spec.BackoffLimit > bindRequest.Status.FailedAttempts { result.RequeueAfter = (1 << bindRequest.Status.FailedAttempts) * time.Second bindRequest.Status.FailedAttempts++ + shouldRetry = true } bindRequest.Status.Phase = schedulingv1alpha2.BindRequestPhaseFailed bindRequest.Status.Reason = err.Error() @@ -252,7 +295,13 @@ func (r *BindRequestReconciler) UpdateStatus( "Namespace", bindRequest.Namespace, "Name", bindRequest.Name) } - return result, err + if shouldRetry { + return result, nil + } + if err != nil { + return result, reconcile.TerminalError(err) + } + return result, nil } func (r *BindRequestReconciler) updatePodCondition( diff --git a/pkg/binder/controllers/bindrequest_controller_test.go b/pkg/binder/controllers/bindrequest_controller_test.go index 8a8266f1e..cd7273715 100644 --- a/pkg/binder/controllers/bindrequest_controller_test.go +++ b/pkg/binder/controllers/bindrequest_controller_test.go @@ -18,10 +18,13 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/client-go/tools/record" + "k8s.io/client-go/util/workqueue" "k8s.io/utils/ptr" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/event" + "sigs.k8s.io/controller-runtime/pkg/reconcile" kubeaischedulerscheme "github.com/kai-scheduler/KAI-scheduler/pkg/apis/client/clientset/versioned/scheme" schedulingv1alpha2 "github.com/kai-scheduler/KAI-scheduler/pkg/apis/scheduling/v1alpha2" @@ -221,7 +224,7 @@ var _ = Describe("BindRequest Controller", func() { bindRequest.Spec.BackoffLimit = ptr.To(int32(1)) return bindRequest }(), - true, true, false, + false, true, false, ), Entry( "missing node but backoff limit already reached", @@ -306,6 +309,67 @@ var _ = Describe("BindRequest Controller", func() { }) }) + Describe("event handlers", func() { + assertCreateQueueLength := func(bindRequest *schedulingv1alpha2.BindRequest, expected int) { + queue := workqueue.NewTypedRateLimitingQueue(workqueue.DefaultTypedControllerRateLimiter[reconcile.Request]()) + DeferCleanup(queue.ShutDown) + + reconciler.eventHandlers().CreateFunc(context.TODO(), event.CreateEvent{Object: bindRequest}, queue) + Expect(queue.Len()).To(Equal(expected)) + } + + It("enqueues only actionable BindRequest creates", func() { + assertCreateQueueLength(baseRequest.DeepCopy(), 1) + + succeeded := baseRequest.DeepCopy() + succeeded.Status.Phase = schedulingv1alpha2.BindRequestPhaseSucceeded + assertCreateQueueLength(succeeded, 0) + + failedWithoutBackoff := baseRequest.DeepCopy() + failedWithoutBackoff.Status.Phase = schedulingv1alpha2.BindRequestPhaseFailed + assertCreateQueueLength(failedWithoutBackoff, 0) + + failedAfterBackoff := baseRequest.DeepCopy() + failedAfterBackoff.Status.Phase = schedulingv1alpha2.BindRequestPhaseFailed + failedAfterBackoff.Spec.BackoffLimit = ptr.To(int32(1)) + failedAfterBackoff.Status.FailedAttempts = 1 + assertCreateQueueLength(failedAfterBackoff, 0) + + failedWithRetry := baseRequest.DeepCopy() + failedWithRetry.Status.Phase = schedulingv1alpha2.BindRequestPhaseFailed + failedWithRetry.Spec.BackoffLimit = ptr.To(int32(2)) + failedWithRetry.Status.FailedAttempts = 1 + assertCreateQueueLength(failedWithRetry, 1) + }) + + It("enqueues only actionable BindRequest spec updates", func() { + queue := workqueue.NewTypedRateLimitingQueue(workqueue.DefaultTypedControllerRateLimiter[reconcile.Request]()) + DeferCleanup(queue.ShutDown) + + oldBindRequest := baseRequest.DeepCopy() + updatedBindRequest := oldBindRequest.DeepCopy() + updatedBindRequest.Generation = oldBindRequest.Generation + 1 + reconciler.eventHandlers().UpdateFunc(context.TODO(), event.UpdateEvent{ + ObjectOld: oldBindRequest, + ObjectNew: updatedBindRequest, + }, queue) + Expect(queue.Len()).To(Equal(1)) + + request, shutdown := queue.Get() + Expect(shutdown).To(BeFalse()) + queue.Done(request) + + terminal := updatedBindRequest.DeepCopy() + terminal.Status.Phase = schedulingv1alpha2.BindRequestPhaseSucceeded + terminal.Generation++ + reconciler.eventHandlers().UpdateFunc(context.TODO(), event.UpdateEvent{ + ObjectOld: updatedBindRequest, + ObjectNew: terminal, + }, queue) + Expect(queue.Len()).To(Equal(0)) + }) + }) + Describe("UpdateStatus", func() { Context("no error", func() { It("Leavs FailedAttempts unchanged", func() { diff --git a/pkg/binder/controllers/pod_controller.go b/pkg/binder/controllers/pod_controller.go index 2857b67ef..876cd0c4c 100644 --- a/pkg/binder/controllers/pod_controller.go +++ b/pkg/binder/controllers/pod_controller.go @@ -17,7 +17,6 @@ import ( "sigs.k8s.io/controller-runtime/pkg/event" "sigs.k8s.io/controller-runtime/pkg/handler" "sigs.k8s.io/controller-runtime/pkg/log" - "sigs.k8s.io/controller-runtime/pkg/reconcile" "github.com/kai-scheduler/KAI-scheduler/pkg/binder/binding/resourcereservation" "github.com/kai-scheduler/KAI-scheduler/pkg/common/resources" @@ -40,7 +39,6 @@ type ReconcilerParams struct { // +kubebuilder:rbac:groups=core,resources=pods,verbs=get;list;watch;create;update;patch;delete // +kubebuilder:rbac:groups=core,resources=pods/status,verbs=get;update;patch // +kubebuilder:rbac:groups=core,resources=pods/finalizers,verbs=update -// +kubebuilder:rbac:groups=core,resources=configmaps,verbs=get;list;watch;create;update;patch;delete // Reconcile is part of the main kubernetes reconciliation loop which aims to // move the current state of the cluster closer to the desired state. @@ -55,7 +53,7 @@ func (r *PodReconciler) SetupWithManager( mgr ctrl.Manager, params *ReconcilerParams, ) error { return ctrl.NewControllerManagedBy(mgr). - For(&corev1.Pod{}). + Named("pod"). Watches(&corev1.Pod{}, r.eventHandlers()). WithOptions(controller.Options{ MaxConcurrentReconciles: params.MaxConcurrentReconciles, @@ -65,44 +63,24 @@ func (r *PodReconciler) SetupWithManager( ), SkipNameValidation: &[]bool{true}[0], }). - Owns(&corev1.ConfigMap{}). Complete(r) } func (r *PodReconciler) eventHandlers() handler.Funcs { return handler.Funcs{ - CreateFunc: func(ctx context.Context, createEvent event.CreateEvent, q workqueue.TypedRateLimitingInterface[reconcile.Request]) { - if !r.isRelevantPod(createEvent.Object) { - return - } - h := handler.EnqueueRequestForObject{} - h.Create(ctx, createEvent, q) - }, - UpdateFunc: func(ctx context.Context, updateEvent event.UpdateEvent, q workqueue.TypedRateLimitingInterface[reconcile.Request]) { + UpdateFunc: func(ctx context.Context, updateEvent event.UpdateEvent, _ workqueue.TypedRateLimitingInterface[ctrl.Request]) { if !r.isRelevantPod(updateEvent.ObjectNew) { return } if isCompletionEvent(updateEvent.ObjectOld, updateEvent.ObjectNew) { r.syncReservationIfNeeded(ctx, updateEvent.ObjectNew) } - h := handler.EnqueueRequestForObject{} - h.Update(ctx, updateEvent, q) }, - DeleteFunc: func(ctx context.Context, deleteEvent event.DeleteEvent, q workqueue.TypedRateLimitingInterface[reconcile.Request]) { + DeleteFunc: func(ctx context.Context, deleteEvent event.DeleteEvent, _ workqueue.TypedRateLimitingInterface[ctrl.Request]) { if !r.isRelevantPod(deleteEvent.Object) { return } r.syncReservationIfNeeded(ctx, deleteEvent.Object) - - h := handler.EnqueueRequestForObject{} - h.Delete(ctx, deleteEvent, q) - }, - GenericFunc: func(ctx context.Context, genericEvent event.GenericEvent, q workqueue.TypedRateLimitingInterface[reconcile.Request]) { - if !r.isRelevantPod(genericEvent.Object) { - return - } - h := handler.EnqueueRequestForObject{} - h.Generic(ctx, genericEvent, q) }, } } diff --git a/pkg/binder/controllers/pod_controller_test.go b/pkg/binder/controllers/pod_controller_test.go new file mode 100644 index 000000000..d888ce758 --- /dev/null +++ b/pkg/binder/controllers/pod_controller_test.go @@ -0,0 +1,56 @@ +// Copyright 2026 NVIDIA CORPORATION +// SPDX-License-Identifier: Apache-2.0 + +package controllers + +import ( + "context" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "go.uber.org/mock/gomock" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/util/workqueue" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/event" + + resourcereservationmock "github.com/kai-scheduler/KAI-scheduler/pkg/binder/binding/resourcereservation/mock" + "github.com/kai-scheduler/KAI-scheduler/pkg/common/constants" +) + +var _ = Describe("Pod Controller", func() { + It("syncs reservations for completed and deleted Pods without enqueueing reconciles", func() { + resourceReservation := resourcereservationmock.NewMockInterface(gomock.NewController(GinkgoT())) + reconciler := &PodReconciler{ + ResourceReservation: resourceReservation, + SchedulerName: "kai-scheduler", + } + queue := workqueue.NewTypedRateLimitingQueue(workqueue.DefaultTypedControllerRateLimiter[ctrl.Request]()) + DeferCleanup(queue.ShutDown) + + completedPod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "pod", + Namespace: "default", + Labels: map[string]string{ + constants.GPUGroup: "group", + }, + }, + Spec: corev1.PodSpec{SchedulerName: "kai-scheduler"}, + Status: corev1.PodStatus{Phase: corev1.PodSucceeded}, + } + pendingPod := completedPod.DeepCopy() + pendingPod.Status.Phase = corev1.PodPending + + resourceReservation.EXPECT().SyncForGpuGroup(gomock.Any(), "group").Return(nil).Times(2) + handlers := reconciler.eventHandlers() + handlers.UpdateFunc(context.TODO(), event.UpdateEvent{ + ObjectOld: pendingPod, + ObjectNew: completedPod, + }, queue) + handlers.DeleteFunc(context.TODO(), event.DeleteEvent{Object: completedPod}, queue) + + Expect(queue.Len()).To(Equal(0)) + }) +})