diff --git a/cmd/kube-apiserver/app/BUILD b/cmd/kube-apiserver/app/BUILD index 338f0d75271..d941df4f13c 100644 --- a/cmd/kube-apiserver/app/BUILD +++ b/cmd/kube-apiserver/app/BUILD @@ -55,6 +55,7 @@ go_library( "//staging/src/k8s.io/apiserver/pkg/server/options:go_default_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/clock: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", diff --git a/cmd/kube-apiserver/app/server.go b/cmd/kube-apiserver/app/server.go index dc31719fec6..851d834b486 100644 --- a/cmd/kube-apiserver/app/server.go +++ b/cmd/kube-apiserver/app/server.go @@ -50,6 +50,7 @@ import ( serveroptions "k8s.io/apiserver/pkg/server/options" serverstorage "k8s.io/apiserver/pkg/server/storage" "k8s.io/apiserver/pkg/storage/etcd3/preflight" + "k8s.io/apiserver/pkg/util/clock" utilfeature "k8s.io/apiserver/pkg/util/feature" utilflowcontrol "k8s.io/apiserver/pkg/util/flowcontrol" "k8s.io/apiserver/pkg/util/term" @@ -559,7 +560,7 @@ func BuildRequestManager(s *options.ServerRunOptions, extclient clientgoclientse return utilflowcontrol.NewRequestManagerWithPreservation( versionedInformer, extclient.FlowcontrolV1alpha1(), - fairqueuing.NewNoRestraintFactory( /* TODO: switch to real implementation */ ), + fairqueuing.NewQueueSetFactory(&clock.RealClock{}, fairqueuing.NoWaitGroup()), s.GenericServerRunOptions.MaxRequestsInFlight+s.GenericServerRunOptions.MaxMutatingRequestsInFlight, s.GenericServerRunOptions.RequestTimeout/4, flowcontrolbootstrap.PreservingFlowSchemas, diff --git a/hack/.golint_failures b/hack/.golint_failures index 9d9b427174b..379dd6b81eb 100644 --- a/hack/.golint_failures +++ b/hack/.golint_failures @@ -421,6 +421,7 @@ staging/src/k8s.io/apiserver/pkg/storage/storagebackend staging/src/k8s.io/apiserver/pkg/storage/testing staging/src/k8s.io/apiserver/pkg/storage/tests staging/src/k8s.io/apiserver/pkg/storage/value +staging/src/k8s.io/apiserver/pkg/util/flowcontrol/metrics staging/src/k8s.io/apiserver/pkg/util/proxy staging/src/k8s.io/apiserver/pkg/util/webhook staging/src/k8s.io/apiserver/pkg/util/wsstream diff --git a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/BUILD b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/BUILD index ee72178b649..fe0a9a392f8 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/BUILD +++ b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/BUILD @@ -23,6 +23,7 @@ go_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/apiserver/pkg/util/flowcontrol/metrics:go_default_library", "//staging/src/k8s.io/client-go/informers:go_default_library", "//staging/src/k8s.io/client-go/kubernetes/typed/flowcontrol/v1alpha1:go_default_library", "//staging/src/k8s.io/client-go/listers/flowcontrol/v1alpha1:go_default_library", @@ -44,6 +45,7 @@ filegroup( srcs = [ ":package-srcs", "//staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing:all-srcs", + "//staging/src/k8s.io/apiserver/pkg/util/flowcontrol/metrics: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 index d2d6ac8ff3a..1e07452de35 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/BUILD +++ b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/BUILD @@ -16,6 +16,7 @@ go_library( deps = [ "//staging/src/k8s.io/apimachinery/pkg/util/waitgroup:go_default_library", "//staging/src/k8s.io/apiserver/pkg/util/clock:go_default_library", + "//staging/src/k8s.io/apiserver/pkg/util/flowcontrol/metrics:go_default_library", "//staging/src/k8s.io/apiserver/pkg/util/shufflesharding:go_default_library", "//vendor/k8s.io/klog:go_default_library", ], diff --git a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/fairqueuing.go b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/fairqueuing.go index 6ba320f8972..6da04e9f3ad 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/fairqueuing.go +++ b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/fairqueuing.go @@ -23,6 +23,7 @@ import ( "k8s.io/apimachinery/pkg/util/waitgroup" "k8s.io/apiserver/pkg/util/clock" + "k8s.io/apiserver/pkg/util/flowcontrol/metrics" "k8s.io/apiserver/pkg/util/shufflesharding" "k8s.io/klog" ) @@ -47,8 +48,8 @@ func NewQueueSetFactory(clk clock.PassiveClock, wg waitgroup.OptionalWaitGroup) // NewQueueSet creates a new QueueSetSystem object // There is a new QueueSet created for each priority level. -func (qsf queueSetFactoryImpl) NewQueueSet(concurrencyLimit, desiredNumQueues, queueLengthLimit int, requestWaitLimit time.Duration) QueueSet { - return newQueueSetImpl(concurrencyLimit, desiredNumQueues, +func (qsf queueSetFactoryImpl) NewQueueSet(name string, concurrencyLimit, desiredNumQueues, queueLengthLimit int, requestWaitLimit time.Duration) QueueSet { + return newQueueSetImpl(name, concurrencyLimit, desiredNumQueues, queueLengthLimit, requestWaitLimit, qsf.clk, qsf.wg) } @@ -96,6 +97,7 @@ func newQueueSetImpl(name string, concurrencyLimit, desiredNumQueues, queueLengt requestWaitLimit time.Duration, clk clock.PassiveClock, wg waitgroup.OptionalWaitGroup) *queueSetImpl { fq := &queueSetImpl{ wg: wg, + name: name, queues: initQueues(desiredNumQueues), clk: clk, vt: 0, @@ -173,6 +175,7 @@ func (qs *queueSetImpl) TimeoutOldRequestsAndRejectOrEnqueue(hashValue uint64, h if ok := qs.rejectOrEnqueue(pkt); !ok { return nil } + metrics.ObserveQueueLength(qs.name, len(queue.Requests)) return pkt } @@ -267,6 +270,8 @@ func (qs *queueSetImpl) enqueue(packet *Request) { queue.Enqueue(packet) qs.updateQueueVirStartTime(packet, queue) qs.numRequestsEnqueued++ + + metrics.UpdateFlowControlRequestsInQueue(qs.name, qs.numRequestsEnqueued) } // Enqueue enqueues a packet directly into an queueSetImpl w/ no restriction @@ -396,8 +401,10 @@ func (qs *queueSetImpl) dequeue() (*Request, bool) { queue.VirStart += qs.estimatedServiceTime packet.StartTime = qs.clk.Now() + // request dequeued, service has started queue.RequestsExecuting++ + metrics.UpdateFlowControlRequestsExecuting(qs.name, queue.RequestsExecuting) } else { // TODO(aaron-prindle) verify this statement is needed... return nil, false @@ -569,6 +576,7 @@ func (qs *queueSetImpl) Wait(hashValue uint64, handSize int32) (quiescent, execu // concurrency shares and at max queue length already if pkt == nil { qs.lock.Unlock() + metrics.AddReject(qs.name, "queue-full") return false, false, func() {} } // ======================================================================== @@ -607,6 +615,7 @@ func (qs *queueSetImpl) Wait(hashValue uint64, handSize int32) (quiescent, execu } // timed out klog.V(5).Infof("pkt.DequeueChannel timed out\n") + metrics.AddReject(qs.name, "time-out") return false, false, func() {} } // ************************************************************************ 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 index db79450e476..c46affa181b 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/interface.go +++ b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/interface.go @@ -26,7 +26,7 @@ import ( // 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 + NewQueueSet(name string, concurrencyLimit, numQueues, queueLengthLimit int, requestWaitLimit time.Duration) QueueSet } // QueueSet is the abstraction for the queuing and dispatching diff --git a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/no-restraint.go b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/no-restraint.go index 9c16d5063ff..59e9923dc0e 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/no-restraint.go +++ b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/no-restraint.go @@ -29,7 +29,7 @@ func NewNoRestraintFactory() QueueSetFactory { type noRestraintFactory struct{} -func (noRestraintFactory) NewQueueSet(concurrencyLimit, numQueues, queueLengthLimit int, requestWaitLimit time.Duration) QueueSet { +func (noRestraintFactory) NewQueueSet(name string, concurrencyLimit, numQueues, queueLengthLimit int, requestWaitLimit time.Duration) QueueSet { return noRestraint{} } diff --git a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/metrics/BUILD b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/metrics/BUILD new file mode 100644 index 00000000000..115224dc108 --- /dev/null +++ b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/metrics/BUILD @@ -0,0 +1,24 @@ +load("@io_bazel_rules_go//go:def.bzl", "go_library") + +go_library( + name = "go_default_library", + srcs = ["metrics.go"], + importmap = "k8s.io/kubernetes/vendor/k8s.io/apiserver/pkg/util/flowcontrol/metrics", + importpath = "k8s.io/apiserver/pkg/util/flowcontrol/metrics", + visibility = ["//visibility:public"], + deps = ["//vendor/github.com/prometheus/client_golang/prometheus: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/metrics/metrics.go b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/metrics/metrics.go new file mode 100644 index 00000000000..3e998268c00 --- /dev/null +++ b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/metrics/metrics.go @@ -0,0 +1,145 @@ +/* +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 metrics + +import ( + "time" + + "github.com/prometheus/client_golang/prometheus" +) + +const ( + namespace = "apiserver" + subsystem = "flowcontrol" +) + +const ( + priorityLevel = "priorityLevel" + flowSchema = "flowSchema" +) + +var ( + queueLengthBuckets = []float64{0, 10, 25, 50, 100, 250, 500, 1000} + requestDurationSecondsBuckets = []float64{0, 0.005, 0.02, 0.05, 0.1, 0.2, 0.5, 1, 2, 5, 10, 30} +) + +func init() { + prometheus.MustRegister(apiserver_rejected_requests) + prometheus.MustRegister(apiserver_current_inqueue_requests) + prometheus.MustRegister(apiserver_request_queue_length) + prometheus.MustRegister(apiserver_request_concurrency_limit) + prometheus.MustRegister(apiserver_current_executing_requests) + prometheus.MustRegister(apiserver_request_waiting_seconds) + prometheus.MustRegister(apiserver_request_execution_seconds) +} + +var ( + apiserver_rejected_requests = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: namespace, + Subsystem: subsystem, + Name: "rejected_requests", + Help: "Number of rejected requests by api priority and fairness system", + }, + []string{priorityLevel, "reason"}, + ) + apiserver_current_inqueue_requests = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: namespace, + Subsystem: subsystem, + Name: "current_inqueue_requests", + Help: "Number of requests currently pending in the queue by the api priority and fairness system", + }, + []string{priorityLevel}, + ) + apiserver_request_queue_length = prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Namespace: namespace, + Subsystem: subsystem, + Name: "request_queue_length", + Help: "Length of queue in the api priority and fairness system", + Buckets: queueLengthBuckets, + }, + []string{priorityLevel}, + ) + apiserver_request_concurrency_limit = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: namespace, + Subsystem: subsystem, + Name: "request_concurrency_limit", + Help: "Shared concurrency limit in the api priority and fairness system", + }, + []string{priorityLevel}, + ) + apiserver_current_executing_requests = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: namespace, + Subsystem: subsystem, + Name: "current_executing_requests", + Help: "Number of requests currently executing in the api priority and fairness system", + }, + []string{priorityLevel}, + ) + apiserver_request_waiting_seconds = prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Namespace: namespace, + Subsystem: subsystem, + Name: "request_wait_duration_seconds", + Help: "Length of time a request spent waiting in its queue", + Buckets: requestDurationSecondsBuckets, + }, + []string{priorityLevel, flowSchema, "execute"}, + ) + apiserver_request_execution_seconds = prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Namespace: namespace, + Subsystem: subsystem, + Name: "request_execution_seconds", + Help: "Time of request executing in the api priority and fairness system", + Buckets: requestDurationSecondsBuckets, + }, + []string{priorityLevel, flowSchema}, + ) +) + +func UpdateFlowControlRequestsInQueue(priorityLevel string, inqueue int) { + apiserver_current_inqueue_requests.WithLabelValues(priorityLevel).Set(float64(inqueue)) +} + +func UpdateFlowControlRequestsExecuting(priorityLevel string, executing int) { + apiserver_current_executing_requests.WithLabelValues(priorityLevel).Set(float64(executing)) +} + +func UpdateSharedConcurrencyLimit(priorityLevel string, limit int) { + apiserver_request_concurrency_limit.WithLabelValues(priorityLevel).Set(float64(limit)) +} + +func AddReject(priorityLevel string, reason string) { + apiserver_rejected_requests.WithLabelValues(priorityLevel, reason).Add(1) +} + +func ObserveQueueLength(priorityLevel string, length int) { + apiserver_request_queue_length.WithLabelValues(priorityLevel).Observe(float64(length)) +} + +func ObserveWaitingDuration(priorityLevel, flowSchema, execute string, waitTime time.Duration) { + apiserver_request_waiting_seconds.WithLabelValues(priorityLevel, flowSchema, execute).Observe(waitTime.Seconds()) +} + +func ObserveExecutionDuration(priorityLevel, flowSchema string, executionTime time.Duration) { + apiserver_request_execution_seconds.WithLabelValues(priorityLevel, flowSchema).Observe(executionTime.Seconds()) +} diff --git a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/reqmgmt-config.go b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/reqmgmt-config.go index 9fd1e3a437b..ea7142e7ec3 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/reqmgmt-config.go +++ b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/reqmgmt-config.go @@ -26,6 +26,7 @@ import ( "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/util/wait" fq "k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing" + "k8s.io/apiserver/pkg/util/flowcontrol/metrics" kubeinformers "k8s.io/client-go/informers" "k8s.io/client-go/tools/cache" "k8s.io/client-go/util/workqueue" @@ -202,9 +203,10 @@ func (reqMgmt *requestManagementSystem) digestConfigObjects(newPLs []*rmtypesv1a continue } plState.concurrencyLimit = int(math.Ceil(float64(reqMgmt.serverConcurrencyLimit) * float64(plState.config.AssuredConcurrencyShares) / shareSum)) + metrics.UpdateSharedConcurrencyLimit(plName, plState.concurrencyLimit) if plState.queues == nil { klog.V(5).Infof("Introducing priority level %s: config=%#+v, concurrencyLimit=%d, quiescent=%v (shares=%v, shareSum=%v)", plName, plState.config, plState.concurrencyLimit, plState.emptyHandler != nil, plState.config.AssuredConcurrencyShares, shareSum) - plState.queues = reqMgmt.queueSetFactory.NewQueueSet(plState.concurrencyLimit, int(plState.config.Queues), int(plState.config.QueueLengthLimit), reqMgmt.requestWaitLimit) + plState.queues = reqMgmt.queueSetFactory.NewQueueSet(plName, plState.concurrencyLimit, int(plState.config.Queues), int(plState.config.QueueLengthLimit), reqMgmt.requestWaitLimit) } else { klog.V(5).Infof("Retaining priority level %s: config=%#+v, concurrencyLimit=%d, quiescent=%v (shares=%v, shareSum=%v)", plName, plState.config, plState.concurrencyLimit, plState.emptyHandler != nil, plState.config.AssuredConcurrencyShares, shareSum) plState.queues.SetConfiguration(plState.concurrencyLimit, int(plState.config.Queues), int(plState.config.QueueLengthLimit), reqMgmt.requestWaitLimit) 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 9a9c6b14e22..f2cc8fbac1b 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/reqmgmt.go +++ b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/reqmgmt.go @@ -35,6 +35,7 @@ import ( "k8s.io/apiserver/pkg/authentication/user" "k8s.io/apiserver/pkg/endpoints/request" fq "k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing" + "k8s.io/apiserver/pkg/util/flowcontrol/metrics" kubeinformers "k8s.io/client-go/informers" "k8s.io/client-go/tools/cache" "k8s.io/client-go/util/workqueue" @@ -220,6 +221,7 @@ type RequestDigest struct { } func (reqMgmt *requestManagementSystem) Wait(requestDigest RequestDigest) (bool, func()) { + startWaitingTime := time.Now() for { rmState := reqMgmt.curState.Load().(*requestManagementState) @@ -235,6 +237,7 @@ func (reqMgmt *requestManagementSystem) Wait(requestDigest RequestDigest) (bool, matchingpriorityLevelName = fs.Spec.PriorityLevelConfiguration.Name matchingFlowDistinguisherMethod = fs.Spec.DistinguisherMethod default: // reject + metrics.AddReject("", "non-match") klog.V(7).Infof("Rejecting requestInfo=%v, user=%v because no FlowSchema matched", requestDigest.RequestInfo, requestDigest.User) return false, func() {} } @@ -249,17 +252,24 @@ func (reqMgmt *requestManagementSystem) Wait(requestDigest RequestDigest) (bool, klog.V(7).Infof("Serving requestInfo=%v, user=%v, fs=%q without delay", requestDigest.RequestInfo, requestDigest.User, fs.Name) return true, func() {} } + quiescent, execute, afterExecute := ps.queues.Wait(hashValue, ps.config.HandSize) if quiescent { klog.V(5).Infof("Request requestInfo=%v, user=%v, fs=%q landed in timing splinter, re-classifying", requestDigest.RequestInfo, requestDigest.User, fs.Name) continue } if execute { + metrics.ObserveWaitingDuration(matchingpriorityLevelName, fs.Name, "true", time.Now().Sub(startWaitingTime)) klog.V(7).Infof("Serving requestInfo=%v, user=%v, fs=%q after fair queuing", requestDigest.RequestInfo, requestDigest.User, fs.Name) } else { + metrics.ObserveWaitingDuration(matchingpriorityLevelName, fs.Name, "false", time.Now().Sub(startWaitingTime)) klog.V(7).Infof("Rejecting requestInfo=%v, user=%v, fs=%q after fair queuing", requestDigest.RequestInfo, requestDigest.User, fs.Name) } - return execute, afterExecute + startExecutionTime := time.Now() + return execute, func() { + metrics.ObserveExecutionDuration(matchingpriorityLevelName, fs.Name, time.Now().Sub(startExecutionTime)) + afterExecute() + } } } diff --git a/vendor/modules.txt b/vendor/modules.txt index 6786b4cd99e..e61d594e7c7 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -1371,6 +1371,7 @@ 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/flowcontrol/metrics k8s.io/apiserver/pkg/util/flushwriter k8s.io/apiserver/pkg/util/openapi k8s.io/apiserver/pkg/util/proxy