Stop Watching when there is encoding error

[tedyu]

What type of PR is this?
/kind cleanup

What this PR does / why we need it:
In WatchServer#HandleWS, if s.EmbeddedEncoder.Encode() fails, s.Watching should be stopped.
This would make encoding error handling consistent with that of s.Encoder.Encode().

kube-apiserver: fixed a bug that could cause a goroutine leak if the apiserver encountered an encoding error serving a watch to a websocket watcher

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

[tedyu]

/assign @liggitt

[caesarxuchao]

/unassign @liggitt
/assign @lavalamp

[tedyu]

/test pull-kubernetes-kubemark-e2e-gce-big

[lavalamp]

/lgtm
/approve

Tests seem broken for unrelated reason. Is there a fix if you rebase?

/retest

[tedyu]

/priority important-soon

[lavalamp]

Good call @liggitt, I'm not sure. Maybe. We should understand the contract better; I can't dig in at the moment. It'd be best if we could defer the call and be certain it'll be called....

[tedyu]

@liggitt
See line 204 w.r.t. the other two places you mentioned.

	defer s.Watching.Stop()

[liggitt]

See line 204 w.r.t. the other two places you mentioned.

defer s.Watching.Stop()

ah, you're right. this seems like it should do the same in this method (but can be a follow up if we want to keep this minimal for back-porting)

[tedyu]

@liggitt @lavalamp
I have pushed new patch where defer is used - this would prevent bugs in the future if some new case is added.
(In the initial comment from Jordan above there was no 'follow up')

[lavalamp]

Is it OK to double-stop this thing? (e.g. for line 298 it is likely already stopped)

[liggitt]

FWIW, that's exactly the path exercised by line 204/226 above in the normal case of the watch channel closing

[lavalamp]

Nifty. Let's get the merge train started and we can verify in parallel.

[lavalamp]

/lgtm

[liggitt]

picks opened to 1.14, 1.15, 1.16

[answer1991]

@tedyu @lavalamp

Could you provide some more detail logs about this bug, as I found even the code has some defect, the watcher will be closed finally.

Because the watcher implement by cacherWatcher is context-aware, when request finished and context is done, the watcher will call watcher.Stop:

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)
	})

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

Lines 1360 to 1375 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
	}
	}
	}

[lavalamp]

@answer1991 Thanks for the investigation; when was the watcher made context aware? I think some versions still in support don't have that change? And this is definitely easy to cherry-pick.

I think we need to have metrics, one for "running watches" and one for "in-flight requests", so we can compare and definitively see that watches are not being leaked.

@tedyu still looking for a test :)

[tedyu]

@lavalamp
Adding test is on my agenda.

[answer1991]

@answer1991 Thanks for the investigation; when was the watcher made context aware? I think some versions still in support don't have that change? And this is definitely easy to cherry-pick.

I think we need to have metrics, one for "running watches" and one for "in-flight requests", so we can compare and definitively see that watches are not being leaked.

@tedyu still looking for a test :)

From v1.10, watcher implement by cacherWatcher had already been context-aware(I did not check earlier codes). But the test codes, FakeWatcher is not context-aware. I think the memory leak is only exist in test codes.