Move into package priorityqueue

Author: alvaroaleman

pkg/controller/controller.go

@@ -25,7 +25,7 @@ import (
 	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog/v2"
 
-	"sigs.k8s.io/controller-runtime/pkg/controllerworkqueue"
+	"sigs.k8s.io/controller-runtime/pkg/controller/priorityqueue"
 	"sigs.k8s.io/controller-runtime/pkg/internal/controller"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
@@ -196,7 +196,7 @@ func NewTypedUnmanaged[request comparable](name string, mgr manager.Manager, opt
 	if options.NewQueue == nil {
 		if mgr.GetControllerOptions().UsePriorityQueue {
 			options.NewQueue = func(controllerName string, rateLimiter workqueue.TypedRateLimiter[request]) workqueue.TypedRateLimitingInterface[request] {
-				return controllerworkqueue.New[request](controllerName)
+				return priorityqueue.New[request](controllerName)
 			}
 		} else {
 			options.NewQueue = func(controllerName string, rateLimiter workqueue.TypedRateLimiter[request]) workqueue.TypedRateLimitingInterface[request] {

pkg/controllerworkqueue/metrics.go renamed to pkg/controller/priorityqueue/metrics.go

@@ -1,4 +1,4 @@
-package controllerworkqueue
+package priorityqueue
 
 import (
 	"sync"

pkg/controllerworkqueue/metrics_test.go renamed to pkg/controller/priorityqueue/metrics_test.go

@@ -1,4 +1,4 @@
-package controllerworkqueue
+package priorityqueue
 
 import (
 	"sync"

pkg/controllerworkqueue/workqueue.go renamed to pkg/controller/priorityqueue/priorityqueue.go

@@ -1,4 +1,4 @@
-package controllerworkqueue
+package priorityqueue
 
 import (
 	"sync"
@@ -56,7 +56,7 @@ func New[T comparable](name string, o ...Opt[T]) PriorityQueue[T] {
 		opts.MetricProvider = metrics.WorkqueueMetricsProvider{}
 	}
 
-	cwq := &controllerworkqueue[T]{
+	cwq := &priorityqueue[T]{
 		items: map[T]*item[T]{},
 		queue: btree.NewG(32, less[T]),
 		// itemOrWaiterAdded indicates that an item or
@@ -77,7 +77,7 @@ func New[T comparable](name string, o ...Opt[T]) PriorityQueue[T] {
 	return wrapWithMetrics(cwq, name, opts.MetricProvider)
 }
 
-type controllerworkqueue[T comparable] struct {
+type priorityqueue[T comparable] struct {
 	// lock has to be acquired for any access to either items or queue
 	lock  sync.Mutex
 	items map[T]*item[T]
@@ -110,7 +110,7 @@ type controllerworkqueue[T comparable] struct {
 	tick func(time.Duration) <-chan time.Time
 }
 
-func (w *controllerworkqueue[T]) AddWithOpts(o AddOpts, items ...T) {
+func (w *priorityqueue[T]) AddWithOpts(o AddOpts, items ...T) {
 	w.lock.Lock()
 	defer w.lock.Unlock()
 
@@ -155,14 +155,14 @@ func (w *controllerworkqueue[T]) AddWithOpts(o AddOpts, items ...T) {
 	}
 }
 
-func (w *controllerworkqueue[T]) notifyItemOrWaiterAdded() {
+func (w *priorityqueue[T]) notifyItemOrWaiterAdded() {
 	select {
 	case w.itemOrWaiterAdded <- struct{}{}:
 	default:
 	}
 }
 
-func (w *controllerworkqueue[T]) spin() {
+func (w *priorityqueue[T]) spin() {
 	blockForever := make(chan time.Time)
 	var nextReady <-chan time.Time
 	nextReady = blockForever
@@ -216,19 +216,19 @@ func (w *controllerworkqueue[T]) spin() {
 	}
 }
 
-func (w *controllerworkqueue[T]) Add(item T) {
+func (w *priorityqueue[T]) Add(item T) {
 	w.AddWithOpts(AddOpts{}, item)
 }
 
-func (w *controllerworkqueue[T]) AddAfter(item T, after time.Duration) {
+func (w *priorityqueue[T]) AddAfter(item T, after time.Duration) {
 	w.AddWithOpts(AddOpts{After: after}, item)
 }
 
-func (w *controllerworkqueue[T]) AddRateLimited(item T) {
+func (w *priorityqueue[T]) AddRateLimited(item T) {
 	w.AddWithOpts(AddOpts{RateLimited: true}, item)
 }
 
-func (w *controllerworkqueue[T]) GetWithPriority() (_ T, priority int, shutdown bool) {
+func (w *priorityqueue[T]) GetWithPriority() (_ T, priority int, shutdown bool) {
 	w.waiters.Add(1)
 
 	w.notifyItemOrWaiterAdded()
@@ -237,40 +237,40 @@ func (w *controllerworkqueue[T]) GetWithPriority() (_ T, priority int, shutdown
 	return item.key, item.priority, w.shutdown.Load()
 }
 
-func (w *controllerworkqueue[T]) Get() (item T, shutdown bool) {
+func (w *priorityqueue[T]) Get() (item T, shutdown bool) {
 	key, _, shutdown := w.GetWithPriority()
 	return key, shutdown
 }
 
-func (w *controllerworkqueue[T]) Forget(item T) {
+func (w *priorityqueue[T]) Forget(item T) {
 	w.rateLimiter.Forget(item)
 }
 
-func (w *controllerworkqueue[T]) NumRequeues(item T) int {
+func (w *priorityqueue[T]) NumRequeues(item T) int {
 	return w.rateLimiter.NumRequeues(item)
 }
 
-func (w *controllerworkqueue[T]) ShuttingDown() bool {
+func (w *priorityqueue[T]) ShuttingDown() bool {
 	return w.shutdown.Load()
 }
 
-func (w *controllerworkqueue[T]) Done(item T) {
+func (w *priorityqueue[T]) Done(item T) {
 	w.lockedLock.Lock()
 	defer w.lockedLock.Unlock()
 	w.locked.Delete(item)
 	w.notifyItemOrWaiterAdded()
 }
 
-func (w *controllerworkqueue[T]) ShutDown() {
+func (w *priorityqueue[T]) ShutDown() {
 	w.shutdown.Store(true)
 	close(w.done)
 }
 
-func (w *controllerworkqueue[T]) ShutDownWithDrain() {
+func (w *priorityqueue[T]) ShutDownWithDrain() {
 	w.ShutDown()
 }
 
-func (w *controllerworkqueue[T]) Len() int {
+func (w *priorityqueue[T]) Len() int {
 	w.lock.Lock()
 	defer w.lock.Unlock()
 
@@ -306,10 +306,10 @@ type item[T comparable] struct {
 	readyAt         *time.Time
 }
 
-func wrapWithMetrics[T comparable](q *controllerworkqueue[T], name string, provider workqueue.MetricsProvider) PriorityQueue[T] {
+func wrapWithMetrics[T comparable](q *priorityqueue[T], name string, provider workqueue.MetricsProvider) PriorityQueue[T] {
 	mwq := &metricWrappedQueue[T]{
-		controllerworkqueue: q,
-		metrics:             newQueueMetrics[T](provider, name, clock.RealClock{}),
+		priorityqueue: q,
+		metrics:       newQueueMetrics[T](provider, name, clock.RealClock{}),
 	}
 
 	go mwq.updateUnfinishedWorkLoop()
@@ -318,19 +318,19 @@ func wrapWithMetrics[T comparable](q *controllerworkqueue[T], name string, provi
 }
 
 type metricWrappedQueue[T comparable] struct {
-	*controllerworkqueue[T]
+	*priorityqueue[T]
 	metrics queueMetrics[T]
 }
 
 func (m *metricWrappedQueue[T]) AddWithOpts(o AddOpts, items ...T) {
 	for _, item := range items {
 		m.metrics.add(item)
 	}
-	m.controllerworkqueue.AddWithOpts(o, items...)
+	m.priorityqueue.AddWithOpts(o, items...)
 }
 
 func (m *metricWrappedQueue[T]) GetWithPriority() (T, int, bool) {
-	item, priority, shutdown := m.controllerworkqueue.GetWithPriority()
+	item, priority, shutdown := m.priorityqueue.GetWithPriority()
 	m.metrics.get(item)
 	return item, priority, shutdown
 }
@@ -342,14 +342,14 @@ func (m *metricWrappedQueue[T]) Get() (T, bool) {
 
 func (m *metricWrappedQueue[T]) Done(item T) {
 	m.metrics.done(item)
-	m.controllerworkqueue.Done(item)
+	m.priorityqueue.Done(item)
 }
 
 func (m *metricWrappedQueue[T]) updateUnfinishedWorkLoop() {
 	t := time.NewTicker(time.Millisecond)
 	defer t.Stop()
 	for range t.C {
-		if m.controllerworkqueue.ShuttingDown() {
+		if m.priorityqueue.ShuttingDown() {
 			return
 		}
 		m.metrics.updateUnfinishedWork()

pkg/controllerworkqueue/workqueue_suite_test.go renamed to pkg/controller/priorityqueue/priorityqueue_suite_test.go

@@ -1,4 +1,4 @@
-package controllerworkqueue
+package priorityqueue
 
 import (
 	"testing"

pkg/controllerworkqueue/workqueue_test.go renamed to pkg/controller/priorityqueue/priorityqueue_test.go

@@ -1,4 +1,4 @@
-package controllerworkqueue
+package priorityqueue
 
 import (
 	"sync"

pkg/handler/eventhandler.go

@@ -22,7 +22,7 @@ import (
 
 	"k8s.io/client-go/util/workqueue"
 	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/controllerworkqueue"
+	"sigs.k8s.io/controller-runtime/pkg/controller/priorityqueue"
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
@@ -140,7 +140,7 @@ func (h TypedFuncs[object, request]) Generic(ctx context.Context, e event.TypedG
 const LowPriority = -100
 
 // WithLowPriorityWhenUnchanged reduces the priority of events stemming from the initial listwatch or from a resync if
-// and only if a controllerworkqueue.PriorityQueue is used. If not, it does nothing.
+// and only if a priorityqueue.PriorityQueue is used. If not, it does nothing.
 func WithLowPriorityWhenUnchanged[object client.Object, request comparable](u TypedEventHandler[object, request]) TypedEventHandler[object, request] {
 	return TypedFuncs[object, request]{
 		CreateFunc: func(ctx context.Context, tce event.TypedCreateEvent[object], trli workqueue.TypedRateLimitingInterface[request]) {
@@ -149,7 +149,7 @@ 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.(controllerworkqueue.PriorityQueue[request])
+					priorityQueue, isPriorityQueue := q.(priorityqueue.PriorityQueue[request])
 					if !isPriorityQueue {
 						q.Add(item)
 						return
@@ -158,15 +158,15 @@ func WithLowPriorityWhenUnchanged[object client.Object, request comparable](u Ty
 					if isObjectUnchanged(tce) {
 						priority = LowPriority
 					}
-					priorityQueue.AddWithOpts(controllerworkqueue.AddOpts{Priority: priority}, item)
+					priorityQueue.AddWithOpts(priorityqueue.AddOpts{Priority: priority}, 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.(controllerworkqueue.PriorityQueue[request])
+					priorityQueue, isPriorityQueue := q.(priorityqueue.PriorityQueue[request])
 					if !isPriorityQueue {
 						q.Add(item)
 						return
@@ -175,7 +175,7 @@ func WithLowPriorityWhenUnchanged[object client.Object, request comparable](u Ty
 					if tue.ObjectOld.GetResourceVersion() == tue.ObjectNew.GetResourceVersion() {
 						priority = LowPriority
 					}
-					priorityQueue.AddWithOpts(controllerworkqueue.AddOpts{Priority: priority}, item)
+					priorityQueue.AddWithOpts(priorityqueue.AddOpts{Priority: priority}, item)
 				},
 			})
 		},