Adding

Author: engedaam

events/controller.go

@@ -37,29 +37,31 @@ func NewController[T client.Object](ctx context.Context, client client.Client, c
 		startTime:  clock.Now(),
 		kubeClient: client,
 		EventCount: eventTotalMetric(strings.ToLower(gvk.Kind)),
-		eventWatch: lo.Must(kubernetesInterface.CoreV1().Events("").Watch(ctx, metav1.ListOptions{})),
+		eventWatch: lo.Must(kubernetesInterface.CoreV1().Events("").Watch(ctx, metav1.ListOptions{
+			FieldSelector: fmt.Sprintf("involvedObject.kind=%s,involvedObject.apiVersion=%s", gvk.Kind, gvk.GroupVersion().String()),
+		})),
 	}
 }
 
 func (c *Controller[T]) Register(ctx context.Context, m manager.Manager) error {
 	return controllerruntime.NewControllerManagedBy(m).
 		Named(fmt.Sprintf("operatorpkg.%s.events", strings.ToLower(c.gvk.Kind))).
-		WatchesRawSource(singleton.ChannelSource[*v1.Event](ctx, c.eventWatch.ResultChan())).
-		Complete(reconcile.AsReconciler(c.kubeClient, c))
+		WatchesRawSource(singleton.Source()).
+		Complete(singleton.AsReconciler(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
-	}
+func (c *Controller[T]) Reconcile(ctx context.Context) (reconcile.Result, error) {
+	for e := range c.eventWatch.ResultChan() {
+		event := e.Object.(*v1.Event)
 
-	// 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) {
-		c.EventCount.Inc(map[string]string{
-			pmetrics.LabelType:   event.Type,
-			pmetrics.LabelReason: event.Reason,
-		})
+		// 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) {
+			c.EventCount.Inc(map[string]string{
+				pmetrics.LabelType:   event.Type,
+				pmetrics.LabelReason: event.Reason,
+			})
+		}
 	}
 
 	return reconcile.Result{}, nil

events/suite_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/awslabs/operatorpkg/events"
 	pmetrics "github.com/awslabs/operatorpkg/metrics"
 	"github.com/awslabs/operatorpkg/object"
+	"github.com/awslabs/operatorpkg/singleton"
 	"github.com/awslabs/operatorpkg/test"
 	. "github.com/awslabs/operatorpkg/test/expectations"
 	"github.com/onsi/ginkgo/v2"
@@ -73,7 +74,7 @@ var _ = Describe("Controller", func() {
 			Expect(GetMetric("operator_customobject_event_total", conditionLabels(fmt.Sprintf("Test-type-%d", i), fmt.Sprintf("Test-reason-%d", i)))).To(BeNil())
 
 			// reconcile on the event
-			_, err := reconcile.AsReconciler(kubeClient, controller).Reconcile(ctx, reconcile.Request{NamespacedName: client.ObjectKeyFromObject(events[i])})
+			_, err := singleton.AsReconciler(controller).Reconcile(ctx, reconcile.Request{NamespacedName: client.ObjectKeyFromObject(events[i])})
 			Expect(err).ToNot(HaveOccurred())
 
 			// expect an emitted metric to for the event
@@ -90,7 +91,7 @@ var _ = Describe("Controller", func() {
 		Expect(GetMetric("operator_ustomobject_event_total", conditionLabels(corev1.EventTypeNormal, "reason"))).To(BeNil())
 
 		// reconcile on the event
-		_, err := reconcile.AsReconciler(kubeClient, controller).Reconcile(ctx, reconcile.Request{NamespacedName: client.ObjectKeyFromObject(event)})
+		_, err := singleton.AsReconciler(controller).Reconcile(ctx, reconcile.Request{NamespacedName: client.ObjectKeyFromObject(event)})
 		Expect(err).ToNot(HaveOccurred())
 
 		// expect not have an emitted metric to for the event
@@ -101,7 +102,7 @@ var _ = Describe("Controller", func() {
 		ExpectApplied(ctx, kubeClient, event)
 
 		// reconcile on the event
-		_, err = reconcile.AsReconciler(kubeClient, controller).Reconcile(ctx, reconcile.Request{NamespacedName: client.ObjectKeyFromObject(event)})
+		_, err = singleton.AsReconciler(controller).Reconcile(ctx, reconcile.Request{NamespacedName: client.ObjectKeyFromObject(event)})
 		Expect(err).ToNot(HaveOccurred())
 
 		// expect an emitted metric to for the event

singleton/controller.go

@@ -4,9 +4,7 @@ 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"
@@ -38,32 +36,3 @@ func Source() source.Source {
 		},
 	})
 }
-
-func ChannelSource[T client.Object](ctx context.Context, objectChan <-chan watch.Event) source.Source {
-	eventSource := make(chan event.GenericEvent, 1000)
-
-	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),
-			})
-		},
-	})
-}