Merge pull request #99344 from wzshiming/fix/server-filters-flake

Fix staging/src/k8s.io/apiserver/pkg/server/filters flake test timeout

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

@@ -380,7 +380,7 @@ func TestPriorityAndFairnessWithPanicRecoveryAndTimeoutFilter(t *testing.T) {
 		})
 		handler := newHandlerChain(t, requestHandler, controller, userName, requestTimeout)
 
-		server, requestGetter := newHTTP2ServerWithClient(handler)
+		server, requestGetter := newHTTP2ServerWithClient(handler, requestTimeout*2)
 		defer server.Close()
 
 		var err error
@@ -388,6 +388,9 @@ func TestPriorityAndFairnessWithPanicRecoveryAndTimeoutFilter(t *testing.T) {
 		if !executed {
 			t.Errorf("Expected inner handler to be executed for request: %q", firstRequestPathPanic)
 		}
+		if isClientTimeout(err) {
+			t.Fatalf("the client has unexpectedly timed out - request: %q error: %s", firstRequestPathPanic, err.Error())
+		}
 		expectResetStreamError(t, err)
 
 		executed = false
@@ -414,8 +417,9 @@ func TestPriorityAndFairnessWithPanicRecoveryAndTimeoutFilter(t *testing.T) {
 	})
 
 	t.Run("priority level concurrency is set to 1, request times out and inner handler hasn't written to the response yet", func(t *testing.T) {
+		t.Parallel()
 		const (
-			requestTimeout                                        = 3 * time.Second
+			requestTimeout                                        = 5 * time.Second
 			userName                                              = "alice"
 			fsName                                                = "test-fs"
 			plName                                                = "test-pl"
@@ -442,9 +446,14 @@ func TestPriorityAndFairnessWithPanicRecoveryAndTimeoutFilter(t *testing.T) {
 		})
 		handler := newHandlerChain(t, requestHandler, controller, userName, requestTimeout)
 
-		server, requestGetter := newHTTP2ServerWithClient(handler)
+		server, requestGetter := newHTTP2ServerWithClient(handler, requestTimeout*2)
 		defer server.Close()
 
+		go hardStop(t, stopCh, requestTimeout*3, func() {
+			// the client timed out unexpectedly, let's clean up and abort.
+			close(reqHandlerCompletedCh)
+		})
+
 		var (
 			response *http.Response
 			err      error
@@ -454,6 +463,9 @@ func TestPriorityAndFairnessWithPanicRecoveryAndTimeoutFilter(t *testing.T) {
 
 			t.Logf("Waiting for the request: %q to time out", rquestTimesOutPath)
 			response, err = requestGetter(rquestTimesOutPath)
+			if isClientTimeout(err) {
+				t.Fatalf("the client has unexpectedly timed out - request: %q error: %s", rquestTimesOutPath, err.Error())
+			}
 		}()
 
 		if !executed {
@@ -479,8 +491,9 @@ func TestPriorityAndFairnessWithPanicRecoveryAndTimeoutFilter(t *testing.T) {
 	})
 
 	t.Run("priority level concurrency is set to 1, inner handler panics after the request times out", func(t *testing.T) {
+		t.Parallel()
 		const (
-			requestTimeout                                        = 3 * time.Second
+			requestTimeout                                        = 5 * time.Second
 			userName                                              = "alice"
 			fsName                                                = "test-fs"
 			plName                                                = "test-pl"
@@ -511,9 +524,14 @@ func TestPriorityAndFairnessWithPanicRecoveryAndTimeoutFilter(t *testing.T) {
 		})
 		handler := newHandlerChain(t, requestHandler, controller, userName, requestTimeout)
 
-		server, requestGetter := newHTTP2ServerWithClient(handler)
+		server, requestGetter := newHTTP2ServerWithClient(handler, requestTimeout*2)
 		defer server.Close()
 
+		go hardStop(t, stopCh, requestTimeout*3, func() {
+			// the client timed out unexpectedly, let's clean up and abort.
+			close(reqHandlerCompletedCh)
+		})
+
 		var (
 			response *http.Response
 			err      error
@@ -522,6 +540,9 @@ func TestPriorityAndFairnessWithPanicRecoveryAndTimeoutFilter(t *testing.T) {
 			defer close(callerRoundTripDoneCh)
 			t.Logf("Waiting for the request: %q to time out", rquestTimesOutPath)
 			response, err = requestGetter(rquestTimesOutPath)
+			if isClientTimeout(err) {
+				t.Fatalf("the client has unexpectedly timed out - request: %q error: %s", rquestTimesOutPath, err.Error())
+			}
 		}()
 
 		t.Logf("Waiting for the inner handler of the request: %q to complete", rquestTimesOutPath)
@@ -544,8 +565,9 @@ func TestPriorityAndFairnessWithPanicRecoveryAndTimeoutFilter(t *testing.T) {
 	})
 
 	t.Run("priority level concurrency is set to 1, inner handler writes to the response before request times out", func(t *testing.T) {
+		t.Parallel()
 		const (
-			requestTimeout                                        = 3 * time.Second
+			requestTimeout                                        = 5 * time.Second
 			userName                                              = "alice"
 			fsName                                                = "test-fs"
 			plName                                                = "test-pl"
@@ -577,14 +599,22 @@ func TestPriorityAndFairnessWithPanicRecoveryAndTimeoutFilter(t *testing.T) {
 		})
 		handler := newHandlerChain(t, requestHandler, controller, userName, requestTimeout)
 
-		server, requestGetter := newHTTP2ServerWithClient(handler)
+		server, requestGetter := newHTTP2ServerWithClient(handler, requestTimeout*2)
 		defer server.Close()
 
+		go hardStop(t, stopCh, requestTimeout*3, func() {
+			// the client timed out unexpectedly, let's clean up and abort.
+			close(reqHandlerCompletedCh)
+		})
+
 		var err error
 		func() {
 			defer close(callerRoundTripDoneCh)
 			t.Logf("Waiting for the request: %q to time out", rquestTimesOutPath)
 			_, err = requestGetter(rquestTimesOutPath)
+			if isClientTimeout(err) {
+				t.Fatalf("the client has unexpectedly timed out - request: %q error: %s", rquestTimesOutPath, err.Error())
+			}
 		}()
 
 		t.Logf("Waiting for the inner handler of the request: %q to complete", rquestTimesOutPath)
@@ -602,8 +632,9 @@ func TestPriorityAndFairnessWithPanicRecoveryAndTimeoutFilter(t *testing.T) {
 	})
 
 	t.Run("priority level concurrency is set to 1, queue length is 1, first request should time out and second (enqueued) request should time out as well", func(t *testing.T) {
+		t.Parallel()
 		const (
-			requestTimeout                                                     = 3 * time.Second
+			requestTimeout                                                     = 5 * time.Second
 			userName                                                           = "alice"
 			fsName                                                             = "test-fs"
 			plName                                                             = "test-pl"
@@ -646,15 +677,24 @@ func TestPriorityAndFairnessWithPanicRecoveryAndTimeoutFilter(t *testing.T) {
 		})
 		handler := newHandlerChain(t, requestHandler, controller, userName, requestTimeout)
 
-		server, requestGetter := newHTTP2ServerWithClient(handler)
+		server, requestGetter := newHTTP2ServerWithClient(handler, requestTimeout*2)
 		defer server.Close()
 
+		go hardStop(t, stopCh, requestTimeout*3, func() {
+			// the client timed out unexpectedly, let's clean up and abort.
+			close(firstReqInProgressCh)
+			close(firstReqHandlerCompletedCh)
+		})
+
 		var firstReqErr, secondReqErr error
 		var resp1, resp2 *http.Response
 		go func() {
 			defer close(firstReqRoundTripDoneCh)
 			t.Logf("Waiting for the request: %q to time out", firstRequestTimesOutPath)
 			resp1, firstReqErr = requestGetter(firstRequestTimesOutPath)
+			if isClientTimeout(firstReqErr) {
+				t.Fatalf("the client has unexpectedly timed out - request: %q error: %s", firstRequestTimesOutPath, firstReqErr.Error())
+			}
 		}()
 		func() {
 			defer close(secondReqRoundTripDoneCh)
@@ -662,6 +702,9 @@ func TestPriorityAndFairnessWithPanicRecoveryAndTimeoutFilter(t *testing.T) {
 			// we must wait for the "first" request to start executing first
 			<-firstReqInProgressCh
 			resp2, secondReqErr = requestGetter(secondRequestEnqueuedPath)
+			if isClientTimeout(secondReqErr) {
+				t.Fatalf("the client has unexpectedly timed out - request: %q error: %s", secondRequestEnqueuedPath, secondReqErr.Error())
+			}
 		}()
 
 		<-firstReqRoundTripDoneCh
@@ -736,13 +779,14 @@ func startAPFController(t *testing.T, stopCh <-chan struct{}, apfConfiguration [
 }
 
 // returns a started http2 server, with a client function to send request to the server.
-func newHTTP2ServerWithClient(handler http.Handler) (*httptest.Server, func(path string) (*http.Response, error)) {
+func newHTTP2ServerWithClient(handler http.Handler, clientTimeout time.Duration) (*httptest.Server, func(path string) (*http.Response, error)) {
 	server := httptest.NewUnstartedServer(handler)
 	server.EnableHTTP2 = true
 	server.StartTLS()
-
+	cli := server.Client()
+	cli.Timeout = clientTimeout
 	return server, func(path string) (*http.Response, error) {
-		return server.Client().Get(server.URL + path)
+		return cli.Get(server.URL + path)
 	}
 }
 
@@ -949,3 +993,20 @@ func gaugeValueMatch(name string, labelFilter map[string]string, wantValue int)
 
 	return nil
 }
+
+func isClientTimeout(err error) bool {
+	if urlErr, ok := err.(*url.Error); ok {
+		return urlErr.Timeout()
+	}
+	return false
+}
+
+func hardStop(t *testing.T, stopCh <-chan struct{}, testTimeout time.Duration, cleanup func()) {
+	select {
+	case <-stopCh:
+		// The test has completed normally.
+	case <-time.After(testTimeout):
+		cleanup()
+		t.Fatalf("the test did not finish within %s", testTimeout)
+	}
+}