Merge pull request kubernetes#129341 from pohly/log-client-go-watch

client-go watch: context support

Author: k8s-ci-robot

staging/src/k8s.io/apiextensions-apiserver/examples/client-go/pkg/client/informers/externalversions/cr/v1/example.go

@@ -51,6 +51,7 @@ type Reporter interface {
 // StreamWatcher turns any stream for which you can write a Decoder interface
 // into a watch.Interface.
 type StreamWatcher struct {
+	logger klog.Logger
 	sync.Mutex
 	source   Decoder
 	reporter Reporter
@@ -59,8 +60,16 @@ type StreamWatcher struct {
 }
 
 // NewStreamWatcher creates a StreamWatcher from the given decoder.
+//
+// Contextual logging: NewStreamWatcherWithLogger should be used instead of NewStreamWatcher in code which supports contextual logging.
 func NewStreamWatcher(d Decoder, r Reporter) *StreamWatcher {
+	return NewStreamWatcherWithLogger(klog.Background(), d, r)
+}
+
+// NewStreamWatcherWithLogger creates a StreamWatcher from the given decoder and logger.
+func NewStreamWatcherWithLogger(logger klog.Logger, d Decoder, r Reporter) *StreamWatcher {
 	sw := &StreamWatcher{
+		logger:   logger,
 		source:   d,
 		reporter: r,
 		// It's easy for a consumer to add buffering via an extra
@@ -98,7 +107,7 @@ func (sw *StreamWatcher) Stop() {
 
 // receive reads result from the decoder in a loop and sends down the result channel.
 func (sw *StreamWatcher) receive() {
-	defer utilruntime.HandleCrash()
+	defer utilruntime.HandleCrashWithLogger(sw.logger)
 	defer close(sw.result)
 	defer sw.Stop()
 	for {
@@ -108,10 +117,10 @@ func (sw *StreamWatcher) receive() {
 			case io.EOF:
 				// watch closed normally
 			case io.ErrUnexpectedEOF:
-				klog.V(1).Infof("Unexpected EOF during watch stream event decoding: %v", err)
+				sw.logger.V(1).Info("Unexpected EOF during watch stream event decoding", "err", err)
 			default:
 				if net.IsProbableEOF(err) || net.IsTimeout(err) {
-					klog.V(5).Infof("Unable to decode an event from the watch stream: %v", err)
+					sw.logger.V(5).Info("Unable to decode an event from the watch stream", "err", err)
 				} else {
 					select {
 					case <-sw.done:

staging/src/k8s.io/apiextensions-apiserver/pkg/client/informers/externalversions/apiextensions/v1/customresourcedefinition.go

@@ -64,6 +64,7 @@ func TestStreamWatcher(t *testing.T) {
 	}
 
 	fd := fakeDecoder{items: make(chan Event, 5)}
+	//nolint:logcheck // Intentionally uses the old API.
 	sw := NewStreamWatcher(fd, nil)
 
 	for _, item := range table {
@@ -87,6 +88,7 @@ func TestStreamWatcher(t *testing.T) {
 func TestStreamWatcherError(t *testing.T) {
 	fd := fakeDecoder{err: fmt.Errorf("test error")}
 	fr := &fakeReporter{}
+	//nolint:logcheck // Intentionally uses the old API.
 	sw := NewStreamWatcher(fd, fr)
 	evt, ok := <-sw.ResultChan()
 	if !ok {
@@ -110,6 +112,7 @@ func TestStreamWatcherError(t *testing.T) {
 func TestStreamWatcherRace(t *testing.T) {
 	fd := fakeDecoder{err: fmt.Errorf("test error")}
 	fr := &fakeReporter{}
+	//nolint:logcheck // Intentionally uses the old API.
 	sw := NewStreamWatcher(fd, fr)
 	time.Sleep(10 * time.Millisecond)
 	sw.Stop()

staging/src/k8s.io/apiextensions-apiserver/pkg/client/informers/externalversions/apiextensions/v1beta1/customresourcedefinition.go

@@ -23,6 +23,7 @@ import (
 	"k8s.io/klog/v2"
 
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/utils/ptr"
 )
 
 // Interface can be implemented by anything that knows how to watch and report changes.
@@ -103,29 +104,42 @@ func (w emptyWatch) ResultChan() <-chan Event {
 
 // FakeWatcher lets you test anything that consumes a watch.Interface; threadsafe.
 type FakeWatcher struct {
+	logger  klog.Logger
 	result  chan Event
 	stopped bool
 	sync.Mutex
 }
 
+var _ Interface = &FakeWatcher{}
+
+// Contextual logging: NewFakeWithOptions and a logger in the FakeOptions should be used instead in code which supports contextual logging.
 func NewFake() *FakeWatcher {
-	return &FakeWatcher{
-		result: make(chan Event),
-	}
+	return NewFakeWithOptions(FakeOptions{})
 }
 
+// Contextual logging: NewFakeWithOptions and a logger in the FakeOptions should be used instead in code which supports contextual logging.
 func NewFakeWithChanSize(size int, blocking bool) *FakeWatcher {
+	return NewFakeWithOptions(FakeOptions{ChannelSize: size})
+}
+
+func NewFakeWithOptions(options FakeOptions) *FakeWatcher {
 	return &FakeWatcher{
-		result: make(chan Event, size),
+		logger: ptr.Deref(options.Logger, klog.Background()),
+		result: make(chan Event, options.ChannelSize),
 	}
 }
 
+type FakeOptions struct {
+	Logger      *klog.Logger
+	ChannelSize int
+}
+
 // Stop implements Interface.Stop().
 func (f *FakeWatcher) Stop() {
 	f.Lock()
 	defer f.Unlock()
 	if !f.stopped {
-		klog.V(4).Infof("Stopping fake watcher.")
+		f.logger.V(4).Info("Stopping fake watcher")
 		close(f.result)
 		f.stopped = true
 	}
@@ -176,13 +190,22 @@ func (f *FakeWatcher) Action(action EventType, obj runtime.Object) {
 
 // RaceFreeFakeWatcher lets you test anything that consumes a watch.Interface; threadsafe.
 type RaceFreeFakeWatcher struct {
+	logger  klog.Logger
 	result  chan Event
 	Stopped bool
 	sync.Mutex
 }
 
+var _ Interface = &RaceFreeFakeWatcher{}
+
+// Contextual logging: RaceFreeFakeWatcherWithLogger should be used instead of NewRaceFreeFake in code which supports contextual logging.
 func NewRaceFreeFake() *RaceFreeFakeWatcher {
+	return NewRaceFreeFakeWithLogger(klog.Background())
+}
+
+func NewRaceFreeFakeWithLogger(logger klog.Logger) *RaceFreeFakeWatcher {
 	return &RaceFreeFakeWatcher{
+		logger: logger,
 		result: make(chan Event, DefaultChanSize),
 	}
 }
@@ -192,7 +215,7 @@ func (f *RaceFreeFakeWatcher) Stop() {
 	f.Lock()
 	defer f.Unlock()
 	if !f.Stopped {
-		klog.V(4).Infof("Stopping fake watcher.")
+		f.logger.V(4).Info("Stopping fake watcher")
 		close(f.result)
 		f.Stopped = true
 	}

staging/src/k8s.io/apimachinery/pkg/watch/streamwatcher.go

@@ -153,13 +153,25 @@ func NewFilteredDynamicInformer(client dynamic.Interface, gvr schema.GroupVersio
 					if tweakListOptions != nil {
 						tweakListOptions(&options)
 					}
-					return client.Resource(gvr).Namespace(namespace).List(context.TODO(), options)
+					return client.Resource(gvr).Namespace(namespace).List(context.Background(), options)
 				},
 				WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
 					if tweakListOptions != nil {
 						tweakListOptions(&options)
 					}
-					return client.Resource(gvr).Namespace(namespace).Watch(context.TODO(), options)
+					return client.Resource(gvr).Namespace(namespace).Watch(context.Background(), options)
+				},
+				ListWithContextFunc: func(ctx context.Context, options metav1.ListOptions) (runtime.Object, error) {
+					if tweakListOptions != nil {
+						tweakListOptions(&options)
+					}
+					return client.Resource(gvr).Namespace(namespace).List(ctx, options)
+				},
+				WatchFuncWithContext: func(ctx context.Context, options metav1.ListOptions) (watch.Interface, error) {
+					if tweakListOptions != nil {
+						tweakListOptions(&options)
+					}
+					return client.Resource(gvr).Namespace(namespace).Watch(ctx, options)
 				},
 			},
 			&unstructured.Unstructured{},