feat: vertical slice application -> provider -> binding (#4)

Co-authored-by: Timo Behrendt <t.behrendt@t00n.de> Co-committed-by: Timo Behrendt <t.behrendt@t00n.de>
2026-05-25 17:14:35 +02:00
parent 2a091df8b9
commit 26bd576690
65 changed files with 4912 additions and 121 deletions
@@ -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())
+}