NLB-6294 The synchronizer uses a typed rate-limited workqueue

The latest versions of the kubernetes libraries recommend using a typed workqueue and this reduces a bit of boilerplate and error handling, because we no longer have to cast the workitems returned by the queue into the desired types.

Author: arussellf5

cmd/nginx-loadbalancer-kubernetes/main.go

@@ -133,13 +133,12 @@ func buildKubernetesClient() (*kubernetes.Clientset, error) {
 	return client, nil
 }
 
-// TODO: NLB-6294 change to use new typed workqueues
-//
-//nolint:staticcheck //ignore deprecation warnings
-func buildWorkQueue(settings configuration.WorkQueueSettings) workqueue.RateLimitingInterface {
+func buildWorkQueue(settings configuration.WorkQueueSettings,
+) workqueue.TypedRateLimitingInterface[synchronization.ServiceKey] {
 	slog.Debug("Watcher::buildSynchronizerWorkQueue")
 
-	//nolint:staticcheck //ignore deprecation warnings
-	rateLimiter := workqueue.NewItemExponentialFailureRateLimiter(settings.RateLimiterBase, settings.RateLimiterMax)
-	return workqueue.NewNamedRateLimitingQueue(rateLimiter, settings.Name)
+	rateLimiter := workqueue.NewTypedItemExponentialFailureRateLimiter[synchronization.ServiceKey](
+		settings.RateLimiterBase, settings.RateLimiterMax)
+	return workqueue.NewTypedRateLimitingQueueWithConfig(
+		rateLimiter, workqueue.TypedRateLimitingQueueConfig[synchronization.ServiceKey]{Name: settings.Name})
 }

internal/synchronization/synchronizer.go

@@ -50,9 +50,7 @@ type ServiceKey struct {
 // a Border Client as specified in the Service annotation for the Upstream.
 // See application/border_client.go and application/application_constants.go for details.
 type Synchronizer struct {
-	// TODO: NLB-6294 change to use new typed workqueues
-	//nolint:staticcheck //ignore deprecation warnings
-	eventQueue    workqueue.RateLimitingInterface
+	eventQueue    workqueue.TypedRateLimitingInterface[ServiceKey]
 	settings      configuration.Settings
 	translator    Translator
 	cache         *cache
@@ -62,8 +60,7 @@ type Synchronizer struct {
 // NewSynchronizer creates a new Synchronizer.
 func NewSynchronizer(
 	settings configuration.Settings,
-	//nolint:staticcheck //ignore deprecation warnings
-	eventQueue workqueue.RateLimitingInterface,
+	eventQueue workqueue.TypedRateLimitingInterface[ServiceKey],
 	translator Translator,
 	serviceLister corelisters.ServiceLister,
 ) (*Synchronizer, error) {
@@ -292,13 +289,7 @@ func (s *Synchronizer) handleNextServiceEvent(ctx context.Context) bool {
 
 	defer s.eventQueue.Done(svc)
 
-	key, ok := svc.(ServiceKey)
-	if !ok {
-		slog.Warn(`Synchronizer::handleNextServiceEvent: invalid service type`, "service", svc)
-		return true
-	}
-
-	s.withRetry(s.handleServiceEvent(ctx, key), key)
+	s.withRetry(s.handleServiceEvent(ctx, svc), svc)
 
 	return true
 }

internal/synchronization/synchronizer_test.go

@@ -22,7 +22,7 @@ import (
 func TestSynchronizer_NewSynchronizer(t *testing.T) {
 	t.Parallel()
 
-	rateLimiter := &mocks.MockRateLimiter{}
+	rateLimiter := &mocks.MockRateLimiter[ServiceKey]{}
 
 	synchronizer, err := NewSynchronizer(
 		configuration.Settings{},
@@ -43,7 +43,7 @@ func TestSynchronizer_AddEventNoHosts(t *testing.T) {
 	t.Parallel()
 	const expectedEventCount = 0
 
-	rateLimiter := &mocks.MockRateLimiter{}
+	rateLimiter := &mocks.MockRateLimiter[ServiceKey]{}
 
 	synchronizer, err := NewSynchronizer(
 		defaultSettings(),
@@ -73,7 +73,7 @@ func TestSynchronizer_AddEventOneHost(t *testing.T) {
 	const expectedEventCount = 1
 	events := buildServerUpdateEvents(1)
 
-	rateLimiter := &mocks.MockRateLimiter{}
+	rateLimiter := &mocks.MockRateLimiter[ServiceKey]{}
 
 	synchronizer, err := NewSynchronizer(
 		defaultSettings("https://localhost:8080"),
@@ -106,7 +106,7 @@ func TestSynchronizer_AddEventManyHosts(t *testing.T) {
 		"https://localhost:8082",
 	}
 
-	rateLimiter := &mocks.MockRateLimiter{}
+	rateLimiter := &mocks.MockRateLimiter[ServiceKey]{}
 
 	synchronizer, err := NewSynchronizer(
 		defaultSettings(hosts...),
@@ -133,7 +133,7 @@ func TestSynchronizer_AddEventsNoHosts(t *testing.T) {
 	t.Parallel()
 	const expectedEventCount = 0
 	events := buildServerUpdateEvents(4)
-	rateLimiter := &mocks.MockRateLimiter{}
+	rateLimiter := &mocks.MockRateLimiter[ServiceKey]{}
 
 	synchronizer, err := NewSynchronizer(
 		defaultSettings(),
@@ -165,7 +165,7 @@ func TestSynchronizer_AddEventsOneHost(t *testing.T) {
 	t.Parallel()
 	const expectedEventCount = 4
 	events := buildServerUpdateEvents(1)
-	rateLimiter := &mocks.MockRateLimiter{}
+	rateLimiter := &mocks.MockRateLimiter[ServiceKey]{}
 
 	synchronizer, err := NewSynchronizer(
 		defaultSettings("https://localhost:8080"),
@@ -195,7 +195,7 @@ func TestSynchronizer_AddEventsManyHosts(t *testing.T) {
 	t.Parallel()
 	const eventCount = 4
 	events := buildServerUpdateEvents(eventCount)
-	rateLimiter := &mocks.MockRateLimiter{}
+	rateLimiter := &mocks.MockRateLimiter[ServiceKey]{}
 
 	hosts := []string{
 		"https://localhost:8080",

test/mocks/mock_ratelimitinginterface.go

@@ -7,51 +7,50 @@ package mocks
 
 import "time"
 
-type MockRateLimiter struct {
-	items []interface{}
+type MockRateLimiter[T any] struct {
+	items []T
 }
 
-func (m *MockRateLimiter) Add(_ interface{}) {
+func (m *MockRateLimiter[T]) Add(_ T) {
 }
 
-func (m *MockRateLimiter) Len() int {
+func (m *MockRateLimiter[T]) Len() int {
 	return len(m.items)
 }
 
-func (m *MockRateLimiter) Get() (item interface{}, shutdown bool) {
+func (m *MockRateLimiter[T]) Get() (item T, shutdown bool) {
 	if len(m.items) > 0 {
 		item = m.items[0]
 		m.items = m.items[1:]
 		return item, false
 	}
-	return nil, false
+	return item, false
 }
 
-func (m *MockRateLimiter) Done(_ interface{}) {
+func (m *MockRateLimiter[T]) Done(_ T) {
 }
 
-func (m *MockRateLimiter) ShutDown() {
+func (m *MockRateLimiter[T]) ShutDown() {
 }
 
-func (m *MockRateLimiter) ShutDownWithDrain() {
+func (m *MockRateLimiter[T]) ShutDownWithDrain() {
 }
 
-func (m *MockRateLimiter) ShuttingDown() bool {
+func (m *MockRateLimiter[T]) ShuttingDown() bool {
 	return true
 }
 
-func (m *MockRateLimiter) AddAfter(item interface{}, _ time.Duration) {
+func (m *MockRateLimiter[T]) AddAfter(item T, _ time.Duration) {
 	m.items = append(m.items, item)
 }
 
-func (m *MockRateLimiter) AddRateLimited(item interface{}) {
+func (m *MockRateLimiter[T]) AddRateLimited(item T) {
 	m.items = append(m.items, item)
 }
 
-func (m *MockRateLimiter) Forget(_ interface{}) {
-
+func (m *MockRateLimiter[T]) Forget(_ T) {
 }
 
-func (m *MockRateLimiter) NumRequeues(_ interface{}) int {
+func (m *MockRateLimiter[T]) NumRequeues(_ T) int {
 	return 0
 }