Removed the source queue and pulled object from the channel

Author: engedaam

events/controller.go

@@ -9,12 +9,9 @@ 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/client"
@@ -23,32 +20,29 @@ import (
 )
 
 type Controller[T client.Object] struct {
-	gvk        schema.GroupVersionKind
-	startTime  time.Time
-	kubeClient client.Client
-	EventCount pmetrics.CounterMetric
-	EventWatch watch.Interface
+	gvk               schema.GroupVersionKind
+	startTime         time.Time
+	kubeClient        client.Client
+	EventCount        pmetrics.CounterMetric
+	EventWatchChannel <-chan watch.Event
 }
 
-func NewController[T client.Object](ctx context.Context, client client.Client, clock clock.Clock, kubernetesInterface kubernetes.Interface) *Controller[T] {
+func NewController[T client.Object](ctx context.Context, client client.Client, clock clock.Clock, channel <-chan watch.Event) *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{
-			// Only reconcile on the object kind we care about
-			FieldSelector: fmt.Sprintf("involvedObject.kind=%s,involvedObject.apiVersion=%s", gvk.Kind, gvk.GroupVersion().String()),
-		})),
+		gvk:               gvk,
+		startTime:         clock.Now(),
+		kubeClient:        client,
+		EventCount:        eventTotalMetric(strings.ToLower(gvk.Kind)),
+		EventWatchChannel: channel,
 	}
 }
 
 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.Source()).
-		Complete(singleton.AsChannelObjectReconciler(c.EventWatch.ResultChan(), c))
+		Complete(singleton.AsChannelObjectReconciler(c.EventWatchChannel, c))
 }
 
 func (c *Controller[T]) Reconcile(ctx context.Context, event *v1.Event) (reconcile.Result, error) {

events/suite_test.go

@@ -20,11 +20,11 @@ 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/apimachinery/pkg/watch"
 	"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/envtest"
+	ctrlfake "sigs.k8s.io/controller-runtime/pkg/client/fake"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
@@ -39,8 +39,8 @@ var (
 var ctx context.Context
 var fakeClock *clock.FakeClock
 var controller *events.Controller[*test.CustomObject]
-var environment envtest.Environment
 var kubeClient client.Client
+var eventChannel chan watch.Event
 
 func Test(t *testing.T) {
 	lo.Must0(SchemeBuilder.AddToScheme(scheme.Scheme))
@@ -52,15 +52,13 @@ var _ = BeforeSuite(func() {
 	ctx = log.IntoContext(context.Background(), ginkgo.GinkgoLogr)
 
 	fakeClock = clock.NewFakeClock(time.Now())
-	environment = envtest.Environment{Scheme: scheme.Scheme}
-	_ = lo.Must(environment.Start())
-	kubeClient = lo.Must(client.New(environment.Config, client.Options{Scheme: scheme.Scheme}))
+	kubeClient = ctrlfake.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](ctx, kubeClient, fakeClock, kubernetes.NewForConfigOrDie(environment.Config))
-})
-
-var _ = AfterSuite(func() {
-	environment.Stop()
+	eventChannel = make(chan watch.Event, 1000)
+	controller = events.NewController[*test.CustomObject](ctx, kubeClient, fakeClock, eventChannel)
 })
 
 var _ = Describe("Controller", func() {
@@ -78,8 +76,11 @@ var _ = Describe("Controller", func() {
 			// expect an metrics for custom object to be zero, waiting on controller reconcile
 			Expect(GetMetric("operator_customobject_event_total", conditionLabels(fmt.Sprintf("Test-type-%d", i), fmt.Sprintf("Test-reason-%d", i)))).To(BeNil())
 
+			eventChannel <- watch.Event{
+				Object: events[i],
+			}
 			// reconcile on the event
-			_, err := singleton.AsChannelObjectReconciler(controller.EventWatch.ResultChan(), controller).Reconcile(ctx, reconcile.Request{})
+			_, err := singleton.AsChannelObjectReconciler(eventChannel, controller).Reconcile(ctx, reconcile.Request{})
 			Expect(err).ToNot(HaveOccurred())
 
 			// expect an emitted metric to for the event
@@ -95,8 +96,11 @@ var _ = Describe("Controller", func() {
 		// expect an metrics for custom object to be zero, waiting on controller reconcile
 		Expect(GetMetric("operator_ustomobject_event_total", conditionLabels(corev1.EventTypeNormal, "reason"))).To(BeNil())
 
+		eventChannel <- watch.Event{
+			Object: event,
+		}
 		// reconcile on the event
-		_, err := singleton.AsChannelObjectReconciler(controller.EventWatch.ResultChan(), controller).Reconcile(ctx, reconcile.Request{NamespacedName: client.ObjectKeyFromObject(event)})
+		_, err := singleton.AsChannelObjectReconciler(eventChannel, controller).Reconcile(ctx, reconcile.Request{NamespacedName: client.ObjectKeyFromObject(event)})
 		Expect(err).ToNot(HaveOccurred())
 
 		// expect not have an emitted metric to for the event
@@ -106,8 +110,11 @@ var _ = Describe("Controller", func() {
 		event.LastTimestamp.Time = time.Now().Add(-30 * time.Minute)
 		ExpectApplied(ctx, kubeClient, event)
 
+		eventChannel <- watch.Event{
+			Object: event,
+		}
 		// reconcile on the event
-		_, err = singleton.AsChannelObjectReconciler(controller.EventWatch.ResultChan(), controller).Reconcile(ctx, reconcile.Request{NamespacedName: client.ObjectKeyFromObject(event)})
+		_, err = singleton.AsChannelObjectReconciler(eventChannel, controller).Reconcile(ctx, reconcile.Request{NamespacedName: client.ObjectKeyFromObject(event)})
 		Expect(err).ToNot(HaveOccurred())
 
 		// expect an emitted metric to for the event

singleton/controller.go

@@ -40,6 +40,9 @@ func AsChannelObjectReconciler[T client.Object](watchEvents <-chan watch.Event,
 		var errs error
 		var results []reconcile.Result
 		e := <-watchEvents
+		if e.Object == nil {
+			return reconcile.Result{RequeueAfter: RequeueImmediately}, nil
+		}
 		res, err := reconciler.Reconcile(ctx, e.Object.(T))
 		errs = multierr.Append(errs, err)
 		results = append(results, res)