Merge pull request kubernetes#120222 from tkashem/apf-queue-wait-ctx

apf: manage request queue wait with context in APF Filter

Author: k8s-ci-robot

pkg/controlplane/apiserver/config.go

@@ -202,7 +202,6 @@ func BuildPriorityAndFairness(s controlplaneapiserver.CompletedOptions, extclien
 		versionedInformer,
 		extclient.FlowcontrolV1beta3(),
 		s.GenericServerRunOptions.MaxRequestsInFlight+s.GenericServerRunOptions.MaxMutatingRequestsInFlight,
-		s.GenericServerRunOptions.RequestTimeout/4,
 	), nil
 }
 

staging/src/k8s.io/apiserver/pkg/server/config.go

@@ -915,7 +915,7 @@ func DefaultBuildHandlerChain(apiHandler http.Handler, c *Config) http.Handler {
 		requestWorkEstimator := flowcontrolrequest.NewWorkEstimator(
 			c.StorageObjectCountTracker.Get, c.FlowControl.GetInterestedWatchCount, workEstimatorCfg, c.FlowControl.GetMaxSeats)
 		handler = filterlatency.TrackCompleted(handler)
-		handler = genericfilters.WithPriorityAndFairness(handler, c.LongRunningFunc, c.FlowControl, requestWorkEstimator)
+		handler = genericfilters.WithPriorityAndFairness(handler, c.LongRunningFunc, c.FlowControl, requestWorkEstimator, c.RequestTimeout/4)
 		handler = filterlatency.TrackStarted(handler, c.TracerProvider, "priorityandfairness")
 	} else {
 		handler = genericfilters.WithMaxInFlightLimit(handler, c.MaxRequestsInFlight, c.MaxMutatingRequestsInFlight, c.LongRunningFunc)

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

@@ -35,6 +35,7 @@ import (
 	fcmetrics "k8s.io/apiserver/pkg/util/flowcontrol/metrics"
 	flowcontrolrequest "k8s.io/apiserver/pkg/util/flowcontrol/request"
 	"k8s.io/klog/v2"
+	utilsclock "k8s.io/utils/clock"
 )
 
 // PriorityAndFairnessClassification identifies the results of
@@ -78,6 +79,10 @@ type priorityAndFairnessHandler struct {
 	// the purpose of computing RetryAfter header to avoid system
 	// overload.
 	droppedRequests utilflowcontrol.DroppedRequestsTracker
+
+	// newReqWaitCtxFn creates a derived context with a deadline
+	// of how long a given request can wait in its queue.
+	newReqWaitCtxFn func(context.Context) (context.Context, context.CancelFunc)
 }
 
 func (h *priorityAndFairnessHandler) Handle(w http.ResponseWriter, r *http.Request) {
@@ -240,16 +245,17 @@ func (h *priorityAndFairnessHandler) Handle(w http.ResponseWriter, r *http.Reque
 				resultCh <- err
 			}()
 
-			// We create handleCtx with explicit cancelation function.
-			// The reason for it is that Handle() underneath may start additional goroutine
+			// We create handleCtx with an adjusted deadline, for two reasons.
+			// One is to limit the time the request waits before its execution starts.
+			// The other reason for it is that Handle() underneath may start additional goroutine
 			// that is blocked on context cancellation. However, from APF point of view,
 			// we don't want to wait until the whole watch request is processed (which is
 			// when it context is actually cancelled) - we want to unblock the goroutine as
 			// soon as the request is processed from the APF point of view.
 			//
 			// Note that we explicitly do NOT call the actuall handler using that context
 			// to avoid cancelling request too early.
-			handleCtx, handleCtxCancel := context.WithCancel(ctx)
+			handleCtx, handleCtxCancel := h.newReqWaitCtxFn(ctx)
 			defer handleCtxCancel()
 
 			// Note that Handle will return irrespective of whether the request
@@ -286,7 +292,11 @@ func (h *priorityAndFairnessHandler) Handle(w http.ResponseWriter, r *http.Reque
 			h.handler.ServeHTTP(w, r)
 		}
 
-		h.fcIfc.Handle(ctx, digest, noteFn, estimateWork, queueNote, execute)
+		func() {
+			handleCtx, cancelFn := h.newReqWaitCtxFn(ctx)
+			defer cancelFn()
+			h.fcIfc.Handle(handleCtx, digest, noteFn, estimateWork, queueNote, execute)
+		}()
 	}
 
 	if !served {
@@ -309,6 +319,7 @@ func WithPriorityAndFairness(
 	longRunningRequestCheck apirequest.LongRunningRequestCheck,
 	fcIfc utilflowcontrol.Interface,
 	workEstimator flowcontrolrequest.WorkEstimatorFunc,
+	defaultRequestWaitLimit time.Duration,
 ) http.Handler {
 	if fcIfc == nil {
 		klog.Warningf("priority and fairness support not found, skipping")
@@ -322,12 +333,18 @@ func WithPriorityAndFairness(
 		waitingMark.mutatingObserver = fcmetrics.GetWaitingMutatingConcurrency()
 	})
 
+	clock := &utilsclock.RealClock{}
+	newReqWaitCtxFn := func(ctx context.Context) (context.Context, context.CancelFunc) {
+		return getRequestWaitContext(ctx, defaultRequestWaitLimit, clock)
+	}
+
 	priorityAndFairnessHandler := &priorityAndFairnessHandler{
 		handler:                 handler,
 		longRunningRequestCheck: longRunningRequestCheck,
 		fcIfc:                   fcIfc,
 		workEstimator:           workEstimator,
 		droppedRequests:         utilflowcontrol.NewDroppedRequestsTracker(),
+		newReqWaitCtxFn:         newReqWaitCtxFn,
 	}
 	return http.HandlerFunc(priorityAndFairnessHandler.Handle)
 }
@@ -356,3 +373,48 @@ func tooManyRequests(req *http.Request, w http.ResponseWriter, retryAfter string
 	w.Header().Set("Retry-After", retryAfter)
 	http.Error(w, "Too many requests, please try again later.", http.StatusTooManyRequests)
 }
+
+// getRequestWaitContext returns a new context with a deadline of how
+// long the request is allowed to wait before it is removed from its
+// queue and rejected.
+// The context.CancelFunc returned must never be nil and the caller is
+// responsible for calling the CancelFunc function for cleanup.
+//   - ctx: the context associated with the request (it may or may
+//     not have a deadline).
+//   - defaultRequestWaitLimit: the default wait duration that is used
+//     if the request context does not have any deadline.
+//     (a) initialization of a watch or
+//     (b) a request whose context has no deadline
+//
+// clock comes in handy for testing the function
+func getRequestWaitContext(ctx context.Context, defaultRequestWaitLimit time.Duration, clock utilsclock.PassiveClock) (context.Context, context.CancelFunc) {
+	if ctx.Err() != nil {
+		return ctx, func() {}
+	}
+
+	reqArrivedAt := clock.Now()
+	if reqReceivedTimestamp, ok := apirequest.ReceivedTimestampFrom(ctx); ok {
+		reqArrivedAt = reqReceivedTimestamp
+	}
+
+	// a) we will allow the request to wait in the queue for one
+	// fourth of the time of its allotted deadline.
+	// b) if the request context does not have any deadline
+	// then we default to 'defaultRequestWaitLimit'
+	// in any case, the wait limit for any request must not
+	// exceed the hard limit of 1m
+	//
+	// request has deadline:
+	//   wait-limit = min(remaining deadline / 4, 1m)
+	// request has no deadline:
+	//   wait-limit = min(defaultRequestWaitLimit, 1m)
+	thisReqWaitLimit := defaultRequestWaitLimit
+	if deadline, ok := ctx.Deadline(); ok {
+		thisReqWaitLimit = deadline.Sub(reqArrivedAt) / 4
+	}
+	if thisReqWaitLimit > time.Minute {
+		thisReqWaitLimit = time.Minute
+	}
+
+	return context.WithDeadline(ctx, reqArrivedAt.Add(thisReqWaitLimit))
+}