Update dependency go modules for k8s v1.29.0-rc.1

vendor/k8s.io/client-go/tools/cache/reflector.go

@@ -334,12 +334,9 @@ func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
 			return nil
 		}
 		if err != nil {
-			if !apierrors.IsInvalid(err) {
-				return err
-			}
-			klog.Warning("the watch-list feature is not supported by the server, falling back to the previous LIST/WATCH semantic")
+			klog.Warningf("The watchlist request ended with an error, falling back to the standard LIST/WATCH semantics because making progress is better than deadlocking, err = %v", err)
 			fallbackToList = true
-			// Ensure that we won't accidentally pass some garbage down the watch.
+			// ensure that we won't accidentally pass some garbage down the watch.
 			w = nil
 		}
 	}
@@ -351,6 +348,8 @@ func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
 		}
 	}
 
+	klog.V(2).Infof("Caches populated for %v from %s", r.typeDescription, r.name)
+
 	resyncerrc := make(chan error, 1)
 	cancelCh := make(chan struct{})
 	defer close(cancelCh)
@@ -395,6 +394,11 @@ func (r *Reflector) watch(w watch.Interface, stopCh <-chan struct{}, resyncerrc
 		// give the stopCh a chance to stop the loop, even in case of continue statements further down on errors
 		select {
 		case <-stopCh:
+			// we can only end up here when the stopCh
+			// was closed after a successful watchlist or list request
+			if w != nil {
+				w.Stop()
+			}
 			return nil
 		default:
 		}
@@ -670,6 +674,12 @@ func (r *Reflector) watchList(stopCh <-chan struct{}) (watch.Interface, error) {
 	// "k8s.io/initial-events-end" bookmark.
 	initTrace.Step("Objects streamed", trace.Field{Key: "count", Value: len(temporaryStore.List())})
 	r.setIsLastSyncResourceVersionUnavailable(false)
+
+	// we utilize the temporaryStore to ensure independence from the current store implementation.
+	// as of today, the store is implemented as a queue and will be drained by the higher-level
+	// component as soon as it finishes replacing the content.
+	checkWatchListConsistencyIfRequested(stopCh, r.name, resourceVersion, r.listerWatcher, temporaryStore)
+
 	if err = r.store.Replace(temporaryStore.List(), resourceVersion); err != nil {
 		return nil, fmt.Errorf("unable to sync watch-list result: %v", err)
 	}
@@ -762,7 +772,7 @@ loop:
 				}
 			case watch.Bookmark:
 				// A `Bookmark` means watch has synced here, just update the resourceVersion
-				if _, ok := meta.GetAnnotations()["k8s.io/initial-events-end"]; ok {
+				if meta.GetAnnotations()["k8s.io/initial-events-end"] == "true" {
 					if exitOnInitialEventsEndBookmark != nil {
 						*exitOnInitialEventsEndBookmark = true
 					}

vendor/k8s.io/client-go/tools/cache/reflector_data_consistency_detector.go

@@ -0,0 +1,119 @@
+/*
+Copyright 2023 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 cache
+
+import (
+	"context"
+	"os"
+	"sort"
+	"strconv"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/klog/v2"
+)
+
+var dataConsistencyDetectionEnabled = false
+
+func init() {
+	dataConsistencyDetectionEnabled, _ = strconv.ParseBool(os.Getenv("KUBE_WATCHLIST_INCONSISTENCY_DETECTOR"))
+}
+
+// checkWatchListConsistencyIfRequested performs a data consistency check only when
+// the KUBE_WATCHLIST_INCONSISTENCY_DETECTOR environment variable was set during a binary startup.
+//
+// The consistency check is meant to be enforced only in the CI, not in production.
+// The check ensures that data retrieved by the watch-list api call
+// is exactly the same as data received by the standard list api call.
+//
+// Note that this function will panic when data inconsistency is detected.
+// This is intentional because we want to catch it in the CI.
+func checkWatchListConsistencyIfRequested(stopCh <-chan struct{}, identity string, lastSyncedResourceVersion string, listerWatcher Lister, store Store) {
+	if !dataConsistencyDetectionEnabled {
+		return
+	}
+	checkWatchListConsistency(stopCh, identity, lastSyncedResourceVersion, listerWatcher, store)
+}
+
+// checkWatchListConsistency exists solely for testing purposes.
+// we cannot use checkWatchListConsistencyIfRequested because
+// it is guarded by an environmental variable.
+// we cannot manipulate the environmental variable because
+// it will affect other tests in this package.
+func checkWatchListConsistency(stopCh <-chan struct{}, identity string, lastSyncedResourceVersion string, listerWatcher Lister, store Store) {
+	klog.Warningf("%s: data consistency check for the watch-list feature is enabled, this will result in an additional call to the API server.", identity)
+	opts := metav1.ListOptions{
+		ResourceVersion:      lastSyncedResourceVersion,
+		ResourceVersionMatch: metav1.ResourceVersionMatchExact,
+	}
+	var list runtime.Object
+	err := wait.PollUntilContextCancel(wait.ContextForChannel(stopCh), time.Second, true, func(_ context.Context) (done bool, err error) {
+		list, err = listerWatcher.List(opts)
+		if err != nil {
+			// the consistency check will only be enabled in the CI
+			// and LIST calls in general will be retired by the client-go library
+			// if we fail simply log and retry
+			klog.Errorf("failed to list data from the server, retrying until stopCh is closed, err: %v", err)
+			return false, nil
+		}
+		return true, nil
+	})
+	if err != nil {
+		klog.Errorf("failed to list data from the server, the watch-list consistency check won't be performed, stopCh was closed, err: %v", err)
+		return
+	}
+
+	rawListItems, err := meta.ExtractListWithAlloc(list)
+	if err != nil {
+		panic(err) // this should never happen
+	}
+
+	listItems := toMetaObjectSliceOrDie(rawListItems)
+	storeItems := toMetaObjectSliceOrDie(store.List())
+
+	sort.Sort(byUID(listItems))
+	sort.Sort(byUID(storeItems))
+
+	if !cmp.Equal(listItems, storeItems) {
+		klog.Infof("%s: data received by the new watch-list api call is different than received by the standard list api call, diff: %v", identity, cmp.Diff(listItems, storeItems))
+		msg := "data inconsistency detected for the watch-list feature, panicking!"
+		panic(msg)
+	}
+}
+
+type byUID []metav1.Object
+
+func (a byUID) Len() int           { return len(a) }
+func (a byUID) Less(i, j int) bool { return a[i].GetUID() < a[j].GetUID() }
+func (a byUID) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
+
+func toMetaObjectSliceOrDie[T any](s []T) []metav1.Object {
+	result := make([]metav1.Object, len(s))
+	for i, v := range s {
+		m, err := meta.Accessor(v)
+		if err != nil {
+			panic(err)
+		}
+		result[i] = m
+	}
+	return result
+}

vendor/k8s.io/client-go/tools/cache/shared_informer.go

@@ -334,11 +334,9 @@ func WaitForCacheSync(stopCh <-chan struct{}, cacheSyncs ...InformerSynced) bool
 		},
 		stopCh)
 	if err != nil {
-		klog.V(2).Infof("stop requested")
 		return false
 	}
 
-	klog.V(4).Infof("caches populated")
 	return true
 }
 

vendor/k8s.io/client-go/tools/clientcmd/merged_client_builder.go

@@ -49,12 +49,12 @@ type InClusterConfig interface {
 	Possible() bool
 }
 
-// NewNonInteractiveDeferredLoadingClientConfig creates a ConfigClientClientConfig using the passed context name
+// NewNonInteractiveDeferredLoadingClientConfig creates a ClientConfig using the passed context name
 func NewNonInteractiveDeferredLoadingClientConfig(loader ClientConfigLoader, overrides *ConfigOverrides) ClientConfig {
 	return &DeferredLoadingClientConfig{loader: loader, overrides: overrides, icc: &inClusterClientConfig{overrides: overrides}}
 }
 
-// NewInteractiveDeferredLoadingClientConfig creates a ConfigClientClientConfig using the passed context name and the fallback auth reader
+// NewInteractiveDeferredLoadingClientConfig creates a ClientConfig using the passed context name and the fallback auth reader
 func NewInteractiveDeferredLoadingClientConfig(loader ClientConfigLoader, overrides *ConfigOverrides, fallbackReader io.Reader) ClientConfig {
 	return &DeferredLoadingClientConfig{loader: loader, overrides: overrides, icc: &inClusterClientConfig{overrides: overrides}, fallbackReader: fallbackReader}
 }

vendor/k8s.io/client-go/tools/internal/events/interfaces.go

@@ -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
+}

vendor/k8s.io/client-go/tools/record/event.go

@@ -29,6 +29,7 @@ import (
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/watch"
 	restclient "k8s.io/client-go/rest"
+	internalevents "k8s.io/client-go/tools/internal/events"
 	"k8s.io/client-go/tools/record/util"
 	ref "k8s.io/client-go/tools/reference"
 	"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{})
 }
 
+// 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.
 type EventBroadcaster interface {
 	// 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
 	// 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
 	// 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
 // implementing the new "k8s.io/client-go/tools/events".EventRecorder interface.
 type EventRecorderAdapter struct {
-	recorder EventRecorder
+	recorder EventRecorderLogger
 }
 
+var _ internalevents.EventRecorder = &EventRecorderAdapter{}
+
 // NewEventRecorderAdapter returns an adapter implementing the new
 // "k8s.io/client-go/tools/events".EventRecorder interface.
-func NewEventRecorderAdapter(recorder EventRecorder) *EventRecorderAdapter {
+func NewEventRecorderAdapter(recorder EventRecorderLogger) *EventRecorderAdapter {
 	return &EventRecorderAdapter{
 		recorder: recorder,
 	}
@@ -158,28 +176,76 @@ func (a *EventRecorderAdapter) Eventf(regarding, _ runtime.Object, eventtype, re
 	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.
-func NewBroadcaster() EventBroadcaster {
-	return newEventBroadcaster(watch.NewLongQueueBroadcaster(maxQueuedEvents, watch.DropIfChannelFull), defaultSleepDuration)
+func NewBroadcaster(opts ...BroadcasterOption) EventBroadcaster {
+	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 {
-	return newEventBroadcaster(watch.NewLongQueueBroadcaster(maxQueuedEvents, watch.DropIfChannelFull), sleepDuration)
+	return NewBroadcaster(WithSleepDuration(sleepDuration))
 }
 
 func NewBroadcasterWithCorrelatorOptions(options CorrelatorOptions) EventBroadcaster {
-	eventBroadcaster := newEventBroadcaster(watch.NewLongQueueBroadcaster(maxQueuedEvents, watch.DropIfChannelFull), defaultSleepDuration)
-	eventBroadcaster.options = options
-	return eventBroadcaster
+	return NewBroadcaster(WithCorrelatorOptions(options))
 }
 
-func newEventBroadcaster(broadcaster *watch.Broadcaster, sleepDuration time.Duration) *eventBroadcasterImpl {
-	eventBroadcaster := &eventBroadcasterImpl{
-		Broadcaster:   broadcaster,
-		sleepDuration: sleepDuration,
+func WithCorrelatorOptions(options CorrelatorOptions) BroadcasterOption {
+	return func(c *config) {
+		c.CorrelatorOptions = options
 	}
-	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 {
@@ -220,12 +286,12 @@ func (e *eventBroadcasterImpl) recordToSink(sink EventSink, event *v1.Event, eve
 	}
 	tries := 0
 	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
 		}
 		tries++
 		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
 		}
 
@@ -237,7 +303,7 @@ func (e *eventBroadcasterImpl) recordToSink(sink EventSink, event *v1.Event, eve
 		}
 		select {
 		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
 		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.
 // If updateExistingEvent is false, it creates a new event, otherwise it updates
 // 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 err error
 	if updateExistingEvent {
@@ -271,13 +337,13 @@ func recordEvent(sink EventSink, event *v1.Event, patch []byte, updateExistingEv
 	switch err.(type) {
 	case *restclient.RequestConstructionError:
 		// 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
 	case *errors.StatusError:
 		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 {
-			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
 	case *errors.UnexpectedObjectError:
@@ -286,7 +352,7 @@ func recordEvent(sink EventSink, event *v1.Event, patch []byte, updateExistingEv
 	default:
 		// 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
 }
 
@@ -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.
 func (e *eventBroadcasterImpl) StartStructuredLogging(verbosity klog.Level) watch.Interface {
+	loggerV := klog.FromContext(e.cancelationCtx).V(int(verbosity))
 	return e.StartEventWatcher(
 		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 {
 	watcher, err := e.Watch()
 	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() {
 		defer utilruntime.HandleCrash()
-		for watchEvent := range watcher.ResultChan() {
-			event, ok := watchEvent.Object.(*v1.Event)
-			if !ok {
-				// This is all local, so there's no reason this should
-				// ever happen.
-				continue
+		for {
+			select {
+			case <-e.cancelationCtx.Done():
+				watcher.Stop()
+				return
+			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
 }
 
 // NewRecorder returns an EventRecorder that records events with the given event source.
-func (e *eventBroadcasterImpl) NewRecorder(scheme *runtime.Scheme, source v1.EventSource) EventRecorder {
-	return &recorderImpl{scheme, source, e.Broadcaster, clock.RealClock{}}
+func (e *eventBroadcasterImpl) NewRecorder(scheme *runtime.Scheme, source v1.EventSource) EventRecorderLogger {
+	return &recorderImplLogger{recorderImpl: &recorderImpl{scheme, source, e.Broadcaster, clock.RealClock{}}, logger: klog.Background()}
 }
 
 type recorderImpl struct {
@@ -342,15 +417,17 @@ type recorderImpl struct {
 	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)
 	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
 	}
 
 	if !util.ValidateEventType(eventtype) {
-		klog.Errorf("Unsupported event type: '%v'", eventtype)
+		logger.Error(nil, "Unsupported event type", "eventType", eventtype)
 		return
 	}
 
@@ -367,16 +444,16 @@ func (recorder *recorderImpl) generateEvent(object runtime.Object, annotations m
 	// outgoing events anyway).
 	sent, err := recorder.ActionOrDrop(watch.Added, event)
 	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
 	}
 	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) {
-	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{}) {
@@ -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{}) {
-	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 {
@@ -408,3 +485,26 @@ func (recorder *recorderImpl) makeEvent(ref *v1.ObjectReference, annotations map
 		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}
+}

vendor/k8s.io/client-go/tools/record/fake.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/klog/v2"
 )
 
 // FakeRecorder is used as a fake during tests. It is thread safe. It is usable
@@ -31,6 +32,8 @@ type FakeRecorder struct {
 	IncludeObject bool
 }
 
+var _ EventRecorderLogger = &FakeRecorder{}
+
 func objectString(object runtime.Object, includeObject bool) string {
 	if !includeObject {
 		return ""
@@ -68,6 +71,10 @@ func (f *FakeRecorder) AnnotatedEventf(object runtime.Object, annotations map[st
 	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
 // buffer of given size.
 func NewFakeRecorder(bufferSize int) *FakeRecorder {