:sparkling: POC of a priority queue

This change contains the POC of a priority workqueue that allows to
prioritize events over one another. It is opt-in and will by default
de-prioritize events originating from the initial listwatch and from
periodic resyncs.

Author: alvaroaleman

.golangci.yml

@@ -165,6 +165,14 @@ issues:
     - linters:
         - dupl
       path: _test\.go
+    - linters:
+        - revive
+      path: .*/internal/.*
+    - linters:
+        - unused
+      # Seems to incorrectly trigger on the two implementations that are only
+      # used through an interface and not directly..?
+      path: pkg/controllerworkqueue/metrics\.go
 
 run:
   go: "1.23"

examples/priorityqueue/main.go

@@ -0,0 +1,73 @@
+/*
+Copyright 2024 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	"sigs.k8s.io/controller-runtime/pkg/builder"
+	kubeconfig "sigs.k8s.io/controller-runtime/pkg/client/config"
+	"sigs.k8s.io/controller-runtime/pkg/config"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+	"sigs.k8s.io/controller-runtime/pkg/manager/signals"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+)
+
+func init() {
+}
+
+func main() {
+	if err := run(); err != nil {
+		fmt.Fprintf(os.Stderr, "%v\n", err)
+		os.Exit(1)
+	}
+}
+
+func run() error {
+	log.SetLogger(zap.New())
+
+	// Setup a Manager
+	mgr, err := manager.New(kubeconfig.GetConfigOrDie(), manager.Options{
+		Controller: config.Controller{UsePriorityQueue: true},
+	})
+	if err != nil {
+		return fmt.Errorf("failed to set up controller-manager: %w", err)
+	}
+
+	if err := builder.ControllerManagedBy(mgr).
+		For(&corev1.ConfigMap{}).
+		Complete(reconcile.Func(func(ctx context.Context, r reconcile.Request) (reconcile.Result, error) {
+			log.FromContext(ctx).Info("Reconciling")
+			time.Sleep(10 * time.Second)
+
+			return reconcile.Result{}, nil
+		})); err != nil {
+		return fmt.Errorf("failed to set up controller: %w", err)
+	}
+
+	if err := mgr.Start(signals.SetupSignalHandler()); err != nil {
+		return fmt.Errorf("failed to start manager: %w", err)
+	}
+
+	return nil
+}

pkg/builder/controller.go

@@ -163,7 +163,7 @@ func (blder *TypedBuilder[request]) Watches(
 ) *TypedBuilder[request] {
 	input := WatchesInput[request]{
 		obj:     object,
-		handler: eventHandler,
+		handler: handler.WithLowPriorityWhenUnchanged(eventHandler),
 	}
 	for _, opt := range opts {
 		opt.ApplyToWatches(&input)
@@ -317,7 +317,7 @@ func (blder *TypedBuilder[request]) doWatch() error {
 		}
 
 		var hdler handler.TypedEventHandler[client.Object, request]
-		reflect.ValueOf(&hdler).Elem().Set(reflect.ValueOf(&handler.EnqueueRequestForObject{}))
+		reflect.ValueOf(&hdler).Elem().Set(reflect.ValueOf(handler.WithLowPriorityWhenUnchanged(&handler.EnqueueRequestForObject{})))
 		allPredicates := append([]predicate.Predicate(nil), blder.globalPredicates...)
 		allPredicates = append(allPredicates, blder.forInput.predicates...)
 		src := source.TypedKind(blder.mgr.GetCache(), obj, hdler, allPredicates...)
@@ -341,11 +341,11 @@ func (blder *TypedBuilder[request]) doWatch() error {
 		}
 
 		var hdler handler.TypedEventHandler[client.Object, request]
-		reflect.ValueOf(&hdler).Elem().Set(reflect.ValueOf(handler.EnqueueRequestForOwner(
+		reflect.ValueOf(&hdler).Elem().Set(reflect.ValueOf(handler.WithLowPriorityWhenUnchanged(handler.EnqueueRequestForOwner(
 			blder.mgr.GetScheme(), blder.mgr.GetRESTMapper(),
 			blder.forInput.object,
 			opts...,
-		)))
+		))))
 		allPredicates := append([]predicate.Predicate(nil), blder.globalPredicates...)
 		allPredicates = append(allPredicates, own.predicates...)
 		src := source.TypedKind(blder.mgr.GetCache(), obj, hdler, allPredicates...)

pkg/config/controller.go

@@ -53,4 +53,8 @@ type Controller struct {
 	// NeedLeaderElection indicates whether the controller needs to use leader election.
 	// Defaults to true, which means the controller will use leader election.
 	NeedLeaderElection *bool
+
+	// UsePriorityQueue is experimental and configures if controllers that do not have a
+	// NewQueue() configured should default to the priority queue.
+	UsePriorityQueue bool
 }

pkg/controller/controller.go

@@ -25,6 +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/internal/controller"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
@@ -193,10 +194,16 @@ func NewTypedUnmanaged[request comparable](name string, mgr manager.Manager, opt
 	}
 
 	if options.NewQueue == nil {
-		options.NewQueue = func(controllerName string, rateLimiter workqueue.TypedRateLimiter[request]) workqueue.TypedRateLimitingInterface[request] {
-			return workqueue.NewTypedRateLimitingQueueWithConfig(rateLimiter, workqueue.TypedRateLimitingQueueConfig[request]{
-				Name: controllerName,
-			})
+		if mgr.GetControllerOptions().UsePriorityQueue {
+			options.NewQueue = func(controllerName string, rateLimiter workqueue.TypedRateLimiter[request]) workqueue.TypedRateLimitingInterface[request] {
+				return controllerworkqueue.New[request](controllerName)
+			}
+		} else {
+			options.NewQueue = func(controllerName string, rateLimiter workqueue.TypedRateLimiter[request]) workqueue.TypedRateLimitingInterface[request] {
+				return workqueue.NewTypedRateLimitingQueueWithConfig(rateLimiter, workqueue.TypedRateLimitingQueueConfig[request]{
+					Name: controllerName,
+				})
+			}
 		}
 	}
 

pkg/controllerworkqueue/metrics.go

@@ -0,0 +1,140 @@
+package controllerworkqueue
+
+import (
+	"sync"
+	"time"
+
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/utils/clock"
+)
+
+// This file is mostly a copy of unexported code from
+// https://github.com/kubernetes/kubernetes/blob/1d8828ce707ed9dd7a6a9756385419cce1d202ac/staging/src/k8s.io/client-go/util/workqueue/metrics.go
+
+type queueMetrics[T comparable] interface {
+	add(item T)
+	get(item T)
+	done(item T)
+	updateUnfinishedWork()
+}
+
+func newQueueMetrics[T comparable](mp workqueue.MetricsProvider, name string, clock clock.Clock) queueMetrics[T] {
+	if len(name) == 0 {
+		return noMetrics[T]{}
+	}
+	return &defaultQueueMetrics[T]{
+		clock:                   clock,
+		depth:                   mp.NewDepthMetric(name),
+		adds:                    mp.NewAddsMetric(name),
+		latency:                 mp.NewLatencyMetric(name),
+		workDuration:            mp.NewWorkDurationMetric(name),
+		unfinishedWorkSeconds:   mp.NewUnfinishedWorkSecondsMetric(name),
+		longestRunningProcessor: mp.NewLongestRunningProcessorSecondsMetric(name),
+		added:                   sets.Set[T]{},
+		addTimes:                map[T]time.Time{},
+		processingStartTimes:    map[T]time.Time{},
+	}
+}
+
+// defaultQueueMetrics expects the caller to lock before setting any metrics.
+type defaultQueueMetrics[T comparable] struct {
+	clock clock.Clock
+
+	// current depth of a workqueue
+	depth workqueue.GaugeMetric
+	// total number of adds handled by a workqueue
+	adds workqueue.CounterMetric
+	// how long an item stays in a workqueue
+	latency workqueue.HistogramMetric
+	// how long processing an item from a workqueue takes
+	workDuration workqueue.HistogramMetric
+
+	mapLock              sync.RWMutex
+	added                sets.Set[T]
+	addTimes             map[T]time.Time
+	processingStartTimes map[T]time.Time
+
+	// how long have current threads been working?
+	unfinishedWorkSeconds   workqueue.SettableGaugeMetric
+	longestRunningProcessor workqueue.SettableGaugeMetric
+}
+
+func (m *defaultQueueMetrics[T]) add(item T) {
+	if m == nil {
+		return
+	}
+
+	m.adds.Inc()
+
+	m.mapLock.Lock()
+	defer m.mapLock.Unlock()
+	if !m.added.Has(item) {
+		m.added.Insert(item)
+		m.depth.Inc()
+	}
+	if _, exists := m.addTimes[item]; !exists {
+		m.addTimes[item] = m.clock.Now()
+	}
+}
+
+func (m *defaultQueueMetrics[T]) get(item T) {
+	if m == nil {
+		return
+	}
+
+	m.mapLock.Lock()
+	defer m.mapLock.Unlock()
+
+	m.depth.Dec()
+	m.added.Delete(item)
+
+	m.processingStartTimes[item] = m.clock.Now()
+	if startTime, exists := m.addTimes[item]; exists {
+		m.latency.Observe(m.sinceInSeconds(startTime))
+		delete(m.addTimes, item)
+	}
+}
+
+func (m *defaultQueueMetrics[T]) done(item T) {
+	if m == nil {
+		return
+	}
+
+	m.mapLock.Lock()
+	defer m.mapLock.Unlock()
+	if startTime, exists := m.processingStartTimes[item]; exists {
+		m.workDuration.Observe(m.sinceInSeconds(startTime))
+		delete(m.processingStartTimes, item)
+	}
+}
+
+func (m *defaultQueueMetrics[T]) updateUnfinishedWork() {
+	m.mapLock.RLock()
+	defer m.mapLock.RUnlock()
+	// Note that a summary metric would be better for this, but prometheus
+	// doesn't seem to have non-hacky ways to reset the summary metrics.
+	var total float64
+	var oldest float64
+	for _, t := range m.processingStartTimes {
+		age := m.sinceInSeconds(t)
+		total += age
+		if age > oldest {
+			oldest = age
+		}
+	}
+	m.unfinishedWorkSeconds.Set(total)
+	m.longestRunningProcessor.Set(oldest)
+}
+
+// Gets the time since the specified start in seconds.
+func (m *defaultQueueMetrics[T]) sinceInSeconds(start time.Time) float64 {
+	return m.clock.Since(start).Seconds()
+}
+
+type noMetrics[T any] struct{}
+
+func (noMetrics[T]) add(item T)            {}
+func (noMetrics[T]) get(item T)            {}
+func (noMetrics[T]) done(item T)           {}
+func (noMetrics[T]) updateUnfinishedWork() {}