prometheus metrics

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",

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,

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

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"],

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",
     ],

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() {}
 	}
 	// ************************************************************************

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

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{}
 }
 

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"],
+)

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())
+}

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)

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("<none>", "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()
+		}
 	}
 }
 

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