Repository: kubernetes/kubernetes (language: go)

Issue:
[Flaking Test] UT k8s.io/apiserver/pkg/storage: etcd3 TestWatchErrResultNotBlockAfterCancel
### Which jobs are flaking?

https://prow.k8s.io/job-history/gs/kubernetes-ci-logs/logs/ci-kubernetes-unit
https://prow.k8s.io/job-history/gs/kubernetes-ci-logs/logs/ci-kubernetes-unit-ppc64le

### Which tests are flaking?

k8s.io/apiserver/pkg/storage: etcd3 TestWatchErrResultNotBlockAfterCancel

### Since when has it been flaking?

Probably since https://github.com/kubernetes/kubernetes/pull/131162 merged.

### Testgrid link

https://testgrid.k8s.io/sig-release-master-blocking#ci-kubernetes-unit

### Reason for failure (if possible)

```
{Failed  === RUN   TestWatchErrResultNotBlockAfterCancel
==================
WARNING: DATA RACE
Read at 0x00c0027dbf90 by goroutine 14383:
  ??()
      -:0 +0x6fafc
  k8s.io/apiserver/pkg/storage/etcd3.(*watchChan).sendError()
      /home/prow/go/src/k8s.io/kubernetes/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher.go:651 +0x13c
  k8s.io/apiserver/pkg/storage/etcd3.TestWatchErrResultNotBlockAfterCancel.func2()
      /home/prow/go/src/k8s.io/kubernetes/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher_test.go:189 +0x60

Previous write at 0x00c0027dbf90 by goroutine 14382:
  ??()
      -:0 +0x7065c
  k8s.io/apiserver/pkg/storage/etcd3.(*watchChan).run()
      /home/prow/go/src/k8s.io/kubernetes/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher.go:245 +0x21c
  k8s.io/apiserver/pkg/storage/etcd3.TestWatchErrResultNotBlockAfterCancel.func1()
      /home/prow/go/src/k8s.io/kubernetes/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher_test.go:184 +0x44

Goroutine 14383 (running) created at:
  k8s.io/apiserver/pkg/storage/etcd3.TestWatchErrResultNotBlockAfterCancel()
      /home/prow/go/src/k8s.io/kubernetes/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher_test.go:188 +0x338
  k8s.io/apiserver/pkg/storage/etcd3.TestWatchErrResultNotBlockAfterCancel()
      /home/prow/go/src/k8s.io/kubernetes/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher_test.go:172 +0x50
  testing.tRunner()
      /usr/local/go/src/testing/testing.go:1792 +0x1b0
  testing.(*T).Run.gowrap1()
      /usr/local/go/src/testing/testing.go:1851 +0x64

Goroutine 14382 (finished) created at:
  k8s.io/apiserver/pkg/storage/etcd3.TestWatchErrResultNotBlockAfterCancel()
      /home/prow/go/src/k8s.io/kubernetes/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher_test.go:183 +0x27c
  k8s.io/apiserver/pkg/storage/etcd3.TestWatchErrResultNotBlockAfterCancel()
      /home/prow/go/src/k8s.io/kubernetes/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher_test.go:172 +0x50
  testing.tRunner()
      /usr/local/go/src/testing/testing.go:1792 +0x1b0
  testing.(*T).Run.gowrap1()
      /usr/local/go/src/testing/testing.go:1851 +0x64
==================
panic: send on closed channel

goroutine 14379 [running]:
k8s.io/apiserver/pkg/storage/etcd
3.(*watchChan).sendError(0xc002d08270, {0x2c9a638, 0xc003a824c0})
	/home/prow/go/src/k8s.io/kubernetes/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher.go:650 +0x16c
k8s.io/apiserver/pkg/storage/etcd3.TestWatchErrResultNotBlockAfterCancel.func2()
	/home/prow/go/src/k8s.io/kubernetes/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher_test.go:189 +0x64
created by k8s.io/apiserver/pkg/storage/etcd3.TestWatchErrResultNotBlockAfterCancel in goroutine 14263
	/home/prow/go/src/k8s.io/kubernetes/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher_test.go:188 +0x33c
}
```

### Anything else we need to know?

_No response_

### Relevant SIG(s)

/sig testing
/sig storage


Discussion / hints:
This issue is currently awaiting triage.

If a SIG or subproject determines this is a relevant issue, they will accept it by applying the `triage/accepted` label and provide further guidance.

The `triage/accepted` label can be added by org members by writing `/triage accepted` in a comment.


<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>
cc @wojtek-t 

Fix patch:
diff --git a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher.go b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher.go
index 17cc24f0db85a..b006f60844abb 100644
--- a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher.go
+++ b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher.go
@@ -226,9 +226,13 @@ func isCancelError(err error) bool {
 
 func (wc *watchChan) run(initialEventsEndBookmarkRequired, forceInitialEvents bool) {
 	watchClosedCh := make(chan struct{})
-	go wc.startWatching(watchClosedCh, initialEventsEndBookmarkRequired, forceInitialEvents)
-
 	var resultChanWG sync.WaitGroup
+
+	resultChanWG.Add(1)
+	go func() {
+		defer resultChanWG.Done()
+		wc.startWatching(watchClosedCh, initialEventsEndBookmarkRequired, forceInitialEvents)
+	}()
 	wc.processEvents(&resultChanWG)
 
 	select {


Test patch:
diff --git a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher_test.go b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher_test.go
index a72ba7dec583b..f31afd89cf0bd 100644
--- a/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher_test.go
+++ b/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher_test.go
@@ -20,7 +20,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"sync"
 	"testing"
 	"time"
 
@@ -168,32 +167,6 @@ func TestWatchErrorEventIsBlockingFurtherEvent(t *testing.T) {
 // impact.
 // =======================================================================
 
-func TestWatchErrResultNotBlockAfterCancel(t *testing.T) {
-	origCtx, store, _ := testSetup(t)
-	ctx, cancel := context.WithCancel(origCtx)
-	w := store.watcher.createWatchChan(ctx, "/abc", 0, false, false, storage.Everything)
-	// make resultChan and errChan blocking to ensure ordering.
-	w.resultChan = make(chan watch.Event)
-	// The event flow goes like:
-	// - first we send an error, it should block on resultChan.
-	// - Then we cancel ctx. The blocking on resultChan should be freed up
-	//   and run() goroutine should return.
-	var wg sync.WaitGroup
-	wg.Add(1)
-	go func() {
-		w.run(false, true)
-		wg.Done()
-	}()
-	wg.Add(1)
-	go func() {
-		w.sendError(fmt.Errorf("some error"))
-		wg.Done()
-	}()
-	cancel()
-	// Ensure that both run() and sendError() don't hung forever.
-	wg.Wait()
-}
-
 // TestWatchErrorIncorrectConfiguration checks if an error
 // will be returned when the storage hasn't been properly
 // initialised for watch requests