Avoid Event cache hydration by using clinet go watcher

Author: engedaam

events/controller.go

@@ -8,15 +8,17 @@ import (
 
 	pmetrics "github.com/awslabs/operatorpkg/metrics"
 	"github.com/awslabs/operatorpkg/object"
+	"github.com/awslabs/operatorpkg/singleton"
+	"github.com/samber/lo"
 	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/utils/clock"
 	controllerruntime "sigs.k8s.io/controller-runtime"
-	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
-	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
 
@@ -25,31 +27,32 @@ type Controller[T client.Object] struct {
 	startTime  time.Time
 	kubeClient client.Client
 	EventCount pmetrics.CounterMetric
+	eventWatch watch.Interface
 }
 
-func NewController[T client.Object](client client.Client, clock clock.Clock) *Controller[T] {
+func NewController[T client.Object](ctx context.Context, client client.Client, clock clock.Clock, kubernetesInterface kubernetes.Interface) *Controller[T] {
 	gvk := object.GVK(object.New[T]())
 	return &Controller[T]{
 		gvk:        gvk,
 		startTime:  clock.Now(),
 		kubeClient: client,
 		EventCount: eventTotalMetric(strings.ToLower(gvk.Kind)),
+		eventWatch: lo.Must(kubernetesInterface.CoreV1().Events("").Watch(ctx, metav1.ListOptions{})),
 	}
 }
 
-func (c *Controller[T]) Register(_ context.Context, m manager.Manager) error {
+func (c *Controller[T]) Register(ctx context.Context, m manager.Manager) error {
 	return controllerruntime.NewControllerManagedBy(m).
-		For(&v1.Event{}, builder.WithPredicates(predicate.NewTypedPredicateFuncs(func(o client.Object) bool {
-			// Only reconcile on the object kind we care about
-			event := o.(*v1.Event)
-			return event.InvolvedObject.Kind == c.gvk.Kind && event.InvolvedObject.APIVersion == c.gvk.GroupVersion().String()
-		}))).
-		WithOptions(controller.Options{MaxConcurrentReconciles: 10}).
 		Named(fmt.Sprintf("operatorpkg.%s.events", strings.ToLower(c.gvk.Kind))).
-		Complete(reconcile.AsReconciler(m.GetClient(), c))
+		WatchesRawSource(singleton.ChannelSource[*v1.Event](ctx, c.eventWatch.ResultChan())).
+		Complete(reconcile.AsReconciler(c.kubeClient, c))
 }
 
 func (c *Controller[T]) Reconcile(ctx context.Context, event *v1.Event) (reconcile.Result, error) {
+	if event.InvolvedObject.Kind != c.gvk.Kind && event.InvolvedObject.APIVersion != c.gvk.GroupVersion().String() {
+		return reconcile.Result{}, nil
+	}
+
 	// We check if the event was created in the lifetime of this controller
 	// since we don't duplicate metrics on controller restart or lease handover
 	if c.startTime.Before(event.LastTimestamp.Time) {

events/suite_test.go

@@ -19,10 +19,12 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/scheme"
 	clock "k8s.io/utils/clock/testing"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+	"sigs.k8s.io/controller-runtime/pkg/envtest"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
@@ -46,13 +48,18 @@ func Test(t *testing.T) {
 }
 
 var _ = BeforeSuite(func() {
+	ctx = context.Background()
+	ctx = log.IntoContext(context.Background(), ginkgo.GinkgoLogr)
+
 	fakeClock = clock.NewFakeClock(time.Now())
 	kubeClient = fake.NewClientBuilder().WithScheme(scheme.Scheme).WithIndex(&corev1.Event{}, "involvedObject.kind", func(o client.Object) []string {
 		evt := o.(*corev1.Event)
 		return []string{evt.InvolvedObject.Kind}
 	}).Build()
-	controller = events.NewController[*test.CustomObject](kubeClient, fakeClock)
-	ctx = log.IntoContext(context.Background(), ginkgo.GinkgoLogr)
+	environment := envtest.Environment{Scheme: scheme.Scheme}
+	_ = lo.Must(environment.Start())
+
+	controller = events.NewController[*test.CustomObject](ctx, kubeClient, fakeClock, kubernetes.NewForConfigOrDie(environment.Config))
 })
 
 var _ = Describe("Controller", func() {

singleton/controller.go

@@ -4,7 +4,9 @@ import (
 	"context"
 	"time"
 
+	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/util/workqueue"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
@@ -36,3 +38,33 @@ func Source() source.Source {
 		},
 	})
 }
+
+func ChannelSource[T client.Object](ctx context.Context, objectChan <-chan watch.Event) source.Source {
+	eventSource := make(chan event.GenericEvent, 1)
+
+	go func(ctx context.Context, objectChan <-chan watch.Event, eventSource chan event.GenericEvent) {
+
+		for {
+			select {
+			case delta, ok := <-objectChan:
+				if !ok {
+					return
+				}
+				// Convert the delta to GenericEvent
+				eventSource <- event.GenericEvent{
+					Object: delta.Object.(T),
+				}
+			case <-ctx.Done():
+				return
+			}
+		}
+	}(ctx, objectChan, eventSource)
+
+	return source.Channel(eventSource, handler.Funcs{
+		GenericFunc: func(_ context.Context, event event.GenericEvent, queue workqueue.TypedRateLimitingInterface[reconcile.Request]) {
+			queue.Add(reconcile.Request{
+				NamespacedName: client.ObjectKeyFromObject(event.Object),
+			})
+		},
+	})
+}