Merge pull request #127089 from sanposhiho/revert-126574

Revert: Fix data race in APF tests

staging/src/k8s.io/apiserver/pkg/server/filters/priority-and-fairness_test.go

@@ -18,7 +18,6 @@ package filters
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"net/http"
 	"net/http/httptest"
@@ -27,7 +26,6 @@ import (
 	"reflect"
 	"strings"
 	"sync"
-	"sync/atomic"
 	"testing"
 	"time"
 
@@ -130,20 +128,20 @@ func newApfServerWithSingleRequest(t *testing.T, decision mockDecision) *httptes
 			t.Errorf("execute should not be invoked")
 		}
 		// atomicReadOnlyExecuting can be either 0 or 1 as we test one request at a time.
-		if want, got := int32(1), atomic.LoadInt32(&atomicReadOnlyExecuting); decision != decisionSkipFilter && want != got {
+		if decision != decisionSkipFilter && atomicReadOnlyExecuting != 1 {
-			t.Errorf("Wanted %d requests executing, got %d", want, got)
+			t.Errorf("Wanted %d requests executing, got %d", 1, atomicReadOnlyExecuting)
 		}
 	}
 	postExecuteFunc := func() {}
 	// atomicReadOnlyWaiting can be either 0 or 1 as we test one request at a time.
 	postEnqueueFunc := func() {
-		if want, got := int32(1), atomic.LoadInt32(&atomicReadOnlyWaiting); want != got {
+		if atomicReadOnlyWaiting != 1 {
-			t.Errorf("Wanted %d requests in queue, got %d", want, got)
+			t.Errorf("Wanted %d requests in queue, got %d", 1, atomicReadOnlyWaiting)
 		}
 	}
 	postDequeueFunc := func() {
-		if want, got := int32(0), atomic.LoadInt32(&atomicReadOnlyWaiting); want != got {
+		if atomicReadOnlyWaiting != 0 {
-			t.Errorf("Wanted %d requests in queue, got %d", want, got)
+			t.Errorf("Wanted %d requests in queue, got %d", 0, atomicReadOnlyWaiting)
 		}
 	}
 	return newApfServerWithHooks(t, decision, onExecuteFunc, postExecuteFunc, postEnqueueFunc, postDequeueFunc)
@@ -179,19 +177,11 @@ func newApfHandlerWithFilter(t *testing.T, flowControlFilter utilflowcontrol.Int
 		r = r.WithContext(apirequest.WithUser(r.Context(), &user.DefaultInfo{
 			Groups: []string{user.AllUnauthenticated},
 		}))
-		func() {
-			// the defer ensures that the following assertion is
-			// executed, even if the APF handler panics
-			// TODO: all test(s) using this filter must run serially to each other
-			defer func() {
-				t.Logf("the APF handler has finished, checking atomicReadOnlyExecuting")
-				if want, got := int32(0), atomic.LoadInt32(&atomicReadOnlyExecuting); want != got {
-					t.Errorf("Wanted %d requests executing, got %d", want, got)
-				}
-			}()
 		apfHandler.ServeHTTP(w, r)
 		postExecute()
-		}()
+		if atomicReadOnlyExecuting != 0 {
+			t.Errorf("Wanted %d requests executing, got %d", 0, atomicReadOnlyExecuting)
+		}
 	}), requestInfoFactory)
 
 	return handler
@@ -280,8 +270,8 @@ func TestApfExecuteMultipleRequests(t *testing.T) {
 	onExecuteFunc := func() {
 		preStartExecute.Done()
 		preStartExecute.Wait()
-		if want, got := int32(concurrentRequests), atomic.LoadInt32(&atomicReadOnlyExecuting); want != got {
+		if int(atomicReadOnlyExecuting) != concurrentRequests {
-			t.Errorf("Wanted %d requests executing, got %d", want, got)
+			t.Errorf("Wanted %d requests executing, got %d", concurrentRequests, atomicReadOnlyExecuting)
 		}
 		postStartExecute.Done()
 		postStartExecute.Wait()
@@ -290,8 +280,8 @@ func TestApfExecuteMultipleRequests(t *testing.T) {
 	postEnqueueFunc := func() {
 		preEnqueue.Done()
 		preEnqueue.Wait()
-		if want, got := int32(concurrentRequests), atomic.LoadInt32(&atomicReadOnlyWaiting); want != got {
+		if int(atomicReadOnlyWaiting) != concurrentRequests {
-			t.Errorf("Wanted %d requests in queue, got %d", want, got)
+			t.Errorf("Wanted %d requests in queue, got %d", 1, atomicReadOnlyWaiting)
 
 		}
 		postEnqueue.Done()
@@ -301,8 +291,8 @@ func TestApfExecuteMultipleRequests(t *testing.T) {
 	postDequeueFunc := func() {
 		preDequeue.Done()
 		preDequeue.Wait()
-		if want, got := int32(0), atomic.LoadInt32(&atomicReadOnlyWaiting); want != got {
+		if atomicReadOnlyWaiting != 0 {
-			t.Errorf("Wanted %d requests in queue, got %d", want, got)
+			t.Errorf("Wanted %d requests in queue, got %d", 0, atomicReadOnlyWaiting)
 		}
 		postDequeue.Done()
 		postDequeue.Wait()
@@ -355,21 +345,19 @@ func TestApfCancelWaitRequest(t *testing.T) {
 }
 
 type fakeWatchApfFilter struct {
-	t        *testing.T
 	lock     sync.Mutex
 	inflight int
 	capacity int
 
-	postExecutePanic error
+	postExecutePanic bool
-	preExecutePanic  error
+	preExecutePanic  bool
 
 	utilflowcontrol.WatchTracker
 	utilflowcontrol.MaxSeatsTracker
 }
 
-func newFakeWatchApfFilter(t *testing.T, capacity int) *fakeWatchApfFilter {
+func newFakeWatchApfFilter(capacity int) *fakeWatchApfFilter {
 	return &fakeWatchApfFilter{
-		t:               t,
 		capacity:        capacity,
 		WatchTracker:    utilflowcontrol.NewWatchTracker(),
 		MaxSeatsTracker: utilflowcontrol.NewMaxSeatsTracker(),
@@ -397,23 +385,17 @@ func (f *fakeWatchApfFilter) Handle(ctx context.Context,
 		return
 	}
 
-	func() {
+	if f.preExecutePanic {
-		defer func() {
+		panic("pre-exec-panic")
+	}
+	execFn()
+	if f.postExecutePanic {
+		panic("post-exec-panic")
+	}
+
 	f.lock.Lock()
 	defer f.lock.Unlock()
 	f.inflight--
-		}()
-
-		if f.preExecutePanic != nil {
-			f.t.Logf("going to panic (pre-exec) as expected with error: %v, fakeWatchApfFilter: %#v", f.preExecutePanic, f)
-			panic(f.preExecutePanic)
-		}
-		execFn()
-		if f.postExecutePanic != nil {
-			f.t.Logf("going to panic (post-exec) as expected with error: %v, fakeWatchApfFilter: %#v", f.postExecutePanic, f)
-			panic(f.postExecutePanic)
-		}
-	}()
 }
 
 func (f *fakeWatchApfFilter) Run(stopCh <-chan struct{}) error {
@@ -465,7 +447,7 @@ func TestApfExecuteWatchRequestsWithInitializationSignal(t *testing.T) {
 	allRunning := sync.WaitGroup{}
 	allRunning.Add(2 * concurrentRequests)
 
-	fakeFilter := newFakeWatchApfFilter(t, concurrentRequests)
+	fakeFilter := newFakeWatchApfFilter(concurrentRequests)
 
 	onExecuteFunc := func() {
 		firstRunning.Done()
@@ -511,7 +493,7 @@ func TestApfExecuteWatchRequestsWithInitializationSignal(t *testing.T) {
 }
 
 func TestApfRejectWatchRequestsWithInitializationSignal(t *testing.T) {
-	fakeFilter := newFakeWatchApfFilter(t, 0)
+	fakeFilter := newFakeWatchApfFilter(0)
 
 	onExecuteFunc := func() {
 		t.Errorf("Request unexepectedly executing")
@@ -530,7 +512,7 @@ func TestApfWatchPanic(t *testing.T) {
 	epmetrics.Register()
 	fcmetrics.Register()
 
-	fakeFilter := newFakeWatchApfFilter(t, 1)
+	fakeFilter := newFakeWatchApfFilter(1)
 
 	onExecuteFunc := func() {
 		panic("test panic")
@@ -557,11 +539,11 @@ func TestApfWatchPanic(t *testing.T) {
 func TestApfWatchHandlePanic(t *testing.T) {
 	epmetrics.Register()
 	fcmetrics.Register()
-	preExecutePanicingFilter := newFakeWatchApfFilter(t, 1)
+	preExecutePanicingFilter := newFakeWatchApfFilter(1)
-	preExecutePanicingFilter.preExecutePanic = http.ErrAbortHandler
+	preExecutePanicingFilter.preExecutePanic = true
 
-	postExecutePanicingFilter := newFakeWatchApfFilter(t, 1)
+	postExecutePanicingFilter := newFakeWatchApfFilter(1)
-	postExecutePanicingFilter.postExecutePanic = http.ErrAbortHandler
+	postExecutePanicingFilter.postExecutePanic = true
 
 	testCases := []struct {
 		name   string
@@ -577,31 +559,18 @@ func TestApfWatchHandlePanic(t *testing.T) {
 		},
 	}
 
-	for _, test := range testCases {
-		t.Run(test.name, func(t *testing.T) {
 	onExecuteFunc := func() {
 		time.Sleep(5 * time.Second)
-
-				// this function should not be executed if
-				// pre-execute panic is set
-				if test.filter.preExecutePanic != nil {
-					t.Errorf("did not expect the execute function to be executed")
-				}
-				t.Logf("on-execute function invoked")
-			}
-
-			// we either panic before the execute function, or after,
-			// so the following function should never be executed.
-			postExecuteFunc := func() {
-				t.Errorf("did not expect the post-execute function to be invoked")
 	}
+	postExecuteFunc := func() {}
 
+	for _, test := range testCases {
+		t.Run(test.name, func(t *testing.T) {
 			apfHandler := newApfHandlerWithFilter(t, test.filter, time.Minute/4, onExecuteFunc, postExecuteFunc)
 			handler := func(w http.ResponseWriter, r *http.Request) {
 				defer func() {
-					recovered := recover()
+					if err := recover(); err == nil {
-					if err, ok := recovered.(error); !ok || !errors.Is(err, http.ErrAbortHandler) {
+						t.Errorf("expected panic, got %v", err)
-						t.Errorf("expected panic with error: %v, but got: %v", http.ErrAbortHandler, err)
 					}
 				}()
 				apfHandler.ServeHTTP(w, r)