refactor: consolidate common controller code
This commit is contained in:
@@ -0,0 +1,94 @@
|
||||
package baseController
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
|
||||
"k8s.io/apimachinery/pkg/util/wait"
|
||||
"k8s.io/client-go/tools/cache"
|
||||
"k8s.io/client-go/tools/record"
|
||||
"k8s.io/client-go/util/workqueue"
|
||||
"k8s.io/klog/v2"
|
||||
)
|
||||
|
||||
type SyncHandler func(ctx context.Context, objRef cache.ObjectName) error
|
||||
|
||||
type Controller struct {
|
||||
workqueue workqueue.TypedRateLimitingInterface[cache.ObjectName]
|
||||
recorder record.EventRecorder
|
||||
synced cache.InformerSynced
|
||||
syncHandler SyncHandler
|
||||
}
|
||||
|
||||
func NewController(
|
||||
ctx context.Context,
|
||||
workqueue workqueue.TypedRateLimitingInterface[cache.ObjectName],
|
||||
recorder record.EventRecorder,
|
||||
synced cache.InformerSynced,
|
||||
syncHandler SyncHandler,
|
||||
) *Controller {
|
||||
return &Controller{
|
||||
workqueue: workqueue,
|
||||
recorder: recorder,
|
||||
synced: synced,
|
||||
syncHandler: syncHandler,
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Controller) Run(ctx context.Context, workers int) error {
|
||||
defer utilruntime.HandleCrash()
|
||||
defer c.workqueue.ShutDown()
|
||||
logger := klog.FromContext(ctx)
|
||||
|
||||
logger.Info("Starting PolicyBinding controller")
|
||||
|
||||
logger.Info("Waiting for informer caches to sync")
|
||||
if ok := cache.WaitForCacheSync(ctx.Done(), c.synced); !ok {
|
||||
return fmt.Errorf("failed to wait for caches to sync")
|
||||
}
|
||||
|
||||
logger.Info("Starting workers", "count", workers)
|
||||
for i := 0; i < workers; i++ {
|
||||
go wait.UntilWithContext(ctx, c.runWorker, time.Second)
|
||||
}
|
||||
|
||||
logger.Info("Started workers")
|
||||
<-ctx.Done()
|
||||
logger.Info("Shutting down workers")
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Controller) runWorker(ctx context.Context) {
|
||||
for c.processNextWorkItem(ctx) {
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Controller) processNextWorkItem(ctx context.Context) bool {
|
||||
objRef, shutdown := c.workqueue.Get()
|
||||
logger := klog.FromContext(ctx)
|
||||
if shutdown {
|
||||
return false
|
||||
}
|
||||
defer c.workqueue.Done(objRef)
|
||||
|
||||
err := c.syncHandler(ctx, objRef)
|
||||
if err == nil {
|
||||
c.workqueue.Forget(objRef)
|
||||
logger.Info("Successfully synced", "objectName", objRef)
|
||||
return true
|
||||
}
|
||||
utilruntime.HandleErrorWithContext(ctx, err, "Error syncing; requeuing for later retry", "objectReference", objRef)
|
||||
c.workqueue.AddRateLimited(objRef)
|
||||
return true
|
||||
}
|
||||
|
||||
func (c *Controller) Enqueue(obj interface{}) {
|
||||
objectRef, err := cache.ObjectToName(obj)
|
||||
if err != nil {
|
||||
utilruntime.HandleError(err)
|
||||
return
|
||||
}
|
||||
c.workqueue.Add(objectRef)
|
||||
}
|
||||
@@ -0,0 +1,190 @@
|
||||
// AI generated tests and not yet reviewed.
|
||||
package baseController
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
corev1 "k8s.io/api/core/v1"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/client-go/tools/cache"
|
||||
"k8s.io/client-go/tools/record"
|
||||
"k8s.io/client-go/util/workqueue"
|
||||
)
|
||||
|
||||
func newTestController(t *testing.T, synced cache.InformerSynced, syncHandler SyncHandler) (*Controller, workqueue.TypedRateLimitingInterface[cache.ObjectName]) {
|
||||
t.Helper()
|
||||
|
||||
ratelimiter := workqueue.NewTypedItemExponentialFailureRateLimiter[cache.ObjectName](time.Millisecond, time.Second)
|
||||
q := workqueue.NewTypedRateLimitingQueue(ratelimiter)
|
||||
t.Cleanup(q.ShutDown)
|
||||
|
||||
if synced == nil {
|
||||
synced = func() bool { return true }
|
||||
}
|
||||
|
||||
ctrl := NewController(
|
||||
context.Background(),
|
||||
q,
|
||||
record.NewFakeRecorder(10),
|
||||
synced,
|
||||
syncHandler,
|
||||
)
|
||||
return ctrl, q
|
||||
}
|
||||
|
||||
func TestController_processNextWorkItem_success(t *testing.T) {
|
||||
objRef := cache.ObjectName{Namespace: "default", Name: "test"}
|
||||
|
||||
var syncedRef cache.ObjectName
|
||||
ctrl, q := newTestController(t, nil, func(_ context.Context, ref cache.ObjectName) error {
|
||||
syncedRef = ref
|
||||
return nil
|
||||
})
|
||||
q.Add(objRef)
|
||||
|
||||
if !ctrl.processNextWorkItem(context.Background()) {
|
||||
t.Fatal("processNextWorkItem() = false, want true")
|
||||
}
|
||||
if syncedRef != objRef {
|
||||
t.Fatalf("syncHandler object = %+v, want %+v", syncedRef, objRef)
|
||||
}
|
||||
if q.Len() != 0 {
|
||||
t.Fatalf("queue length = %d, want 0 after successful sync", q.Len())
|
||||
}
|
||||
if q.NumRequeues(objRef) != 0 {
|
||||
t.Fatalf("requeues = %d, want 0 after successful sync", q.NumRequeues(objRef))
|
||||
}
|
||||
}
|
||||
|
||||
func TestController_processNextWorkItem_syncError(t *testing.T) {
|
||||
objRef := cache.ObjectName{Namespace: "default", Name: "test"}
|
||||
syncErr := errors.New("sync failed")
|
||||
|
||||
ctrl, q := newTestController(t, nil, func(context.Context, cache.ObjectName) error {
|
||||
return syncErr
|
||||
})
|
||||
q.Add(objRef)
|
||||
|
||||
if !ctrl.processNextWorkItem(context.Background()) {
|
||||
t.Fatal("processNextWorkItem() = false, want true")
|
||||
}
|
||||
if q.NumRequeues(objRef) != 1 {
|
||||
t.Fatalf("requeues = %d, want 1 after sync error", q.NumRequeues(objRef))
|
||||
}
|
||||
}
|
||||
|
||||
func TestController_processNextWorkItem_shutdown(t *testing.T) {
|
||||
ctrl, q := newTestController(t, nil, func(context.Context, cache.ObjectName) error {
|
||||
return nil
|
||||
})
|
||||
q.ShutDown()
|
||||
|
||||
if ctrl.processNextWorkItem(context.Background()) {
|
||||
t.Fatal("processNextWorkItem() = true, want false on shutdown")
|
||||
}
|
||||
}
|
||||
|
||||
func TestController_Enqueue(t *testing.T) {
|
||||
ctrl, q := newTestController(t, nil, func(context.Context, cache.ObjectName) error {
|
||||
return nil
|
||||
})
|
||||
|
||||
obj := &corev1.ConfigMap{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Namespace: "default",
|
||||
Name: "test",
|
||||
},
|
||||
}
|
||||
ctrl.Enqueue(obj)
|
||||
|
||||
if q.Len() != 1 {
|
||||
t.Fatalf("queue length = %d, want 1 after Enqueue", q.Len())
|
||||
}
|
||||
}
|
||||
|
||||
func TestController_Enqueue_invalidObject(t *testing.T) {
|
||||
ctrl, q := newTestController(t, nil, func(context.Context, cache.ObjectName) error {
|
||||
return nil
|
||||
})
|
||||
|
||||
ctrl.Enqueue("not-a-kubernetes-object")
|
||||
|
||||
if q.Len() != 0 {
|
||||
t.Fatalf("queue length = %d, want 0 for invalid object", q.Len())
|
||||
}
|
||||
}
|
||||
|
||||
func TestController_Run_cacheSyncFails(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
ctrl, _ := newTestController(t, func() bool { return false }, func(context.Context, cache.ObjectName) error {
|
||||
return nil
|
||||
})
|
||||
|
||||
go func() {
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
cancel()
|
||||
}()
|
||||
|
||||
err := ctrl.Run(ctx, 1)
|
||||
if err == nil {
|
||||
t.Fatal("Run() error = nil, want cache sync failure")
|
||||
}
|
||||
}
|
||||
|
||||
func TestController_Run_shutsDownOnCancel(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
ctrl, _ := newTestController(t, nil, func(context.Context, cache.ObjectName) error {
|
||||
return nil
|
||||
})
|
||||
|
||||
errCh := make(chan error, 1)
|
||||
go func() {
|
||||
errCh <- ctrl.Run(ctx, 1)
|
||||
}()
|
||||
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
cancel()
|
||||
|
||||
select {
|
||||
case err := <-errCh:
|
||||
if err != nil {
|
||||
t.Fatalf("Run() error = %v, want nil on context cancel", err)
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("Run() did not return after context cancellation")
|
||||
}
|
||||
}
|
||||
|
||||
func TestController_runWorker_processesQueuedItem(t *testing.T) {
|
||||
objRef := cache.ObjectName{Namespace: "default", Name: "test"}
|
||||
var calls atomic.Int32
|
||||
|
||||
ctrl, q := newTestController(t, nil, func(context.Context, cache.ObjectName) error {
|
||||
calls.Add(1)
|
||||
return nil
|
||||
})
|
||||
q.Add(objRef)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
go ctrl.runWorker(ctx)
|
||||
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
if calls.Load() == 1 && q.Len() == 0 {
|
||||
cancel()
|
||||
return
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
cancel()
|
||||
t.Fatalf("runWorker did not process queued item: calls=%d queueLen=%d", calls.Load(), q.Len())
|
||||
}
|
||||
Reference in New Issue
Block a user