From 3926050733d5c6b6bf163b2a3030b59daf875ea8 Mon Sep 17 00:00:00 2001 From: Mike Spreitzer Date: Sun, 4 Aug 2019 00:41:10 -0400 Subject: [PATCH] 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). --- cmd/kube-apiserver/app/BUILD | 2 + cmd/kube-apiserver/app/server.go | 7 + .../apiserver/pkg/server/filters/reqmgmt.go | 50 ++---- .../k8s.io/apiserver/pkg/server/options/BUILD | 2 + .../pkg/server/options/recommended.go | 7 + .../apiserver/pkg/util/flowcontrol/BUILD | 7 +- .../pkg/util/flowcontrol/fairqueuing/BUILD | 23 +++ .../util/flowcontrol/fairqueuing/interface.go | 82 ++++++++++ .../apiserver/pkg/util/flowcontrol/reqmgmt.go | 154 +++++------------- vendor/modules.txt | 1 + 10 files changed, 186 insertions(+), 149 deletions(-) create mode 100644 staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/BUILD create mode 100644 staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/interface.go diff --git a/cmd/kube-apiserver/app/BUILD b/cmd/kube-apiserver/app/BUILD index 1fdfe1ba015..e9b841c24de 100644 --- 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", diff --git a/cmd/kube-apiserver/app/server.go b/cmd/kube-apiserver/app/server.go index f014af7e752..a76d169aa9c 100644 --- 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 } 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 a81009b6438..c918f58bcb8 100644 --- 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 }) diff --git a/staging/src/k8s.io/apiserver/pkg/server/options/BUILD b/staging/src/k8s.io/apiserver/pkg/server/options/BUILD index a1ae49080f9..82329c48f88 100644 --- 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", diff --git a/staging/src/k8s.io/apiserver/pkg/server/options/recommended.go b/staging/src/k8s.io/apiserver/pkg/server/options/recommended.go index 2ddb901a922..ed5ced3f18b 100644 --- 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 } diff --git a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/BUILD b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/BUILD index 70b1b84e599..37befac346a 100644 --- 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"], ) diff --git a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/BUILD b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/BUILD new file mode 100644 index 00000000000..30dbd260df1 --- /dev/null +++ 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"], +) diff --git a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/interface.go b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/interface.go new file mode 100644 index 00000000000..db79450e476 --- /dev/null +++ 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() +} diff --git a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/reqmgmt.go b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/reqmgmt.go index 1814bc47677..c28bc43d6ec 100644 --- 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 diff --git a/vendor/modules.txt b/vendor/modules.txt index d6950f425e2..423a29fc945 100644 --- 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