Support both events APIs

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

Author: clebs

pkg/cluster/cluster.go

@@ -25,8 +25,10 @@ import (
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/kubernetes/scheme"
+	eventsv1client "k8s.io/client-go/kubernetes/typed/events/v1"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/events"
+	"k8s.io/client-go/tools/record"
 
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -126,10 +128,16 @@ type Options struct {
 	//
 	// Deprecated: using this may cause goroutine leaks if the lifetime of your manager or controllers
 	// is shorter than the lifetime of your process.
-	EventBroadcaster events.EventBroadcaster
+	EventBroadcaster record.EventBroadcaster
+
+	// makeBroadcaster allows deferring the creation of the broadcaster to
+	// avoid leaking goroutines if we never call Start on this manager.  It also
+	// returns whether or not this is a "owned" broadcaster, and as such should be
+	// stopped with the manager.
+	makeBroadcaster intrec.EventBroadcasterProducer
 
 	// Dependency injection for testing
-	newRecorderProvider func(config *rest.Config, httpClient *http.Client, scheme *runtime.Scheme, logger logr.Logger, broadcaster events.EventBroadcaster, stopWithProvider bool) (*intrec.Provider, error)
+	newRecorderProvider func(config *rest.Config, httpClient *http.Client, scheme *runtime.Scheme, logger logr.Logger, makeBroadcaster intrec.EventBroadcasterProducer) (*intrec.Provider, error)
 }
 
 // Option can be used to manipulate Options.
@@ -223,7 +231,7 @@ func New(config *rest.Config, opts ...Option) (Cluster, error) {
 	// 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.EventBroadcaster, options.EventBroadcaster == nil)
+	recorderProvider, err := options.newRecorderProvider(config, options.HTTPClient, options.Scheme, options.Logger.WithName("events"), options.makeBroadcaster)
 	if err != nil {
 		return nil, err
 	}
@@ -276,6 +284,27 @@ func setOptionsDefaults(options Options, config *rest.Config) (Options, error) {
 		options.newRecorderProvider = intrec.NewProvider
 	}
 
+	// This is duplicated with pkg/manager, we need it here to provide
+	// the user with an EventBroadcaster and there for the Leader election
+	evtCl, err := eventsv1client.NewForConfigAndClient(config, options.HTTPClient)
+	if err != nil {
+		return options, err
+	}
+
+	// This is duplicated with pkg/manager, we need it here to provide
+	// the user with an EventBroadcaster and there for the Leader election
+	if options.EventBroadcaster == nil {
+		// defer initialization to avoid leaking by default
+		options.makeBroadcaster = func() (record.EventBroadcaster, events.EventBroadcaster, bool) {
+			return record.NewBroadcaster(), events.NewBroadcaster(&events.EventSinkImpl{Interface: evtCl}), true
+		}
+	} else {
+		// keep supporting the options.EventBroadcaster in the old API, but do not introduce it for the new one.
+		options.makeBroadcaster = func() (record.EventBroadcaster, events.EventBroadcaster, bool) {
+			return options.EventBroadcaster, events.NewBroadcaster(&events.EventSinkImpl{Interface: evtCl}), false
+		}
+	}
+
 	if options.Logger.GetSink() == nil {
 		options.Logger = logf.RuntimeLog.WithName("cluster")
 	}

pkg/cluster/cluster_test.go

@@ -29,7 +29,6 @@ import (
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/rest"
-	"k8s.io/client-go/tools/events"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	intrec "sigs.k8s.io/controller-runtime/pkg/internal/recorder"
@@ -87,7 +86,7 @@ var _ = Describe("cluster.Cluster", func() {
 
 		It("should return an error it can't create a recorder.Provider", func() {
 			c, err := New(cfg, func(o *Options) {
-				o.newRecorderProvider = func(_ *rest.Config, _ *http.Client, _ *runtime.Scheme, _ logr.Logger, _ events.EventBroadcaster, _ bool) (*intrec.Provider, error) {
+				o.newRecorderProvider = func(_ *rest.Config, _ *http.Client, _ *runtime.Scheme, _ logr.Logger, _ intrec.EventBroadcasterProducer) (*intrec.Provider, error) {
 					return nil, fmt.Errorf("expected error")
 				}
 			})

pkg/internal/recorder/recorder.go

@@ -23,13 +23,20 @@ import (
 	"sync"
 
 	"github.com/go-logr/logr"
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
-	eventsv1client "k8s.io/client-go/kubernetes/typed/events/v1"
+	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/events"
 	"k8s.io/client-go/tools/record"
 )
 
+// 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
+type EventBroadcasterProducer func() (deprecatedCaster record.EventBroadcaster, caster events.EventBroadcaster, stopWithProvider bool)
+
 // Provider is a recorder.Provider that records events to the k8s API server
 // and to a logr Logger.
 type Provider struct {
@@ -39,12 +46,15 @@ type Provider struct {
 	// scheme to specify when creating a recorder
 	scheme *runtime.Scheme
 	// logger is the logger to use when logging diagnostic event info
-	logger    logr.Logger
-	evtClient eventsv1client.EventsV1Interface
+	logger          logr.Logger
+	evtClient       corev1client.EventInterface
+	makeBroadcaster EventBroadcasterProducer
 
 	broadcasterOnce sync.Once
 	broadcaster     events.EventBroadcaster
-	stopBroadcaster bool
+	// Deprecated: will be removed in a future release. Use the broadcaster above instead.
+	deprecatedBroadcaster record.EventBroadcaster
+	stopBroadcaster       bool
 }
 
 // NB(directxman12): this manually implements Stop instead of Being a runnable because we need to
@@ -65,10 +75,11 @@ func (p *Provider) Stop(shutdownCtx context.Context) {
 		// almost certainly already been started (e.g. by leader election).  We
 		// need to invoke this to ensure that we don't inadvertently race with
 		// an invocation of getBroadcaster.
-		broadcaster := p.getBroadcaster()
+		deprecatedBroadcaster, broadcaster := p.getBroadcaster()
 		if p.stopBroadcaster {
 			p.lock.Lock()
 			broadcaster.Shutdown()
+			deprecatedBroadcaster.Shutdown()
 			p.stopped = true
 			p.lock.Unlock()
 		}
@@ -83,47 +94,52 @@ func (p *Provider) Stop(shutdownCtx context.Context) {
 
 // getBroadcaster ensures that a broadcaster is started for this
 // provider, and returns it.  It's threadsafe.
-func (p *Provider) getBroadcaster() events.EventBroadcaster {
+func (p *Provider) getBroadcaster() (record.EventBroadcaster, events.EventBroadcaster) {
 	// NB(directxman12): this can technically still leak if something calls
 	// "getBroadcaster" (i.e. Emits an Event) but never calls Start, but if we
 	// create the broadcaster in start, we could race with other things that
 	// are started at the same time & want to emit events.  The alternative is
 	// silently swallowing events and more locking, but that seems suboptimal.
 
 	p.broadcasterOnce.Do(func() {
-		if p.broadcaster == nil {
-			p.broadcaster = events.NewBroadcaster(&events.EventSinkImpl{Interface: p.evtClient})
-		}
+		p.deprecatedBroadcaster, p.broadcaster, p.stopBroadcaster = p.makeBroadcaster()
+
+		// init old 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
 		// TODO(clebs): figure out how to manage the context/channel that StartRecordingToSink needs inside the provider.
 		_ = p.broadcaster.StartRecordingToSinkWithContext(context.TODO())
 	})
 
-	return p.broadcaster
+	return p.deprecatedBroadcaster, p.broadcaster
 }
 
 // NewProvider create a new Provider instance.
-func NewProvider(config *rest.Config, httpClient *http.Client, scheme *runtime.Scheme, logger logr.Logger, broadcaster events.EventBroadcaster, stopWithProvider bool) (*Provider, error) {
+func NewProvider(config *rest.Config, httpClient *http.Client, scheme *runtime.Scheme, logger logr.Logger, makeBroadcaster EventBroadcasterProducer) (*Provider, error) {
 	if httpClient == nil {
 		panic("httpClient must not be nil")
 	}
 
-	eventsv1Client, err := eventsv1client.NewForConfigAndClient(config, httpClient)
+	corev1Client, err := corev1client.NewForConfigAndClient(config, httpClient)
 	if err != nil {
 		return nil, fmt.Errorf("failed to init client: %w", err)
 	}
 
-	p := &Provider{scheme: scheme, logger: logger, broadcaster: broadcaster, stopBroadcaster: stopWithProvider, evtClient: eventsv1Client}
+	p := &Provider{scheme: scheme, logger: logger, makeBroadcaster: makeBroadcaster, evtClient: corev1Client.Events("")}
 	return p, nil
 }
 
 // GetEventRecorderFor returns an event recorder that broadcasts to this provider's
 // broadcaster.  All events will be associated with a component of the given name.
 func (p *Provider) GetEventRecorderFor(name string) record.EventRecorder {
-	return &oldRecorder{
-		newRecorder: &lazyRecorder{
-			prov: p,
-			name: name,
-		},
+	return &deprecatedRecorder{
+		prov: p,
+		name: name,
 	}
 }
 
@@ -149,7 +165,7 @@ type lazyRecorder struct {
 // ensureRecording ensures that a concrete recorder is populated for this recorder.
 func (l *lazyRecorder) ensureRecording() {
 	l.recOnce.Do(func() {
-		broadcaster := l.prov.getBroadcaster()
+		_, broadcaster := l.prov.getBroadcaster()
 		l.rec = broadcaster.NewRecorder(l.prov.scheme, l.name)
 	})
 }
@@ -164,21 +180,50 @@ func (l *lazyRecorder) Eventf(regarding runtime.Object, related runtime.Object,
 	l.prov.lock.RUnlock()
 }
 
-// oldRecorder is a wrapper around the events.EventRecorder that implements the old record.EventRecorder API.
-// This is a temporary solution to support both the old and new events APIs without duplicating everything.
-// Internally it calls the new events API from the old API funcs and no longer supported parameters are ignored (e.g. annotations).
-type oldRecorder struct {
-	newRecorder *lazyRecorder
+// deprecatedRecorder implements the old events API during the tranisiton and will be removed in a future release.
+// Deprecated: will be removed in a future release.
+type deprecatedRecorder struct {
+	prov *Provider
+	name string
+
+	recOnce sync.Once
+	rec     record.EventRecorder
 }
 
-func (l *oldRecorder) Event(object runtime.Object, eventtype, reason, message string) {
-	l.newRecorder.Eventf(object, nil, eventtype, reason, "no action", message)
+// ensureRecording ensures that a concrete recorder is populated for this recorder.
+func (l *deprecatedRecorder) ensureRecording() {
+	l.recOnce.Do(func() {
+		deprecatedBroadcaster, _ := l.prov.getBroadcaster()
+		l.rec = deprecatedBroadcaster.NewRecorder(l.prov.scheme, corev1.EventSource{Component: l.name})
+	})
 }
 
-func (l *oldRecorder) Eventf(object runtime.Object, eventtype, reason, messageFmt string, args ...any) {
-	l.newRecorder.Eventf(object, nil, eventtype, reason, "no action", messageFmt, args...)
+func (l *deprecatedRecorder) Event(object runtime.Object, eventtype, reason, message string) {
+	l.ensureRecording()
+
+	l.prov.lock.RLock()
+	if !l.prov.stopped {
+		l.rec.Event(object, eventtype, reason, message)
+	}
+	l.prov.lock.RUnlock()
+}
+
+func (l *deprecatedRecorder) Eventf(object runtime.Object, eventtype, reason, messageFmt string, args ...any) {
+	l.ensureRecording()
+
+	l.prov.lock.RLock()
+	if !l.prov.stopped {
+		l.rec.Eventf(object, eventtype, reason, messageFmt, args...)
+	}
+	l.prov.lock.RUnlock()
 }
 
-func (l *oldRecorder) AnnotatedEventf(object runtime.Object, annotations map[string]string, eventtype, reason, messageFmt string, args ...any) {
-	l.newRecorder.Eventf(object, nil, eventtype, reason, "no action", messageFmt, args...)
+func (l *deprecatedRecorder) AnnotatedEventf(object runtime.Object, annotations map[string]string, eventtype, reason, messageFmt string, args ...any) {
+	l.ensureRecording()
+
+	l.prov.lock.RLock()
+	if !l.prov.stopped {
+		l.rec.AnnotatedEventf(object, annotations, eventtype, reason, messageFmt, args...)
+	}
+	l.prov.lock.RUnlock()
 }

pkg/internal/recorder/recorder_test.go

@@ -21,13 +21,23 @@ import (
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	"k8s.io/client-go/kubernetes/scheme"
+	eventsv1client "k8s.io/client-go/kubernetes/typed/events/v1"
+	"k8s.io/client-go/tools/events"
+	"k8s.io/client-go/tools/record"
 	"sigs.k8s.io/controller-runtime/pkg/internal/recorder"
 )
 
 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(), nil, true)
+			provider, err := recorder.NewProvider(cfg, httpClient, scheme.Scheme, logr.Discard(), makeBroadcaster)
 			Expect(provider).NotTo(BeNil())
 			Expect(err).NotTo(HaveOccurred())
 		})
@@ -36,14 +46,14 @@ var _ = Describe("recorder.Provider", func() {
 			// Invalid the config
 			cfg1 := *cfg
 			cfg1.Host = "invalid host"
-			_, err := recorder.NewProvider(&cfg1, httpClient, scheme.Scheme, logr.Discard(), nil, true)
+			_, 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(), nil, true)
+			provider, err := recorder.NewProvider(cfg, httpClient, scheme.Scheme, logr.Discard(), makeBroadcaster)
 			Expect(err).NotTo(HaveOccurred())
 
 			recorder := provider.GetEventRecorderFor("test")