Simplified FairQueuingSystem interface

.. and introduced a tiny bit of starting to work with config objects.

staging/src/k8s.io/apiserver/pkg/server/filters/BUILD

@@ -56,6 +56,9 @@ go_library(
         "//staging/src/k8s.io/apiserver/pkg/endpoints/metrics:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/endpoints/request:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/server/httplog:go_default_library",
+        "//staging/src/k8s.io/client-go/informers:go_default_library",
+        "//staging/src/k8s.io/client-go/kubernetes:go_default_library",
+        "//staging/src/k8s.io/client-go/listers/flowcontrol/v1alpha1:go_default_library",
         "//staging/src/k8s.io/client-go/rest: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",

staging/src/k8s.io/apiserver/pkg/server/filters/reqmgmt.go

@@ -32,11 +32,14 @@ import (
 	// "k8s.io/apiserver/pkg/endpoints/metrics"
 
 	apirequest "k8s.io/apiserver/pkg/endpoints/request"
+	kubeinformers "k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
 	restclient "k8s.io/client-go/rest"
 	cache "k8s.io/client-go/tools/cache"
 	workqueue "k8s.io/client-go/util/workqueue"
 
 	rmtypesv1a1 "k8s.io/api/flowcontrol/v1alpha1"
+	rmlisterv1a1 "k8s.io/client-go/listers/flowcontrol/v1alpha1"
 
 	"k8s.io/klog"
 )
@@ -58,26 +61,18 @@ type FairQueuingFactory interface {
 // today is not that day.
 type FairQueuingSystem interface {
 
-	// The following methods modify or query the configuration
+	// SetConfiguration updates the configuration
+	SetConfiguration(concurrencyLimit, desiredNumQueues, queueLengthLimit int, requestWaitLimit time.Duration)
 
-	SetConcurrencyLimit(int)
-	SetDesiredNumQueues(int)
-	GetDesiredNumQueues() int
-	GetActualNumQueues() int
-	SetQueueLengthLimit(int)
-	SetRequestWaitLimit(time.Duration)
-
-	// DoOnEmpty records a function to be called when all the queues
-	// have no requests waiting nor executing.  This is setting a
-	// notification handler.  A call to `DoOnEmpty(fn)` requires the
-	// implementation to guarantee that if `fn != nil` then either (1)
-	// eventually there is another call to `DoOnEmpty` or (2) if
-	// eventually all the queues have no requests waiting nor
-	// executing then eventually there will be a call to `fn()` with
-	// no mutexes locked.  Note that (2) effectively has two
-	// interesting intervals of time (until empty, then until the call
-	// to fn), and that (1) is about how the client can effectively
-	// remove the promise of (2).
+	// 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
@@ -85,17 +80,29 @@ type FairQueuingSystem interface {
 	// 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.
-	DoOnEmpty(func())
+	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 `execute==true`.  In the
-	// 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) (execute bool, afterExecution func())
+	// 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()
 }
 
 // RMState is the variable state that this filter is working with at a
@@ -130,6 +137,8 @@ type PriorityLevelState struct {
 // requestManagement holds all the state and infrastructure of this
 // filter
 type requestManagement struct {
+	clk clock.Clock
+
 	fairQueuingFactory FairQueuingFactory
 
 	// configQueue holds TypedConfigObjectReference values, identifying
@@ -139,11 +148,12 @@ type requestManagement struct {
 	// plInformer is the informer for priority level config objects
 	plInformer cache.SharedIndexInformer
 
+	plLister rmlisterv1a1.PriorityLevelConfigurationLister
+
 	// fsInformer is the informer for flow schema config objects
 	fsInformer cache.SharedIndexInformer
 
-	// plLister belongs here too
-	// fsLister belongs here too
+	fsLister rmlisterv1a1.FlowSchemaLister
 
 	// serverConcurrencyLimit is the limit on the server's total
 	// number of non-exempt requests being served at once.  This comes
@@ -182,9 +192,18 @@ func (tr *TypedConfigObjectReference) String() string {
 }
 
 // rmSetup is invoked at startup to create the infrastructure of this filter
-func rmSetup(loopbackClientConfig *restclient.Config, serverConcurrencyLimit int, requestWaitLimit time.Duration) *requestManagement {
+func rmSetup(kubeClient kubernetes.Interface, serverConcurrencyLimit int, requestWaitLimit time.Duration, clk clock.Clock) *requestManagement {
+	kubeInformerFactory := kubeinformers.NewSharedInformerFactory(kubeClient, 0)
+	fci := kubeInformerFactory.Flowcontrol().V1alpha1()
+	pli := fci.PriorityLevelConfigurations()
+	fsi := fci.FlowSchemas()
 	reqMgmt := &requestManagement{
+		clk:                    clk,
 		configQueue:            workqueue.NewNamedRateLimitingQueue(workqueue.NewItemExponentialFailureRateLimiter(200*time.Millisecond, 8*time.Hour), "req_mgmt_config_queue"),
+		plInformer:             pli.Informer(),
+		plLister:               pli.Lister(),
+		fsInformer:             fsi.Informer(),
+		fsLister:               fsi.Lister(),
 		serverConcurrencyLimit: serverConcurrencyLimit,
 		requestWaitLimit:       requestWaitLimit,
 	}
@@ -195,12 +214,31 @@ func rmSetup(loopbackClientConfig *restclient.Config, serverConcurrencyLimit int
 // WithRequestManagement limits the number of in-flight requests in a fine-grained way
 func WithRequestManagement(
 	handler http.Handler,
-	loopbackClientConfig *restclient.Config,
+	clientConfig *restclient.Config,
 	serverConcurrencyLimit int,
 	requestWaitLimit time.Duration,
 	longRunningRequestCheck apirequest.LongRunningRequestCheck,
 ) http.Handler {
-	reqMgmt := rmSetup(loopbackClientConfig, serverConcurrencyLimit, requestWaitLimit)
+	kubeClient, err := kubernetes.NewForConfig(clientConfig)
+	if err != nil {
+		klog.Errorf("Failed to construct Kubernetes client (%s), skipping prioritization and fairness filter", err.Error())
+		return handler
+	}
+	return WithRequestManagementByClient(handler, kubeClient, serverConcurrencyLimit, requestWaitLimit, longRunningRequestCheck, clock.RealClock{})
+}
+
+// WithRequestManagementByClient limits the number of in-flight
+// requests in a fine-grained way and is more appropriate than
+// WithRequestManagement for testing
+func WithRequestManagementByClient(
+	handler http.Handler,
+	kubeClient kubernetes.Interface,
+	serverConcurrencyLimit int,
+	requestWaitLimit time.Duration,
+	longRunningRequestCheck apirequest.LongRunningRequestCheck,
+	clk clock.Clock,
+) http.Handler {
+	reqMgmt := rmSetup(kubeClient, serverConcurrencyLimit, requestWaitLimit, clk)
 
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		ctx := r.Context()
@@ -216,34 +254,41 @@ func WithRequestManagement(
 			return
 		}
 
-		rmState := reqMgmt.curState.Load().(*RMState)
-		fs := reqMgmt.pickFlowSchema(r, rmState.flowSchemas, rmState.priorityLevelStates)
-		ps := reqMgmt.requestPriorityState(r, fs, rmState.priorityLevelStates)
-		if ps.config.Exempt {
-			klog.V(5).Infof("Serving %v without delay\n", r)
-			handler.ServeHTTP(w, r)
-			return
-		}
-		flowDistinguisher := reqMgmt.computeFlowDistinguisher(r, fs)
-		hashValue := reqMgmt.hashFlowID(fs.Name, flowDistinguisher)
-		execute, afterExecute := ps.fqs.Wait(hashValue, ps.config.HandSize)
-		if execute {
-			klog.V(5).Infof("Serving %v after queuing\n", r)
-			timedOut := ctx.Done()
-			finished := make(chan struct{})
-			go func() {
+		for {
+			rmState := reqMgmt.curState.Load().(*RMState)
+			fs := reqMgmt.pickFlowSchema(r, rmState.flowSchemas, rmState.priorityLevelStates)
+			ps := reqMgmt.requestPriorityState(r, fs, rmState.priorityLevelStates)
+			if ps.config.Exempt {
+				klog.V(5).Infof("Serving %v without delay\n", r)
 				handler.ServeHTTP(w, r)
-				close(finished)
-			}()
-			select {
-			case <-timedOut:
-				klog.V(5).Infof("Timed out waiting for %v to finish\n", r)
-			case <-finished:
+				return
+			}
+			flowDistinguisher := reqMgmt.computeFlowDistinguisher(r, fs)
+			hashValue := reqMgmt.hashFlowID(fs.Name, flowDistinguisher)
+			quiescent, execute, afterExecute := ps.fqs.Wait(hashValue, ps.config.HandSize)
+			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)
+
+				tooManyRequests(r, w)
 			}
-			afterExecute()
-		} else {
-			klog.V(5).Infof("Rejecting %v\n", r)
-			tooManyRequests(r, w)
 		}
 
 		return