Repository: kubernetes-sigs/controller-runtime

Issue:
WithLowPriorityWhenUnchanged only sets priority for Add
I took a closer look at workqueueWithCustomAddFunc. It currently only overwrites the Add func. Would it also make sense to overwrite AddRateLimited, AddAfter and AddWithOpts?

For AddWithOpts ideally only set priority if not already set, but atm we can't differentiate between priority intentionally set to 0 and the zero value, because Priority in AddOpts is not a pointer (probably it should be).

I know that for our builtin event handler we only have to overwrite Add. But I assume if folks use WithLowPriorityWhenUnchanged they would probably expect that it works for all Add funcs. 

Tasks:
* [x] Consider to make AddOpts.Priority a pointer so we can differentiate between not set and priority 0 (the zero value)

Discussion:
@sbueringer: cc @alvaroaleman @alvaroaleman: Seems okay to me but I think its pretty uncommon to implement a full handler rather than using `handler.Funcs` or such@sbueringer: If I'm not missing anything even if you use handler.Funcs you would implement a func like this: `func(context.Context, event.TypedCreateEvent[object], workqueue.TypedRateLimitingInterface[request])`

The `workqueue.TypedRateLimitingInterface[request]` parameter then provides all Add* funcs, but only `Add`will set the low priority@sbueringer: Note to myself. When looking into this we might also want to make AddOpts.Priority a pointer so we can differentiate between not set and priority 0 (the zero value)@k8s-triage-robot: The Kubernetes project currently lacks enough contributors to adequately respond to all issues.

This bot triages un-triaged issues according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Mark this issue as fresh with `/remove-lifecycle stale`
- Close this issue with `/close`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/lifecycle stale

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/@sbueringer: /remove-lifecycle stale

Solution patch:
diff --git a/pkg/cache/cache.go b/pkg/cache/cache.go
index 02d77387..a7e49185 100644
--- a/pkg/cache/cache.go
+++ b/pkg/cache/cache.go
@@ -173,16 +173,14 @@ type Options struct {
 	// recommend the `Reconcile` function return `reconcile.Result{RequeueAfter: t}`,
 	// instead of `reconcile.Result{}`.
 	//
-	// SyncPeriod will trigger update events with the old object being equal to the new
-	// object, except when the cache was out of sync.
-	// If you filter update events like this:
-	// Controller.Watch(
-	//     &source.Kind{Type: v1.MyCustomKind},
-	//     &handler.EnqueueRequestForObject{},
-	//     predicate.Or(predicate.GenerationChangedPredicate{},
-	//         predicate.LabelChangedPredicate{}))
-	// then the SyncPeriod will not trigger a Reconcile call, because the update event
-	// will be ignored by the predicate.
+	// SyncPeriod will locally trigger an artificial Update event with the same
+	// object in both ObjectOld and ObjectNew for everything that is in the
+	// cache.
+	//
+	// Predicates or Handlers that expect ObjectOld and ObjectNew to be different
+	// (such as GenerationChangedPredicate) will filter out this event, preventing
+	// it from triggering a reconciliation.
+	// SyncPeriod does not sync between the local cache and the server.
 	SyncPeriod *time.Duration
 
 	// ReaderFailOnMissingInformer configures the cache to return a ErrResourceNotCached error when a user
diff --git a/pkg/handler/eventhandler.go b/pkg/handler/eventhandler.go
index 9a05a73d..88510d29 100644
--- a/pkg/handler/eventhandler.go
+++ b/pkg/handler/eventhandler.go
@@ -19,8 +19,10 @@ package handler
 import (
 	"context"
 	"reflect"
+	"time"
 
 	"k8s.io/client-go/util/workqueue"
+	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/priorityqueue"
 	"sigs.k8s.io/controller-runtime/pkg/event"
@@ -126,20 +128,14 @@ func (h TypedFuncs[object, request]) Create(ctx context.Context, e event.TypedCr
 			h.CreateFunc(ctx, e, q)
 			return
 		}
-		wq := workqueueWithCustomAddFunc[request]{
-			TypedRateLimitingInterface: q,
+
+		wq := workqueueWithDefaultPriority[request]{
 			// We already know that we have a priority queue, that event.Object implements
 			// client.Object and that its not nil
-			addFunc: func(item request, q workqueue.TypedRateLimitingInterface[request]) {
-				var priority int
-				if e.IsInInitialList {
-					priority = LowPriority
-				}
-				q.(priorityqueue.PriorityQueue[request]).AddWithOpts(
-					priorityqueue.AddOpts{Priority: &priority},
-					item,
-				)
-			},
+			PriorityQueue: q.(priorityqueue.PriorityQueue[request]),
+		}
+		if e.IsInInitialList {
+			wq.priority = ptr.To(LowPriority)
 		}
 		h.CreateFunc(ctx, e, wq)
 	}
@@ -160,20 +156,13 @@ func (h TypedFuncs[object, request]) Update(ctx context.Context, e event.TypedUp
 			return
 		}
 
-		wq := workqueueWithCustomAddFunc[request]{
-			TypedRateLimitingInterface: q,
+		wq := workqueueWithDefaultPriority[request]{
 			// We already know that we have a priority queue, that event.ObjectOld and ObjectNew implement
 			// client.Object and that they are  not nil
-			addFunc: func(item request, q workqueue.TypedRateLimitingInterface[request]) {
-				var priority int
-				if any(e.ObjectOld).(client.Object).GetResourceVersion() == any(e.ObjectNew).(client.Object).GetResourceVersion() {
-					priority = LowPriority
-				}
-				q.(priorityqueue.PriorityQueue[request]).AddWithOpts(
-					priorityqueue.AddOpts{Priority: &priority},
-					item,
-				)
-			},
+			PriorityQueue: q.(priorityqueue.PriorityQueue[request]),
+		}
+		if any(e.ObjectOld).(client.Object).GetResourceVersion() == any(e.ObjectNew).(client.Object).GetResourceVersion() {
+			wq.priority = ptr.To(LowPriority)
 		}
 		h.UpdateFunc(ctx, e, wq)
 	}
@@ -201,13 +190,28 @@ func WithLowPriorityWhenUnchanged[object client.Object, request comparable](u Ty
 	}
 }
 
-type workqueueWithCustomAddFunc[request comparable] struct {
-	workqueue.TypedRateLimitingInterface[request]
-	addFunc func(item request, q workqueue.TypedRateLimitingInterface[request])
+type workqueueWithDefaultPriority[request comparable] struct {
+	priorityqueue.PriorityQueue[request]
+	priority *int
+}
+
+func (w workqueueWithDefaultPriority[request]) Add(item request) {
+	w.PriorityQueue.AddWithOpts(priorityqueue.AddOpts{Priority: w.priority}, item)
 }
 
-func (w workqueueWithCustomAddFunc[request]) Add(item request) {
-	w.addFunc(item, w.TypedRateLimitingInterface)
+func (w workqueueWithDefaultPriority[request]) AddAfter(item request, after time.Duration) {
+	w.PriorityQueue.AddWithOpts(priorityqueue.AddOpts{Priority: w.priority, After: after}, item)
+}
+
+func (w workqueueWithDefaultPriority[request]) AddRateLimited(item request) {
+	w.PriorityQueue.AddWithOpts(priorityqueue.AddOpts{Priority: w.priority, RateLimited: true}, item)
+}
+
+func (w workqueueWithDefaultPriority[request]) AddWithOpts(o priorityqueue.AddOpts, items ...request) {
+	if o.Priority == nil {
+		o.Priority = w.priority
+	}
+	w.PriorityQueue.AddWithOpts(o, items...)
 }
 
 // addToQueueCreate adds the reconcile.Request to the priorityqueue in the handler
@@ -219,11 +223,11 @@ func addToQueueCreate[T client.Object, request comparable](q workqueue.TypedRate
 		return
 	}
 
-	var priority int
+	var priority *int
 	if evt.IsInInitialList {
-		priority = LowPriority
+		priority = ptr.To(LowPriority)
 	}
-	priorityQueue.AddWithOpts(priorityqueue.AddOpts{Priority: &priority}, item)
+	priorityQueue.AddWithOpts(priorityqueue.AddOpts{Priority: priority}, item)
 }
 
 // addToQueueUpdate adds the reconcile.Request to the priorityqueue in the handler
@@ -235,9 +239,9 @@ func addToQueueUpdate[T client.Object, request comparable](q workqueue.TypedRate
 		return
 	}
 
-	var priority int
+	var priority *int
 	if evt.ObjectOld.GetResourceVersion() == evt.ObjectNew.GetResourceVersion() {
-		priority = LowPriority
+		priority = ptr.To(LowPriority)
 	}
-	priorityQueue.AddWithOpts(priorityqueue.AddOpts{Priority: &priority}, item)
+	priorityQueue.AddWithOpts(priorityqueue.AddOpts{Priority: priority}, item)
 }



Test patch:
diff --git a/pkg/handler/eventhandler_test.go b/pkg/handler/eventhandler_test.go
index 3b0417ad..2a7453f7 100644
--- a/pkg/handler/eventhandler_test.go
+++ b/pkg/handler/eventhandler_test.go
@@ -18,6 +18,7 @@ package handler_test
 
 import (
 	"context"
+	"time"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -776,8 +777,11 @@ var _ = Describe("Eventhandler", func() {
 
 	Describe("WithLowPriorityWhenUnchanged", func() {
 		handlerPriorityTests := []struct {
-			name    string
-			handler func() handler.EventHandler
+			name             string
+			handler          func() handler.EventHandler
+			after            time.Duration
+			ratelimited      bool
+			overridePriority int
 		}{
 			{
 				name:    "WithLowPriorityWhenUnchanged wrapper",
@@ -837,6 +841,103 @@ var _ = Describe("Eventhandler", func() {
 					})
 				},
 			},
+			{
+				name: "WithLowPriorityWhenUnchanged - Add",
+				handler: func() handler.EventHandler {
+					return handler.WithLowPriorityWhenUnchanged(
+						handler.TypedFuncs[client.Object, reconcile.Request]{
+							CreateFunc: func(ctx context.Context, tce event.TypedCreateEvent[client.Object], wq workqueue.TypedRateLimitingInterface[reconcile.Request]) {
+								wq.Add(reconcile.Request{NamespacedName: types.NamespacedName{
+									Namespace: tce.Object.GetNamespace(),
+									Name:      tce.Object.GetName(),
+								}})
+							},
+							UpdateFunc: func(ctx context.Context, tue event.TypedUpdateEvent[client.Object], wq workqueue.TypedRateLimitingInterface[reconcile.Request]) {
+								wq.Add(reconcile.Request{NamespacedName: types.NamespacedName{
+									Namespace: tue.ObjectNew.GetNamespace(),
+									Name:      tue.ObjectNew.GetName(),
+								}})
+							},
+						})
+				},
+			},
+			{
+				name: "WithLowPriorityWhenUnchanged - AddAfter",
+				handler: func() handler.EventHandler {
+					return handler.WithLowPriorityWhenUnchanged(
+						handler.TypedFuncs[client.Object, reconcile.Request]{
+							CreateFunc: func(ctx context.Context, tce event.TypedCreateEvent[client.Object], wq workqueue.TypedRateLimitingInterface[reconcile.Request]) {
+								wq.AddAfter(reconcile.Request{NamespacedName: types.NamespacedName{
+									Namespace: tce.Object.GetNamespace(),
+									Name:      tce.Object.GetName(),
+								}}, time.Second)
+							},
+							UpdateFunc: func(ctx context.Context, tue event.TypedUpdateEvent[client.Object], wq workqueue.TypedRateLimitingInterface[reconcile.Request]) {
+								wq.AddAfter(reconcile.Request{NamespacedName: types.NamespacedName{
+									Namespace: tue.ObjectNew.GetNamespace(),
+									Name:      tue.ObjectNew.GetName(),
+								}}, time.Second)
+							},
+						})
+				},
+				after: time.Second,
+			},
+			{
+				name: "WithLowPriorityWhenUnchanged - AddRateLimited",
+				handler: func() handler.EventHandler {
+					return handler.WithLowPriorityWhenUnchanged(
+						handler.TypedFuncs[client.Object, reconcile.Request]{
+							CreateFunc: func(ctx context.Context, tce event.TypedCreateEvent[client.Object], wq workqueue.TypedRateLimitingInterface[reconcile.Request]) {
+								wq.AddRateLimited(reconcile.Request{NamespacedName: types.NamespacedName{
+									Namespace: tce.Object.GetNamespace(),
+									Name:      tce.Object.GetName(),
+								}})
+							},
+							UpdateFunc: func(ctx context.Context, tue event.TypedUpdateEvent[client.Object], wq workqueue.TypedRateLimitingInterface[reconcile.Request]) {
+								wq.AddRateLimited(reconcile.Request{NamespacedName: types.NamespacedName{
+									Namespace: tue.ObjectNew.GetNamespace(),
+									Name:      tue.ObjectNew.GetName(),
+								}})
+							},
+						})
+				},
+				ratelimited: true,
+			},
+			{
+				name: "WithLowPriorityWhenUnchanged - AddWithOpts priority is retained",
+				handler: func() handler.EventHandler {
+					return handler.WithLowPriorityWhenUnchanged(
+						handler.TypedFuncs[client.Object, reconcile.Request]{
+							CreateFunc: func(ctx context.Context, tce event.TypedCreateEvent[client.Object], wq workqueue.TypedRateLimitingInterface[reconcile.Request]) {
+								if pq, isPQ := wq.(priorityqueue.PriorityQueue[reconcile.Request]); isPQ {
+									pq.AddWithOpts(priorityqueue.AddOpts{Priority: ptr.To(100)}, reconcile.Request{NamespacedName: types.NamespacedName{
+										Namespace: tce.Object.GetNamespace(),
+										Name:      tce.Object.GetName(),
+									}})
+									return
+								}
+								wq.Add(reconcile.Request{NamespacedName: types.NamespacedName{
+									Namespace: tce.Object.GetNamespace(),
+									Name:      tce.Object.GetName(),
+								}})
+							},
+							UpdateFunc: func(ctx context.Context, tue event.TypedUpdateEvent[client.Object], wq workqueue.TypedRateLimitingInterface[reconcile.Request]) {
+								if pq, isPQ := wq.(priorityqueue.PriorityQueue[reconcile.Request]); isPQ {
+									pq.AddWithOpts(priorityqueue.AddOpts{Priority: ptr.To(100)}, reconcile.Request{NamespacedName: types.NamespacedName{
+										Namespace: tue.ObjectNew.GetNamespace(),
+										Name:      tue.ObjectNew.GetName(),
+									}})
+									return
+								}
+								wq.Add(reconcile.Request{NamespacedName: types.NamespacedName{
+									Namespace: tue.ObjectNew.GetNamespace(),
+									Name:      tue.ObjectNew.GetName(),
+								}})
+							},
+						})
+				},
+				overridePriority: 100,
+			},
 		}
 		for _, test := range handlerPriorityTests {
 			When("handler is "+test.name, func() {
@@ -862,7 +963,16 @@ var _ = Describe("Eventhandler", func() {
 						IsInInitialList: true,
 					}, wq)
 
-					Expect(actualOpts).To(Equal(priorityqueue.AddOpts{Priority: ptr.To(handler.LowPriority)}))
+					expected := handler.LowPriority
+					if test.overridePriority != 0 {
+						expected = test.overridePriority
+					}
+
+					Expect(actualOpts).To(Equal(priorityqueue.AddOpts{
+						Priority:    ptr.To(expected),
+						After:       test.after,
+						RateLimited: test.ratelimited,
+					}))
 					Expect(actualRequests).To(Equal([]reconcile.Request{{NamespacedName: types.NamespacedName{Name: "my-pod"}}}))
 				})
 
@@ -888,10 +998,12 @@ var _ = Describe("Eventhandler", func() {
 						IsInInitialList: false,
 					}, wq)
 
-					Expect(actualOpts).To(Or(
-						Equal(priorityqueue.AddOpts{}),
-						Equal(priorityqueue.AddOpts{Priority: ptr.To(0)}),
-					))
+					var expectedPriority *int
+					if test.overridePriority != 0 {
+						expectedPriority = &test.overridePriority
+					}
+
+					Expect(actualOpts).To(Equal(priorityqueue.AddOpts{After: test.after, RateLimited: test.ratelimited, Priority: expectedPriority}))
 					Expect(actualRequests).To(Equal([]reconcile.Request{{NamespacedName: types.NamespacedName{Name: "my-pod"}}}))
 				})
 
@@ -922,7 +1034,12 @@ var _ = Describe("Eventhandler", func() {
 						}},
 					}, wq)
 
-					Expect(actualOpts).To(Equal(priorityqueue.AddOpts{Priority: ptr.To(handler.LowPriority)}))
+					expectedPriority := handler.LowPriority
+					if test.overridePriority != 0 {
+						expectedPriority = test.overridePriority
+					}
+
+					Expect(actualOpts).To(Equal(priorityqueue.AddOpts{After: test.after, RateLimited: test.ratelimited, Priority: ptr.To(expectedPriority)}))
 					Expect(actualRequests).To(Equal([]reconcile.Request{{NamespacedName: types.NamespacedName{Name: "my-pod"}}}))
 				})
 
@@ -954,10 +1071,11 @@ var _ = Describe("Eventhandler", func() {
 						}},
 					}, wq)
 
-					Expect(actualOpts).To(Or(
-						Equal(priorityqueue.AddOpts{}),
-						Equal(priorityqueue.AddOpts{Priority: ptr.To(0)}),
-					))
+					var expectedPriority *int
+					if test.overridePriority != 0 {
+						expectedPriority = &test.overridePriority
+					}
+					Expect(actualOpts).To(Equal(priorityqueue.AddOpts{After: test.after, RateLimited: test.ratelimited, Priority: expectedPriority}))
 					Expect(actualRequests).To(Equal([]reconcile.Request{{NamespacedName: types.NamespacedName{Name: "my-pod"}}}))
 				})