github/client-go
1
0
Fork 0
mirror of https://github.com/kubernetes/client-go.git synced 2025-07-20 18:10:24 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #120729 from pohly/events-context

Browse Source 
k8s.io/client-go/tools/[events|record]: support context

Kubernetes-commit: f936f69cf994146650c287a95fa436d1d7352835
This commit is contained in:
Kubernetes Publisher 2023-10-04 15:45:16 +02:00
commit 0a782d6adb
10 changed files with 352 additions and 136 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

121
tools/events/event_broadcaster.go
View File

@ -81,27 +81,27 @@ type EventSinkImpl struct {
} }
// Create takes the representation of a event and creates it. Returns the server's representation of the event, and an error, if there is any. // Create takes the representation of a event and creates it. Returns the server's representation of the event, and an error, if there is any.
func (e *EventSinkImpl) Create(event *eventsv1.Event) (*eventsv1.Event, error) { func (e *EventSinkImpl) Create(ctx context.Context, event *eventsv1.Event) (*eventsv1.Event, error) {
if event.Namespace == "" { if event.Namespace == "" {
return nil, fmt.Errorf("can't create an event with empty namespace") return nil, fmt.Errorf("can't create an event with empty namespace")
} }
return e.Interface.Events(event.Namespace).Create(context.TODO(), event, metav1.CreateOptions{}) return e.Interface.Events(event.Namespace).Create(ctx, event, metav1.CreateOptions{})
} }
// Update takes the representation of a event and updates it. Returns the server's representation of the event, and an error, if there is any. // Update takes the representation of a event and updates it. Returns the server's representation of the event, and an error, if there is any.
func (e *EventSinkImpl) Update(event *eventsv1.Event) (*eventsv1.Event, error) { func (e *EventSinkImpl) Update(ctx context.Context, event *eventsv1.Event) (*eventsv1.Event, error) {
if event.Namespace == "" { if event.Namespace == "" {
return nil, fmt.Errorf("can't update an event with empty namespace") return nil, fmt.Errorf("can't update an event with empty namespace")
} }
return e.Interface.Events(event.Namespace).Update(context.TODO(), event, metav1.UpdateOptions{}) return e.Interface.Events(event.Namespace).Update(ctx, event, metav1.UpdateOptions{})
} }
// Patch applies the patch and returns the patched event, and an error, if there is any. // Patch applies the patch and returns the patched event, and an error, if there is any.
func (e *EventSinkImpl) Patch(event *eventsv1.Event, data []byte) (*eventsv1.Event, error) { func (e *EventSinkImpl) Patch(ctx context.Context, event *eventsv1.Event, data []byte) (*eventsv1.Event, error) {
if event.Namespace == "" { if event.Namespace == "" {
return nil, fmt.Errorf("can't patch an event with empty namespace") return nil, fmt.Errorf("can't patch an event with empty namespace")
} }
return e.Interface.Events(event.Namespace).Patch(context.TODO(), event.Name, types.StrategicMergePatchType, data, metav1.PatchOptions{}) return e.Interface.Events(event.Namespace).Patch(ctx, event.Name, types.StrategicMergePatchType, data, metav1.PatchOptions{})
} }
// NewBroadcaster Creates a new event broadcaster. // NewBroadcaster Creates a new event broadcaster.
@ -124,13 +124,13 @@ func (e *eventBroadcasterImpl) Shutdown() {
} }
// refreshExistingEventSeries refresh events TTL // refreshExistingEventSeries refresh events TTL
func (e *eventBroadcasterImpl) refreshExistingEventSeries() { func (e *eventBroadcasterImpl) refreshExistingEventSeries(ctx context.Context) {
// TODO: Investigate whether lock contention won't be a problem // TODO: Investigate whether lock contention won't be a problem
e.mu.Lock() e.mu.Lock()
defer e.mu.Unlock() defer e.mu.Unlock()
for isomorphicKey, event := range e.eventCache { for isomorphicKey, event := range e.eventCache {
if event.Series != nil { if event.Series != nil {
if recordedEvent, retry := recordEvent(e.sink, event); !retry { if recordedEvent, retry := recordEvent(ctx, e.sink, event); !retry {
if recordedEvent != nil { if recordedEvent != nil {
e.eventCache[isomorphicKey] = recordedEvent e.eventCache[isomorphicKey] = recordedEvent
} }
@ -142,7 +142,7 @@ func (e *eventBroadcasterImpl) refreshExistingEventSeries() {
// finishSeries checks if a series has ended and either: // finishSeries checks if a series has ended and either:
// - write final count to the apiserver // - write final count to the apiserver
// - delete a singleton event (i.e. series field is nil) from the cache // - delete a singleton event (i.e. series field is nil) from the cache
func (e *eventBroadcasterImpl) finishSeries() { func (e *eventBroadcasterImpl) finishSeries(ctx context.Context) {
// TODO: Investigate whether lock contention won't be a problem // TODO: Investigate whether lock contention won't be a problem
e.mu.Lock() e.mu.Lock()
defer e.mu.Unlock() defer e.mu.Unlock()
@ -150,7 +150,7 @@ func (e *eventBroadcasterImpl) finishSeries() {
eventSerie := event.Series eventSerie := event.Series
if eventSerie != nil { if eventSerie != nil {
if eventSerie.LastObservedTime.Time.Before(time.Now().Add(-finishTime)) { if eventSerie.LastObservedTime.Time.Before(time.Now().Add(-finishTime)) {
if _, retry := recordEvent(e.sink, event); !retry { if _, retry := recordEvent(ctx, e.sink, event); !retry {
delete(e.eventCache, isomorphicKey) delete(e.eventCache, isomorphicKey)
} }
} }
@ -161,13 +161,13 @@ func (e *eventBroadcasterImpl) finishSeries() {
} }
// NewRecorder returns an EventRecorder that records events with the given event source. // NewRecorder returns an EventRecorder that records events with the given event source.
func (e *eventBroadcasterImpl) NewRecorder(scheme *runtime.Scheme, reportingController string) EventRecorder { func (e *eventBroadcasterImpl) NewRecorder(scheme *runtime.Scheme, reportingController string) EventRecorderLogger {
hostname, _ := os.Hostname() hostname, _ := os.Hostname()
reportingInstance := reportingController + "-" + hostname reportingInstance := reportingController + "-" + hostname
return &recorderImpl{scheme, reportingController, reportingInstance, e.Broadcaster, clock.RealClock{}} return &recorderImplLogger{recorderImpl: &recorderImpl{scheme, reportingController, reportingInstance, e.Broadcaster, clock.RealClock{}}, logger: klog.Background()}
} }
func (e *eventBroadcasterImpl) recordToSink(event *eventsv1.Event, clock clock.Clock) { func (e *eventBroadcasterImpl) recordToSink(ctx context.Context, event *eventsv1.Event, clock clock.Clock) {
// Make a copy before modification, because there could be multiple listeners. // Make a copy before modification, because there could be multiple listeners.
eventCopy := event.DeepCopy() eventCopy := event.DeepCopy()
go func() { go func() {
@ -197,7 +197,7 @@ func (e *eventBroadcasterImpl) recordToSink(event *eventsv1.Event, clock clock.C
}() }()
if evToRecord != nil { if evToRecord != nil {
// TODO: Add a metric counting the number of recording attempts // TODO: Add a metric counting the number of recording attempts
e.attemptRecording(evToRecord) e.attemptRecording(ctx, evToRecord)
// We don't want the new recorded Event to be reflected in the // We don't want the new recorded Event to be reflected in the
// client's cache because server-side mutations could mess with the // client's cache because server-side mutations could mess with the
// aggregation mechanism used by the client. // aggregation mechanism used by the client.
@ -205,40 +205,45 @@ func (e *eventBroadcasterImpl) recordToSink(event *eventsv1.Event, clock clock.C
}() }()
} }
func (e *eventBroadcasterImpl) attemptRecording(event *eventsv1.Event) *eventsv1.Event { func (e *eventBroadcasterImpl) attemptRecording(ctx context.Context, event *eventsv1.Event) {
tries := 0 tries := 0
for { for {
if recordedEvent, retry := recordEvent(e.sink, event); !retry { if _, retry := recordEvent(ctx, e.sink, event); !retry {
return recordedEvent return
} }
tries++ tries++
if tries >= maxTriesPerEvent { if tries >= maxTriesPerEvent {
klog.Errorf("Unable to write event '%#v' (retry limit exceeded!)", event) klog.FromContext(ctx).Error(nil, "Unable to write event (retry limit exceeded!)", "event", event)
return nil return
} }
// Randomize sleep so that various clients won't all be // Randomize sleep so that various clients won't all be
// synced up if the master goes down. // synced up if the master goes down. Give up when
time.Sleep(wait.Jitter(e.sleepDuration, 0.25)) // the context is canceled.
select {
case <-ctx.Done():
return
case <-time.After(wait.Jitter(e.sleepDuration, 0.25)):
}
} }
} }
func recordEvent(sink EventSink, event *eventsv1.Event) (*eventsv1.Event, bool) { func recordEvent(ctx context.Context, sink EventSink, event *eventsv1.Event) (*eventsv1.Event, bool) {
var newEvent *eventsv1.Event var newEvent *eventsv1.Event
var err error var err error
isEventSeries := event.Series != nil isEventSeries := event.Series != nil
if isEventSeries { if isEventSeries {
patch, patchBytesErr := createPatchBytesForSeries(event) patch, patchBytesErr := createPatchBytesForSeries(event)
if patchBytesErr != nil { if patchBytesErr != nil {
klog.Errorf("Unable to calculate diff, no merge is possible: %v", patchBytesErr) klog.FromContext(ctx).Error(patchBytesErr, "Unable to calculate diff, no merge is possible")
return nil, false return nil, false
} }
newEvent, err = sink.Patch(event, patch) newEvent, err = sink.Patch(ctx, event, patch)
} }
// Update can fail because the event may have been removed and it no longer exists. // Update can fail because the event may have been removed and it no longer exists.
if !isEventSeries || (isEventSeries && util.IsKeyNotFoundError(err)) { if !isEventSeries || (isEventSeries && util.IsKeyNotFoundError(err)) {
// Making sure that ResourceVersion is empty on creation // Making sure that ResourceVersion is empty on creation
event.ResourceVersion = "" event.ResourceVersion = ""
newEvent, err = sink.Create(event) newEvent, err = sink.Create(ctx, event)
} }
if err == nil { if err == nil {
return newEvent, false return newEvent, false
@ -248,7 +253,7 @@ func recordEvent(sink EventSink, event *eventsv1.Event) (*eventsv1.Event, bool)
switch err.(type) { switch err.(type) {
case *restclient.RequestConstructionError: case *restclient.RequestConstructionError:
// We will construct the request the same next time, so don't keep trying. // We will construct the request the same next time, so don't keep trying.
klog.Errorf("Unable to construct event '%#v': '%v' (will not retry!)", event, err) klog.FromContext(ctx).Error(err, "Unable to construct event (will not retry!)", "event", event)
return nil, false return nil, false
case *errors.StatusError: case *errors.StatusError:
if errors.IsAlreadyExists(err) { if errors.IsAlreadyExists(err) {
@ -260,9 +265,9 @@ func recordEvent(sink EventSink, event *eventsv1.Event) (*eventsv1.Event, bool)
if isEventSeries { if isEventSeries {
return nil, true return nil, true
} }
klog.V(5).Infof("Server rejected event '%#v': '%v' (will not retry!)", event, err) klog.FromContext(ctx).V(5).Info("Server rejected event (will not retry!)", "event", event, "err", err)
} else { } else {
klog.Errorf("Server rejected event '%#v': '%v' (will not retry!)", event, err) klog.FromContext(ctx).Error(err, "Server rejected event (will not retry!)", "event", event)
} }
return nil, false return nil, false
case *errors.UnexpectedObjectError: case *errors.UnexpectedObjectError:
@ -271,7 +276,7 @@ func recordEvent(sink EventSink, event *eventsv1.Event) (*eventsv1.Event, bool)
default: default:
// This case includes actual http transport errors. Go ahead and retry. // This case includes actual http transport errors. Go ahead and retry.
} }
klog.Errorf("Unable to write event: '%v' (may retry after sleeping)", err) klog.FromContext(ctx).Error(err, "Unable to write event (may retry after sleeping)")
return nil, true return nil, true
} }
@ -307,29 +312,38 @@ func getKey(event *eventsv1.Event) eventKey {
// StartStructuredLogging starts sending events received from this EventBroadcaster to the structured logging function. // StartStructuredLogging starts sending events received from this EventBroadcaster to the structured logging function.
// The return value can be ignored or used to stop recording, if desired. // The return value can be ignored or used to stop recording, if desired.
// TODO: this function should also return an error. // TODO: this function should also return an error.
//
// Deprecated: use StartLogging instead.
func (e *eventBroadcasterImpl) StartStructuredLogging(verbosity klog.Level) func() { func (e *eventBroadcasterImpl) StartStructuredLogging(verbosity klog.Level) func() {
stopWatcher, err := e.StartEventWatcher( logger := klog.Background().V(int(verbosity))
func(obj runtime.Object) { stopWatcher, err := e.StartLogging(logger)
event, ok := obj.(*eventsv1.Event)
if !ok {
klog.Errorf("unexpected type, expected eventsv1.Event")
return
}
klog.V(verbosity).InfoS("Event occurred", "object", klog.KRef(event.Regarding.Namespace, event.Regarding.Name), "kind", event.Regarding.Kind, "apiVersion", event.Regarding.APIVersion, "type", event.Type, "reason", event.Reason, "action", event.Action, "note", event.Note)
})
if err != nil { if err != nil {
klog.Errorf("failed to start event watcher: '%v'", err) logger.Error(err, "Failed to start event watcher")
return func() {} return func() {}
} }
return stopWatcher return stopWatcher
} }
// StartLogging starts sending events received from this EventBroadcaster to the structured logger.
// To adjust verbosity, use the logger's V method (i.e. pass `logger.V(3)` instead of `logger`).
// The returned function can be ignored or used to stop recording, if desired.
func (e *eventBroadcasterImpl) StartLogging(logger klog.Logger) (func(), error) {
return e.StartEventWatcher(
func(obj runtime.Object) {
event, ok := obj.(*eventsv1.Event)
if !ok {
logger.Error(nil, "unexpected type, expected eventsv1.Event")
return
}
logger.Info("Event occurred", "object", klog.KRef(event.Regarding.Namespace, event.Regarding.Name), "kind", event.Regarding.Kind, "apiVersion", event.Regarding.APIVersion, "type", event.Type, "reason", event.Reason, "action", event.Action, "note", event.Note)
})
}
// StartEventWatcher starts sending events received from this EventBroadcaster to the given event handler function. // StartEventWatcher starts sending events received from this EventBroadcaster to the given event handler function.
// The return value is used to stop recording // The return value is used to stop recording
func (e *eventBroadcasterImpl) StartEventWatcher(eventHandler func(event runtime.Object)) (func(), error) { func (e *eventBroadcasterImpl) StartEventWatcher(eventHandler func(event runtime.Object)) (func(), error) {
watcher, err := e.Watch() watcher, err := e.Watch()
if err != nil { if err != nil {
klog.Errorf("Unable start event watcher: '%v' (will not retry!)", err)
return nil, err return nil, err
} }
go func() { go func() {
@ -345,37 +359,42 @@ func (e *eventBroadcasterImpl) StartEventWatcher(eventHandler func(event runtime
return watcher.Stop, nil return watcher.Stop, nil
} }
func (e *eventBroadcasterImpl) startRecordingEvents(stopCh <-chan struct{}) error { func (e *eventBroadcasterImpl) startRecordingEvents(ctx context.Context) error {
eventHandler := func(obj runtime.Object) { eventHandler := func(obj runtime.Object) {
event, ok := obj.(*eventsv1.Event) event, ok := obj.(*eventsv1.Event)
if !ok { if !ok {
klog.Errorf("unexpected type, expected eventsv1.Event") klog.FromContext(ctx).Error(nil, "unexpected type, expected eventsv1.Event")
return return
} }
e.recordToSink(event, clock.RealClock{}) e.recordToSink(ctx, event, clock.RealClock{})
} }
stopWatcher, err := e.StartEventWatcher(eventHandler) stopWatcher, err := e.StartEventWatcher(eventHandler)
if err != nil { if err != nil {
return err return err
} }
go func() { go func() {
<-stopCh <-ctx.Done()
stopWatcher() stopWatcher()
}() }()
return nil return nil
} }
// StartRecordingToSink starts sending events received from the specified eventBroadcaster to the given sink. // StartRecordingToSink starts sending events received from the specified eventBroadcaster to the given sink.
// Deprecated: use StartRecordingToSinkWithContext instead.
func (e *eventBroadcasterImpl) StartRecordingToSink(stopCh <-chan struct{}) { func (e *eventBroadcasterImpl) StartRecordingToSink(stopCh <-chan struct{}) {
go wait.Until(e.refreshExistingEventSeries, refreshTime, stopCh) err := e.StartRecordingToSinkWithContext(wait.ContextForChannel(stopCh))
go wait.Until(e.finishSeries, finishTime, stopCh)
err := e.startRecordingEvents(stopCh)
if err != nil { if err != nil {
klog.Errorf("unexpected type, expected eventsv1.Event") klog.Background().Error(err, "Failed to start recording to sink")
return
} }
} }
// StartRecordingToSinkWithContext starts sending events received from the specified eventBroadcaster to the given sink.
func (e *eventBroadcasterImpl) StartRecordingToSinkWithContext(ctx context.Context) error {
go wait.UntilWithContext(ctx, e.refreshExistingEventSeries, refreshTime)
go wait.UntilWithContext(ctx, e.finishSeries, finishTime)
return e.startRecordingEvents(ctx)
}
type eventBroadcasterAdapterImpl struct { type eventBroadcasterAdapterImpl struct {
coreClient typedv1core.EventsGetter coreClient typedv1core.EventsGetter
coreBroadcaster record.EventBroadcaster coreBroadcaster record.EventBroadcaster
@ -409,14 +428,14 @@ func (e *eventBroadcasterAdapterImpl) StartRecordingToSink(stopCh <-chan struct{
} }
} }
func (e *eventBroadcasterAdapterImpl) NewRecorder(name string) EventRecorder { func (e *eventBroadcasterAdapterImpl) NewRecorder(name string) EventRecorderLogger {
if e.eventsv1Broadcaster != nil && e.eventsv1Client != nil { if e.eventsv1Broadcaster != nil && e.eventsv1Client != nil {
return e.eventsv1Broadcaster.NewRecorder(scheme.Scheme, name) return e.eventsv1Broadcaster.NewRecorder(scheme.Scheme, name)
} }
return record.NewEventRecorderAdapter(e.DeprecatedNewLegacyRecorder(name)) return record.NewEventRecorderAdapter(e.DeprecatedNewLegacyRecorder(name))
} }
func (e *eventBroadcasterAdapterImpl) DeprecatedNewLegacyRecorder(name string) record.EventRecorder { func (e *eventBroadcasterAdapterImpl) DeprecatedNewLegacyRecorder(name string) record.EventRecorderLogger {
return e.coreBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: name}) return e.coreBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: name})
} }

4
tools/events/event_broadcaster_test.go
View File

@ -25,6 +25,7 @@ import (
eventsv1 "k8s.io/api/events/v1" eventsv1 "k8s.io/api/events/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/kubernetes/fake" "k8s.io/client-go/kubernetes/fake"
"k8s.io/klog/v2/ktesting"
) )
func TestRecordEventToSink(t *testing.T) { func TestRecordEventToSink(t *testing.T) {
@ -78,11 +79,12 @@ func TestRecordEventToSink(t *testing.T) {
for _, tc := range testCases { for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) { t.Run(tc.name, func(t *testing.T) {
_, ctx := ktesting.NewTestContext(t)
kubeClient := fake.NewSimpleClientset() kubeClient := fake.NewSimpleClientset()
eventSink := &EventSinkImpl{Interface: kubeClient.EventsV1()} eventSink := &EventSinkImpl{Interface: kubeClient.EventsV1()}
for _, ev := range tc.eventsToRecord { for _, ev := range tc.eventsToRecord {
recordEvent(eventSink, &ev) recordEvent(ctx, eventSink, &ev)
} }
recordedEvents, err := kubeClient.EventsV1().Events(metav1.NamespaceDefault).List(context.TODO(), metav1.ListOptions{}) recordedEvents, err := kubeClient.EventsV1().Events(metav1.NamespaceDefault).List(context.TODO(), metav1.ListOptions{})

27
tools/events/event_recorder.go
View File

@ -40,12 +40,33 @@ type recorderImpl struct {
clock clock.Clock clock clock.Clock
} }
var _ EventRecorder = &recorderImpl{}
func (recorder *recorderImpl) Eventf(regarding runtime.Object, related runtime.Object, eventtype, reason, action, note string, args ...interface{}) { func (recorder *recorderImpl) Eventf(regarding runtime.Object, related runtime.Object, eventtype, reason, action, note string, args ...interface{}) {
recorder.eventf(klog.Background(), regarding, related, eventtype, reason, action, note, args...)
}
type recorderImplLogger struct {
*recorderImpl
logger klog.Logger
}
var _ EventRecorderLogger = &recorderImplLogger{}
func (recorder *recorderImplLogger) Eventf(regarding runtime.Object, related runtime.Object, eventtype, reason, action, note string, args ...interface{}) {
recorder.eventf(recorder.logger, regarding, related, eventtype, reason, action, note, args...)
}
func (recorder *recorderImplLogger) WithLogger(logger klog.Logger) EventRecorderLogger {
return &recorderImplLogger{recorderImpl: recorder.recorderImpl, logger: logger}
}
func (recorder *recorderImpl) eventf(logger klog.Logger, regarding runtime.Object, related runtime.Object, eventtype, reason, action, note string, args ...interface{}) {
timestamp := metav1.MicroTime{Time: time.Now()} timestamp := metav1.MicroTime{Time: time.Now()}
message := fmt.Sprintf(note, args...) message := fmt.Sprintf(note, args...)
refRegarding, err := reference.GetReference(recorder.scheme, regarding) refRegarding, err := reference.GetReference(recorder.scheme, regarding)
if err != nil { if err != nil {
klog.Errorf("Could not construct reference to: '%#v' due to: '%v'. Will not report event: '%v' '%v' '%v'", regarding, err, eventtype, reason, message) logger.Error(err, "Could not construct reference, will not report event", "object", regarding, "eventType", eventtype, "reason", reason, "message", message)
return return
} }
@ -53,11 +74,11 @@ func (recorder *recorderImpl) Eventf(regarding runtime.Object, related runtime.O
if related != nil { if related != nil {
refRelated, err = reference.GetReference(recorder.scheme, related) refRelated, err = reference.GetReference(recorder.scheme, related)
if err != nil { if err != nil {
klog.V(9).Infof("Could not construct reference to: '%#v' due to: '%v'.", related, err) logger.V(9).Info("Could not construct reference", "object", related, "err", err)
} }
} }
if !util.ValidateEventType(eventtype) { if !util.ValidateEventType(eventtype) {
klog.Errorf("Unsupported event type: '%v'", eventtype) logger.Error(nil, "Unsupported event type", "eventType", eventtype)
return return
} }
event := recorder.makeEvent(refRegarding, refRelated, timestamp, eventtype, reason, message, recorder.reportingController, recorder.reportingInstance, action) event := recorder.makeEvent(refRegarding, refRelated, timestamp, eventtype, reason, message, recorder.reportingController, recorder.reportingInstance, action)

24
tools/events/eventseries_test.go
View File

@ -34,6 +34,7 @@ import (
"k8s.io/client-go/kubernetes/scheme" "k8s.io/client-go/kubernetes/scheme"
restclient "k8s.io/client-go/rest" restclient "k8s.io/client-go/rest"
ref "k8s.io/client-go/tools/reference" ref "k8s.io/client-go/tools/reference"
"k8s.io/klog/v2/ktesting"
) )
type testEventSeriesSink struct { type testEventSeriesSink struct {
@ -43,7 +44,7 @@ type testEventSeriesSink struct {
} }
// Create records the event for testing. // Create records the event for testing.
func (t *testEventSeriesSink) Create(e *eventsv1.Event) (*eventsv1.Event, error) { func (t *testEventSeriesSink) Create(ctx context.Context, e *eventsv1.Event) (*eventsv1.Event, error) {
if t.OnCreate != nil { if t.OnCreate != nil {
return t.OnCreate(e) return t.OnCreate(e)
} }
@ -51,7 +52,7 @@ func (t *testEventSeriesSink) Create(e *eventsv1.Event) (*eventsv1.Event, error)
} }
// Update records the event for testing. // Update records the event for testing.
func (t *testEventSeriesSink) Update(e *eventsv1.Event) (*eventsv1.Event, error) { func (t *testEventSeriesSink) Update(ctx context.Context, e *eventsv1.Event) (*eventsv1.Event, error) {
if t.OnUpdate != nil { if t.OnUpdate != nil {
return t.OnUpdate(e) return t.OnUpdate(e)
} }
@ -59,7 +60,7 @@ func (t *testEventSeriesSink) Update(e *eventsv1.Event) (*eventsv1.Event, error)
} }
// Patch records the event for testing. // Patch records the event for testing.
func (t *testEventSeriesSink) Patch(e *eventsv1.Event, p []byte) (*eventsv1.Event, error) { func (t *testEventSeriesSink) Patch(ctx context.Context, e *eventsv1.Event, p []byte) (*eventsv1.Event, error) {
if t.OnPatch != nil { if t.OnPatch != nil {
return t.OnPatch(e, p) return t.OnPatch(e, p)
} }
@ -135,7 +136,9 @@ func TestEventSeriesf(t *testing.T) {
}, },
} }
stopCh := make(chan struct{}) _, ctx := ktesting.NewTestContext(t)
ctx, cancel := context.WithCancel(ctx)
defer cancel()
createEvent := make(chan *eventsv1.Event) createEvent := make(chan *eventsv1.Event)
updateEvent := make(chan *eventsv1.Event) updateEvent := make(chan *eventsv1.Event)
@ -163,7 +166,7 @@ func TestEventSeriesf(t *testing.T) {
// Don't call StartRecordingToSink, as we don't need neither refreshing event // Don't call StartRecordingToSink, as we don't need neither refreshing event
// series nor finishing them in this tests and additional events updated would // series nor finishing them in this tests and additional events updated would
// race with our expected ones. // race with our expected ones.
err = broadcaster.startRecordingEvents(stopCh) err = broadcaster.startRecordingEvents(ctx)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@ -184,7 +187,6 @@ func TestEventSeriesf(t *testing.T) {
validateEvent(strconv.Itoa(index), false, actualEvent, item.expect, t) validateEvent(strconv.Itoa(index), false, actualEvent, item.expect, t)
} }
} }
close(stopCh)
} }
// TestEventSeriesWithEventSinkImplRace verifies that when Events are emitted to // TestEventSeriesWithEventSinkImplRace verifies that when Events are emitted to
@ -256,6 +258,7 @@ func validateEvent(messagePrefix string, expectedUpdate bool, actualEvent *event
} }
func TestFinishSeries(t *testing.T) { func TestFinishSeries(t *testing.T) {
_, ctx := ktesting.NewTestContext(t)
hostname, _ := os.Hostname() hostname, _ := os.Hostname()
testPod := &v1.Pod{ testPod := &v1.Pod{
ObjectMeta: metav1.ObjectMeta{ ObjectMeta: metav1.ObjectMeta{
@ -295,7 +298,7 @@ func TestFinishSeries(t *testing.T) {
} }
cache := map[eventKey]*eventsv1.Event{} cache := map[eventKey]*eventsv1.Event{}
eventBroadcaster := newBroadcaster(&testEvents, 0, cache).(*eventBroadcasterImpl) eventBroadcaster := newBroadcaster(&testEvents, 0, cache).(*eventBroadcasterImpl)
recorder := eventBroadcaster.NewRecorder(scheme.Scheme, "k8s.io/kube-foo").(*recorderImpl) recorder := eventBroadcaster.NewRecorder(scheme.Scheme, "k8s.io/kube-foo").(*recorderImplLogger)
cachedEvent := recorder.makeEvent(regarding, related, metav1.MicroTime{Time: time.Now()}, v1.EventTypeNormal, "test", "some verbose message: 1", "eventTest", "eventTest-"+hostname, "started") cachedEvent := recorder.makeEvent(regarding, related, metav1.MicroTime{Time: time.Now()}, v1.EventTypeNormal, "test", "some verbose message: 1", "eventTest", "eventTest-"+hostname, "started")
nonFinishedEvent := cachedEvent.DeepCopy() nonFinishedEvent := cachedEvent.DeepCopy()
nonFinishedEvent.ReportingController = "nonFinished-controller" nonFinishedEvent.ReportingController = "nonFinished-controller"
@ -305,7 +308,7 @@ func TestFinishSeries(t *testing.T) {
} }
cache[getKey(cachedEvent)] = cachedEvent cache[getKey(cachedEvent)] = cachedEvent
cache[getKey(nonFinishedEvent)] = nonFinishedEvent cache[getKey(nonFinishedEvent)] = nonFinishedEvent
eventBroadcaster.finishSeries() eventBroadcaster.finishSeries(ctx)
select { select {
case actualEvent := <-patchEvent: case actualEvent := <-patchEvent:
t.Logf("validating event affected by patch request") t.Logf("validating event affected by patch request")
@ -327,6 +330,7 @@ func TestFinishSeries(t *testing.T) {
} }
func TestRefreshExistingEventSeries(t *testing.T) { func TestRefreshExistingEventSeries(t *testing.T) {
_, ctx := ktesting.NewTestContext(t)
hostname, _ := os.Hostname() hostname, _ := os.Hostname()
testPod := &v1.Pod{ testPod := &v1.Pod{
ObjectMeta: metav1.ObjectMeta{ ObjectMeta: metav1.ObjectMeta{
@ -381,7 +385,7 @@ func TestRefreshExistingEventSeries(t *testing.T) {
} }
cache := map[eventKey]*eventsv1.Event{} cache := map[eventKey]*eventsv1.Event{}
eventBroadcaster := newBroadcaster(&testEvents, 0, cache).(*eventBroadcasterImpl) eventBroadcaster := newBroadcaster(&testEvents, 0, cache).(*eventBroadcasterImpl)
recorder := eventBroadcaster.NewRecorder(scheme.Scheme, "k8s.io/kube-foo").(*recorderImpl) recorder := eventBroadcaster.NewRecorder(scheme.Scheme, "k8s.io/kube-foo").(*recorderImplLogger)
cachedEvent := recorder.makeEvent(regarding, related, metav1.MicroTime{Time: time.Now()}, v1.EventTypeNormal, "test", "some verbose message: 1", "eventTest", "eventTest-"+hostname, "started") cachedEvent := recorder.makeEvent(regarding, related, metav1.MicroTime{Time: time.Now()}, v1.EventTypeNormal, "test", "some verbose message: 1", "eventTest", "eventTest-"+hostname, "started")
cachedEvent.Series = &eventsv1.EventSeries{ cachedEvent.Series = &eventsv1.EventSeries{
Count: 10, Count: 10,
@ -390,7 +394,7 @@ func TestRefreshExistingEventSeries(t *testing.T) {
cacheKey := getKey(cachedEvent) cacheKey := getKey(cachedEvent)
cache[cacheKey] = cachedEvent cache[cacheKey] = cachedEvent
eventBroadcaster.refreshExistingEventSeries() eventBroadcaster.refreshExistingEventSeries(ctx)
select { select {
case <-patchEvent: case <-patchEvent:
t.Logf("validating event affected by patch request") t.Logf("validating event affected by patch request")

7
tools/events/fake.go
View File

@ -20,6 +20,7 @@ import (
"fmt" "fmt"
"k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime"
"k8s.io/klog/v2"
) )
// FakeRecorder is used as a fake during tests. It is thread safe. It is usable // FakeRecorder is used as a fake during tests. It is thread safe. It is usable
@ -29,6 +30,8 @@ type FakeRecorder struct {
Events chan string Events chan string
} }
var _ EventRecorderLogger = &FakeRecorder{}
// Eventf emits an event // Eventf emits an event
func (f *FakeRecorder) Eventf(regarding runtime.Object, related runtime.Object, eventtype, reason, action, note string, args ...interface{}) { func (f *FakeRecorder) Eventf(regarding runtime.Object, related runtime.Object, eventtype, reason, action, note string, args ...interface{}) {
if f.Events != nil { if f.Events != nil {
@ -36,6 +39,10 @@ func (f *FakeRecorder) Eventf(regarding runtime.Object, related runtime.Object,
} }
} }
func (f *FakeRecorder) WithLogger(logger klog.Logger) EventRecorderLogger {
return f
}
// NewFakeRecorder creates new fake event recorder with event channel with // NewFakeRecorder creates new fake event recorder with event channel with
// buffer of given size. // buffer of given size.
func NewFakeRecorder(bufferSize int) *FakeRecorder { func NewFakeRecorder(bufferSize int) *FakeRecorder {

45
tools/events/interfaces.go
View File

@ -17,39 +17,30 @@ limitations under the License.
package events package events
import ( import (
"context"
eventsv1 "k8s.io/api/events/v1" eventsv1 "k8s.io/api/events/v1"
"k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime"
internalevents "k8s.io/client-go/tools/internal/events"
"k8s.io/client-go/tools/record" "k8s.io/client-go/tools/record"
"k8s.io/klog/v2" "k8s.io/klog/v2"
) )
// EventRecorder knows how to record events on behalf of an EventSource. type EventRecorder = internalevents.EventRecorder
type EventRecorder interface { type EventRecorderLogger = internalevents.EventRecorderLogger
// Eventf constructs an event from the given information and puts it in the queue for sending.
// 'regarding' is the object this event is about. Event will make a reference-- or you may also
// pass a reference to the object directly.
// 'related' is the secondary object for more complex actions. E.g. when regarding object triggers
// a creation or deletion of related object.
// 'type' of this event, and can be one of Normal, Warning. New types could be added in future
// 'reason' is the reason this event is generated. 'reason' should be short and unique; it
// should be in UpperCamelCase format (starting with a capital letter). "reason" will be used
// to automate handling of events, so imagine people writing switch statements to handle them.
// You want to make that easy.
// 'action' explains what happened with regarding/what action did the ReportingController
// (ReportingController is a type of a Controller reporting an Event, e.g. k8s.io/node-controller, k8s.io/kubelet.)
// take in regarding's name; it should be in UpperCamelCase format (starting with a capital letter).
// 'note' is intended to be human readable.
Eventf(regarding runtime.Object, related runtime.Object, eventtype, reason, action, note string, args ...interface{})
}
// EventBroadcaster knows how to receive events and send them to any EventSink, watcher, or log. // EventBroadcaster knows how to receive events and send them to any EventSink, watcher, or log.
type EventBroadcaster interface { type EventBroadcaster interface {
// StartRecordingToSink starts sending events received from the specified eventBroadcaster. // StartRecordingToSink starts sending events received from the specified eventBroadcaster.
// Deprecated: use StartRecordingToSinkWithContext instead.
StartRecordingToSink(stopCh <-chan struct{}) StartRecordingToSink(stopCh <-chan struct{})
// StartRecordingToSink starts sending events received from the specified eventBroadcaster.
StartRecordingToSinkWithContext(ctx context.Context) error
// NewRecorder returns an EventRecorder that can be used to send events to this EventBroadcaster // NewRecorder returns an EventRecorder that can be used to send events to this EventBroadcaster
// with the event source set to the given event source. // with the event source set to the given event source.
NewRecorder(scheme *runtime.Scheme, reportingController string) EventRecorder NewRecorder(scheme *runtime.Scheme, reportingController string) EventRecorderLogger
// StartEventWatcher enables you to watch for emitted events without usage // StartEventWatcher enables you to watch for emitted events without usage
// of StartRecordingToSink. This lets you also process events in a custom way (e.g. in tests). // of StartRecordingToSink. This lets you also process events in a custom way (e.g. in tests).
@ -59,8 +50,14 @@ type EventBroadcaster interface {
// StartStructuredLogging starts sending events received from this EventBroadcaster to the structured // StartStructuredLogging starts sending events received from this EventBroadcaster to the structured
// logging function. The return value can be ignored or used to stop recording, if desired. // logging function. The return value can be ignored or used to stop recording, if desired.
// Deprecated: use StartLogging instead.
StartStructuredLogging(verbosity klog.Level) func() StartStructuredLogging(verbosity klog.Level) func()
// StartLogging starts sending events received from this EventBroadcaster to the structured logger.
// To adjust verbosity, use the logger's V method (i.e. pass `logger.V(3)` instead of `logger`).
// The returned function can be ignored or used to stop recording, if desired.
StartLogging(logger klog.Logger) (func(), error)
// Shutdown shuts down the broadcaster // Shutdown shuts down the broadcaster
Shutdown() Shutdown()
} }
@ -70,9 +67,9 @@ type EventBroadcaster interface {
// It is assumed that EventSink will return the same sorts of errors as // It is assumed that EventSink will return the same sorts of errors as
// client-go's REST client. // client-go's REST client.
type EventSink interface { type EventSink interface {
Create(event *eventsv1.Event) (*eventsv1.Event, error) Create(ctx context.Context, event *eventsv1.Event) (*eventsv1.Event, error)
Update(event *eventsv1.Event) (*eventsv1.Event, error) Update(ctx context.Context, event *eventsv1.Event) (*eventsv1.Event, error)
Patch(oldEvent *eventsv1.Event, data []byte) (*eventsv1.Event, error) Patch(ctx context.Context, oldEvent *eventsv1.Event, data []byte) (*eventsv1.Event, error)
} }
// EventBroadcasterAdapter is a auxiliary interface to simplify migration to // EventBroadcasterAdapter is a auxiliary interface to simplify migration to
@ -85,10 +82,10 @@ type EventBroadcasterAdapter interface {
StartRecordingToSink(stopCh <-chan struct{}) StartRecordingToSink(stopCh <-chan struct{})
// NewRecorder creates a new Event Recorder with specified name. // NewRecorder creates a new Event Recorder with specified name.
NewRecorder(name string) EventRecorder NewRecorder(name string) EventRecorderLogger
// DeprecatedNewLegacyRecorder creates a legacy Event Recorder with specific name. // DeprecatedNewLegacyRecorder creates a legacy Event Recorder with specific name.
DeprecatedNewLegacyRecorder(name string) record.EventRecorder DeprecatedNewLegacyRecorder(name string) record.EventRecorderLogger
// Shutdown shuts down the broadcaster. // Shutdown shuts down the broadcaster.
Shutdown() Shutdown()

59
tools/internal/events/interfaces.go Normal file
View File

@ -0,0 +1,59 @@
/*
Copyright 2019 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
// Package internal is needed to break an import cycle: record.EventRecorderAdapter
// needs this interface definition to implement it, but event.NewEventBroadcasterAdapter
// needs record.NewBroadcaster. Therefore this interface cannot be in event/interfaces.go.
package internal
import (
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/klog/v2"
)
// EventRecorder knows how to record events on behalf of an EventSource.
type EventRecorder interface {
// Eventf constructs an event from the given information and puts it in the queue for sending.
// 'regarding' is the object this event is about. Event will make a reference-- or you may also
// pass a reference to the object directly.
// 'related' is the secondary object for more complex actions. E.g. when regarding object triggers
// a creation or deletion of related object.
// 'type' of this event, and can be one of Normal, Warning. New types could be added in future
// 'reason' is the reason this event is generated. 'reason' should be short and unique; it
// should be in UpperCamelCase format (starting with a capital letter). "reason" will be used
// to automate handling of events, so imagine people writing switch statements to handle them.
// You want to make that easy.
// 'action' explains what happened with regarding/what action did the ReportingController
// (ReportingController is a type of a Controller reporting an Event, e.g. k8s.io/node-controller, k8s.io/kubelet.)
// take in regarding's name; it should be in UpperCamelCase format (starting with a capital letter).
// 'note' is intended to be human readable.
Eventf(regarding runtime.Object, related runtime.Object, eventtype, reason, action, note string, args ...interface{})
}
// EventRecorderLogger extends EventRecorder such that a logger can
// be set for methods in EventRecorder. Normally, those methods
// uses the global default logger to record errors and debug messages.
// If that is not desired, use WithLogger to provide a logger instance.
type EventRecorderLogger interface {
EventRecorder
// WithLogger replaces the context used for logging. This is a cheap call
// and meant to be used for contextual logging:
// recorder := ...
// logger := klog.FromContext(ctx)
// recorder.WithLogger(logger).Eventf(...)
WithLogger(logger klog.Logger) EventRecorderLogger
}

184
tools/record/event.go
View File

@ -29,6 +29,7 @@ import (
utilruntime "k8s.io/apimachinery/pkg/util/runtime" utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/watch" "k8s.io/apimachinery/pkg/watch"
restclient "k8s.io/client-go/rest" restclient "k8s.io/client-go/rest"
internalevents "k8s.io/client-go/tools/internal/events"
"k8s.io/client-go/tools/record/util" "k8s.io/client-go/tools/record/util"
ref "k8s.io/client-go/tools/reference" ref "k8s.io/client-go/tools/reference"
"k8s.io/klog/v2" "k8s.io/klog/v2"
@ -110,6 +111,21 @@ type EventRecorder interface {
AnnotatedEventf(object runtime.Object, annotations map[string]string, eventtype, reason, messageFmt string, args ...interface{}) AnnotatedEventf(object runtime.Object, annotations map[string]string, eventtype, reason, messageFmt string, args ...interface{})
} }
// EventRecorderLogger extends EventRecorder such that a logger can
// be set for methods in EventRecorder. Normally, those methods
// uses the global default logger to record errors and debug messages.
// If that is not desired, use WithLogger to provide a logger instance.
type EventRecorderLogger interface {
EventRecorder
// WithLogger replaces the context used for logging. This is a cheap call
// and meant to be used for contextual logging:
// recorder := ...
// logger := klog.FromContext(ctx)
// recorder.WithLogger(logger).Eventf(...)
WithLogger(logger klog.Logger) EventRecorderLogger
}
// EventBroadcaster knows how to receive events and send them to any EventSink, watcher, or log. // EventBroadcaster knows how to receive events and send them to any EventSink, watcher, or log.
type EventBroadcaster interface { type EventBroadcaster interface {
// StartEventWatcher starts sending events received from this EventBroadcaster to the given // StartEventWatcher starts sending events received from this EventBroadcaster to the given
@ -131,7 +147,7 @@ type EventBroadcaster interface {
// NewRecorder returns an EventRecorder that can be used to send events to this EventBroadcaster // NewRecorder returns an EventRecorder that can be used to send events to this EventBroadcaster
// with the event source set to the given event source. // with the event source set to the given event source.
NewRecorder(scheme *runtime.Scheme, source v1.EventSource) EventRecorder NewRecorder(scheme *runtime.Scheme, source v1.EventSource) EventRecorderLogger
// Shutdown shuts down the broadcaster. Once the broadcaster is shut // Shutdown shuts down the broadcaster. Once the broadcaster is shut
// down, it will only try to record an event in a sink once before // down, it will only try to record an event in a sink once before
@ -142,12 +158,14 @@ type EventBroadcaster interface {
// EventRecorderAdapter is a wrapper around a "k8s.io/client-go/tools/record".EventRecorder // EventRecorderAdapter is a wrapper around a "k8s.io/client-go/tools/record".EventRecorder
// implementing the new "k8s.io/client-go/tools/events".EventRecorder interface. // implementing the new "k8s.io/client-go/tools/events".EventRecorder interface.
type EventRecorderAdapter struct { type EventRecorderAdapter struct {
recorder EventRecorder recorder EventRecorderLogger
} }
var _ internalevents.EventRecorder = &EventRecorderAdapter{}
// NewEventRecorderAdapter returns an adapter implementing the new // NewEventRecorderAdapter returns an adapter implementing the new
// "k8s.io/client-go/tools/events".EventRecorder interface. // "k8s.io/client-go/tools/events".EventRecorder interface.
func NewEventRecorderAdapter(recorder EventRecorder) *EventRecorderAdapter { func NewEventRecorderAdapter(recorder EventRecorderLogger) *EventRecorderAdapter {
return &EventRecorderAdapter{ return &EventRecorderAdapter{
recorder: recorder, recorder: recorder,
} }
@ -158,28 +176,76 @@ func (a *EventRecorderAdapter) Eventf(regarding, _ runtime.Object, eventtype, re
a.recorder.Eventf(regarding, eventtype, reason, note, args...) a.recorder.Eventf(regarding, eventtype, reason, note, args...)
} }
func (a *EventRecorderAdapter) WithLogger(logger klog.Logger) internalevents.EventRecorderLogger {
return &EventRecorderAdapter{
recorder: a.recorder.WithLogger(logger),
}
}
// Creates a new event broadcaster. // Creates a new event broadcaster.
func NewBroadcaster() EventBroadcaster { func NewBroadcaster(opts ...BroadcasterOption) EventBroadcaster {
return newEventBroadcaster(watch.NewLongQueueBroadcaster(maxQueuedEvents, watch.DropIfChannelFull), defaultSleepDuration) c := config{
sleepDuration: defaultSleepDuration,
}
for _, opt := range opts {
opt(&c)
}
eventBroadcaster := &eventBroadcasterImpl{
Broadcaster: watch.NewLongQueueBroadcaster(maxQueuedEvents, watch.DropIfChannelFull),
sleepDuration: c.sleepDuration,
options: c.CorrelatorOptions,
}
ctx := c.Context
if ctx == nil {
ctx = context.Background()
} else {
// Calling Shutdown is not required when a context was provided:
// when the context is canceled, this goroutine will shut down
// the broadcaster.
go func() {
<-ctx.Done()
eventBroadcaster.Broadcaster.Shutdown()
}()
}
eventBroadcaster.cancelationCtx, eventBroadcaster.cancel = context.WithCancel(ctx)
return eventBroadcaster
} }
func NewBroadcasterForTests(sleepDuration time.Duration) EventBroadcaster { func NewBroadcasterForTests(sleepDuration time.Duration) EventBroadcaster {
return newEventBroadcaster(watch.NewLongQueueBroadcaster(maxQueuedEvents, watch.DropIfChannelFull), sleepDuration) return NewBroadcaster(WithSleepDuration(sleepDuration))
} }
func NewBroadcasterWithCorrelatorOptions(options CorrelatorOptions) EventBroadcaster { func NewBroadcasterWithCorrelatorOptions(options CorrelatorOptions) EventBroadcaster {
eventBroadcaster := newEventBroadcaster(watch.NewLongQueueBroadcaster(maxQueuedEvents, watch.DropIfChannelFull), defaultSleepDuration) return NewBroadcaster(WithCorrelatorOptions(options))
eventBroadcaster.options = options
return eventBroadcaster
} }
func newEventBroadcaster(broadcaster *watch.Broadcaster, sleepDuration time.Duration) *eventBroadcasterImpl { func WithCorrelatorOptions(options CorrelatorOptions) BroadcasterOption {
eventBroadcaster := &eventBroadcasterImpl{ return func(c *config) {
Broadcaster: broadcaster, c.CorrelatorOptions = options
sleepDuration: sleepDuration,
} }
eventBroadcaster.cancelationCtx, eventBroadcaster.cancel = context.WithCancel(context.Background()) }
return eventBroadcaster
// WithContext sets a context for the broadcaster. Canceling the context will
// shut down the broadcaster, Shutdown doesn't need to be called. The context
// can also be used to provide a logger.
func WithContext(ctx context.Context) BroadcasterOption {
return func(c *config) {
c.Context = ctx
}
}
func WithSleepDuration(sleepDuration time.Duration) BroadcasterOption {
return func(c *config) {
c.sleepDuration = sleepDuration
}
}
type BroadcasterOption func(*config)
type config struct {
CorrelatorOptions
context.Context
sleepDuration time.Duration
} }
type eventBroadcasterImpl struct { type eventBroadcasterImpl struct {
@ -220,12 +286,12 @@ func (e *eventBroadcasterImpl) recordToSink(sink EventSink, event *v1.Event, eve
} }
tries := 0 tries := 0
for { for {
if recordEvent(sink, result.Event, result.Patch, result.Event.Count > 1, eventCorrelator) { if recordEvent(e.cancelationCtx, sink, result.Event, result.Patch, result.Event.Count > 1, eventCorrelator) {
break break
} }
tries++ tries++
if tries >= maxTriesPerEvent { if tries >= maxTriesPerEvent {
klog.Errorf("Unable to write event '%#v' (retry limit exceeded!)", event) klog.FromContext(e.cancelationCtx).Error(nil, "Unable to write event (retry limit exceeded!)", "event", event)
break break
} }
@ -237,7 +303,7 @@ func (e *eventBroadcasterImpl) recordToSink(sink EventSink, event *v1.Event, eve
} }
select { select {
case <-e.cancelationCtx.Done(): case <-e.cancelationCtx.Done():
klog.Errorf("Unable to write event '%#v' (broadcaster is shut down)", event) klog.FromContext(e.cancelationCtx).Error(nil, "Unable to write event (broadcaster is shut down)", "event", event)
return return
case <-time.After(delay): case <-time.After(delay):
} }
@ -248,7 +314,7 @@ func (e *eventBroadcasterImpl) recordToSink(sink EventSink, event *v1.Event, eve
// was successfully recorded or discarded, false if it should be retried. // was successfully recorded or discarded, false if it should be retried.
// If updateExistingEvent is false, it creates a new event, otherwise it updates // If updateExistingEvent is false, it creates a new event, otherwise it updates
// existing event. // existing event.
func recordEvent(sink EventSink, event *v1.Event, patch []byte, updateExistingEvent bool, eventCorrelator *EventCorrelator) bool { func recordEvent(ctx context.Context, sink EventSink, event *v1.Event, patch []byte, updateExistingEvent bool, eventCorrelator *EventCorrelator) bool {
var newEvent *v1.Event var newEvent *v1.Event
var err error var err error
if updateExistingEvent { if updateExistingEvent {
@ -271,13 +337,13 @@ func recordEvent(sink EventSink, event *v1.Event, patch []byte, updateExistingEv
switch err.(type) { switch err.(type) {
case *restclient.RequestConstructionError: case *restclient.RequestConstructionError:
// We will construct the request the same next time, so don't keep trying. // We will construct the request the same next time, so don't keep trying.
klog.Errorf("Unable to construct event '%#v': '%v' (will not retry!)", event, err) klog.FromContext(ctx).Error(err, "Unable to construct event (will not retry!)", "event", event)
return true return true
case *errors.StatusError: case *errors.StatusError:
if errors.IsAlreadyExists(err) || errors.HasStatusCause(err, v1.NamespaceTerminatingCause) { if errors.IsAlreadyExists(err) || errors.HasStatusCause(err, v1.NamespaceTerminatingCause) {
klog.V(5).Infof("Server rejected event '%#v': '%v' (will not retry!)", event, err) klog.FromContext(ctx).V(5).Info("Server rejected event (will not retry!)", "event", event, "err", err)
} else { } else {
klog.Errorf("Server rejected event '%#v': '%v' (will not retry!)", event, err) klog.FromContext(ctx).Error(err, "Server rejected event (will not retry!)", "event", event)
} }
return true return true
case *errors.UnexpectedObjectError: case *errors.UnexpectedObjectError:
@ -286,7 +352,7 @@ func recordEvent(sink EventSink, event *v1.Event, patch []byte, updateExistingEv
default: default:
// This case includes actual http transport errors. Go ahead and retry. // This case includes actual http transport errors. Go ahead and retry.
} }
klog.Errorf("Unable to write event: '%#v': '%v'(may retry after sleeping)", event, err) klog.FromContext(ctx).Error(err, "Unable to write event (may retry after sleeping)", "event", event)
return false return false
} }
@ -299,12 +365,15 @@ func (e *eventBroadcasterImpl) StartLogging(logf func(format string, args ...int
}) })
} }
// StartStructuredLogging starts sending events received from this EventBroadcaster to the structured logging function. // StartStructuredLogging starts sending events received from this EventBroadcaster to a structured logger.
// The logger is retrieved from a context if the broadcaster was constructed with a context, otherwise
// the global default is used.
// The return value can be ignored or used to stop recording, if desired. // The return value can be ignored or used to stop recording, if desired.
func (e *eventBroadcasterImpl) StartStructuredLogging(verbosity klog.Level) watch.Interface { func (e *eventBroadcasterImpl) StartStructuredLogging(verbosity klog.Level) watch.Interface {
loggerV := klog.FromContext(e.cancelationCtx).V(int(verbosity))
return e.StartEventWatcher( return e.StartEventWatcher(
func(e *v1.Event) { func(e *v1.Event) {
klog.V(verbosity).InfoS("Event occurred", "object", klog.KRef(e.InvolvedObject.Namespace, e.InvolvedObject.Name), "fieldPath", e.InvolvedObject.FieldPath, "kind", e.InvolvedObject.Kind, "apiVersion", e.InvolvedObject.APIVersion, "type", e.Type, "reason", e.Reason, "message", e.Message) loggerV.Info("Event occurred", "object", klog.KRef(e.InvolvedObject.Namespace, e.InvolvedObject.Name), "fieldPath", e.InvolvedObject.FieldPath, "kind", e.InvolvedObject.Kind, "apiVersion", e.InvolvedObject.APIVersion, "type", e.Type, "reason", e.Reason, "message", e.Message)
}) })
} }
@ -313,26 +382,32 @@ func (e *eventBroadcasterImpl) StartStructuredLogging(verbosity klog.Level) watc
func (e *eventBroadcasterImpl) StartEventWatcher(eventHandler func(*v1.Event)) watch.Interface { func (e *eventBroadcasterImpl) StartEventWatcher(eventHandler func(*v1.Event)) watch.Interface {
watcher, err := e.Watch() watcher, err := e.Watch()
if err != nil { if err != nil {
klog.Errorf("Unable start event watcher: '%v' (will not retry!)", err) klog.FromContext(e.cancelationCtx).Error(err, "Unable start event watcher (will not retry!)")
} }
go func() { go func() {
defer utilruntime.HandleCrash() defer utilruntime.HandleCrash()
for watchEvent := range watcher.ResultChan() { for {
event, ok := watchEvent.Object.(*v1.Event) select {
if !ok { case <-e.cancelationCtx.Done():
// This is all local, so there's no reason this should watcher.Stop()
// ever happen. return
continue case watchEvent := <-watcher.ResultChan():
event, ok := watchEvent.Object.(*v1.Event)
if !ok {
// This is all local, so there's no reason this should
// ever happen.
continue
}
eventHandler(event)
} }
eventHandler(event)
} }
}() }()
return watcher return watcher
} }
// NewRecorder returns an EventRecorder that records events with the given event source. // NewRecorder returns an EventRecorder that records events with the given event source.
func (e *eventBroadcasterImpl) NewRecorder(scheme *runtime.Scheme, source v1.EventSource) EventRecorder { func (e *eventBroadcasterImpl) NewRecorder(scheme *runtime.Scheme, source v1.EventSource) EventRecorderLogger {
return &recorderImpl{scheme, source, e.Broadcaster, clock.RealClock{}} return &recorderImplLogger{recorderImpl: &recorderImpl{scheme, source, e.Broadcaster, clock.RealClock{}}, logger: klog.Background()}
} }
type recorderImpl struct { type recorderImpl struct {
@ -342,15 +417,17 @@ type recorderImpl struct {
clock clock.PassiveClock clock clock.PassiveClock
} }
func (recorder *recorderImpl) generateEvent(object runtime.Object, annotations map[string]string, eventtype, reason, message string) { var _ EventRecorder = &recorderImpl{}
func (recorder *recorderImpl) generateEvent(logger klog.Logger, object runtime.Object, annotations map[string]string, eventtype, reason, message string) {
ref, err := ref.GetReference(recorder.scheme, object) ref, err := ref.GetReference(recorder.scheme, object)
if err != nil { if err != nil {
klog.Errorf("Could not construct reference to: '%#v' due to: '%v'. Will not report event: '%v' '%v' '%v'", object, err, eventtype, reason, message) logger.Error(err, "Could not construct reference, will not report event", "object", object, "eventType", eventtype, "reason", reason, "message", message)
return return
} }
if !util.ValidateEventType(eventtype) { if !util.ValidateEventType(eventtype) {
klog.Errorf("Unsupported event type: '%v'", eventtype) logger.Error(nil, "Unsupported event type", "eventType", eventtype)
return return
} }
@ -367,16 +444,16 @@ func (recorder *recorderImpl) generateEvent(object runtime.Object, annotations m
// outgoing events anyway). // outgoing events anyway).
sent, err := recorder.ActionOrDrop(watch.Added, event) sent, err := recorder.ActionOrDrop(watch.Added, event)
if err != nil { if err != nil {
klog.Errorf("unable to record event: %v (will not retry!)", err) logger.Error(err, "Unable to record event (will not retry!)")
return return
} }
if !sent { if !sent {
klog.Errorf("unable to record event: too many queued events, dropped event %#v", event) logger.Error(nil, "Unable to record event: too many queued events, dropped event", "event", event)
} }
} }
func (recorder *recorderImpl) Event(object runtime.Object, eventtype, reason, message string) { func (recorder *recorderImpl) Event(object runtime.Object, eventtype, reason, message string) {
recorder.generateEvent(object, nil, eventtype, reason, message) recorder.generateEvent(klog.Background(), object, nil, eventtype, reason, message)
} }
func (recorder *recorderImpl) Eventf(object runtime.Object, eventtype, reason, messageFmt string, args ...interface{}) { func (recorder *recorderImpl) Eventf(object runtime.Object, eventtype, reason, messageFmt string, args ...interface{}) {
@ -384,7 +461,7 @@ func (recorder *recorderImpl) Eventf(object runtime.Object, eventtype, reason, m
} }
func (recorder *recorderImpl) AnnotatedEventf(object runtime.Object, annotations map[string]string, eventtype, reason, messageFmt string, args ...interface{}) { func (recorder *recorderImpl) AnnotatedEventf(object runtime.Object, annotations map[string]string, eventtype, reason, messageFmt string, args ...interface{}) {
recorder.generateEvent(object, annotations, eventtype, reason, fmt.Sprintf(messageFmt, args...)) recorder.generateEvent(klog.Background(), object, annotations, eventtype, reason, fmt.Sprintf(messageFmt, args...))
} }
func (recorder *recorderImpl) makeEvent(ref *v1.ObjectReference, annotations map[string]string, eventtype, reason, message string) *v1.Event { func (recorder *recorderImpl) makeEvent(ref *v1.ObjectReference, annotations map[string]string, eventtype, reason, message string) *v1.Event {
@ -408,3 +485,26 @@ func (recorder *recorderImpl) makeEvent(ref *v1.ObjectReference, annotations map
Type: eventtype, Type: eventtype,
} }
} }
type recorderImplLogger struct {
*recorderImpl
logger klog.Logger
}
var _ EventRecorderLogger = &recorderImplLogger{}
func (recorder recorderImplLogger) Event(object runtime.Object, eventtype, reason, message string) {
recorder.recorderImpl.generateEvent(recorder.logger, object, nil, eventtype, reason, message)
}
func (recorder recorderImplLogger) Eventf(object runtime.Object, eventtype, reason, messageFmt string, args ...interface{}) {
recorder.Event(object, eventtype, reason, fmt.Sprintf(messageFmt, args...))
}
func (recorder recorderImplLogger) AnnotatedEventf(object runtime.Object, annotations map[string]string, eventtype, reason, messageFmt string, args ...interface{}) {
recorder.generateEvent(recorder.logger, object, annotations, eventtype, reason, fmt.Sprintf(messageFmt, args...))
}
func (recorder recorderImplLogger) WithLogger(logger klog.Logger) EventRecorderLogger {
return recorderImplLogger{recorderImpl: recorder.recorderImpl, logger: logger}
}

10
tools/record/event_test.go
View File

@ -112,7 +112,7 @@ func TestNonRacyShutdown(t *testing.T) {
caster := NewBroadcasterForTests(0) caster := NewBroadcasterForTests(0)
clock := testclocks.NewFakeClock(time.Now()) clock := testclocks.NewFakeClock(time.Now())
recorder := recorderWithFakeClock(v1.EventSource{Component: "eventTest"}, caster, clock) recorder := recorderWithFakeClock(t, v1.EventSource{Component: "eventTest"}, caster, clock)
var wg sync.WaitGroup var wg sync.WaitGroup
wg.Add(100) wg.Add(100)
@ -381,7 +381,7 @@ func TestEventf(t *testing.T) {
sinkWatcher := eventBroadcaster.StartRecordingToSink(&testEvents) sinkWatcher := eventBroadcaster.StartRecordingToSink(&testEvents)
clock := testclocks.NewFakeClock(time.Now()) clock := testclocks.NewFakeClock(time.Now())
recorder := recorderWithFakeClock(v1.EventSource{Component: "eventTest"}, eventBroadcaster, clock) recorder := recorderWithFakeClock(t, v1.EventSource{Component: "eventTest"}, eventBroadcaster, clock)
for index, item := range table { for index, item := range table {
clock.Step(1 * time.Second) clock.Step(1 * time.Second)
logWatcher := eventBroadcaster.StartLogging(func(formatter string, args ...interface{}) { logWatcher := eventBroadcaster.StartLogging(func(formatter string, args ...interface{}) {
@ -407,7 +407,7 @@ func TestEventf(t *testing.T) {
sinkWatcher.Stop() sinkWatcher.Stop()
} }
func recorderWithFakeClock(eventSource v1.EventSource, eventBroadcaster EventBroadcaster, clock clock.Clock) EventRecorder { func recorderWithFakeClock(t *testing.T, eventSource v1.EventSource, eventBroadcaster EventBroadcaster, clock clock.Clock) EventRecorder {
return &recorderImpl{scheme.Scheme, eventSource, eventBroadcaster.(*eventBroadcasterImpl).Broadcaster, clock} return &recorderImpl{scheme.Scheme, eventSource, eventBroadcaster.(*eventBroadcasterImpl).Broadcaster, clock}
} }
@ -662,7 +662,7 @@ func TestEventfNoNamespace(t *testing.T) {
sinkWatcher := eventBroadcaster.StartRecordingToSink(&testEvents) sinkWatcher := eventBroadcaster.StartRecordingToSink(&testEvents)
clock := testclocks.NewFakeClock(time.Now()) clock := testclocks.NewFakeClock(time.Now())
recorder := recorderWithFakeClock(v1.EventSource{Component: "eventTest"}, eventBroadcaster, clock) recorder := recorderWithFakeClock(t, v1.EventSource{Component: "eventTest"}, eventBroadcaster, clock)
for index, item := range table { for index, item := range table {
clock.Step(1 * time.Second) clock.Step(1 * time.Second)
@ -955,7 +955,7 @@ func TestMultiSinkCache(t *testing.T) {
eventBroadcaster := NewBroadcasterForTests(0) eventBroadcaster := NewBroadcasterForTests(0)
clock := testclocks.NewFakeClock(time.Now()) clock := testclocks.NewFakeClock(time.Now())
recorder := recorderWithFakeClock(v1.EventSource{Component: "eventTest"}, eventBroadcaster, clock) recorder := recorderWithFakeClock(t, v1.EventSource{Component: "eventTest"}, eventBroadcaster, clock)
sinkWatcher := eventBroadcaster.StartRecordingToSink(&testEvents) sinkWatcher := eventBroadcaster.StartRecordingToSink(&testEvents)
for index, item := range table { for index, item := range table {

7
tools/record/fake.go
View File

@ -20,6 +20,7 @@ import (
"fmt" "fmt"
"k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime"
"k8s.io/klog/v2"
) )
// FakeRecorder is used as a fake during tests. It is thread safe. It is usable // FakeRecorder is used as a fake during tests. It is thread safe. It is usable
@ -31,6 +32,8 @@ type FakeRecorder struct {
IncludeObject bool IncludeObject bool
} }
var _ EventRecorderLogger = &FakeRecorder{}
func objectString(object runtime.Object, includeObject bool) string { func objectString(object runtime.Object, includeObject bool) string {
if !includeObject { if !includeObject {
return "" return ""
@ -68,6 +71,10 @@ func (f *FakeRecorder) AnnotatedEventf(object runtime.Object, annotations map[st
f.writeEvent(object, annotations, eventtype, reason, messageFmt, args...) f.writeEvent(object, annotations, eventtype, reason, messageFmt, args...)
} }
func (f *FakeRecorder) WithLogger(logger klog.Logger) EventRecorderLogger {
return f
}
// NewFakeRecorder creates new fake event recorder with event channel with // NewFakeRecorder creates new fake event recorder with event channel with
// buffer of given size. // buffer of given size.
func NewFakeRecorder(bufferSize int) *FakeRecorder { func NewFakeRecorder(bufferSize int) *FakeRecorder {