Repository: kubernetes
Issue: Goroutine leak in filteredWatch
Labels: kind/bug,sig/api-machinery,triage/accepted

### What happened?

I created a some ListWatches with filters to filter out uninteresting events before processing. And I did this by introducing the filteredWatch in the apimachinery.
https://github.com/kubernetes/kubernetes/blob/master/staging/src/k8s.io/apimachinery/pkg/watch/filter.go#45
```
type filteredWatch struct {
	incoming Interface
	result   chan Event
	f        FilterFunc
}
```
And then I stopped ListWatches. Of which I invoked the 'Stop' function to close filteredWatches.
https://github.com/kubernetes/kubernetes/blob/master/staging/src/k8s.io/apimachinery/pkg/watch/filter.go#56
```
// Stop stops the upstream watch, which will eventually stop this watch.
func (fw *filteredWatch) Stop() {
	fw.incoming.Stop()
}
```
I repeated the process many times. Then I noticed that my program memory was increasing.

After analysis, it is found that the problem is caused by incomplete resources release of filteredWatch.

I created the filteredWatch object using the Filter function. In which a blocking channel called 'result' and a goroutine called 'loop' were created.
https://github.com/kubernetes/kubernetes/blob/master/staging/src/k8s.io/apimachinery/pkg/watch/filter.go#35
```
func Filter(w Interface, f FilterFunc) Interface {
	fw := &filteredWatch{
		incoming: w,
		result:   make(chan Event),
		f:        f,
	}
	go fw.loop()
	return fw
}
```

https://github.com/kubernetes/kubernetes/blob/master/staging/src/k8s.io/apimachinery/pkg/watch/filter.go#61
```
// loop waits for new values, filters them, and resends them.
func (fw *filteredWatch) loop() {
	defer close(fw.result)
	for event := range fw.incoming.ResultChan() {
		filtered, keep := fw.f(event)
		if keep {
			fw.result <- filtered
		}
	}
}
```

Assume that there were two events(A & B) in the incoming channel. They all meet the filter.

Time 1: A was popped from the fw.incoming.ResultChan() and was sent to the fw.result.
Time 2: B was just popped from the fw.incoming.ResultChan().
Time 3: The Stop function of the filteredWatch was invoked and the fw.incoming.ResultChan() was closed.

Since A has not been taken from the fw.result, the loop goroutine will stuck at line 'fw.result <- filtered' and leads to goroutine leaks.

### What did you expect to happen?

No goroutine leak when filteredWatch is closed.

### How can we reproduce it (as minimally and precisely as possible)?

We can use a test case to reproduce the problem.
1. Add some observable points to the loop method.
```
var (
	PopFromIncoming = make(chan struct{}) // added
	ExitLoop = make(chan struct{}) // added
)

// loop waits for new values, filters them, and resends them.
func (fw *filteredWatch) loop() {
	defer close(fw.result)
	for event := range fw.incoming.ResultChan() {
		PopFromIncoming <- struct{}{} // added
		filtered, keep := fw.f(event)
		if keep {
			fw.result <- filtered
		}
	}
	ExitLoop <- struct{}{} // added
}
```
2. Add a test case.
```
func TestFilterLoopStuck(t *testing.T) {
	source := NewFake()
	filtered := Filter(source, func(e Event) (Event, bool) {
		return e, e.Object.(testType)[0] != 'b'
	})

	source.Add(testType("foo"))

	<-PopFromIncoming // Ensure event was popped from the fw.incoming.ResultChan()

	<-filtered.ResultChan() // Take out event

	filtered.Stop()
	<-ExitLoop
}
```
When we take out event from the filtered.ResultChan() before closing the filteredWatch, the test case do not stuck.
![image](https://user-images.githubusercontent.com/112469714/197201261-df06ff9e-faac-4d7d-8a09-c2405fc220e5.png)
And when we directly close filteredWatch (Simulate the scenario where the ListWatch is closed without handling the event in a timely manner), the use case is stuck.
![image](https://user-images.githubusercontent.com/112469714/197201874-64d4c11a-8cf5-4a49-9b54-ae2a21a41fba.png)

### Anything else we need to know?

_No response_

### Kubernetes version

kubernetes GitVersion: v1.22.1

### Cloud provider

<details>

</details>


### OS version

<details>

```console
# On Linux:
$ cat /etc/os-release
# paste output here
$ uname -a
# paste output here

# On Windows:
C:\> wmic os get Caption, Version, BuildNumber, OSArchitecture
# paste output here
```

</details>


### Install tools

<details>

</details>


### Container runtime (CRI) and version (if applicable)

<details>

</details>


### Related plugins (CNI, CSI, ...) and versions (if applicable)

<details>

</details>