implement priority queue

Signed-off-by: Troy Connor <[email protected]>

Author: troy0820

pkg/handler/enqueue_mapped.go

@@ -19,10 +19,8 @@ package handler
 import (
 	"context"
 
-	"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"
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
@@ -48,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 WithLowPriorityWhenUnchanged(TypedEnqueueRequestsFromMapFunc(fn))
+	return TypedEnqueueRequestsFromMapFunc(fn)
 }
 
 // TypedEnqueueRequestsFromMapFunc enqueues Requests by running a transformation function that outputs a collection
@@ -82,23 +80,8 @@ func (e *enqueueRequestsFromMapFunc[object, request]) Create(
 	evt event.TypedCreateEvent[object],
 	q workqueue.TypedRateLimitingInterface[request],
 ) {
-	wq, ok := q.(workqueue.TypedRateLimitingInterface[reconcile.Request])
-	if ok {
-		evnt, ok := any(e).(event.TypedCreateEvent[client.Object])
-		if ok {
-			priorityQueue, isPriorityQueue := wq.(priorityqueue.PriorityQueue[reconcile.Request])
-			if isPriorityQueue {
-				item := reconcile.Request{NamespacedName: types.NamespacedName{
-					Name:      evnt.Object.GetName(),
-					Namespace: evnt.Object.GetNamespace(),
-				}}
-				addToQueueCreate(priorityQueue, evnt, item)
-				return
-			}
-			reqs := map[request]empty{}
-			e.mapAndEnqueue(ctx, q, evt.Object, reqs)
-		}
-	}
+	reqs := map[request]empty{}
+	e.mapAndEnqueue(ctx, q, evt.Object, reqs)
 }
 
 // Update implements EventHandler.
@@ -107,40 +90,6 @@ func (e *enqueueRequestsFromMapFunc[object, request]) Update(
 	evt event.TypedUpdateEvent[object],
 	q workqueue.TypedRateLimitingInterface[request],
 ) {
-	wq, ok := q.(workqueue.TypedRateLimitingInterface[reconcile.Request])
-	if ok {
-		evnt, ok := any(e).(event.TypedUpdateEvent[client.Object])
-		if ok {
-			switch {
-			case !isNil(evt.ObjectNew):
-				priorityQueue, isPriorityQueue := wq.(priorityqueue.PriorityQueue[reconcile.Request])
-				if isPriorityQueue {
-					item := reconcile.Request{NamespacedName: types.NamespacedName{
-						Name:      evnt.ObjectNew.GetName(),
-						Namespace: evnt.ObjectNew.GetNamespace(),
-					}}
-					addToQueueUpdate(priorityQueue, evnt, item)
-					return
-				}
-				reqs := map[request]empty{}
-				e.mapAndEnqueue(ctx, q, evt.ObjectNew, reqs)
-			case !isNil(evt.ObjectOld):
-				priorityQueue, isPriorityQueue := wq.(priorityqueue.PriorityQueue[reconcile.Request])
-				if isPriorityQueue {
-					item := reconcile.Request{NamespacedName: types.NamespacedName{
-						Name:      evnt.ObjectOld.GetName(),
-						Namespace: evnt.ObjectOld.GetNamespace(),
-					}}
-					addToQueueUpdate(priorityQueue, evnt, item)
-					return
-				}
-				reqs := map[request]empty{}
-				e.mapAndEnqueue(ctx, q, evt.ObjectOld, reqs)
-			default:
-				enqueueLog.Error(nil, "UpdateEvent received with no metadata", "event", evt)
-			}
-		}
-	}
 	reqs := map[request]empty{}
 	e.mapAndEnqueue(ctx, q, evt.ObjectOld, reqs)
 	e.mapAndEnqueue(ctx, q, evt.ObjectNew, reqs)