diff --git a/staging/src/k8s.io/apiserver/pkg/server/filters/BUILD b/staging/src/k8s.io/apiserver/pkg/server/filters/BUILD index d2f3ef0f697..232e384fa07 100644 --- a/staging/src/k8s.io/apiserver/pkg/server/filters/BUILD +++ b/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", 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 fb67968afe6..e0918a1bcc1 100644 --- a/staging/src/k8s.io/apiserver/pkg/server/filters/reqmgmt.go +++ b/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