⚠️ Migration to the new events API #3262
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -24,16 +24,19 @@ 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" | ||
| "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 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 | ||
| // and to a logr Logger. | ||
|
|
@@ -48,9 +51,13 @@ type Provider struct { | |
| evtClient corev1client.EventInterface | ||
| makeBroadcaster EventBroadcasterProducer | ||
|
|
||
| broadcasterOnce sync.Once | ||
| broadcaster events.EventBroadcaster | ||
| cancelSinkRecordingFunc context.CancelFunc | ||
| stopWatcherFunc func() | ||
| // 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 | ||
|
|
@@ -71,10 +78,13 @@ 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. | ||
| deprecatedBroadcaster, broadcaster := p.getBroadcaster() | ||
| if p.stopBroadcaster { | ||
| p.lock.Lock() | ||
| broadcaster.Shutdown() | ||
| p.cancelSinkRecordingFunc() | ||
| p.stopWatcherFunc() | ||
| deprecatedBroadcaster.Shutdown() | ||
| p.stopped = true | ||
| p.lock.Unlock() | ||
| } | ||
|
|
@@ -89,25 +99,45 @@ 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, 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() { | ||
| p.deprecatedBroadcaster, p.broadcaster, p.stopBroadcaster = p.makeBroadcaster() | ||
|
|
||
| // 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 | ||
| ctx, cancel := context.WithCancel(context.Background()) | ||
| p.cancelSinkRecordingFunc = cancel | ||
| if err := p.broadcaster.StartRecordingToSinkWithContext(ctx); err != nil { | ||
| p.logger.Error(err, "error starting recording for broadcaster") | ||
|
There was a problem hiding this comment. Sorry for the back and forth on this. I thought a bit more about it. Can you please check if we can return the error here (and below) I think otherwise a controller just silently won't produce events if this fails and that seems bad There was a problem hiding this comment. I checked this out and we can propagate the errors by using a closure on the
We could probably refactor the provider to move the broadcaster initialisation to There was a problem hiding this comment. I'll take another look next week. There was a problem hiding this comment. Fun :). Apparently when designing this Events API it was not a concern that there is no way to return errors when events cannot be created (It's not just about the broadcaster creation here, also later when we actually try to send events) There was a problem hiding this comment. @alvaroaleman What do you think? Apparently event publishing is entirely best effort and apart from a few logs nobody would ever notice if it doesn't work (not specific to the current PR, in general). I personally think that is pretty bad, but I don't see a reasonable way to solve it without building our own thing directly on top of client-go. On the client-go level we get back errors if sending Events fails, but I think there is no way with Broadcaster etc. So I think I would keep the current implementation on this PR For these two error cases here. Looks like with the current implementation they can only fail if we hit this error "broadcaster already stopped" in pkg/watch.Broadcaster.Watch. That shouldn't really happen I guess There was a problem hiding this comment.
Yeah I was actually aware of that. I think an error is logged somewhere in client-go, at least when this fails due to permission issues. I do think this is somewhat okay insofar as that almost everywhere the error handling for this would be to just log it There was a problem hiding this comment. Alrighty. Let's go with that for now |
||
| return | ||
| } | ||
|
|
||
| stopWatcher, err := 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) | ||
| } | ||
| }) | ||
| if err != nil { | ||
| p.logger.Error(err, "error starting event watcher for broadcaster") | ||
| } | ||
|
|
||
| p.stopWatcherFunc = stopWatcher | ||
| }) | ||
|
|
||
| return p.deprecatedBroadcaster, p.broadcaster | ||
| } | ||
|
|
||
| // NewProvider create a new Provider instance. | ||
|
|
@@ -128,6 +158,15 @@ func NewProvider(config *rest.Config, httpClient *http.Client, scheme *runtime.S | |
| // 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 &deprecatedRecorder{ | ||
| prov: p, | ||
| name: name, | ||
| } | ||
| } | ||
|
|
||
| // 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) GetEventRecorder(name string) events.EventRecorder { | ||
| return &lazyRecorder{ | ||
| prov: p, | ||
| name: name, | ||
|
|
@@ -141,18 +180,47 @@ type lazyRecorder struct { | |
| name string | ||
|
|
||
| recOnce sync.Once | ||
| 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, l.name) | ||
| }) | ||
| } | ||
|
|
||
| func (l *lazyRecorder) Eventf(regarding runtime.Object, related runtime.Object, eventtype, reason, action, note string, args ...any) { | ||
| l.ensureRecording() | ||
|
|
||
| l.prov.lock.RLock() | ||
| if !l.prov.stopped { | ||
| l.rec.Eventf(regarding, related, eventtype, reason, action, note, args...) | ||
| } | ||
| l.prov.lock.RUnlock() | ||
| } | ||
|
|
||
| // 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 | ||
| } | ||
|
|
||
| // 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 *deprecatedRecorder) Event(object runtime.Object, eventtype, reason, message string) { | ||
| l.ensureRecording() | ||
|
|
||
| l.prov.lock.RLock() | ||
|
|
@@ -161,7 +229,8 @@ func (l *lazyRecorder) Event(object runtime.Object, eventtype, reason, message s | |
| } | ||
| l.prov.lock.RUnlock() | ||
| } | ||
|
|
||
| func (l *deprecatedRecorder) Eventf(object runtime.Object, eventtype, reason, messageFmt string, args ...any) { | ||
| l.ensureRecording() | ||
|
|
||
| l.prov.lock.RLock() | ||
|
|
@@ -170,7 +239,8 @@ func (l *lazyRecorder) Eventf(object runtime.Object, eventtype, reason, messageF | |
| } | ||
| l.prov.lock.RUnlock() | ||
| } | ||
|
|
||
| func (l *deprecatedRecorder) AnnotatedEventf(object runtime.Object, annotations map[string]string, eventtype, reason, messageFmt string, args ...any) { | ||
| l.ensureRecording() | ||
|
|
||
| l.prov.lock.RLock() | ||
|
|
||
Uh oh!
There was an error while loading. Please reload this page.