Add unit tests for the new events API

Signed-off-by: Borja Clemente <borja.clemente@gmail.com>

Author: clebs

pkg/cluster/cluster_test.go

@@ -155,6 +155,12 @@ var _ = Describe("cluster.Cluster", func() {
 	})
 
 	It("should provide a function to get the EventRecorder", func() {
+		c, err := New(cfg)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(c.GetEventRecorder("test")).NotTo(BeNil())
+	})
+
+	It("should provide a function to get the deprecated EventRecorder", func() {
 		c, err := New(cfg)
 		Expect(err).NotTo(HaveOccurred())
 		Expect(c.GetEventRecorderFor("test")).NotTo(BeNil()) //nolint:staticcheck

pkg/internal/recorder/recorder.go

@@ -104,16 +104,12 @@ func (p *Provider) getBroadcaster() (record.EventBroadcaster, events.EventBroadc
 	p.broadcasterOnce.Do(func() {
 		p.deprecatedBroadcaster, p.broadcaster, p.stopBroadcaster = p.makeBroadcaster()
 
-		// init old broadcaster
+		// init deprecated broadcaster (new broadcaster does not need to be initialised)
 		p.deprecatedBroadcaster.StartRecordingToSink(&corev1client.EventSinkImpl{Interface: p.evtClient})
 		p.deprecatedBroadcaster.StartEventWatcher(
 			func(e *corev1.Event) {
 				p.logger.V(1).Info(e.Message, "type", e.Type, "object", e.InvolvedObject, "reason", e.Reason)
 			})
-
-		// init new broadcaster
-		// TODO(clebs): figure out how to manage the context/channel that StartRecordingToSink needs inside the provider.
-		_ = p.broadcaster.StartRecordingToSinkWithContext(context.TODO())
 	})
 
 	return p.deprecatedBroadcaster, p.broadcaster

pkg/internal/recorder/recorder_test.go

@@ -28,16 +28,9 @@ import (
 )
 
 var _ = Describe("recorder.Provider", func() {
-	evtCl, err := eventsv1client.NewForConfigAndClient(cfg, httpClient)
-	Expect(err).NotTo(HaveOccurred())
-
-	makeBroadcaster := func() (record.EventBroadcaster, events.EventBroadcaster, bool) {
-		return record.NewBroadcaster(), events.NewBroadcaster(&events.EventSinkImpl{Interface: evtCl}), true
-	}
-
 	Describe("NewProvider", func() {
 		It("should return a provider instance and a nil error.", func() {
-			provider, err := recorder.NewProvider(cfg, httpClient, scheme.Scheme, logr.Discard(), makeBroadcaster)
+			provider, err := recorder.NewProvider(cfg, httpClient, scheme.Scheme, logr.Discard(), makeBroadcaster())
 			Expect(provider).NotTo(BeNil())
 			Expect(err).NotTo(HaveOccurred())
 		})
@@ -46,18 +39,36 @@ var _ = Describe("recorder.Provider", func() {
 			// Invalid the config
 			cfg1 := *cfg
 			cfg1.Host = "invalid host"
-			_, err := recorder.NewProvider(&cfg1, httpClient, scheme.Scheme, logr.Discard(), makeBroadcaster)
+			_, err := recorder.NewProvider(&cfg1, httpClient, scheme.Scheme, logr.Discard(), makeBroadcaster())
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("failed to init client"))
 		})
 	})
 	Describe("GetEventRecorderFor", func() {
-		It("should return a recorder instance.", func() {
-			provider, err := recorder.NewProvider(cfg, httpClient, scheme.Scheme, logr.Discard(), makeBroadcaster)
+		It("should return a deprecated recorder instance.", func() {
+			provider, err := recorder.NewProvider(cfg, httpClient, scheme.Scheme, logr.Discard(), makeBroadcaster())
 			Expect(err).NotTo(HaveOccurred())
 
 			recorder := provider.GetEventRecorderFor("test")
 			Expect(recorder).NotTo(BeNil())
 		})
 	})
+	Describe("GetEventRecorder", func() {
+		It("should return a recorder instance.", func() {
+			provider, err := recorder.NewProvider(cfg, httpClient, scheme.Scheme, logr.Discard(), makeBroadcaster())
+			Expect(err).NotTo(HaveOccurred())
+
+			recorder := provider.GetEventRecorder("test")
+			Expect(recorder).NotTo(BeNil())
+		})
+	})
 })
+
+func makeBroadcaster() func() (record.EventBroadcaster, events.EventBroadcaster, bool) {
+	evtCl, err := eventsv1client.NewForConfigAndClient(cfg, httpClient)
+	Expect(err).NotTo(HaveOccurred())
+
+	return func() (record.EventBroadcaster, events.EventBroadcaster, bool) {
+		return record.NewBroadcaster(), events.NewBroadcaster(&events.EventSinkImpl{Interface: evtCl}), true
+	}
+}

pkg/manager/manager_test.go

@@ -1804,7 +1804,7 @@ var _ = Describe("manger.Manager", func() {
 		Eventually(func() error { return goleak.Find(currentGRs) }).Should(Succeed())
 	})
 
-	It("should not leak goroutines if the default event broadcaster is used & events are emitted", func(specCtx SpecContext) {
+	It("should not leak goroutines if the deprecated event broadcaster is used & events are emitted", func(specCtx SpecContext) {
 		currentGRs := goleak.IgnoreCurrent()
 
 		m, err := New(cfg, Options{ /* implicit: default setting for EventBroadcaster */ })
@@ -1852,6 +1852,54 @@ var _ = Describe("manger.Manager", func() {
 		Eventually(func() error { return goleak.Find(currentGRs) }).Should(Succeed())
 	})
 
+	It("should not leak goroutines if the default event broadcaster is used & events are emitted", func(specCtx SpecContext) {
+		currentGRs := goleak.IgnoreCurrent()
+
+		m, err := New(cfg, Options{ /* implicit: default setting for EventBroadcaster */ })
+		Expect(err).NotTo(HaveOccurred())
+
+		By("adding a runnable that emits an event")
+		ns := corev1.Namespace{}
+		ns.Name = "default"
+
+		recorder := m.GetEventRecorder("rock-and-roll")
+		Expect(m.Add(RunnableFunc(func(_ context.Context) error {
+			recorder.Eventf(&ns, nil, "Warning", "BallroomBlitz", "dance action", "yeah, yeah, yeah-yeah-yeah")
+			return nil
+		}))).To(Succeed())
+
+		By("starting the manager & waiting till we've sent our event")
+		ctx, cancel := context.WithCancel(specCtx)
+		doneCh := make(chan struct{})
+		go func() {
+			defer GinkgoRecover()
+			defer close(doneCh)
+			Expect(m.Start(ctx)).To(Succeed())
+		}()
+		<-m.Elected()
+
+		Eventually(func() *corev1.Event {
+			evts, err := clientset.CoreV1().Events("").SearchWithContext(ctx, m.GetScheme(), &ns)
+			Expect(err).NotTo(HaveOccurred())
+
+			for i, evt := range evts.Items {
+				if evt.Reason == "BallroomBlitz" {
+					return &evts.Items[i]
+				}
+			}
+			return nil
+		}).ShouldNot(BeNil())
+
+		By("making sure there's no extra go routines still running after we stop")
+		cancel()
+		<-doneCh
+
+		// force-close keep-alive connections.  These'll time anyway (after
+		// like 30s or so) but force it to speed up the tests.
+		clientTransport.CloseIdleConnections()
+		Eventually(func() error { return goleak.Find(currentGRs) }).Should(Succeed())
+	})
+
 	It("should not leak goroutines when a runnable returns error slowly after being signaled to stop", func(specCtx SpecContext) {
 		// This test reproduces the race condition where the manager's Start method
 		// exits due to context cancellation, leaving no one to drain errChan
@@ -1932,11 +1980,16 @@ var _ = Describe("manger.Manager", func() {
 		Expect(m.GetFieldIndexer()).To(Equal(mgr.cluster.GetFieldIndexer()))
 	})
 
-	It("should provide a function to get the EventRecorder", func() {
+	It("should provide a function to get the deprecated EventRecorder", func() {
 		m, err := New(cfg, Options{})
 		Expect(err).NotTo(HaveOccurred())
 		Expect(m.GetEventRecorderFor("test")).NotTo(BeNil()) //nolint:staticcheck
 	})
+	It("should provide a function to get the EventRecorder", func() {
+		m, err := New(cfg, Options{})
+		Expect(err).NotTo(HaveOccurred())
+		Expect(m.GetEventRecorder("test")).NotTo(BeNil())
+	})
 	It("should provide a function to get the APIReader", func() {
 		m, err := New(cfg, Options{})
 		Expect(err).NotTo(HaveOccurred())

pkg/recorder/example_test.go

@@ -18,31 +18,39 @@ package recorder_test
 
 import (
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 
 	_ "github.com/onsi/ginkgo/v2"
 	"sigs.k8s.io/controller-runtime/pkg/recorder"
 )
 
 var (
-	recorderProvider recorder.Provider
-	somePod          *corev1.Pod // the object you're reconciling, for example
+	recorderProvider  recorder.Provider
+	somePod           *corev1.Pod    // the object you're reconciling, for example
+	someRelatedObject runtime.Object // another object related to the reconciled object and the event.
 )
 
 func Example_event() {
 	// recorderProvider is a recorder.Provider
-	recorder := recorderProvider.GetEventRecorderFor("my-controller")
+	deprecatedRecorder := recorderProvider.GetEventRecorderFor("my-controller")
 
 	// emit an event with a fixed message
-	recorder.Event(somePod, corev1.EventTypeWarning,
+	deprecatedRecorder.Event(somePod, corev1.EventTypeWarning,
 		"WrongTrousers", "It's the wrong trousers, Gromit!")
 }
 
 func Example_eventf() {
 	// recorderProvider is a recorder.Provider
-	recorder := recorderProvider.GetEventRecorderFor("my-controller")
+	deprecatedRecorder := recorderProvider.GetEventRecorderFor("my-controller")
 
 	// emit an event with a variable message
 	mildCheese := "Wensleydale"
-	recorder.Eventf(somePod, corev1.EventTypeNormal,
+	deprecatedRecorder.Eventf(somePod, corev1.EventTypeNormal,
 		"DislikesCheese", "Not even %s?", mildCheese)
+
+	recorder := recorderProvider.GetEventRecorder("my-controller")
+
+	// emit an event with a fixed message
+	recorder.Eventf(somePod, someRelatedObject, corev1.EventTypeWarning,
+		"WrongTrousers", "getting dressed", "It's the wrong trousers, Gromit!")
 }