Fix new API not broadcasting events and add integration test

Signed-off-by: Borja Clemente <[email protected]>

Author: clebs

pkg/cluster/cluster.go

@@ -230,7 +230,6 @@ func New(config *rest.Config, opts ...Option) (Cluster, error) {
 	// Create the recorder provider to inject event recorders for the components.
 	// TODO(directxman12): the log for the event provider should have a context (name, tags, etc) specific
 	// to the particular controller that it's being injected into, rather than a generic one like is here.
-	// Stop the broadcaster with the provider only if the broadcaster is externally given (aka non-nil).
 	recorderProvider, err := options.newRecorderProvider(config, options.HTTPClient, options.Scheme, options.Logger.WithName("events"), options.makeBroadcaster)
 	if err != nil {
 		return nil, err

pkg/internal/recorder/recorder.go

@@ -24,6 +24,7 @@ import (
 
 	"github.com/go-logr/logr"
 	corev1 "k8s.io/api/core/v1"
+	eventsv1 "k8s.io/api/events/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
 	"k8s.io/client-go/rest"
@@ -34,7 +35,7 @@ import (
 // EventBroadcasterProducer makes an event broadcaster, returning
 // whether or not the broadcaster should be stopped with the Provider,
 // or not (e.g. if it's shared, it shouldn't be stopped with the Provider).
-// This producer currently produces both a
+// This producer currently produces both an old API and a new API broadcaster.
 type EventBroadcasterProducer func() (deprecatedCaster record.EventBroadcaster, caster events.EventBroadcaster, stopWithProvider bool)
 
 // Provider is a recorder.Provider that records events to the k8s API server
@@ -52,6 +53,7 @@ type Provider struct {
 
 	broadcasterOnce sync.Once
 	broadcaster     events.EventBroadcaster
+	stopCh          chan struct{}
 	// Deprecated: will be removed in a future release. Use the broadcaster above instead.
 	deprecatedBroadcaster record.EventBroadcaster
 	stopBroadcaster       bool
@@ -79,6 +81,7 @@ func (p *Provider) Stop(shutdownCtx context.Context) {
 		if p.stopBroadcaster {
 			p.lock.Lock()
 			broadcaster.Shutdown()
+			close(p.stopCh)
 			deprecatedBroadcaster.Shutdown()
 			p.stopped = true
 			p.lock.Unlock()
@@ -104,12 +107,22 @@ func (p *Provider) getBroadcaster() (record.EventBroadcaster, events.EventBroadc
 	p.broadcasterOnce.Do(func() {
 		p.deprecatedBroadcaster, p.broadcaster, p.stopBroadcaster = p.makeBroadcaster()
 
-		// init deprecated broadcaster (new broadcaster does not need to be initialised)
+		// init deprecated broadcaster
 		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
+		p.stopCh = make(chan struct{})
+		p.broadcaster.StartRecordingToSink(p.stopCh)
+		_, _ = p.broadcaster.StartEventWatcher(func(event runtime.Object) {
+			e, isEvt := event.(*eventsv1.Event)
+			if isEvt {
+				p.logger.V(1).Info(e.Note, "type", e.Type, "object", e.Related, "action", e.Action, "reason", e.Reason)
+			}
+		})
 	})
 
 	return p.deprecatedBroadcaster, p.broadcaster

pkg/internal/recorder/recorder_integration_test.go

@@ -21,11 +21,14 @@ import (
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	eventsv1 "k8s.io/api/events/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/scheme"
 	ref "k8s.io/client-go/tools/reference"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/envtest"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
@@ -36,14 +39,14 @@ import (
 )
 
 var _ = Describe("recorder", func() {
-	Describe("recorder", func() {
+	Describe("deprecated recorder", func() {
 		It("should publish events", func(ctx SpecContext) {
 			By("Creating the Manager")
 			cm, err := manager.New(cfg, manager.Options{})
 			Expect(err).NotTo(HaveOccurred())
 
 			By("Creating the Controller")
-			recorder := cm.GetEventRecorderFor("test-recorder") //nolint:staticcheck
+			recorder := cm.GetEventRecorderFor("test-deprecated-recorder") //nolint:staticcheck
 			instance, err := controller.New("foo-controller", cm, controller.Options{
 				Reconciler: reconcile.Func(
 					func(ctx context.Context, request reconcile.Request) (reconcile.Result, error) {
@@ -66,7 +69,7 @@ var _ = Describe("recorder", func() {
 			}()
 
 			deployment := &appsv1.Deployment{
-				ObjectMeta: metav1.ObjectMeta{Name: "deployment-name"},
+				ObjectMeta: metav1.ObjectMeta{Name: "deprecated-deployment-name"},
 				Spec: appsv1.DeploymentSpec{
 					Selector: &metav1.LabelSelector{
 						MatchLabels: map[string]string{"foo": "bar"},
@@ -108,4 +111,88 @@ var _ = Describe("recorder", func() {
 			Expect(evt.Message).To(Equal("test-msg"))
 		})
 	})
+
+	Describe("recorder", func() {
+		It("should publish events", func(ctx SpecContext) {
+			By("Creating the Manager")
+			// this test needs its own env for now to not interfere with the previous one.
+			// Once the deprecated API is removed this can be removed.
+			testenv := &envtest.Environment{}
+
+			cfg, err := testenv.Start()
+			Expect(err).NotTo(HaveOccurred())
+			defer testenv.Stop() //nolint:errcheck
+
+			clientset, err := kubernetes.NewForConfig(cfg)
+			Expect(err).NotTo(HaveOccurred())
+
+			cm, err := manager.New(cfg, manager.Options{})
+			Expect(err).NotTo(HaveOccurred())
+
+			By("Creating the Controller")
+			recorder := cm.GetEventRecorder("test-recorder")
+			instance, err := controller.New("bar-controller", cm, controller.Options{
+				Reconciler: reconcile.Func(
+					func(ctx context.Context, request reconcile.Request) (reconcile.Result, error) {
+						dp, err := clientset.AppsV1().Deployments(request.Namespace).Get(ctx, request.Name, metav1.GetOptions{})
+						Expect(err).NotTo(HaveOccurred())
+						recorder.Eventf(dp, nil, corev1.EventTypeNormal, "test-reason", "test-action", "test-msg")
+						return reconcile.Result{}, nil
+					}),
+			})
+			Expect(err).NotTo(HaveOccurred())
+
+			By("Watching Resources")
+			err = instance.Watch(source.Kind(cm.GetCache(), &appsv1.Deployment{}, &handler.TypedEnqueueRequestForObject[*appsv1.Deployment]{}))
+			Expect(err).NotTo(HaveOccurred())
+
+			By("Starting the Manager")
+			go func() {
+				defer GinkgoRecover()
+				Expect(cm.Start(ctx)).NotTo(HaveOccurred())
+			}()
+
+			deployment := &appsv1.Deployment{
+				ObjectMeta: metav1.ObjectMeta{Name: "deployment-name"},
+				Spec: appsv1.DeploymentSpec{
+					Selector: &metav1.LabelSelector{
+						MatchLabels: map[string]string{"foo": "bar"},
+					},
+					Template: corev1.PodTemplateSpec{
+						ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{"foo": "bar"}},
+						Spec: corev1.PodSpec{
+							Containers: []corev1.Container{
+								{
+									Name:  "nginx",
+									Image: "nginx",
+								},
+							},
+						},
+					},
+				},
+			}
+
+			By("Invoking Reconciling")
+			deployment, err = clientset.AppsV1().Deployments("default").Create(ctx, deployment, metav1.CreateOptions{})
+			Expect(err).NotTo(HaveOccurred())
+
+			By("Validate event is published as expected")
+			evtWatcher, err := clientset.EventsV1().Events("default").Watch(ctx, metav1.ListOptions{})
+			Expect(err).NotTo(HaveOccurred())
+
+			resultEvent := <-evtWatcher.ResultChan()
+
+			Expect(resultEvent.Type).To(Equal(watch.Added))
+			evt, isEvent := resultEvent.Object.(*eventsv1.Event)
+			Expect(isEvent).To(BeTrue())
+
+			dpRef, err := ref.GetReference(scheme.Scheme, deployment)
+			Expect(err).NotTo(HaveOccurred())
+
+			Expect(evt.Regarding).To(Equal(*dpRef))
+			Expect(evt.Type).To(Equal(corev1.EventTypeNormal))
+			Expect(evt.Reason).To(Equal("test-reason"))
+			Expect(evt.Note).To(Equal("test-msg"))
+		})
+	})
 })

pkg/recorder/recorder.go

@@ -27,11 +27,11 @@ import (
 
 // Provider knows how to generate new event recorders with given name.
 type Provider interface {
-	// GetEventRecorder returns a EventRecorder with given name.
+	// GetEventRecorderFor returns an EventRecorder for the old events API.
 	//
 	// Deprecated: this uses the old events API and will be removed in a future release. Please use GetEventRecorder instead.
 	GetEventRecorderFor(name string) record.EventRecorder
-	// GetEventRecorder returns an EventRecorder for the old events API.
+	// GetEventRecorder returns a EventRecorder with given name.
 	// The old API is not 100% supported anymore, use the new one whenever possible.
 	GetEventRecorder(name string) events.EventRecorder
 }