refactor request management filter

staging/src/k8s.io/apiserver/BUILD

@@ -41,6 +41,7 @@ filegroup(
         "//staging/src/k8s.io/apiserver/pkg/storage:all-srcs",
         "//staging/src/k8s.io/apiserver/pkg/util/dryrun:all-srcs",
         "//staging/src/k8s.io/apiserver/pkg/util/feature:all-srcs",
+        "//staging/src/k8s.io/apiserver/pkg/util/flowcontrol:all-srcs",
         "//staging/src/k8s.io/apiserver/pkg/util/flushwriter:all-srcs",
         "//staging/src/k8s.io/apiserver/pkg/util/openapi:all-srcs",
         "//staging/src/k8s.io/apiserver/pkg/util/proxy:all-srcs",

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

@@ -96,6 +96,7 @@ go_library(
         "//staging/src/k8s.io/apiserver/pkg/endpoints/handlers/responsewriters:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/endpoints/openapi:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/endpoints/request:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/features:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/registry/generic:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/registry/rest:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/server/egressselector:go_default_library",
@@ -104,6 +105,8 @@ go_library(
         "//staging/src/k8s.io/apiserver/pkg/server/mux:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/server/routes:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/server/storage: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/openapi:go_default_library",
         "//staging/src/k8s.io/client-go/informers:go_default_library",
         "//staging/src/k8s.io/client-go/rest:go_default_library",

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

@@ -54,12 +54,15 @@ import (
 	genericapifilters "k8s.io/apiserver/pkg/endpoints/filters"
 	apiopenapi "k8s.io/apiserver/pkg/endpoints/openapi"
 	apirequest "k8s.io/apiserver/pkg/endpoints/request"
+	"k8s.io/apiserver/pkg/features"
 	genericregistry "k8s.io/apiserver/pkg/registry/generic"
 	"k8s.io/apiserver/pkg/server/egressselector"
 	genericfilters "k8s.io/apiserver/pkg/server/filters"
 	"k8s.io/apiserver/pkg/server/healthz"
 	"k8s.io/apiserver/pkg/server/routes"
 	serverstore "k8s.io/apiserver/pkg/server/storage"
+	"k8s.io/apiserver/pkg/util/feature"
+	utilflowcontrol "k8s.io/apiserver/pkg/util/flowcontrol"
 	"k8s.io/client-go/informers"
 	restclient "k8s.io/client-go/rest"
 	certutil "k8s.io/client-go/util/cert"
@@ -108,6 +111,9 @@ type Config struct {
 	AdmissionControl      admission.Interface
 	CorsAllowedOriginList []string
 
+	// RequestManagement performs flow-control of a given request
+	RequestManagement utilflowcontrol.Interface
+
 	EnableIndex     bool
 	EnableProfiling bool
 	EnableDiscovery bool
@@ -571,6 +577,17 @@ func (c completedConfig) New(name string, delegationTarget DelegationTarget) (*G
 		}
 	}
 
+	if feature.DefaultFeatureGate.Enabled(features.RequestManagement) {
+		requestManagementHookName := "generic-apiserver-request-management"
+		err := s.AddPostStartHook(requestManagementHookName, func(context PostStartHookContext) error {
+			return c.RequestManagement.Run(context.StopCh)
+		})
+		if err != nil {
+			return nil, err
+		}
+		// TODO(yue9944882): plumb pre-shutdown-hook for request-management system?
+	}
+
 	for _, delegateCheck := range delegationTarget.HealthzChecks() {
 		skip := false
 		for _, existingCheck := range c.HealthzChecks {
@@ -603,7 +620,11 @@ func (c completedConfig) New(name string, delegationTarget DelegationTarget) (*G
 
 func DefaultBuildHandlerChain(apiHandler http.Handler, c *Config) http.Handler {
 	handler := genericapifilters.WithAuthorization(apiHandler, c.Authorization.Authorizer, c.Serializer)
-	handler = genericfilters.WithMaxInFlightLimit(handler, c.MaxRequestsInFlight, c.MaxMutatingRequestsInFlight, c.LongRunningFunc)
+	if feature.DefaultFeatureGate.Enabled(features.RequestManagement) {
+		handler = genericfilters.WithRequestManagement(handler, c.LongRunningFunc, c.RequestManagement)
+	} else {
+		handler = genericfilters.WithMaxInFlightLimit(handler, c.MaxRequestsInFlight, c.MaxMutatingRequestsInFlight, c.LongRunningFunc)
+	}
 	handler = genericapifilters.WithImpersonation(handler, c.Authorization.Authorizer, c.Serializer)
 	handler = genericapifilters.WithAudit(handler, c.AuditBackend, c.AuditPolicyChecker, c.LongRunningFunc)
 	failedHandler := genericapifilters.Unauthorized(c.Serializer, c.Authentication.SupportsBasicAuth)

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

@@ -44,9 +44,7 @@ go_library(
     importmap = "k8s.io/kubernetes/vendor/k8s.io/apiserver/pkg/server/filters",
     importpath = "k8s.io/apiserver/pkg/server/filters",
     deps = [
-        "//staging/src/k8s.io/api/flowcontrol/v1alpha1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
-        "//staging/src/k8s.io/apimachinery/pkg/util/clock: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",
         "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
@@ -56,12 +54,7 @@ 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/util/flowcontrol:go_default_library",
         "//vendor/k8s.io/klog:go_default_library",
     ],
 )

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

@@ -19,10 +19,6 @@ package filters
 import (
 	"fmt"
 	"net/http"
-	"sync/atomic"
-	"time"
-
-	"k8s.io/apimachinery/pkg/util/clock"
 
 	// TODO: decide whether to use the existing metrics, which
 	// categorize according to mutating vs readonly, or make new
@@ -32,221 +28,36 @@ 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"
-
+	utilflowcontrol "k8s.io/apiserver/pkg/util/flowcontrol"
 	"k8s.io/klog"
 )
 
-// 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()
-}
-
-// RMState is the variable state that this filter is working with at a
-// given point in time.
-type RMState struct {
-	// flowSchemas holds the flow schema objects, sorted by increasing
-	// numerical (decreasing logical) matching precedence
-	flowSchemas FlowSchemaSeq
-
-	// priorityLevelStates maps the PriorityLevelConfiguration object
-	// name to the state for that level
-	priorityLevelStates map[string]*PriorityLevelState
-}
-
-// FlowSchemaSeq holds sorted set of pointers to FlowSchema objects.
-// FLowSchemaSeq implements `sort.Interface` (TODO: implement this).
-type FlowSchemaSeq []*rmtypesv1a1.FlowSchema
-
-// PriorityLevelState holds the state specific to a priority level.
-// golint requires that I write something here,
-// even if I can not think of something better than a tautology.
-type PriorityLevelState struct {
-	// config holds the configuration after defaulting logic has been applied
-	config rmtypesv1a1.PriorityLevelConfigurationSpec
-
-	// concurrencyLimit is the limit on number executing
-	concurrencyLimit int
-
-	fqs FairQueuingSystem
-}
-
-// requestManagement holds all the state and infrastructure of this
-// filter
-type requestManagement struct {
-	clk clock.Clock
-
-	fairQueuingFactory FairQueuingFactory
-
-	// configQueue holds TypedConfigObjectReference values, identifying
-	// config objects that need to be processed
-	configQueue workqueue.RateLimitingInterface
-
-	// 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
-
-	fsLister rmlisterv1a1.FlowSchemaLister
-
-	// serverConcurrencyLimit is the limit on the server's total
-	// number of non-exempt requests being served at once.  This comes
-	// from server configuration.
-	serverConcurrencyLimit int
-
-	// requestWaitLimit comes from server configuration.
-	requestWaitLimit time.Duration
-
-	// curState holds a pointer to the current RMState.  That is,
-	// `Load()` produces a `*RMState`.  When a config work queue worker
-	// processes a configuration change, it stores a new pointer here ---
-	// it does NOT side-effect the old `RMState` value.  The new `RMState`
-	// has a freshly constructed slice of FlowSchema pointers and a
-	// freshly constructed map of priority level states.  But the new
-	// `RMState.priorityLevelStates` includes in its range at least all the
-	// `*PriorityLevelState` values of the old `RMState.priorityLevelStates`.
-	// Consequently the filter can load a `*RMState` and work with it
-	// without concern for concurrent updates.  When a priority level is
-	// finally deleted, this will also involve storing a new `*RMState`
-	// pointer here, but in this case the range of the
-	// `RMState.priorityLevels` will be reduced --- by removal of the
-	// priority level that is no longer in use.
-	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
-}
-
-// rmSetup is invoked at startup to create the infrastructure of this filter
-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,
-	}
-	// TODO: finish implementation
-	return reqMgmt
-}
-
-// WithRequestManagement limits the number of in-flight requests in a fine-grained way
-func WithRequestManagement(
-	handler http.Handler,
-	clientConfig *restclient.Config,
-	serverConcurrencyLimit int,
-	requestWaitLimit time.Duration,
-	longRunningRequestCheck apirequest.LongRunningRequestCheck,
-) http.Handler {
-	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
+// WithRequestManagement limits the number of in-flight
 // requests in a fine-grained way and is more appropriate than
 // WithRequestManagement for testing
-func WithRequestManagementByClient(
+func WithRequestManagement(
 	handler http.Handler,
-	kubeClient kubernetes.Interface,
-	serverConcurrencyLimit int,
-	requestWaitLimit time.Duration,
 	longRunningRequestCheck apirequest.LongRunningRequestCheck,
-	clk clock.Clock,
+	reqMgmt utilflowcontrol.Interface,
 ) http.Handler {
-	reqMgmt := rmSetup(kubeClient, serverConcurrencyLimit, requestWaitLimit, clk)
+	if reqMgmt == nil {
+		klog.Warningf("request management system not found, skipping setup flow-control system")
+		return handler
+	}
 
 	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, handler chain must be wrong"))
+			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
+		}
+		requestDigest := utilflowcontrol.RequestDigest{requestInfo, user}
 
 		// Skip tracking long running events.
 		if longRunningRequestCheck != nil && longRunningRequestCheck(r, requestInfo) {
@@ -255,17 +66,16 @@ func WithRequestManagementByClient(
 		}
 
 		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 {
+			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)
 				handler.ServeHTTP(w, r)
 				return
 			}
-			flowDistinguisher := reqMgmt.computeFlowDistinguisher(r, fs)
-			hashValue := reqMgmt.hashFlowID(fs.Name, flowDistinguisher)
-			quiescent, execute, afterExecute := ps.fqs.Wait(hashValue, ps.config.HandSize)
+			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
@@ -290,27 +100,6 @@ func WithRequestManagementByClient(
 				tooManyRequests(r, w)
 			}
 		}
-
 		return
 	})
 }
-
-func (requestManagement) computeFlowDistinguisher(r *http.Request, fs *rmtypesv1a1.FlowSchema) string {
-	// TODO: implement
-	return ""
-}
-
-func (requestManagement) hashFlowID(fsName, fDistinguisher string) uint64 {
-	// TODO: implement
-	return 0
-}
-
-func (requestManagement) pickFlowSchema(r *http.Request, flowSchemas FlowSchemaSeq, priorityLevelStates map[string]*PriorityLevelState) *rmtypesv1a1.FlowSchema {
-	// TODO: implement
-	return nil
-}
-
-func (requestManagement) requestPriorityState(r *http.Request, fs *rmtypesv1a1.FlowSchema, priorityLevelStates map[string]*PriorityLevelState) *PriorityLevelState {
-	// TODO: implement
-	return nil
-}

staging/src/k8s.io/apiserver/pkg/util/flowcontrol/BUILD

@@ -0,0 +1,33 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["reqmgmt.go"],
+    importmap = "k8s.io/kubernetes/vendor/k8s.io/apiserver/pkg/util/flowcontrol",
+    importpath = "k8s.io/apiserver/pkg/util/flowcontrol",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//staging/src/k8s.io/api/flowcontrol/v1alpha1:go_default_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/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",
+    ],
+)
+
+filegroup(
+    name = "package-srcs",
+    srcs = glob(["**"]),
+    tags = ["automanaged"],
+    visibility = ["//visibility:private"],
+)
+
+filegroup(
+    name = "all-srcs",
+    srcs = [":package-srcs"],
+    tags = ["automanaged"],
+    visibility = ["//visibility:public"],
+)

staging/src/k8s.io/apiserver/pkg/util/flowcontrol/reqmgmt.go

@@ -0,0 +1,278 @@
+/*
+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 filters
+
+import (
+	"sync/atomic"
+	"time"
+
+	"k8s.io/apimachinery/pkg/util/clock"
+
+	// TODO: decide whether to use the existing metrics, which
+	// categorize according to mutating vs readonly, or make new
+	// metrics because this filter does not pay attention to that
+	// distinction
+
+	// "k8s.io/apiserver/pkg/endpoints/metrics"
+
+	"k8s.io/apiserver/pkg/authentication/user"
+	"k8s.io/apiserver/pkg/endpoints/request"
+	kubeinformers "k8s.io/client-go/informers"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/workqueue"
+
+	rmtypesv1alpha1 "k8s.io/api/flowcontrol/v1alpha1"
+	rmlistersv1alpha1 "k8s.io/client-go/listers/flowcontrol/v1alpha1"
+)
+
+// Interface defines how the request-management filter interacts with the underlying system.
+type Interface interface {
+	GetCurrentState() *RequestManagementState
+	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 {
+	// flowSchemas holds the flow schema objects, sorted by increasing
+	// numerical (decreasing logical) matching precedence
+	flowSchemas FlowSchemaSequence
+
+	// priorityLevelStates maps the PriorityLevelConfiguration object
+	// name to the state for that level
+	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
+
+// PriorityLevelState holds the state specific to a priority level.
+// golint requires that I write something here,
+// even if I can not think of something better than a tautology.
+type PriorityLevelState struct {
+	// config holds the configuration after defaulting logic has been applied
+	config rmtypesv1alpha1.PriorityLevelConfigurationSpec
+
+	// 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
+}
+
+// requestManagement holds all the state and infrastructure of this
+// filter
+type requestManagementSystem struct {
+	clk clock.Clock
+
+	fairQueuingFactory FairQueuingFactory
+
+	// configQueue holds TypedConfigObjectReference values, identifying
+	// config objects that need to be processed
+	configQueue workqueue.RateLimitingInterface
+
+	// plInformer is the informer for priority level config objects
+	plInformer cache.SharedIndexInformer
+
+	plLister rmlistersv1alpha1.PriorityLevelConfigurationLister
+
+	// fsInformer is the informer for flow schema config objects
+	fsInformer cache.SharedIndexInformer
+
+	fsLister rmlistersv1alpha1.FlowSchemaLister
+
+	// serverConcurrencyLimit is the limit on the server's total
+	// number of non-exempt requests being served at once.  This comes
+	// from server configuration.
+	serverConcurrencyLimit int
+
+	// requestWaitLimit comes from server configuration.
+	requestWaitLimit time.Duration
+
+	// curState holds a pointer to the current RMState.  That is,
+	// `Load()` produces a `*RMState`.  When a config work queue worker
+	// processes a configuration change, it stores a new pointer here ---
+	// it does NOT side-effect the old `RMState` value.  The new `RMState`
+	// has a freshly constructed slice of FlowSchema pointers and a
+	// freshly constructed map of priority level states.  But the new
+	// `RMState.priorityLevelStates` includes in its range at least all the
+	// `*PriorityLevelState` values of the old `RMState.priorityLevelStates`.
+	// Consequently the filter can load a `*RMState` and work with it
+	// without concern for concurrent updates.  When a priority level is
+	// finally deleted, this will also involve storing a new `*RMState`
+	// pointer here, but in this case the range of the
+	// `RMState.priorityLevels` will be reduced --- by removal of the
+	// priority level that is no longer in use.
+	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,
+	serverConcurrencyLimit int,
+	requestWaitLimit time.Duration,
+	clk clock.Clock,
+) Interface {
+	reqMgmt := &requestManagementSystem{
+		clk:                    clk,
+		configQueue:            workqueue.NewNamedRateLimitingQueue(workqueue.NewItemExponentialFailureRateLimiter(200*time.Millisecond, 8*time.Hour), "req_mgmt_config_queue"),
+		plLister:               informerFactory.Flowcontrol().V1alpha1().PriorityLevelConfigurations().Lister(),
+		fsLister:               informerFactory.Flowcontrol().V1alpha1().FlowSchemas().Lister(),
+		serverConcurrencyLimit: serverConcurrencyLimit,
+		requestWaitLimit:       requestWaitLimit,
+	}
+	// TODO: finish implementation
+	return reqMgmt
+}
+
+// Run bootstraps the request-management system
+func (r *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
+}
+
+// ComputeFlowDistinguisher computes flow-distinguisher according to request information for a flow-schema
+func ComputeFlowDistinguisher(digest RequestDigest, flowSchema *rmtypesv1alpha1.FlowSchema) uint64 {
+	var fDistinguisher string
+	// 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 {
+	// TODO: implement
+	return nil
+}
+
+// HashFlowID hashes the inputs into 64-bits
+func hashFlowID(fsName, fDistinguisher string) uint64 {
+	// TODO: implement
+	return 0
+}

vendor/modules.txt

@@ -1369,6 +1369,7 @@ k8s.io/apiserver/pkg/storage/value/encrypt/identity
 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/flushwriter
 k8s.io/apiserver/pkg/util/openapi
 k8s.io/apiserver/pkg/util/proxy