fix potential memory leak issue in processing watch request

[answer1991]

What type of PR is this?

/kind bug

What this PR does / why we need it:

Fix API Server potential memory leak issue in processing watch request

Which issue(s) this PR fixes:

Related Issue: #84001
Related PR: #84693

Special notes for your reviewer:

After #84693 picked, API Server processing watch request still may leak memory(does not stop watcher).

Does this PR introduce a user-facing change?:

Fix API Server potential memory leak issue in processing watch request.

Additional documentation e.g., KEPs (Kubernetes Enhancement Proposals), usage docs, etc.:

None

[k8s-ci-robot]

Hi @answer1991. Thanks for your PR.

I'm waiting for a kubernetes member to verify that this patch is reasonable to test. If it is, they should reply with /ok-to-test on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should join the org to skip this step.

Once the patch is verified, the new status will be reflected by the ok-to-test label.

I understand the commands that are listed here.

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/test-infra repository.

[answer1991]

CC @lavalamp @tedyu

[neolit123]

/ok-to-test
/priority backlog

[lavalamp]

/lgtm
/approve

Let's cherry pick this back. Thanks!

[k8s-ci-robot]

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: answer1991, lavalamp

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/apiserver/OWNERS [lavalamp]

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

[hzxuzhonghu]

Doesn't the cancel almost equals to watcher.Stop here?

kubernetes/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/get.go

Lines 252 to 262 in 81af5ba

	ctx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()
	watcher, err := rw.Watch(ctx, &opts)
	if err != nil {
	scope.err(err, w, req)
	return
	}
	requestInfo, _ := request.RequestInfoFrom(ctx)
	metrics.RecordLongRunning(req, requestInfo, metrics.APIServerComponent, func() {
	serveWatch(watcher, scope, outputMediaType, req, w, timeout)
	})

[tedyu]

Why is this addition needed ?

[answer1991]

watcher.Stop has moved to serveWatch function. The test codes now is tricky, but I have not found a better way to call watcher.Stop in test codes.

[riking]

But you're stopping the watcher on any request, instead of just the relevant ones.

[answer1991]

/hold

Let me refactor codes again.

[answer1991]

@tedyu @riking

I had already improved the PR, many thanks for a review.

BTW, I do not think it's a bug any more, as the real Watcher will be closed if context is done:

kubernetes/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go

Lines 1360 to 1376 in c97baa3

	defer c.Stop()
	for {
	select {
	case event, ok := <-c.input:
	if !ok {
	return
	}
	// only send events newer than resourceVersion
	if event.ResourceVersion > resourceVersion {
	c.sendWatchCacheEvent(event)
	}
	case <-ctx.Done():
	return
	}
	}
	}

[wojtek-t]

@answer1991 - the test failure seems related

[answer1991]

@wojtek-t

The tests passed in my environment. Let me try run test again, if also failed let me debug try to resolve it :-P

/test pull-kubernetes-bazel-test

[answer1991]

@wojtek-t

Found the test failed root cause, we should call FakeWatcher.IsStopped() instead of FakeWatcher.Stopped: #86120

/hold cancel

[wojtek-t]

/lgtm

Thanks!

[tedyu]

I ran the new test after reverting the change to staging/src/k8s.io/apiserver/pkg/endpoints/handlers/watch.go

The test still passed.

[answer1991]

@tedyu

Yes, it is. As we dose not test serveWatch in staging/src/k8s.io/apiserver/pkg/endpoints/handlers/watch.go, but the func serveWatch has a mock in

kubernetes/staging/src/k8s.io/apiserver/pkg/endpoints/watch_test.go

Lines 611 to 624 in 1fd2137

	// serveWatch will serve a watch response according to the watcher and watchServer.
	// Before watchServer.ServeHTTP, an error may occur like k8s.io/apiserver/pkg/endpoints/handlers/watch.go#serveWatch does.
	func serveWatch(watcher watch.Interface, watchServer *handlers.WatchServer, preServeErr error) http.HandlerFunc {
	return func(w http.ResponseWriter, req *http.Request) {
	defer watcher.Stop()
	if preServeErr != nil {
	responsewriters.ErrorNegotiated(preServeErr, watchServer.Scope.Serializer, watchServer.Scope.Kind.GroupVersion(), w, req)
	return
	}
	watchServer.ServeHTTP(w, req)
	}
	}

If we use the older method to ServeHTTP, then the test case TestWatchHTTPErrorsBeforeServe will be failed:

	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
                if preServeErr != nil { 
 			responsewriters.ErrorNegotiated(preServeErr, watchServer.Scope.Serializer, watchServer.Scope.Kind.GroupVersion(), w, req) 
 			return 
 		} 
		watchServer.ServeHTTP(w, req)
	}))
	defer s.Close()

[tedyu]

With the above modification:

$ GO111MODULE=on go test -run=TestWatchHTTPErrors
^Csignal: interrupt
FAIL	k8s.io/apiserver/pkg/endpoints	129.222s

I would expect that existing tests would pass.

[answer1991]

@tedyu

If you want existing tests would pass, you should call serveWatch function which wrapper defer watcher.Stop().

Call watchServer.ServeHTTP directly will leak watcher goroutine and test will be failed, for both cases TestWatchHTTPErrors and TestWatchHTTPErrorsBeforeServe. As we had moved defer watcher.Stop() to serveWatch function.

[jfbai]

Hi @answer1991 , why do we need this when we already have the watcher to be closed after timeout in cacheWatcher.process().

[MikeSpreitzer]

+1 on #85410 (comment)
As far as I can tell, this fix was never put in the earlier supported releases.
@lavalamp
@answer1991

[wojtek-t]

It's part of 1.18, so with 1.19 out, we can just cherrypick it to 1.17. But that sounds reasonable to me.

@answer1991 - can you please open 1.17 cherrypick?

[answer1991]

@wojtek-t sure, is there any automated command in github to help do this? or should I cherry-pick it manually？

[wojtek-t]

https://github.com/kubernetes/kubernetes/blob/master/hack/cherry_pick_pull.sh

[MikeSpreitzer]

I thought I heard that our policy now is to support the previous 3 minor releases.

[wojtek-t]

Isn't it that starting with 1.19 we will be supporting a release for 1 year? So it will be effectively 4 releases, but when 1.19 will become the oldest one?

[MikeSpreitzer]

I see I remembered wrong. https://kubernetes.io/docs/setup/release/version-skew-policy/#supported-versions has the statement. So 1.17 is the oldest supported one now.