Continue refactoring started in PR 80628

Moved the top-level logic associated with the data in k8s.io/apiserver/pkg/util/flowcontrol into the same package, greatly simplifying its interface. Also plumbed the call to NewRequestManagementSystem (1) into `RecommendedOptions::ApplyTo(RecommendedConfig)` for running in aggregated servers and (2) into buildGenericConfig in the kube-apiserver code. Also moved the definitions of FairQueuingSystem and FairQueuingFactory into the directory where Aaron is developing his implementation of those. Also did some renaming to satisfy golint: FairQueuingSytem became QueueSet, and FairQueuingFactory became QueueSetFactory. Also realized that the clock should be passed in every Create... call to the Factory, the Factory should hold the clock to use (as well as the WaitGroup that will be helpful for testing).
2026-07-17 02:06:23 +00:00 · 2019-08-04 00:41:10 -04:00
parent ab696f6ba3
commit 3926050733
10 changed files with 186 additions and 149 deletions
--- a/cmd/kube-apiserver/app/BUILD
+++ b/cmd/kube-apiserver/app/BUILD
@@ -39,6 +39,7 @@ go_library(
        "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
        "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
        "//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/util/clock:go_default_library",
        "//staging/src/k8s.io/apimachinery/pkg/util/errors:go_default_library",
        "//staging/src/k8s.io/apimachinery/pkg/util/net:go_default_library",
        "//staging/src/k8s.io/apimachinery/pkg/util/sets:go_default_library",
@@ -55,6 +56,7 @@ go_library(
        "//staging/src/k8s.io/apiserver/pkg/server/storage:go_default_library",
        "//staging/src/k8s.io/apiserver/pkg/storage/etcd3/preflight:go_default_library",
        "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/util/flowcontrol:go_default_library",
        "//staging/src/k8s.io/apiserver/pkg/util/term:go_default_library",
        "//staging/src/k8s.io/apiserver/pkg/util/webhook:go_default_library",
        "//staging/src/k8s.io/client-go/informers:go_default_library",
--- a/cmd/kube-apiserver/app/server.go
+++ b/cmd/kube-apiserver/app/server.go
@@ -36,6 +36,7 @@ import (

 	extensionsapiserver "k8s.io/apiextensions-apiserver/pkg/apiserver"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/clock"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	utilnet "k8s.io/apimachinery/pkg/util/net"
 	"k8s.io/apimachinery/pkg/util/sets"
@@ -44,12 +45,14 @@ import (
 	"k8s.io/apiserver/pkg/authentication/authenticator"
 	"k8s.io/apiserver/pkg/authorization/authorizer"
 	openapinamer "k8s.io/apiserver/pkg/endpoints/openapi"
+	genericfeatures "k8s.io/apiserver/pkg/features"
 	genericapiserver "k8s.io/apiserver/pkg/server"
 	"k8s.io/apiserver/pkg/server/filters"
 	serveroptions "k8s.io/apiserver/pkg/server/options"
 	serverstorage "k8s.io/apiserver/pkg/server/storage"
 	"k8s.io/apiserver/pkg/storage/etcd3/preflight"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	utilflowcontrol "k8s.io/apiserver/pkg/util/flowcontrol"
 	"k8s.io/apiserver/pkg/util/term"
 	"k8s.io/apiserver/pkg/util/webhook"
 	clientgoinformers "k8s.io/client-go/informers"
@@ -520,6 +523,10 @@ func buildGenericConfig(
 		lastErr = fmt.Errorf("failed to initialize admission: %v", err)
 	}

+	if utilfeature.DefaultFeatureGate.Enabled(genericfeatures.RequestManagement) {
+		genericConfig.RequestManagement = utilflowcontrol.NewRequestManagementSystem(versionedInformers, genericConfig.MaxRequestsInFlight+genericConfig.MaxMutatingRequestsInFlight, genericConfig.RequestTimeout/4, clock.RealClock{})
+	}
+
 	return
 }

--- a/staging/src/k8s.io/apiserver/pkg/server/filters/reqmgmt.go
+++ b/staging/src/k8s.io/apiserver/pkg/server/filters/reqmgmt.go
@@ -33,8 +33,7 @@ import (
 )

 // WithRequestManagement limits the number of in-flight
-// requests in a fine-grained way and is more appropriate than
-// WithRequestManagement for testing
+// requests in a fine-grained way.
 func WithRequestManagement(
 	handler http.Handler,
 	longRunningRequestCheck apirequest.LongRunningRequestCheck,
@@ -65,40 +64,25 @@ func WithRequestManagement(
 			return
 		}

-		for {
-			rmState := reqMgmt.GetCurrentState()
-			fs := utilflowcontrol.PickFlowSchema(requestDigest, rmState.GetFlowSchemas(), rmState.GetPriorityLevelStates())
-			ps := utilflowcontrol.RequestPriorityState(requestDigest, fs, rmState.GetPriorityLevelStates())
-			if ps.IsExempt() {
-				klog.V(5).Infof("Serving %v without delay\n", r)
+		execute, afterExecute := reqMgmt.Wait(requestDigest)
+		if execute {
+			klog.V(5).Infof("Serving %v after queuing\n", r)
+			timedOut := ctx.Done()
+			finished := make(chan struct{})
+			go func() {
 				handler.ServeHTTP(w, r)
-				return
+				close(finished)
+			}()
+			select {
+			case <-timedOut:
+				klog.V(5).Infof("Timed out waiting for %v to finish\n", r)
+			case <-finished:
 			}
-			hashValue := utilflowcontrol.ComputeFlowDistinguisher(requestDigest, fs)
-			quiescent, execute, afterExecute := ps.GetFairQueuingSystem().Wait(hashValue, ps.GetHandSize())
-			if quiescent {
-				klog.V(3).Infof("Request %v landed in timing splinter, re-classifying", r)
-				continue
-			}
-			if execute {
-				klog.V(5).Infof("Serving %v after queuing\n", r)
-				timedOut := ctx.Done()
-				finished := make(chan struct{})
-				go func() {
-					handler.ServeHTTP(w, r)
-					close(finished)
-				}()
-				select {
-				case <-timedOut:
-					klog.V(5).Infof("Timed out waiting for %v to finish\n", r)
-				case <-finished:
-				}
-				afterExecute()
-			} else {
-				klog.V(5).Infof("Rejecting %v\n", r)
+			afterExecute()
+		} else {
+			klog.V(5).Infof("Rejecting %v\n", r)

-				tooManyRequests(r, w)
-			}
+			tooManyRequests(r, w)
 		}
 		return
 	})
--- a/staging/src/k8s.io/apiserver/pkg/server/options/BUILD
+++ b/staging/src/k8s.io/apiserver/pkg/server/options/BUILD
@@ -31,6 +31,7 @@ go_library(
        "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
        "//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
        "//staging/src/k8s.io/apimachinery/pkg/runtime/serializer:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/util/clock:go_default_library",
        "//staging/src/k8s.io/apimachinery/pkg/util/net:go_default_library",
        "//staging/src/k8s.io/apimachinery/pkg/util/runtime:go_default_library",
        "//staging/src/k8s.io/apimachinery/pkg/util/sets:go_default_library",
@@ -65,6 +66,7 @@ go_library(
        "//staging/src/k8s.io/apiserver/pkg/storage/storagebackend:go_default_library",
        "//staging/src/k8s.io/apiserver/pkg/storage/storagebackend/factory:go_default_library",
        "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/util/flowcontrol:go_default_library",
        "//staging/src/k8s.io/apiserver/pkg/util/webhook:go_default_library",
        "//staging/src/k8s.io/apiserver/plugin/pkg/audit/buffered:go_default_library",
        "//staging/src/k8s.io/apiserver/plugin/pkg/audit/dynamic:go_default_library",
--- a/staging/src/k8s.io/apiserver/pkg/server/options/recommended.go
+++ b/staging/src/k8s.io/apiserver/pkg/server/options/recommended.go
@@ -20,9 +20,13 @@ import (
 	"github.com/spf13/pflag"

 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/clock"
 	"k8s.io/apiserver/pkg/admission"
+	"k8s.io/apiserver/pkg/features"
 	"k8s.io/apiserver/pkg/server"
 	"k8s.io/apiserver/pkg/storage/storagebackend"
+	"k8s.io/apiserver/pkg/util/feature"
+	utilflowcontrol "k8s.io/apiserver/pkg/util/flowcontrol"
 )

 // RecommendedOptions contains the recommended options for running an API server.
@@ -117,6 +121,9 @@ func (o *RecommendedOptions) ApplyTo(config *server.RecommendedConfig) error {
 	if err := o.EgressSelector.ApplyTo(&config.Config); err != nil {
 		return err
 	}
+	if feature.DefaultFeatureGate.Enabled(features.RequestManagement) {
+		config.RequestManagement = utilflowcontrol.NewRequestManagementSystem(config.SharedInformerFactory, config.MaxRequestsInFlight+config.MaxMutatingRequestsInFlight, config.RequestTimeout/4, clock.RealClock{})
+	}

 	return nil
 }
--- a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/BUILD
+++ b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/BUILD
@@ -11,10 +11,12 @@ go_library(
        "//staging/src/k8s.io/apimachinery/pkg/util/clock:go_default_library",
        "//staging/src/k8s.io/apiserver/pkg/authentication/user:go_default_library",
        "//staging/src/k8s.io/apiserver/pkg/endpoints/request:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing:go_default_library",
        "//staging/src/k8s.io/client-go/informers:go_default_library",
        "//staging/src/k8s.io/client-go/listers/flowcontrol/v1alpha1:go_default_library",
        "//staging/src/k8s.io/client-go/tools/cache:go_default_library",
        "//staging/src/k8s.io/client-go/util/workqueue:go_default_library",
+        "//vendor/k8s.io/klog:go_default_library",
    ],
 )

@@ -27,7 +29,10 @@ filegroup(

 filegroup(
    name = "all-srcs",
-    srcs = [":package-srcs"],
+    srcs = [
+        ":package-srcs",
+        "//staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing:all-srcs",
+    ],
    tags = ["automanaged"],
    visibility = ["//visibility:public"],
 )
--- a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/BUILD
+++ b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/BUILD
@@ -0,0 +1,23 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["interface.go"],
+    importmap = "k8s.io/kubernetes/vendor/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing",
+    importpath = "k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing",
+    visibility = ["//visibility:public"],
+)
+
+filegroup(
+    name = "package-srcs",
+    srcs = glob(["**"]),
+    tags = ["automanaged"],
+    visibility = ["//visibility:private"],
+)
+
+filegroup(
+    name = "all-srcs",
+    srcs = [":package-srcs"],
+    tags = ["automanaged"],
+    visibility = ["//visibility:public"],
+)
--- a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/interface.go
+++ b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/interface.go
@@ -0,0 +1,82 @@
+/*
+Copyright 2019 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package fairqueuing
+
+import (
+	"time"
+)
+
+// QueueSetFactory knows how to make QueueSet objects.  The request
+// management filter makes a QueueSet for each priority level.  The
+// clock, and any other test-facilitating infrastructure, to use in
+// each QueueSet is known to the factory, the client does not need to
+// pass this stuff in each call to NewQueueSet.
+type QueueSetFactory interface {
+	NewQueueSet(concurrencyLimit, numQueues, queueLengthLimit int, requestWaitLimit time.Duration) QueueSet
+}
+
+// QueueSet is the abstraction for the queuing and dispatching
+// functionality of one non-exempt priority level.  It covers the
+// functionality described in the "Assignment to a Queue", "Queuing",
+// and "Dispatching" sections of
+// https://github.com/kubernetes/enhancements/blob/master/keps/sig-api-machinery/20190228-priority-and-fairness.md
+// .  Some day we may have connections between priority levels, but
+// today is not that day.
+type QueueSet interface {
+	// SetConfiguration updates the configuration
+	SetConfiguration(concurrencyLimit, desiredNumQueues, queueLengthLimit int, requestWaitLimit time.Duration)
+
+	// Quiesce controls whether this system is quiescing.  Passing a
+	// non-nil handler means the system should become quiescent, a nil
+	// handler means the system should become non-quiescent.  A call
+	// to Wait while the system is quiescent will be rebuffed by
+	// returning `quiescent=true`.  If all the queues have no requests
+	// waiting nor executing while the system is quiescent then the
+	// handler will eventually be called with no locks held (even if
+	// the system becomes non-quiescent between the triggering state
+	// and the required call).
+	//
+	// The filter uses this for a priority level that has become
+	// undesired, setting a handler that will cause the priority level
+	// to eventually be removed from the filter if the filter still
+	// wants that.  If the filter later changes its mind and wants to
+	// preserve the priority level then the filter can use this to
+	// cancel the handler registration.
+	Quiesce(EmptyHandler)
+
+	// Wait, in the happy case, shuffle shards the given request into
+	// a queue and eventually dispatches the request from that queue.
+	// Dispatching means to return with `quiescent==false` and
+	// `execute==true`.  In one unhappy case the request is
+	// immediately rebuffed with `quiescent==true` (which tells the
+	// filter that there has been a timing splinter and the filter
+	// re-calcuates the priority level to use); in all other cases
+	// `quiescent` will be returned `false` (even if the system is
+	// quiescent by then).  In the non-quiescent unhappy cases the
+	// request is eventually rejected, which means to return with
+	// `execute=false`.  In the happy case the caller is required to
+	// invoke the returned `afterExecution` after the request is done
+	// executing.  The hash value and hand size are used to do the
+	// shuffle sharding.
+	Wait(hashValue uint64, handSize int32) (quiescent, execute bool, afterExecution func())
+}
+
+// EmptyHandler can be notified that something is empty
+type EmptyHandler interface {
+	// HandleEmpty is called to deliver the notification
+	HandleEmpty()
+}
--- a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/reqmgmt.go
+++ b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/reqmgmt.go
@@ -31,9 +31,11 @@ import (

 	"k8s.io/apiserver/pkg/authentication/user"
 	"k8s.io/apiserver/pkg/endpoints/request"
+	fq "k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing"
 	kubeinformers "k8s.io/client-go/informers"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/workqueue"
+	"k8s.io/klog"

 	rmtypesv1alpha1 "k8s.io/api/flowcontrol/v1alpha1"
 	rmlistersv1alpha1 "k8s.io/client-go/listers/flowcontrol/v1alpha1"
@@ -41,70 +43,20 @@ import (

 // Interface defines how the request-management filter interacts with the underlying system.
 type Interface interface {
-	GetCurrentState() *RequestManagementState
+	// Wait decides what to do about the request with the given digest
+	// and, if appropriate, enqueues that request and waits for it to be
+	// dequeued before returning.  If `execute == false` then the request
+	// is being rejected.  If `execute == true` then the caller should
+	// handle the request and then call `afterExecute()`.
+	Wait(requestDigest RequestDigest) (execute bool, afterExecute func())
+
+	// Run monitors config objects from the main apiservers and causes
+	// any needed changes to local behavior
 	Run(stopCh <-chan struct{}) error
 }

 // This request filter implements https://github.com/kubernetes/enhancements/blob/master/keps/sig-api-machinery/20190228-priority-and-fairness.md

-// FairQueuingFactory knows how to make FairQueueingSystem objects.
-// This filter makes a FairQueuingSystem for each priority level.
-type FairQueuingFactory interface {
-	NewFairQueuingSystem(concurrencyLimit, numQueues, queueLengthLimit int, requestWaitLimit time.Duration, clk clock.Clock) FairQueuingSystem
-}
-
-// FairQueuingSystem is the abstraction for the queuing and
-// dispatching functionality of one non-exempt priority level.  It
-// covers the functionality described in the "Assignment to a Queue",
-// "Queuing", and "Dispatching" sections of
-// https://github.com/kubernetes/enhancements/blob/master/keps/sig-api-machinery/20190228-priority-and-fairness.md
-// .  Some day we may have connections between priority levels, but
-// today is not that day.
-type FairQueuingSystem interface {
-	// SetConfiguration updates the configuration
-	SetConfiguration(concurrencyLimit, desiredNumQueues, queueLengthLimit int, requestWaitLimit time.Duration)
-
-	// Quiesce controls whether this system is quiescing.  Passing a
-	// non-nil handler means the system should become quiescent, a nil
-	// handler means the system should become non-quiescent.  A call
-	// to Wait while the system is quiescent will be rebuffed by
-	// returning `quiescent=true`.  If all the queues have no requests
-	// waiting nor executing while the system is quiescent then the
-	// handler will eventually be called with no locks held (even if
-	// the system becomes non-quiescent between the triggering state
-	// and the required call).
-	//
-	// The filter uses this for a priority level that has become
-	// undesired, setting a handler that will cause the priority level
-	// to eventually be removed from the filter if the filter still
-	// wants that.  If the filter later changes its mind and wants to
-	// preserve the priority level then the filter can use this to
-	// cancel the handler registration.
-	Quiesce(EmptyHandler)
-
-	// Wait, in the happy case, shuffle shards the given request into
-	// a queue and eventually dispatches the request from that queue.
-	// Dispatching means to return with `quiescent==false` and
-	// `execute==true`.  In one unhappy case the request is
-	// immediately rebuffed with `quiescent==true` (which tells the
-	// filter that there has been a timing splinter and the filter
-	// re-calcuates the priority level to use); in all other cases
-	// `quiescent` will be returned `false` (even if the system is
-	// quiescent by then).  In the non-quiescent unhappy cases the
-	// request is eventually rejected, which means to return with
-	// `execute=false`.  In the happy case the caller is required to
-	// invoke the returned `afterExecution` after the request is done
-	// executing.  The hash value and hand size are used to do the
-	// shuffle sharding.
-	Wait(hashValue uint64, handSize int32) (quiescent, execute bool, afterExecution func())
-}
-
-// EmptyHandler can be notified that something is empty
-type EmptyHandler interface {
-	// HandleEmpty is called to deliver the notification
-	HandleEmpty()
-}
-
 // RequestManagementState is the variable state that this filter is working with at a
 // given point in time.
 type RequestManagementState struct {
@@ -117,16 +69,6 @@ type RequestManagementState struct {
 	priorityLevelStates map[string]*PriorityLevelState
 }

-// GetFlowSchemas returns an array of latest observed flow-schemas by the system
-func (s *RequestManagementState) GetFlowSchemas() []*rmtypesv1alpha1.FlowSchema {
-	return s.flowSchemas
-}
-
-// GetPriorityLevelStates returns a map of latest states of the existing priority-levels
-func (s *RequestManagementState) GetPriorityLevelStates() map[string]*PriorityLevelState {
-	return s.priorityLevelStates
-}
-
 // FlowSchemaSequence holds sorted set of pointers to FlowSchema objects.
 // FlowSchemaSequence implements `sort.Interface` (TODO: implement this).
 type FlowSchemaSequence []*rmtypesv1alpha1.FlowSchema
@@ -141,17 +83,7 @@ type PriorityLevelState struct {
 	// concurrencyLimit is the limit on number executing
 	concurrencyLimit int

-	fairQueuingSystem FairQueuingSystem
-}
-
-// IsExempt returns if the priority-level is exempt
-func (s *PriorityLevelState) IsExempt() bool {
-	return s.config.Exempt
-}
-
-// GetHandSize returns the hand-size of the priority-level
-func (s *PriorityLevelState) GetHandSize() int32 {
-	return s.config.HandSize
+	queues fq.QueueSet
 }

 // requestManagement holds all the state and infrastructure of this
@@ -159,7 +91,7 @@ func (s *PriorityLevelState) GetHandSize() int32 {
 type requestManagementSystem struct {
 	clk clock.Clock

-	fairQueuingFactory FairQueuingFactory
+	fairQueuingFactory fq.QueueSetFactory

 	// configQueue holds TypedConfigObjectReference values, identifying
 	// config objects that need to be processed
@@ -200,17 +132,6 @@ type requestManagementSystem struct {
 	curState atomic.Value
 }

-// TypedConfigObjectReference is a reference to a relevant config API object.
-// No namespace is needed because none of these objects is namespaced.
-type TypedConfigObjectReference struct {
-	Kind string
-	Name string
-}
-
-func (tr *TypedConfigObjectReference) String() string {
-	return tr.Kind + "/" + tr.Name
-}
-
 // NewRequestManagementSystem creates a new instance of request-management system
 func NewRequestManagementSystem(
 	informerFactory kubeinformers.SharedInformerFactory,
@@ -230,45 +151,48 @@ func NewRequestManagementSystem(
 	return reqMgmt
 }

-// Run bootstraps the request-management system
-func (r *requestManagementSystem) Run(stopCh <-chan struct{}) error {
+// Run runs the controller that deals with config API objects
+func (reqMgmt *requestManagementSystem) Run(stopCh <-chan struct{}) error {
 	// TODO: implement
 	return nil
 }

-// GetCurrentState returns the current state of the request-management system
-func (r *requestManagementSystem) GetCurrentState() *RequestManagementState {
-	return r.curState.Load().(*RequestManagementState)
-}
-
-// GetFairQueuingSystem returns the fair-queuing system of the priority-level
-func (s *PriorityLevelState) GetFairQueuingSystem() FairQueuingSystem {
-	return s.fairQueuingSystem
-}
-
 // RequestDigest holds necessary info from request for flow-control
 type RequestDigest struct {
 	RequestInfo *request.RequestInfo
 	User        user.Info
 }

-// PickFlowSchema returns a best-matching flow-schema according to the request
-func PickFlowSchema(digest RequestDigest, flowSchemas FlowSchemaSequence, priorityLevelStates map[string]*PriorityLevelState) *rmtypesv1alpha1.FlowSchema {
-	// TODO: implement
-	return nil
+func (reqMgmt *requestManagementSystem) Wait(requestDigest RequestDigest) (execute bool, afterExecute func()) {
+	for {
+		rmState := reqMgmt.curState.Load().(*RequestManagementState)
+		fs := rmState.pickFlowSchema(requestDigest)
+		// RequestManagementState is constructed to guarantee that the following succeeds
+		ps := rmState.priorityLevelStates[fs.Spec.PriorityLevelConfiguration.Name]
+		if ps.config.Exempt {
+			klog.V(7).Infof("Serving %v without delay", requestDigest)
+			return true, func() {}
+		}
+		flowDistinguisher := requestDigest.ComputeFlowDistinguisher(fs.Spec.DistinguisherMethod)
+		hashValue := hashFlowID(fs.Name, flowDistinguisher)
+		quiescent, execute, afterExecute := ps.queues.Wait(hashValue, ps.config.HandSize)
+		if quiescent {
+			klog.V(5).Infof("Request %v landed in timing splinter, re-classifying", requestDigest)
+			continue
+		}
+		return execute, afterExecute
+	}
 }

-// ComputeFlowDistinguisher computes flow-distinguisher according to request information for a flow-schema
-func ComputeFlowDistinguisher(digest RequestDigest, flowSchema *rmtypesv1alpha1.FlowSchema) uint64 {
-	var fDistinguisher string
+func (rmState *RequestManagementState) pickFlowSchema(rd RequestDigest) *rmtypesv1alpha1.FlowSchema {
+	return nil
 	// TODO: implement
-	return hashFlowID(flowSchema.Name, fDistinguisher)
 }

-// RequestPriorityState requests for the current state of a priority-level
-func RequestPriorityState(digest RequestDigest, fs *rmtypesv1alpha1.FlowSchema, priorityLevelStates map[string]*PriorityLevelState) *PriorityLevelState {
+// ComputeFlowDistinguisher extracts the flow distinguisher according to the given method
+func (rd RequestDigest) ComputeFlowDistinguisher(method *rmtypesv1alpha1.FlowDistinguisherMethod) string {
 	// TODO: implement
-	return nil
+	return ""
 }

 // HashFlowID hashes the inputs into 64-bits
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1370,6 +1370,7 @@ k8s.io/apiserver/pkg/storage/value/encrypt/secretbox
 k8s.io/apiserver/pkg/util/dryrun
 k8s.io/apiserver/pkg/util/feature
 k8s.io/apiserver/pkg/util/flowcontrol
+k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing
 k8s.io/apiserver/pkg/util/flushwriter
 k8s.io/apiserver/pkg/util/openapi
 k8s.io/apiserver/pkg/util/proxy