func WithPriorityAndFairness(
	handler http.Handler,
	longRunningRequestCheck apirequest.LongRunningRequestCheck,
	fcIfc utilflowcontrol.Interface,
) http.Handler {
	if fcIfc == nil {
		klog.Warningf("priority and fairness support not found, skipping")
		return handler
	}
	startOnce.Do(func() {
		startRecordingUsage(watermark)
		startRecordingUsage(waitingMark)
	})
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		ctx := r.Context()
		requestInfo, ok := apirequest.RequestInfoFrom(ctx)
		if !ok {
			handleError(w, r, fmt.Errorf("no RequestInfo found in context"))
			return
		}
		user, ok := apirequest.UserFrom(ctx)
		if !ok {
			handleError(w, r, fmt.Errorf("no User found in context"))
			return
		}

		// Skip tracking long running requests.
		if longRunningRequestCheck != nil && longRunningRequestCheck(r, requestInfo) {
			klog.V(6).Infof("Serving RequestInfo=%#+v, user.Info=%#+v as longrunning\n", requestInfo, user)
			handler.ServeHTTP(w, r)
			return
		}

		var classification *PriorityAndFairnessClassification
		note := func(fs *fcv1a1.FlowSchema, pl *fcv1a1.PriorityLevelConfiguration) {
			classification = &PriorityAndFairnessClassification{
				FlowSchemaName:    fs.Name,
				FlowSchemaUID:     fs.UID,
				PriorityLevelName: pl.Name,
				PriorityLevelUID:  pl.UID}
		}

		var served bool
		isMutatingRequest := !nonMutatingRequestVerbs.Has(requestInfo.Verb)
		noteExecutingDelta := func(delta int32) {
			if isMutatingRequest {
				watermark.recordMutating(int(atomic.AddInt32(&atomicMutatingExecuting, delta)))
			} else {
				watermark.recordReadOnly(int(atomic.AddInt32(&atomicReadOnlyExecuting, delta)))
			}
		}
		noteWaitingDelta := func(delta int32) {
			if isMutatingRequest {
				waitingMark.recordMutating(int(atomic.AddInt32(&atomicMutatingWaiting, delta)))
			} else {
				waitingMark.recordReadOnly(int(atomic.AddInt32(&atomicReadOnlyWaiting, delta)))
			}
		}
		execute := func() {
			noteExecutingDelta(1)
			defer noteExecutingDelta(-1)
			served = true
			innerCtx := context.WithValue(ctx, priorityAndFairnessKey, classification)
			innerReq := r.Clone(innerCtx)
			w.Header().Set(responseHeaderMatchedPriorityLevelConfigurationUID, string(classification.PriorityLevelUID))
			w.Header().Set(responseHeaderMatchedFlowSchemaUID, string(classification.FlowSchemaUID))
			handler.ServeHTTP(w, innerReq)
		}
		digest := utilflowcontrol.RequestDigest{requestInfo, user}
		fcIfc.Handle(ctx, digest, note, func(inQueue bool) {
			if inQueue {
				noteWaitingDelta(1)
			} else {
				noteWaitingDelta(-1)
			}
		}, execute)
		if !served {
			tooManyRequests(r, w)
		}

	})
}
// WithPriorityAndFairness limits the number of in-flight
// requests in a fine-grained way.
WithPriorityAndFairness limits the number of in-flight
requests in a fine-grained way.
// WithPriorityAndFairness limits the number of in - flight // requests in a fine - grained way .
func WithPriorityAndFairness(
	handler http.Handler,
	longRunningRequestCheck apirequest.LongRunningRequestCheck,
	fcIfc utilflowcontrol.Interface,
) http.Handler {
	if fcIfc == nil {
		klog.Warningf("priority and fairness support not found, skipping")
		return handler
	}
	startOnce.Do(func() {
		startRecordingUsage(watermark)
		startRecordingUsage(waitingMark)
	})
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		ctx := r.Context()
		requestInfo, ok := apirequest.RequestInfoFrom(ctx)
		if !ok {
			handleError(w, r, fmt.Errorf("no RequestInfo found in context"))
			return
		}
		user, ok := apirequest.UserFrom(ctx)
		if !ok {
			handleError(w, r, fmt.Errorf("no User found in context"))
			return
		}
		if longRunningRequestCheck != nil && longRunningRequestCheck(r, requestInfo) {
			klog.V(6).Infof("Serving RequestInfo=%#+v, user.Info=%#+v as longrunning\n", requestInfo, user)
			handler.ServeHTTP(w, r)
			return
		}
		var classification *PriorityAndFairnessClassification
		note := func(fs *fcv1a1.FlowSchema, pl *fcv1a1.PriorityLevelConfiguration) {
			classification = &PriorityAndFairnessClassification{
				FlowSchemaName:    fs.Name,
				FlowSchemaUID:     fs.UID,
				PriorityLevelName: pl.Name,
				PriorityLevelUID:  pl.UID}
		}
		var served bool
		isMutatingRequest := !nonMutatingRequestVerbs.Has(requestInfo.Verb)
		noteExecutingDelta := func(delta int32) {
			if isMutatingRequest {
				watermark.recordMutating(int(atomic.AddInt32(&atomicMutatingExecuting, delta)))
			} else {
				watermark.recordReadOnly(int(atomic.AddInt32(&atomicReadOnlyExecuting, delta)))
			}
		}
		noteWaitingDelta := func(delta int32) {
			if isMutatingRequest {
				waitingMark.recordMutating(int(atomic.AddInt32(&atomicMutatingWaiting, delta)))
			} else {
				waitingMark.recordReadOnly(int(atomic.AddInt32(&atomicReadOnlyWaiting, delta)))
			}
		}
		execute := func() {
			noteExecutingDelta(1)
			defer noteExecutingDelta(-1)
			served = true
			innerCtx := context.WithValue(ctx, priorityAndFairnessKey, classification)
			innerReq := r.Clone(innerCtx)
			w.Header().Set(responseHeaderMatchedPriorityLevelConfigurationUID, string(classification.PriorityLevelUID))
			w.Header().Set(responseHeaderMatchedFlowSchemaUID, string(classification.FlowSchemaUID))
			handler.ServeHTTP(w, innerReq)
		}
		digest := utilflowcontrol.RequestDigest{requestInfo, user}
		fcIfc.Handle(ctx, digest, note, func(inQueue bool) {
			if inQueue {
				noteWaitingDelta(1)
			} else {
				noteWaitingDelta(-1)
			}
		}, execute)
		if !served {
			tooManyRequests(r, w)
		}
	})
}
func WithPriorityAndFairness ( handler http . Handler , longRunningRequestCheck apirequest . LongRunningRequestCheck , fcIfc utilflowcontrol . Interface , ) http . Handler { if fcIfc == nil { klog . Warningf ( " " ) return handler } startOnce . Do ( func ( ) { startRecordingUsage ( watermark ) startRecordingUsage ( waitingMark ) } ) return http . HandlerFunc ( func ( w http . ResponseWriter , r * http . Request ) { ctx := r . Context ( ) requestInfo , ok := apirequest . RequestInfoFrom ( ctx ) if ! ok { handleError ( w , r , fmt . Errorf ( " " ) ) return } user , ok := apirequest . UserFrom ( ctx ) if ! ok { handleError ( w , r , fmt . Errorf ( " " ) ) return } if longRunningRequestCheck != nil && longRunningRequestCheck ( r , requestInfo ) { klog . V ( 6 ) . Infof ( " \n " , requestInfo , user ) handler . ServeHTTP ( w , r ) return } var classification * PriorityAndFairnessClassification note := func ( fs * fcv1a1 . FlowSchema , pl * fcv1a1 . PriorityLevelConfiguration ) { classification = & PriorityAndFairnessClassification { FlowSchemaName : fs . Name , FlowSchemaUID : fs . UID , PriorityLevelName : pl . Name , PriorityLevelUID : pl . UID } } var served bool isMutatingRequest := ! nonMutatingRequestVerbs . Has ( requestInfo . Verb ) noteExecutingDelta := func ( delta int32 ) { if isMutatingRequest { watermark . recordMutating ( int ( atomic . AddInt32 ( & atomicMutatingExecuting , delta ) ) ) } else { watermark . recordReadOnly ( int ( atomic . AddInt32 ( & atomicReadOnlyExecuting , delta ) ) ) } } noteWaitingDelta := func ( delta int32 ) { if isMutatingRequest { waitingMark . recordMutating ( int ( atomic . AddInt32 ( & atomicMutatingWaiting , delta ) ) ) } else { waitingMark . recordReadOnly ( int ( atomic . AddInt32 ( & atomicReadOnlyWaiting , delta ) ) ) } } execute := func ( ) { noteExecutingDelta ( 1 ) defer noteExecutingDelta ( - 1 ) served = true innerCtx := context . WithValue ( ctx , priorityAndFairnessKey , classification ) innerReq := r . Clone ( innerCtx ) w . Header ( ) . Set ( responseHeaderMatchedPriorityLevelConfigurationUID , string ( classification . PriorityLevelUID ) ) w . Header ( ) . Set ( responseHeaderMatchedFlowSchemaUID , string ( classification . FlowSchemaUID ) ) handler . ServeHTTP ( w , innerReq ) } digest := utilflowcontrol . RequestDigest { requestInfo , user } fcIfc . Handle ( ctx , digest , note , func ( inQueue bool ) { if inQueue { noteWaitingDelta ( 1 ) } else { noteWaitingDelta ( - 1 ) } } , execute ) if ! served { tooManyRequests ( r , w ) } } ) }
WithPriorityAndFairness limits the number of in-flight
requests in a fine-grained way.
WithPriorityAndFairness limits the number of in - flight requests in a fine - grained way .
// Skip tracking long running requests.