implement priority queue for handlers

Signed-off-by: Troy Connor <troy0820@users.noreply.github.com>

Author: troy0820

pkg/handler/enqueue.go

@@ -52,25 +52,32 @@ func (e *TypedEnqueueRequestForObject[T]) Create(ctx context.Context, evt event.
 enqueueLog.Error(nil, "CreateEvent received with no metadata", "event", evt)
 return
 }
-q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
+
+item := reconcile.Request{NamespacedName: types.NamespacedName{
 Name: evt.Object.GetName(),
 Namespace: evt.Object.GetNamespace(),
-}})
+}}
+
+addToQueueCreate(q, evt, item)
 }
 
 // Update implements EventHandler.
 func (e *TypedEnqueueRequestForObject[T]) Update(ctx context.Context, evt event.TypedUpdateEvent[T], q workqueue.TypedRateLimitingInterface[reconcile.Request]) {
 switch {
 case !isNil(evt.ObjectNew):
-q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
+item := reconcile.Request{NamespacedName: types.NamespacedName{
 Name: evt.ObjectNew.GetName(),
 Namespace: evt.ObjectNew.GetNamespace(),
-}})
+}}
+
+addToQueueUpdate(q, evt, item)
 case !isNil(evt.ObjectOld):
-q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
+item := reconcile.Request{NamespacedName: types.NamespacedName{
 Name: evt.ObjectOld.GetName(),
 Namespace: evt.ObjectOld.GetNamespace(),
-}})
+}}
+
+addToQueueUpdate(q, evt, item)
 default:
 enqueueLog.Error(nil, "UpdateEvent received with no metadata", "event", evt)
 }

pkg/handler/enqueue_mapped.go

@@ -46,7 +46,7 @@ type TypedMapFunc[object any, request comparable] func(context.Context, object)
 // For UpdateEvents which contain both a new and old object, the transformation function is run on both
 // objects and both sets of Requests are enqueue.
 func EnqueueRequestsFromMapFunc(fn MapFunc) EventHandler {
-return TypedEnqueueRequestsFromMapFunc(fn)
+return WithLowPriorityWhenUnchanged(TypedEnqueueRequestsFromMapFunc(fn))
 }
 
 // TypedEnqueueRequestsFromMapFunc enqueues Requests by running a transformation function that outputs a collection

pkg/handler/enqueue_owner.go

@@ -48,7 +48,7 @@ type OwnerOption func(e enqueueRequestForOwnerInterface)
 //
 // - a handler.enqueueRequestForOwner EventHandler with an OwnerType of ReplicaSet and OnlyControllerOwner set to true.
 func EnqueueRequestForOwner(scheme *runtime.Scheme, mapper meta.RESTMapper, ownerType client.Object, opts ...OwnerOption) EventHandler {
-return TypedEnqueueRequestForOwner[client.Object](scheme, mapper, ownerType, opts...)
+return WithLowPriorityWhenUnchanged(TypedEnqueueRequestForOwner[client.Object](scheme, mapper, ownerType, opts...))
 }
 
 // TypedEnqueueRequestForOwner enqueues Requests for the Owners of an object. E.g. the object that created

pkg/handler/eventhandler.go

@@ -18,8 +18,10 @@ package handler
 
 import (
 "context"
+"reflect"
 "time"
 
+"k8s.io/apimachinery/pkg/types"
 "k8s.io/client-go/util/workqueue"
 "sigs.k8s.io/controller-runtime/pkg/client"
 "sigs.k8s.io/controller-runtime/pkg/controller/priorityqueue"
@@ -65,7 +67,7 @@ type EventHandler = TypedEventHandler[client.Object, reconcile.Request]
 //
 // Unless you are implementing your own TypedEventHandler, you can ignore the functions on the TypedEventHandler interface.
 // Most users shouldn't need to implement their own TypedEventHandler.
-//
+
 // TypedEventHandler is experimental and subject to future change.
 type TypedEventHandler[object any, request comparable] interface {
 // Create is called in response to a create event - e.g. Pod Creation.
@@ -111,7 +113,19 @@ type TypedFuncs[object any, request comparable] struct {
 // Create implements EventHandler.
 func (h TypedFuncs[object, request]) Create(ctx context.Context, e event.TypedCreateEvent[object], q workqueue.TypedRateLimitingInterface[request]) {
 if h.CreateFunc != nil {
-h.CreateFunc(ctx, e, q)
+if reflect.TypeFor[request]() != reflect.TypeOf(reconcile.Request{}) || !reflect.TypeFor[object]().Implements(reflect.TypeFor[client.Object]()) {
+h.CreateFunc(ctx, e, q)
+}
+
+wq, ok := q.(workqueue.TypedRateLimitingInterface[reconcile.Request])
+if ok {
+evt := any(e).(event.TypedCreateEvent[client.Object])
+item := reconcile.Request{NamespacedName: types.NamespacedName{
+Name: evt.Object.GetName(),
+Namespace: evt.Object.GetNamespace(),
+}}
+addToQueueCreate(wq, evt, item)
+}
 }
 }
 
@@ -125,7 +139,30 @@ func (h TypedFuncs[object, request]) Delete(ctx context.Context, e event.TypedDe
 // Update implements EventHandler.
 func (h TypedFuncs[object, request]) Update(ctx context.Context, e event.TypedUpdateEvent[object], q workqueue.TypedRateLimitingInterface[request]) {
 if h.UpdateFunc != nil {
-h.UpdateFunc(ctx, e, q)
+if reflect.TypeFor[request]() != reflect.TypeOf(reconcile.Request{}) || !reflect.TypeFor[object]().Implements(reflect.TypeFor[client.Object]()) {
+h.UpdateFunc(ctx, e, q)
+}
+
+wq, ok := q.(workqueue.TypedRateLimitingInterface[reconcile.Request])
+if ok {
+evt := any(e).(event.TypedUpdateEvent[client.Object])
+switch {
+case !isNil(evt.ObjectNew):
+item := reconcile.Request{NamespacedName: types.NamespacedName{
+Name: evt.ObjectNew.GetName(),
+Namespace: evt.ObjectNew.GetNamespace(),
+}}
+addToQueueUpdate(wq, evt, item)
+case !isNil(evt.ObjectOld):
+item := reconcile.Request{NamespacedName: types.NamespacedName{
+Name: evt.ObjectOld.GetName(),
+Namespace: evt.ObjectOld.GetNamespace(),
+}}
+addToQueueUpdate(wq, evt, item)
+default:
+enqueueLog.Error(nil, "UpdateEvent received with no metadata", "event", evt)
+}
+}
 }
 }
 
@@ -149,33 +186,15 @@ func WithLowPriorityWhenUnchanged[object client.Object, request comparable](u Ty
 u.Create(ctx, tce, workqueueWithCustomAddFunc[request]{
 TypedRateLimitingInterface: trli,
 addFunc: func(item request, q workqueue.TypedRateLimitingInterface[request]) {
-priorityQueue, isPriorityQueue := q.(priorityqueue.PriorityQueue[request])
-if !isPriorityQueue {
-q.Add(item)
-return
-}
-var priority int
-if isObjectUnchanged(tce) {
-priority = LowPriority
-}
-priorityQueue.AddWithOpts(priorityqueue.AddOpts{Priority: priority}, item)
+addToQueueCreate(q, tce, item)
 },
 })
 },
 UpdateFunc: func(ctx context.Context, tue event.TypedUpdateEvent[object], trli workqueue.TypedRateLimitingInterface[request]) {
 u.Update(ctx, tue, workqueueWithCustomAddFunc[request]{
 TypedRateLimitingInterface: trli,
 addFunc: func(item request, q workqueue.TypedRateLimitingInterface[request]) {
-priorityQueue, isPriorityQueue := q.(priorityqueue.PriorityQueue[request])
-if !isPriorityQueue {
-q.Add(item)
-return
-}
-var priority int
-if tue.ObjectOld.GetResourceVersion() == tue.ObjectNew.GetResourceVersion() {
-priority = LowPriority
-}
-priorityQueue.AddWithOpts(priorityqueue.AddOpts{Priority: priority}, item)
+addToQueueUpdate(q, tue, item)
 },
 })
 },
@@ -199,3 +218,35 @@ func (w workqueueWithCustomAddFunc[request]) Add(item request) {
 func isObjectUnchanged[object client.Object](e event.TypedCreateEvent[object]) bool {
 return e.Object.GetCreationTimestamp().Time.Before(time.Now().Add(-time.Minute))
 }
+
+// addToQueueCreate adds the reconcile.Request to the priorityqueue in the handler
+// for Create requests if and only if the workqueue being used is of type priorityqueue.PriorityQueue[reconcile.Request]
+func addToQueueCreate[T client.Object, request comparable](q workqueue.TypedRateLimitingInterface[request], evt event.TypedCreateEvent[T], item request) {
+priorityQueue, isPriorityQueue := q.(priorityqueue.PriorityQueue[request])
+if !isPriorityQueue {
+q.Add(item)
+return
+}
+
+var priority int
+if isObjectUnchanged(evt) {
+priority = LowPriority
+}
+priorityQueue.AddWithOpts(priorityqueue.AddOpts{Priority: priority}, item)
+}
+
+// addToQueueUpdate adds the reconcile.Request to the priorityqueue in the handler
+// for Update requests if and only if the workqueue being used is of type priorityqueue.PriorityQueue[reconcile.Request]
+func addToQueueUpdate[T client.Object, request comparable](q workqueue.TypedRateLimitingInterface[request], evt event.TypedUpdateEvent[T], item request) {
+priorityQueue, isPriorityQueue := q.(priorityqueue.PriorityQueue[request])
+if !isPriorityQueue {
+q.Add(item)
+return
+}
+
+var priority int
+if evt.ObjectOld.GetResourceVersion() == evt.ObjectNew.GetResourceVersion() {
+priority = LowPriority
+}
+priorityQueue.AddWithOpts(priorityqueue.AddOpts{Priority: priority}, item)
+}

pkg/handler/eventhandler_test.go

@@ -659,7 +659,7 @@ var _ = Describe("Eventhandler", func() {
 })
 
 Describe("Funcs", func() {
-failingFuncs := handler.Funcs{
+failingFuncs := handler.TypedFuncs[client.Object, reconcile.Request]{
 CreateFunc: func(context.Context, event.CreateEvent, workqueue.TypedRateLimitingInterface[reconcile.Request]) {
 defer GinkgoRecover()
 Fail("Did not expect CreateEvent to be called.")
@@ -797,6 +797,27 @@ var _ = Describe("Eventhandler", func() {
 Expect(actualRequests).To(Equal([]reconcile.Request{{NamespacedName: types.NamespacedName{Name: "my-pod"}}}))
 })
 
+It("should lower the priority of a create request for an object that was created more than one minute in the past without the WithLowPriorityWrapper", func() {
+actualOpts := priorityqueue.AddOpts{}
+var actualRequests []reconcile.Request
+wq := &fakePriorityQueue{
+addWithOpts: func(o priorityqueue.AddOpts, items ...reconcile.Request) {
+actualOpts = o
+actualRequests = items
+},
+}
+
+h := &handler.EnqueueRequestForObject{}
+h.Create(ctx, event.CreateEvent{
+Object: &corev1.Pod{ObjectMeta: metav1.ObjectMeta{
+Name: "my-pod",
+}},
+}, wq)
+
+Expect(actualOpts).To(Equal(priorityqueue.AddOpts{Priority: handler.LowPriority}))
+Expect(actualRequests).To(Equal([]reconcile.Request{{NamespacedName: types.NamespacedName{Name: "my-pod"}}}))
+})
+
 It("should not lower the priority of a create request for an object that was created less than one minute in the past", func() {
 actualOpts := priorityqueue.AddOpts{}
 var actualRequests []reconcile.Request
@@ -819,6 +840,28 @@ var _ = Describe("Eventhandler", func() {
 Expect(actualRequests).To(Equal([]reconcile.Request{{NamespacedName: types.NamespacedName{Name: "my-pod"}}}))
 })
 
+It("should not lower the priority of a create request for an object that was created less than one minute in the past without the WithLowPriority wrapperr", func() {
+actualOpts := priorityqueue.AddOpts{}
+var actualRequests []reconcile.Request
+wq := &fakePriorityQueue{
+addWithOpts: func(o priorityqueue.AddOpts, items ...reconcile.Request) {
+actualOpts = o
+actualRequests = items
+},
+}
+
+h := &handler.EnqueueRequestForObject{}
+h.Create(ctx, event.CreateEvent{
+Object: &corev1.Pod{ObjectMeta: metav1.ObjectMeta{
+Name: "my-pod",
+CreationTimestamp: metav1.Now(),
+}},
+}, wq)
+
+Expect(actualOpts).To(Equal(priorityqueue.AddOpts{}))
+Expect(actualRequests).To(Equal([]reconcile.Request{{NamespacedName: types.NamespacedName{Name: "my-pod"}}}))
+})
+
 It("should lower the priority of an update request with unchanged RV", func() {
 actualOpts := priorityqueue.AddOpts{}
 var actualRequests []reconcile.Request
@@ -843,6 +886,30 @@ var _ = Describe("Eventhandler", func() {
 Expect(actualRequests).To(Equal([]reconcile.Request{{NamespacedName: types.NamespacedName{Name: "my-pod"}}}))
 })
 
+It("should lower the priority of an update request with unchanged RV without the WithLowPriority wrapper", func() {
+actualOpts := priorityqueue.AddOpts{}
+var actualRequests []reconcile.Request
+wq := &fakePriorityQueue{
+addWithOpts: func(o priorityqueue.AddOpts, items ...reconcile.Request) {
+actualOpts = o
+actualRequests = items
+},
+}
+
+h := &handler.EnqueueRequestForObject{}
+h.Update(ctx, event.UpdateEvent{
+ObjectOld: &corev1.Pod{ObjectMeta: metav1.ObjectMeta{
+Name: "my-pod",
+}},
+ObjectNew: &corev1.Pod{ObjectMeta: metav1.ObjectMeta{
+Name: "my-pod",
+}},
+}, wq)
+
+Expect(actualOpts).To(Equal(priorityqueue.AddOpts{Priority: handler.LowPriority}))
+Expect(actualRequests).To(Equal([]reconcile.Request{{NamespacedName: types.NamespacedName{Name: "my-pod"}}}))
+})
+
 It("should not lower the priority of an update request with changed RV", func() {
 actualOpts := priorityqueue.AddOpts{}
 var actualRequests []reconcile.Request
@@ -868,6 +935,31 @@ var _ = Describe("Eventhandler", func() {
 Expect(actualRequests).To(Equal([]reconcile.Request{{NamespacedName: types.NamespacedName{Name: "my-pod"}}}))
 })
 
+It("should not lower the priority of an update request with changed RV without the WithLowPriority wrapper", func() {
+actualOpts := priorityqueue.AddOpts{}
+var actualRequests []reconcile.Request
+wq := &fakePriorityQueue{
+addWithOpts: func(o priorityqueue.AddOpts, items ...reconcile.Request) {
+actualOpts = o
+actualRequests = items
+},
+}
+
+h := &handler.EnqueueRequestForObject{}
+h.Update(ctx, event.UpdateEvent{
+ObjectOld: &corev1.Pod{ObjectMeta: metav1.ObjectMeta{
+Name: "my-pod",
+}},
+ObjectNew: &corev1.Pod{ObjectMeta: metav1.ObjectMeta{
+Name: "my-pod",
+ResourceVersion: "1",
+}},
+}, wq)
+
+Expect(actualOpts).To(Equal(priorityqueue.AddOpts{}))
+Expect(actualRequests).To(Equal([]reconcile.Request{{NamespacedName: types.NamespacedName{Name: "my-pod"}}}))
+})
+
 It("should have no effect on create if the workqueue is not a priorityqueue", func() {
 h := handler.WithLowPriorityWhenUnchanged(&handler.EnqueueRequestForObject{})
 h.Create(ctx, event.CreateEvent{
@@ -881,6 +973,19 @@ var _ = Describe("Eventhandler", func() {
 Expect(item).To(Equal(reconcile.Request{NamespacedName: types.NamespacedName{Name: "my-pod"}}))
 })
 
+It("should have no effect on create if the workqueue is not a priorityqueue without the WithLowPriority wrapper", func() {
+h := &handler.EnqueueRequestForObject{}
+h.Create(ctx, event.CreateEvent{
+Object: &corev1.Pod{ObjectMeta: metav1.ObjectMeta{
+Name: "my-pod",
+}},
+}, q)
+
+Expect(q.Len()).To(Equal(1))
+item, _ := q.Get()
+Expect(item).To(Equal(reconcile.Request{NamespacedName: types.NamespacedName{Name: "my-pod"}}))
+})
+
 It("should have no effect on Update if the workqueue is not a priorityqueue", func() {
 h := handler.WithLowPriorityWhenUnchanged(&handler.EnqueueRequestForObject{})
 h.Update(ctx, event.UpdateEvent{
@@ -896,8 +1001,23 @@ var _ = Describe("Eventhandler", func() {
 item, _ := q.Get()
 Expect(item).To(Equal(reconcile.Request{NamespacedName: types.NamespacedName{Name: "my-pod"}}))
 })
-})
 
+It("should have no effect on Update if the workqueue is not a priorityqueue without the WithLowPriority wrapper", func() {
+h := &handler.EnqueueRequestForObject{}
+h.Update(ctx, event.UpdateEvent{
+ObjectOld: &corev1.Pod{ObjectMeta: metav1.ObjectMeta{
+Name: "my-pod",
+}},
+ObjectNew: &corev1.Pod{ObjectMeta: metav1.ObjectMeta{
+Name: "my-pod",
+}},
+}, q)
+
+Expect(q.Len()).To(Equal(1))
+item, _ := q.Get()
+Expect(item).To(Equal(reconcile.Request{NamespacedName: types.NamespacedName{Name: "my-pod"}}))
+})
+})
 })
 
 type fakePriorityQueue struct {