From 7b6ef2555cbc8933ead2958fba49e5f6caa46ff1 Mon Sep 17 00:00:00 2001 From: yue9944882 <291271447@qq.com> Date: Fri, 26 Jul 2019 17:33:47 +0800 Subject: [PATCH] refactor request management filter --- staging/src/k8s.io/apiserver/BUILD | 1 + staging/src/k8s.io/apiserver/pkg/server/BUILD | 3 + .../src/k8s.io/apiserver/pkg/server/config.go | 23 +- .../k8s.io/apiserver/pkg/server/filters/BUILD | 9 +- .../apiserver/pkg/server/filters/reqmgmt.go | 253 ++-------------- .../apiserver/pkg/util/flowcontrol/BUILD | 33 +++ .../apiserver/pkg/util/flowcontrol/reqmgmt.go | 278 ++++++++++++++++++ vendor/modules.txt | 1 + 8 files changed, 360 insertions(+), 241 deletions(-) create mode 100644 staging/src/k8s.io/apiserver/pkg/util/flowcontrol/BUILD create mode 100644 staging/src/k8s.io/apiserver/pkg/util/flowcontrol/reqmgmt.go diff --git a/staging/src/k8s.io/apiserver/BUILD b/staging/src/k8s.io/apiserver/BUILD index 9b457fb28ba..7b190dcf4d5 100644 --- a/staging/src/k8s.io/apiserver/BUILD +++ b/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", diff --git a/staging/src/k8s.io/apiserver/pkg/server/BUILD b/staging/src/k8s.io/apiserver/pkg/server/BUILD index b7d06e11fb8..5c0a95ea88b 100644 --- a/staging/src/k8s.io/apiserver/pkg/server/BUILD +++ b/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", diff --git a/staging/src/k8s.io/apiserver/pkg/server/config.go b/staging/src/k8s.io/apiserver/pkg/server/config.go index b75422a6e09..ef7ac8e526c 100644 --- a/staging/src/k8s.io/apiserver/pkg/server/config.go +++ b/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) diff --git a/staging/src/k8s.io/apiserver/pkg/server/filters/BUILD b/staging/src/k8s.io/apiserver/pkg/server/filters/BUILD index 232e384fa07..145e9ed9f85 100644 --- a/staging/src/k8s.io/apiserver/pkg/server/filters/BUILD +++ b/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", ], ) diff --git a/staging/src/k8s.io/apiserver/pkg/server/filters/reqmgmt.go b/staging/src/k8s.io/apiserver/pkg/server/filters/reqmgmt.go index e0918a1bcc1..a81009b6438 100644 --- a/staging/src/k8s.io/apiserver/pkg/server/filters/reqmgmt.go +++ b/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 -} diff --git a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/BUILD b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/BUILD new file mode 100644 index 00000000000..70b1b84e599 --- /dev/null +++ b/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"], +) diff --git a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/reqmgmt.go b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/reqmgmt.go new file mode 100644 index 00000000000..1814bc47677 --- /dev/null +++ b/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 +} diff --git a/vendor/modules.txt b/vendor/modules.txt index b5751b7437c..d6950f425e2 100644 --- a/vendor/modules.txt +++ b/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