Base migration to the new events API

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

Author: clebs

designs/move-cluster-specific-code-out-of-manager.md

@@ -61,8 +61,8 @@ type Cluster interface {
 	// GetCache returns a cache.Cache
 	GetCache() cache.Cache
 
-	// GetEventRecorderFor returns a new EventRecorder for the provided name
-	GetEventRecorderFor(name string) record.EventRecorder
+	//GetEventRecorder returns a new EventRecorder for the provided name
+	GetEventRecorder(name string) record.EventRecorder
 
 	// GetRESTMapper returns a RESTMapper
 	GetRESTMapper() meta.RESTMapper

pkg/cluster/cluster.go

@@ -26,7 +26,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
-	"k8s.io/client-go/tools/record"
+	"k8s.io/client-go/tools/events"
 
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -58,8 +58,8 @@ type Cluster interface {
 	// GetFieldIndexer returns a client.FieldIndexer configured with the client
 	GetFieldIndexer() client.FieldIndexer
 
-	// GetEventRecorderFor returns a new EventRecorder for the provided name
-	GetEventRecorderFor(name string) record.EventRecorder
+	// GetEventRecorder returns a new EventRecorder for the provided name
+	GetEventRecorder(name string) events.EventRecorder
 
 	// GetRESTMapper returns a RESTMapper
 	GetRESTMapper() meta.RESTMapper
@@ -126,16 +126,10 @@ 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 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
+	EventBroadcaster events.EventBroadcaster
 
 	// Dependency injection for testing
-	newRecorderProvider func(config *rest.Config, httpClient *http.Client, scheme *runtime.Scheme, logger logr.Logger, makeBroadcaster intrec.EventBroadcasterProducer) (*intrec.Provider, error)
+	newRecorderProvider func(config *rest.Config, httpClient *http.Client, scheme *runtime.Scheme, logger logr.Logger, broadcaster events.EventBroadcaster, stopWithProvider bool) (*intrec.Provider, error)
 }
 
 // Option can be used to manipulate Options.
@@ -228,7 +222,8 @@ 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.
-	recorderProvider, err := options.newRecorderProvider(config, options.HTTPClient, options.Scheme, options.Logger.WithName("events"), options.makeBroadcaster)
+	// 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)
 	if err != nil {
 		return nil, err
 	}
@@ -281,19 +276,6 @@ 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
-	if options.EventBroadcaster == nil {
-		// defer initialization to avoid leaking by default
-		options.makeBroadcaster = func() (record.EventBroadcaster, bool) {
-			return record.NewBroadcaster(), true
-		}
-	} else {
-		options.makeBroadcaster = func() (record.EventBroadcaster, bool) {
-			return options.EventBroadcaster, false
-		}
-	}
-
 	if options.Logger.GetSink() == nil {
 		options.Logger = logf.RuntimeLog.WithName("cluster")
 	}

pkg/cluster/cluster_test.go

@@ -40,7 +40,6 @@ var _ = Describe("cluster.Cluster", func() {
 			c, err := New(nil)
 			Expect(c).To(BeNil())
 			Expect(err.Error()).To(ContainSubstring("must specify Config"))
-
 		})
 
 		It("should return an error if it can't create a RestMapper", func() {
@@ -50,7 +49,6 @@ var _ = Describe("cluster.Cluster", func() {
 			})
 			Expect(c).To(BeNil())
 			Expect(err).To(Equal(expected))
-
 		})
 
 		It("should return an error it can't create a client.Client", func() {
@@ -96,7 +94,6 @@ var _ = Describe("cluster.Cluster", func() {
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("expected error"))
 		})
-
 	})
 
 	Describe("Start", func() {
@@ -160,7 +157,7 @@ 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.GetEventRecorderFor("test")).NotTo(BeNil())
+		Expect(c.GetEventRecorder("test")).NotTo(BeNil())
 	})
 	It("should provide a function to get the APIReader", func() {
 		c, err := New(cfg)

pkg/cluster/internal.go

@@ -24,7 +24,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/rest"
-	"k8s.io/client-go/tools/record"
+	"k8s.io/client-go/tools/events"
 
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -83,8 +83,8 @@ func (c *cluster) GetCache() cache.Cache {
 	return c.cache
 }
 
-func (c *cluster) GetEventRecorderFor(name string) record.EventRecorder {
-	return c.recorderProvider.GetEventRecorderFor(name)
+func (c *cluster) GetEventRecorder(name string) events.EventRecorder {
+	return c.recorderProvider.GetEventRecorder(name)
 }
 
 func (c *cluster) GetRESTMapper() meta.RESTMapper {

pkg/internal/recorder/recorder.go

@@ -23,17 +23,16 @@ import (
 	"sync"
 
 	"github.com/go-logr/logr"
-	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
-	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
+	eventsv1client "k8s.io/client-go/kubernetes/typed/events/v1"
 	"k8s.io/client-go/rest"
-	"k8s.io/client-go/tools/record"
+	"k8s.io/client-go/tools/events"
 )
 
 // 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).
-type EventBroadcasterProducer func() (caster record.EventBroadcaster, stopWithProvider bool)
+type EventBroadcasterProducer func() (caster events.EventBroadcaster, stopWithProvider bool)
 
 // Provider is a recorder.Provider that records events to the k8s API server
 // and to a logr Logger.
@@ -45,11 +44,11 @@ type Provider struct {
 	scheme *runtime.Scheme
 	// logger is the logger to use when logging diagnostic event info
 	logger          logr.Logger
-	evtClient       corev1client.EventInterface
+	evtClient       eventsv1client.EventsV1Interface
 	makeBroadcaster EventBroadcasterProducer
 
 	broadcasterOnce sync.Once
-	broadcaster     record.EventBroadcaster
+	broadcaster     events.EventBroadcaster
 	stopBroadcaster bool
 }
 
@@ -89,45 +88,48 @@ 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() record.EventBroadcaster {
+func (p *Provider) getBroadcaster() 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() {
-		broadcaster, stop := p.makeBroadcaster()
-		broadcaster.StartRecordingToSink(&corev1client.EventSinkImpl{Interface: p.evtClient})
-		broadcaster.StartEventWatcher(
-			func(e *corev1.Event) {
+		if p.broadcaster == nil {
+			p.broadcaster = events.NewBroadcaster(&events.EventSinkImpl{Interface: p.evtClient})
+		}
+		// TODO(clebs): figure out how to manage the context/channel that StartRecordingToSink needs inside the provider.
+		p.broadcaster.StartRecordingToSink(nil)
+
+		// TODO(clebs): figure out if we still need this and how the change would make sense.
+		p.broadcaster.StartEventWatcher(
+			func(e runtime.Object) {
 				p.logger.V(1).Info(e.Message, "type", e.Type, "object", e.InvolvedObject, "reason", e.Reason)
 			})
-		p.broadcaster = broadcaster
-		p.stopBroadcaster = stop
 	})
 
 	return p.broadcaster
 }
 
 // NewProvider create a new Provider instance.
-func NewProvider(config *rest.Config, httpClient *http.Client, scheme *runtime.Scheme, logger logr.Logger, makeBroadcaster EventBroadcasterProducer) (*Provider, error) {
+func NewProvider(config *rest.Config, httpClient *http.Client, scheme *runtime.Scheme, logger logr.Logger, broadcaster events.EventBroadcaster, stopWithProvider bool) (*Provider, error) {
 	if httpClient == nil {
 		panic("httpClient must not be nil")
 	}
 
-	corev1Client, err := corev1client.NewForConfigAndClient(config, httpClient)
+	eventsv1Client, err := eventsv1client.NewForConfigAndClient(config, httpClient)
 	if err != nil {
 		return nil, fmt.Errorf("failed to init client: %w", err)
 	}
 
-	p := &Provider{scheme: scheme, logger: logger, makeBroadcaster: makeBroadcaster, evtClient: corev1Client.Events("")}
+	p := &Provider{scheme: scheme, logger: logger, broadcaster: broadcaster, stopBroadcaster: stopWithProvider, evtClient: eventsv1Client}
 	return p, nil
 }
 
-// GetEventRecorderFor returns an event recorder that broadcasts to this provider's
+// GetEventRecorder 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 {
+func (p *Provider) GetEventRecorder(name string) events.EventRecorder {
 	return &lazyRecorder{
 		prov: p,
 		name: name,
@@ -141,41 +143,23 @@ type lazyRecorder struct {
 	name string
 
 	recOnce sync.Once
-	rec     record.EventRecorder
+	rec     events.EventRecorder
 }
 
 // ensureRecording ensures that a concrete recorder is populated for this recorder.
 func (l *lazyRecorder) ensureRecording() {
 	l.recOnce.Do(func() {
 		broadcaster := l.prov.getBroadcaster()
-		l.rec = broadcaster.NewRecorder(l.prov.scheme, corev1.EventSource{Component: l.name})
+		l.rec = broadcaster.NewRecorder(l.prov.scheme, l.name)
 	})
 }
 
-func (l *lazyRecorder) 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 *lazyRecorder) Eventf(object runtime.Object, eventtype, reason, messageFmt string, args ...interface{}) {
-	l.ensureRecording()
-
-	l.prov.lock.RLock()
-	if !l.prov.stopped {
-		l.rec.Eventf(object, eventtype, reason, messageFmt, args...)
-	}
-	l.prov.lock.RUnlock()
-}
-func (l *lazyRecorder) AnnotatedEventf(object runtime.Object, annotations map[string]string, eventtype, reason, messageFmt string, args ...interface{}) {
+func (l *lazyRecorder) Eventf(regarding runtime.Object, related runtime.Object, eventtype, reason, action, note string, args ...interface{}) {
 	l.ensureRecording()
 
 	l.prov.lock.RLock()
 	if !l.prov.stopped {
-		l.rec.AnnotatedEventf(object, annotations, eventtype, reason, messageFmt, args...)
+		l.rec.Eventf(regarding, related, eventtype, reason, action, note, args...)
 	}
 	l.prov.lock.RUnlock()
 }

pkg/internal/recorder/recorder_integration_test.go

@@ -43,7 +43,7 @@ var _ = Describe("recorder", func() {
 			Expect(err).NotTo(HaveOccurred())
 
 			By("Creating the Controller")
-			recorder := cm.GetEventRecorderFor("test-recorder")
+			recorder := cm.GetEventRecorder("test-recorder")
 			instance, err := controller.New("foo-controller", cm, controller.Options{
 				Reconciler: reconcile.Func(
 					func(ctx context.Context, request reconcile.Request) (reconcile.Result, error) {

pkg/internal/recorder/recorder_test.go

@@ -48,7 +48,7 @@ var _ = Describe("recorder.Provider", func() {
 			provider, err := recorder.NewProvider(cfg, httpClient, scheme.Scheme, logr.Discard(), makeBroadcaster)
 			Expect(err).NotTo(HaveOccurred())
 
-			recorder := provider.GetEventRecorderFor("test")
+			recorder := provider.GetEventRecorder("test")
 			Expect(recorder).NotTo(BeNil())
 		})
 	})

pkg/leaderelection/leader_election.go

@@ -126,9 +126,10 @@ func NewResourceLock(config *rest.Config, recorderProvider recorder.Provider, op
 		options.LeaderElectionID,
 		corev1Client,
 		coordinationClient,
+		// TODO(clebs): figure out how to solve this.
 		resourcelock.ResourceLockConfig{
 			Identity:      id,
-			EventRecorder: recorderProvider.GetEventRecorderFor(id),
+			EventRecorder: recorderProvider.GetEventRecorder(id),
 		},
 		options.LeaderLabels,
 	)

pkg/manager/internal.go

@@ -32,9 +32,9 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	kerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/events"
 	"k8s.io/client-go/tools/leaderelection"
 	"k8s.io/client-go/tools/leaderelection/resourcelock"
-	"k8s.io/client-go/tools/record"
 
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -255,8 +255,8 @@ func (cm *controllerManager) GetCache() cache.Cache {
 	return cm.cluster.GetCache()
 }
 
-func (cm *controllerManager) GetEventRecorderFor(name string) record.EventRecorder {
-	return cm.cluster.GetEventRecorderFor(name)
+func (cm *controllerManager) GetEventRecorder(name string) events.EventRecorder {
+	return cm.cluster.GetEventRecorder(name)
 }
 
 func (cm *controllerManager) GetRESTMapper() meta.RESTMapper {