Merge pull request #127029 from tkashem/apf-fix-watch-panic-handling

apf: request handler must wait for watch init goroutine to return

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

@@ -266,17 +266,23 @@ func (h *priorityAndFairnessHandler) Handle(w http.ResponseWriter, r *http.Reque
 
 		select {
 		case <-shouldStartWatchCh:
-			watchCtx := utilflowcontrol.WithInitializationSignal(ctx, watchInitializationSignal)
-			watchReq = r.WithContext(watchCtx)
-			h.handler.ServeHTTP(w, watchReq)
-			// Protect from the situation when request will not reach storage layer
-			// and the initialization signal will not be send.
-			// It has to happen before waiting on the resultCh below.
-			watchInitializationSignal.Signal()
-			// TODO: Consider finishing the request as soon as Handle call panics.
-			if err := <-resultCh; err != nil {
-				panic(err)
-			}
+			func() {
+				// TODO: if both goroutines panic, propagate the stack traces from both
+				// goroutines so they are logged properly:
+				defer func() {
+					// Protect from the situation when request will not reach storage layer
+					// and the initialization signal will not be send.
+					// It has to happen before waiting on the resultCh below.
+					watchInitializationSignal.Signal()
+					// TODO: Consider finishing the request as soon as Handle call panics.
+					if err := <-resultCh; err != nil {
+						panic(err)
+					}
+				}()
+				watchCtx := utilflowcontrol.WithInitializationSignal(ctx, watchInitializationSignal)
+				watchReq = r.WithContext(watchCtx)
+				h.handler.ServeHTTP(w, watchReq)
+			}()
 		case err := <-resultCh:
 			if err != nil {
 				panic(err)

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

@@ -177,11 +177,21 @@ func newApfHandlerWithFilter(t *testing.T, flowControlFilter utilflowcontrol.Int
 		r = r.WithContext(apirequest.WithUser(r.Context(), &user.DefaultInfo{
 			Groups: []string{user.AllUnauthenticated},
 		}))
-		apfHandler.ServeHTTP(w, r)
-		postExecute()
-		if atomicReadOnlyExecuting != 0 {
-			t.Errorf("Wanted %d requests executing, got %d", 0, atomicReadOnlyExecuting)
-		}
+		func() {
+			// the APF handler completes its run, either normally or
+			// with a panic, in either case, all APF book keeping must
+			// be completed by now. Also, whether the request is
+			// executed or rejected, we expect the counter to be zero.
+			// TODO: all test(s) using this filter must run
+			// serially to each other
+			defer func() {
+				if atomicReadOnlyExecuting != 0 {
+					t.Errorf("Wanted %d requests executing, got %d", 0, atomicReadOnlyExecuting)
+				}
+			}()
+			apfHandler.ServeHTTP(w, r)
+			postExecute()
+		}()
 	}), requestInfoFactory)
 
 	return handler