client-go watch: context support #129341

pohly · 2024-12-20T17:49:39Z

What type of PR is this?

/kind cleanup

What this PR does / why we need it:

The Lister and Watcher interfaces only supported methods without context, but
were typically implemented with client-go API calls which need a context. New
interfaces get added using the same approach as in
#129109.

For StreamWatcher and RaceFreeFakeWatcher it was easier to add WithLogger
configuration methods which override the default global logger. The downside of
this approach is that it cannot be enforced by logcheck. This is okay for
Kubernetes in this case because all existing usages of those two types are
local to client-go and get updated immediately.

Which issue(s) this PR fixes:

Part of #129125

Special notes for your reviewer:

Generated informers: for compatibility reasons, the old functions without the ctx parameter still
get generated, now with context.Background instead of context.TODO. In practice
that code won't be used by the client-go reflector code because it prefers
the *WithContext functions, but it cannot be ruled out that some other code
only supports the old fields.

Does this PR introduce a user-facing change?

NONE

/wg structured-logging

sttts · 2025-01-30T09:55:54Z

staging/src/k8s.io/client-go/tools/watch/retrywatcher.go

-	default:
-		close(rw.stopChan)
-	}
+	rw.cancel(errors.New("asked to stop"))


The Lister and Watcher interfaces only supported methods without context, but were typically implemented with client-go API calls which need a context. New interfaces get added using the same approach as in kubernetes#129109.

For compatibility reasons, the old functions without the ctx parameter still get generated, now with context.Background instead of context.TODO. In practice that code won't be used by the client-go reflector code because it prefers the *WithContext functions, but it cannot be ruled out that some other code only supports the old fields.

soltysh · 2025-01-30T12:08:50Z

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

+}
+
+// NewStreamWatcherWithLogger creates a StreamWatcher from the given decoder and logger.
+func NewStreamWatcherWithLogger(logger klog.Logger, d Decoder, r Reporter) *StreamWatcher {


I'd like to avoid a situation where we have *WithLogger and then *WithContext like we have in

kubernetes/staging/src/k8s.io/apimachinery/pkg/util/runtime/runtime.go

Lines 46 to 93 in 07cc230

// HandleCrash simply catches a crash and logs an error. Meant to be called via

// defer. Additional context-specific handlers can be provided, and will be

// called in case of panic. HandleCrash actually crashes, after calling the

// handlers and logging the panic message.

//

// E.g., you can provide one or more additional handlers for something like shutting down go routines gracefully.

//

// Contextual logging: HandleCrashWithContext or HandleCrashWithLogger should be used instead of HandleCrash in code which supports contextual logging.

func HandleCrash(additionalHandlers ...func(interface{})) {

if r := recover(); r != nil {

additionalHandlersWithContext := make([]func(context.Context, interface{}), len(additionalHandlers))

for i, handler := range additionalHandlers {

handler := handler // capture loop variable

additionalHandlersWithContext[i] = func(_ context.Context, r interface{}) {

handler(r)

}

}

handleCrash(context.Background(), r, additionalHandlersWithContext...)

}

}

// HandleCrashWithContext simply catches a crash and logs an error. Meant to be called via

// defer. Additional context-specific handlers can be provided, and will be

// called in case of panic. HandleCrash actually crashes, after calling the

// handlers and logging the panic message.

//

// E.g., you can provide one or more additional handlers for something like shutting down go routines gracefully.

//

// The context is used to determine how to log.

func HandleCrashWithContext(ctx context.Context, additionalHandlers ...func(context.Context, interface{})) {

if r := recover(); r != nil {

handleCrash(ctx, r, additionalHandlers...)

}

}

// HandleCrashWithLogger simply catches a crash and logs an error. Meant to be called via

// defer. Additional context-specific handlers can be provided, and will be

// called in case of panic. HandleCrash actually crashes, after calling the

// handlers and logging the panic message.

//

// E.g., you can provide one or more additional handlers for something like shutting down go routines gracefully.

func HandleCrashWithLogger(logger klog.Logger, additionalHandlers ...func(context.Context, interface{})) {

if r := recover(); r != nil {

ctx := klog.NewContext(context.Background(), logger)

handleCrash(ctx, r, additionalHandlers...)

}

}

although that is a bit different situation. Anyway, StreamWatcher has a done channel, which we allow users to invoke through Stop method. So the question is do we want to just pass a context and wire it such that both Stop and ctx.Done are equivalent? We've done something similar in

kubernetes/staging/src/k8s.io/client-go/tools/cache/controller.go

Lines 143 to 148 in 07cc230

func (c *controller) Run(stopCh <-chan struct{}) {

c.RunWithContext(wait.ContextForChannel(stopCh))

}

// RunWithContext implements [Controller.RunWithContext].

func (c *controller) RunWithContext(ctx context.Context) {

for example.

A NewStreamWatcherWithContext could use a similar construct as NewIndexerInformerWatcherWithContext in this PR to map context cancellation to Stop. It definitely would be more consistent and IMHO useful.

I don't remember any particular reason for deciding not to do that also here. I'll change it.

However, such a change then begs the question: should the context that is getting passed to this NewStreamWatcherWithContext be the same context that is passed to Request.Watch?

kubernetes/staging/src/k8s.io/client-go/rest/request.go

Lines 756 to 761 in 07cc230

// Watch attempts to begin watching the requested location.

// Returns a watch.Interface, or an error.

func (r *Request) Watch(ctx context.Context) (watch.Interface, error) {

w, _, e := r.watchInternal(ctx)

return w, e

}

The call chain is Watch -> watchInternal -> newStreamWatcher -> NewStreamWatcherWithLogger.

That would be a functional change of Watch with regards to how it uses the context that it is being passed. Someone might currently pass a short-lived context that gets canceled after Watch returns, which then breaks the returned StreamWatcher.

So newStreamWatcher would have to call NewStreamWatcherWithContext with klog.NewContext(context.Background(), klog.FromContext(ctx)). Doable, but more complicated.

This problem does not exist for NewIndexerInformerWatcherWithContext in our code base because where the PR passes the parent's context the watcher is expected to stop before the caller returns:

kubernetes/staging/src/k8s.io/client-go/tools/watch/until.go

Lines 128 to 135 in 07cc230

func UntilWithSync(ctx context.Context, lw cache.ListerWatcher, objType runtime.Object, precondition PreconditionFunc, conditions ...ConditionFunc) (*watch.Event, error) {

indexer, informer, watcher, done := NewIndexerInformerWatcher(lw, objType)

// We need to wait for the internal informers to fully stop so it's easier to reason about

// and it works with non-thread safe clients.

defer func() { <-done }()

// Proxy watcher can be stopped multiple times so it's fine to use defer here to cover alternative branches and

// let UntilWithoutRetry to stop it

defer watcher.Stop()

Should NewIndexerInformerWatcherWithContext perhaps get replaced by a simpler NewIndexerInformerWatcherWithLogger?

Watch is and always will be long running function, even the doc says on Watch method says:

// Watch attempts to begin watching the requested location.

so it's a begin, not a short-lived request. So I'm inclined to say it's not a functional change, but maybe that's a question for sig-apimachinery?

It doesn't say anything about the context, though. Usually, contexts are only passed through functions call, so my expectation as a Go developer is that the context is only used while the Watch call runs.

For example, code might set a timeout like this to stop the attempt to set up the watch:

ctx, cancel := context.Timeout(time.Minute) defer cancel() return req.Watch(ctx)

Should NewIndexerInformerWatcherWithContext perhaps get replaced by a simpler NewIndexerInformerWatcherWithLogger?

I convinced myself that this is the right way to go and therefore I included it in this PR as an additional commit. I can squash or remove it, depending on how we decide to do this.

We haven't heard from sig-apimachinery, but given we're early in 1.33 cycle, I don't want to block the work here. So we'll continue the discussion on slack but this can move forward

Can you LGTM again? My last commit merely updated the unit test for the no longer supported context cancellation instead of removing it completely - fixed. Sorry, should have checked test results before pinging you!

soltysh · 2025-01-30T12:15:03Z

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

 	return &FakeWatcher{
-		result: make(chan Event, size),
+		logger: ptr.Deref(options.Logger, klog.Background()),


Identical comment as the other wrt passing a context rather than logger here, which you can then wire to stopping the watcher.

soltysh

/lgtm

k8s-ci-robot · 2025-02-04T13:03:59Z

LGTM label has been added.

Git tree hash: 5e2caa3eef455cd5a9a7031a900043e772997756

k8s-ci-robot · 2025-02-04T14:36:57Z

@pohly: The following tests failed, say /retest to rerun all failed tests or /retest-required to rerun all mandatory failed tests:

Test name	Commit	Details	Required	Rerun command
pull-kubernetes-apidiff	`a54f7fd`	link	false	`/test pull-kubernetes-apidiff`
pull-kubernetes-apidiff-client-go	`1a8d8c9`	link	false	`/test pull-kubernetes-apidiff-client-go`

Full PR test history. Your PR dashboard. Please help us cut down on flakes by linking to an open issue when you hit one in your PR.

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. I understand the commands that are listed here.

pohly · 2025-02-04T14:36:57Z

/assign @deads2k @sttts

For approval.

The ability to automatically stop on context cancellation was new functionality that adds complexity and wasn't really used in Kubernetes. If someone wants this, they can add it outside of the function. A *WithLogger variant avoids the complexity and is consistent with NewStreamWatcherWithLogger over in apimachinery.

sttts · 2025-02-07T08:21:41Z

/lgtm
/approve

k8s-ci-robot · 2025-02-07T08:21:48Z

LGTM label has been added.

Git tree hash: c2fed5f35b3b68e3612ea556f7dec32119303cf9

k8s-ci-robot · 2025-02-07T08:21:57Z

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: pohly, soltysh, sttts

The full list of commands accepted by this bot can be found here.

The pull request process is described here

Needs approval from an approver in each of these files:

~~staging/src/k8s.io/apiextensions-apiserver/OWNERS~~ [sttts]
~~staging/src/k8s.io/apimachinery/pkg/OWNERS~~ [sttts]
~~staging/src/k8s.io/client-go/OWNERS~~ [sttts]
~~staging/src/k8s.io/code-generator/OWNERS~~ [sttts]
~~staging/src/k8s.io/kube-aggregator/OWNERS~~ [sttts]
~~staging/src/k8s.io/sample-apiserver/OWNERS~~ [sttts]
~~staging/src/k8s.io/sample-controller/OWNERS~~ [sttts]

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

wzshiming · 2025-04-29T11:44:46Z

kubernetes/staging/src/k8s.io/client-go/tools/cache/listwatch.go

Lines 134 to 148 in 5be7941

    
           // ListFunc knows how to list resources 
        
           // 
        
           // Deprecated: use ListWithContextFunc instead. 
        
           type ListFunc func(options metav1.ListOptions) (runtime.Object, error) 
        
           // ListWithContextFunc knows how to list resources 
        
           type ListWithContextFunc func(ctx context.Context, options metav1.ListOptions) (runtime.Object, error) 
        
           // WatchFunc knows how to watch resources 
        
           // 
        
           // Deprecated: use WatchFuncWithContext instead. 
        
           type WatchFunc func(options metav1.ListOptions) (watch.Interface, error) 
        
           // WatchFuncWithContext knows how to watch resources 
        
           type WatchFuncWithContext func(ctx context.Context, options metav1.ListOptions) (watch.Interface, error)

I noticed a slight inconsistency in naming conventions while reviewing the context-aware interfaces:

ListWithContextFunc uses WithContext as an infix
WatchFuncWithContext uses WithContext as a suffix

I sent a PR for fix this #131529

pohly changed the title ~~client-go watch: context support~~ WIP: client-go watch: context support Dec 20, 2024

k8s-ci-robot added the wg/device-management Categorizes an issue or PR as relevant to WG Device Management. label Dec 20, 2024

k8s-ci-robot requested review from aojea, caesarxuchao and a team December 20, 2024 17:50

k8s-ci-robot added the do-not-merge/work-in-progress Indicates that a PR should not merge because it is a work in progress. label Dec 20, 2024

This was referenced Dec 20, 2024

client-go event: add WithContext expansion methods #129343

Merged

apimachinery: contextual logging in network util code #129344

Open

client-go workqueue: add optional logger #129345

Merged

apimachinery wait: support contextual logging #129346

Merged

pohly added a commit to pohly/kubernetes that referenced this pull request Dec 20, 2024

merge log-client-go-watch (kubernetes#129341)

46686da

sttts reviewed Jan 30, 2025

View reviewed changes

pohly added 2 commits January 30, 2025 11:54

client-go + apimachinery watch: context support

6688ada

The Lister and Watcher interfaces only supported methods without context, but were typically implemented with client-go API calls which need a context. New interfaces get added using the same approach as in kubernetes#129109.

pohly force-pushed the log-client-go-watch branch from ba2d90b to 8cc74e8 Compare January 30, 2025 11:00

soltysh reviewed Jan 30, 2025

View reviewed changes

soltysh approved these changes Feb 4, 2025

View reviewed changes

k8s-ci-robot assigned soltysh Feb 4, 2025

k8s-ci-robot added the lgtm "Looks good to me", indicates that a PR is ready to be merged. label Feb 4, 2025

pohly force-pushed the log-client-go-watch branch from 91f8403 to 1a8d8c9 Compare February 4, 2025 14:33

k8s-ci-robot removed the lgtm "Looks good to me", indicates that a PR is ready to be merged. label Feb 4, 2025

k8s-ci-robot requested review from rexagod and soltysh February 4, 2025 14:33

k8s-ci-robot assigned deads2k and sttts Feb 4, 2025

k8s-ci-robot added the lgtm "Looks good to me", indicates that a PR is ready to be merged. label Feb 7, 2025

k8s-ci-robot added the approved Indicates a PR has been approved by an approver from all required OWNERS files. label Feb 7, 2025

k8s-ci-robot merged commit a02fe24 into kubernetes:master Feb 7, 2025
13 of 14 checks passed

k8s-ci-robot added this to the v1.33 milestone Feb 7, 2025

soltysh mentioned this pull request Feb 7, 2025

restmapper + discovery: add context-aware APIs #129109

Open

hashim21223445 approved these changes Feb 9, 2025

View reviewed changes

wzshiming mentioned this pull request Apr 29, 2025

client-go listwatch: Add WatchWithContextFunc and deprecate WatchFuncWithContext #131529

Closed

pohly mentioned this pull request Apr 29, 2025

Added stacktrace and logging for list and watch panics in reflector.go #131000

Open

	// HandleCrash simply catches a crash and logs an error. Meant to be called via
	// defer. Additional context-specific handlers can be provided, and will be
	// called in case of panic. HandleCrash actually crashes, after calling the
	// handlers and logging the panic message.
	//
	// E.g., you can provide one or more additional handlers for something like shutting down go routines gracefully.
	//
	// Contextual logging: HandleCrashWithContext or HandleCrashWithLogger should be used instead of HandleCrash in code which supports contextual logging.
	func HandleCrash(additionalHandlers ...func(interface{})) {
	if r := recover(); r != nil {
	additionalHandlersWithContext := make([]func(context.Context, interface{}), len(additionalHandlers))
	for i, handler := range additionalHandlers {
	handler := handler // capture loop variable
	additionalHandlersWithContext[i] = func(_ context.Context, r interface{}) {
	handler(r)
	}
	}

	handleCrash(context.Background(), r, additionalHandlersWithContext...)
	}
	}

	// HandleCrashWithContext simply catches a crash and logs an error. Meant to be called via
	// defer. Additional context-specific handlers can be provided, and will be
	// called in case of panic. HandleCrash actually crashes, after calling the
	// handlers and logging the panic message.
	//
	// E.g., you can provide one or more additional handlers for something like shutting down go routines gracefully.
	//
	// The context is used to determine how to log.
	func HandleCrashWithContext(ctx context.Context, additionalHandlers ...func(context.Context, interface{})) {
	if r := recover(); r != nil {
	handleCrash(ctx, r, additionalHandlers...)
	}
	}

	// HandleCrashWithLogger simply catches a crash and logs an error. Meant to be called via
	// defer. Additional context-specific handlers can be provided, and will be
	// called in case of panic. HandleCrash actually crashes, after calling the
	// handlers and logging the panic message.
	//
	// E.g., you can provide one or more additional handlers for something like shutting down go routines gracefully.
	func HandleCrashWithLogger(logger klog.Logger, additionalHandlers ...func(context.Context, interface{})) {
	if r := recover(); r != nil {
	ctx := klog.NewContext(context.Background(), logger)
	handleCrash(ctx, r, additionalHandlers...)
	}
	}

	func (c *controller) Run(stopCh <-chan struct{}) {
	c.RunWithContext(wait.ContextForChannel(stopCh))
	}

	// RunWithContext implements [Controller.RunWithContext].
	func (c *controller) RunWithContext(ctx context.Context) {

	// Watch attempts to begin watching the requested location.
	// Returns a watch.Interface, or an error.
	func (r *Request) Watch(ctx context.Context) (watch.Interface, error) {
	w, _, e := r.watchInternal(ctx)
	return w, e
	}

	func UntilWithSync(ctx context.Context, lw cache.ListerWatcher, objType runtime.Object, precondition PreconditionFunc, conditions ...ConditionFunc) (*watch.Event, error) {
	indexer, informer, watcher, done := NewIndexerInformerWatcher(lw, objType)
	// We need to wait for the internal informers to fully stop so it's easier to reason about
	// and it works with non-thread safe clients.
	defer func() { <-done }()
	// Proxy watcher can be stopped multiple times so it's fine to use defer here to cover alternative branches and
	// let UntilWithoutRetry to stop it
	defer watcher.Stop()

client-go watch: context support #129341

client-go watch: context support #129341

Uh oh!

Conversation

pohly commented Dec 20, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What type of PR is this?

What this PR does / why we need it:

Which issue(s) this PR fixes:

Special notes for your reviewer:

Does this PR introduce a user-facing change?

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

pohly Jan 30, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

soltysh left a comment

Choose a reason for hiding this comment

Uh oh!

k8s-ci-robot commented Feb 4, 2025

Uh oh!

k8s-ci-robot commented Feb 4, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

pohly commented Feb 4, 2025

Uh oh!

sttts commented Feb 7, 2025

Uh oh!

k8s-ci-robot commented Feb 7, 2025

Uh oh!

k8s-ci-robot commented Feb 7, 2025

Uh oh!

Uh oh!

wzshiming commented Apr 29, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

pohly commented Dec 20, 2024 •

edited

Loading

pohly Jan 30, 2025 •

edited

Loading

k8s-ci-robot commented Feb 4, 2025 •

edited

Loading

wzshiming commented Apr 29, 2025 •

edited

Loading