Introduce more metrics on concurrency

Introduce min, average, and standard deviation for the number of
executing mutating and readOnly requests.

Introduce min, max, average, and standard deviation for the number
waiting and number waiting per priority level.

Later:

Revised to use a series of windows

Use three individuals instead of array of powers

Later:

Add coarse queue count metrics, removed windowed avg and stddev

Add metrics for number of queued mutating and readOnly requests,
to complement metrics for number executing.

Later:

Removed windowed average and standard deviation because consumers can
derive such from integrals of consumer's chosen window.

Also replaced "requestKind" Prometheus label with "request_kind".

Later:

Revised to focus on sampling

Make the clock intrinsic to a TimedObserver

... so that the clock can be read while holding the observer's lock;
otherwise, forward progress is not guaranteed (and violations were
observed in testing).

Bug fixes and histogram buckets revision

SetX1 to 1 when queue length limit is zero, beause dividing by zero is nasty.

Remove obsolete argument in gen_test.go.

Add a bucket boundary at 0 for sample-and-water-mark histograms, to
distinguish zeroes from non-zeros.

This includes adding Integrator test.

Simplified test code.

More pervasively used "ctlr" instead of "ctl" as abbreviation for
"controller".

Author: MikeSpreitzer

staging/src/k8s.io/apiserver/pkg/endpoints/metrics/metrics.go

@@ -122,7 +122,7 @@ var (
 			Help:           "Number of requests dropped with 'Try again later' response",
 			StabilityLevel: compbasemetrics.ALPHA,
 		},
-		[]string{"requestKind"},
+		[]string{"request_kind"},
 	)
 	// TLSHandshakeErrors is a number of requests dropped with 'TLS handshake error from' error
 	TLSHandshakeErrors = compbasemetrics.NewCounter(
@@ -166,7 +166,15 @@ var (
 			Help:           "Maximal number of currently used inflight request limit of this apiserver per request kind in last second.",
 			StabilityLevel: compbasemetrics.ALPHA,
 		},
-		[]string{"requestKind"},
+		[]string{"request_kind"},
+	)
+	currentInqueueRequests = compbasemetrics.NewGaugeVec(
+		&compbasemetrics.GaugeOpts{
+			Name:           "apiserver_current_inqueue_requests",
+			Help:           "Maximal number of queued requests in this apiserver per request kind in last second.",
+			StabilityLevel: compbasemetrics.ALPHA,
+		},
+		[]string{"request_kind"},
 	)
 
 	requestTerminationsTotal = compbasemetrics.NewCounterVec(
@@ -191,6 +199,7 @@ var (
 		WatchEvents,
 		WatchEventsSizes,
 		currentInflightRequests,
+		currentInqueueRequests,
 		requestTerminationsTotal,
 	}
 
@@ -231,6 +240,11 @@ const (
 	ReadOnlyKind = "readOnly"
 	// MutatingKind is a string identifying mutating request kind
 	MutatingKind = "mutating"
+
+	// WaitingPhase is the phase value for a request waiting in a queue
+	WaitingPhase = "waiting"
+	// ExecutingPhase is the phase value for an executing request
+	ExecutingPhase = "executing"
 )
 
 const (
@@ -261,9 +275,19 @@ func Reset() {
 	}
 }
 
-func UpdateInflightRequestMetrics(nonmutating, mutating int) {
-	currentInflightRequests.WithLabelValues(ReadOnlyKind).Set(float64(nonmutating))
-	currentInflightRequests.WithLabelValues(MutatingKind).Set(float64(mutating))
+// UpdateInflightRequestMetrics reports concurrency metrics classified by
+// mutating vs Readonly.
+func UpdateInflightRequestMetrics(phase string, nonmutating, mutating int) {
+	for _, kc := range []struct {
+		kind  string
+		count int
+	}{{ReadOnlyKind, nonmutating}, {MutatingKind, mutating}} {
+		if phase == ExecutingPhase {
+			currentInflightRequests.WithLabelValues(kc.kind).Set(float64(kc.count))
+		} else {
+			currentInqueueRequests.WithLabelValues(kc.kind).Set(float64(kc.count))
+		}
+	}
 }
 
 // RecordRequestTermination records that the request was terminated early as part of a resource

staging/src/k8s.io/apiserver/pkg/server/filters/BUILD

@@ -62,6 +62,7 @@ go_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/apiserver/pkg/util/flowcontrol:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/util/flowcontrol/metrics:go_default_library",
         "//staging/src/k8s.io/client-go/kubernetes/scheme:go_default_library",
         "//vendor/k8s.io/klog/v2:go_default_library",
     ],

staging/src/k8s.io/apiserver/pkg/server/filters/maxinflight.go

@@ -27,6 +27,7 @@ import (
 	"k8s.io/apiserver/pkg/authentication/user"
 	"k8s.io/apiserver/pkg/endpoints/metrics"
 	apirequest "k8s.io/apiserver/pkg/endpoints/request"
+	fcmetrics "k8s.io/apiserver/pkg/util/flowcontrol/metrics"
 
 	"k8s.io/klog/v2"
 )
@@ -50,13 +51,17 @@ func handleError(w http.ResponseWriter, r *http.Request, err error) {
 	klog.Errorf(err.Error())
 }
 
-// requestWatermark is used to trak maximal usage of inflight requests.
+// requestWatermark is used to track maximal numbers of requests in a particular phase of handling
 type requestWatermark struct {
+	phase                                string
+	readOnlyObserver, mutatingObserver   fcmetrics.TimedObserver
 	lock                                 sync.Mutex
 	readOnlyWatermark, mutatingWatermark int
 }
 
 func (w *requestWatermark) recordMutating(mutatingVal int) {
+	w.mutatingObserver.Set(float64(mutatingVal))
+
 	w.lock.Lock()
 	defer w.lock.Unlock()
 
@@ -66,6 +71,8 @@ func (w *requestWatermark) recordMutating(mutatingVal int) {
 }
 
 func (w *requestWatermark) recordReadOnly(readOnlyVal int) {
+	w.readOnlyObserver.Set(float64(readOnlyVal))
+
 	w.lock.Lock()
 	defer w.lock.Unlock()
 
@@ -74,9 +81,14 @@ func (w *requestWatermark) recordReadOnly(readOnlyVal int) {
 	}
 }
 
-var watermark = &requestWatermark{}
+// watermark tracks requests being executed (not waiting in a queue)
+var watermark = &requestWatermark{
+	phase:            metrics.ExecutingPhase,
+	readOnlyObserver: fcmetrics.ReadWriteConcurrencyObserverPairGenerator.Generate(1, 1, []string{metrics.ReadOnlyKind}).RequestsExecuting,
+	mutatingObserver: fcmetrics.ReadWriteConcurrencyObserverPairGenerator.Generate(1, 1, []string{metrics.MutatingKind}).RequestsExecuting,
+}
 
-func startRecordingUsage() {
+func startRecordingUsage(watermark *requestWatermark) {
 	go func() {
 		wait.Forever(func() {
 			watermark.lock.Lock()
@@ -86,7 +98,7 @@ func startRecordingUsage() {
 			watermark.mutatingWatermark = 0
 			watermark.lock.Unlock()
 
-			metrics.UpdateInflightRequestMetrics(readOnlyWatermark, mutatingWatermark)
+			metrics.UpdateInflightRequestMetrics(watermark.phase, readOnlyWatermark, mutatingWatermark)
 		}, inflightUsageMetricUpdatePeriod)
 	}()
 }
@@ -100,17 +112,19 @@ func WithMaxInFlightLimit(
 	mutatingLimit int,
 	longRunningRequestCheck apirequest.LongRunningRequestCheck,
 ) http.Handler {
-	startOnce.Do(startRecordingUsage)
+	startOnce.Do(func() { startRecordingUsage(watermark) })
 	if nonMutatingLimit == 0 && mutatingLimit == 0 {
 		return handler
 	}
 	var nonMutatingChan chan bool
 	var mutatingChan chan bool
 	if nonMutatingLimit != 0 {
 		nonMutatingChan = make(chan bool, nonMutatingLimit)
+		watermark.readOnlyObserver.SetX1(float64(nonMutatingLimit))
 	}
 	if mutatingLimit != 0 {
 		mutatingChan = make(chan bool, mutatingLimit)
+		watermark.mutatingObserver.SetX1(float64(mutatingLimit))
 	}
 
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -141,21 +155,22 @@ func WithMaxInFlightLimit(
 
 			select {
 			case c <- true:
-				var mutatingLen, readOnlyLen int
+				// We note the concurrency level both while the
+				// request is being served and after it is done being
+				// served, because both states contribute to the
+				// sampled stats on concurrency.
 				if isMutatingRequest {
-					mutatingLen = len(mutatingChan)
+					watermark.recordMutating(len(c))
 				} else {
-					readOnlyLen = len(nonMutatingChan)
+					watermark.recordReadOnly(len(c))
 				}
-
 				defer func() {
 					<-c
 					if isMutatingRequest {
-						watermark.recordMutating(mutatingLen)
+						watermark.recordMutating(len(c))
 					} else {
-						watermark.recordReadOnly(readOnlyLen)
+						watermark.recordReadOnly(len(c))
 					}
-
 				}()
 				handler.ServeHTTP(w, r)
 

staging/src/k8s.io/apiserver/pkg/server/filters/priority-and-fairness.go

@@ -24,8 +24,10 @@ import (
 
 	fcv1a1 "k8s.io/api/flowcontrol/v1alpha1"
 	apitypes "k8s.io/apimachinery/pkg/types"
+	epmetrics "k8s.io/apiserver/pkg/endpoints/metrics"
 	apirequest "k8s.io/apiserver/pkg/endpoints/request"
 	utilflowcontrol "k8s.io/apiserver/pkg/util/flowcontrol"
+	fcmetrics "k8s.io/apiserver/pkg/util/flowcontrol/metrics"
 	"k8s.io/klog/v2"
 )
 
@@ -53,7 +55,15 @@ func GetClassification(ctx context.Context) *PriorityAndFairnessClassification {
 	return ctx.Value(priorityAndFairnessKey).(*PriorityAndFairnessClassification)
 }
 
-var atomicMutatingLen, atomicNonMutatingLen int32
+// waitingMark tracks requests waiting rather than being executed
+var waitingMark = &requestWatermark{
+	phase:            epmetrics.WaitingPhase,
+	readOnlyObserver: fcmetrics.ReadWriteConcurrencyObserverPairGenerator.Generate(1, 1, []string{epmetrics.ReadOnlyKind}).RequestsWaiting,
+	mutatingObserver: fcmetrics.ReadWriteConcurrencyObserverPairGenerator.Generate(1, 1, []string{epmetrics.MutatingKind}).RequestsWaiting,
+}
+
+var atomicMutatingExecuting, atomicReadOnlyExecuting int32
+var atomicMutatingWaiting, atomicReadOnlyWaiting int32
 
 // WithPriorityAndFairness limits the number of in-flight
 // requests in a fine-grained way.
@@ -66,7 +76,10 @@ func WithPriorityAndFairness(
 		klog.Warningf("priority and fairness support not found, skipping")
 		return handler
 	}
-	startOnce.Do(startRecordingUsage)
+	startOnce.Do(func() {
+		startRecordingUsage(watermark)
+		startRecordingUsage(waitingMark)
+	})
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		ctx := r.Context()
 		requestInfo, ok := apirequest.RequestInfoFrom(ctx)
@@ -98,22 +111,23 @@ func WithPriorityAndFairness(
 
 		var served bool
 		isMutatingRequest := !nonMutatingRequestVerbs.Has(requestInfo.Verb)
-		execute := func() {
-			var mutatingLen, readOnlyLen int
+		noteExecutingDelta := func(delta int32) {
 			if isMutatingRequest {
-				mutatingLen = int(atomic.AddInt32(&atomicMutatingLen, 1))
+				watermark.recordMutating(int(atomic.AddInt32(&atomicMutatingExecuting, delta)))
 			} else {
-				readOnlyLen = int(atomic.AddInt32(&atomicNonMutatingLen, 1))
+				watermark.recordReadOnly(int(atomic.AddInt32(&atomicReadOnlyExecuting, delta)))
 			}
-			defer func() {
-				if isMutatingRequest {
-					atomic.AddInt32(&atomicMutatingLen, -11)
-					watermark.recordMutating(mutatingLen)
-				} else {
-					atomic.AddInt32(&atomicNonMutatingLen, -1)
-					watermark.recordReadOnly(readOnlyLen)
-				}
-			}()
+		}
+		noteWaitingDelta := func(delta int32) {
+			if isMutatingRequest {
+				waitingMark.recordMutating(int(atomic.AddInt32(&atomicMutatingWaiting, delta)))
+			} else {
+				waitingMark.recordReadOnly(int(atomic.AddInt32(&atomicReadOnlyWaiting, delta)))
+			}
+		}
+		execute := func() {
+			noteExecutingDelta(1)
+			defer noteExecutingDelta(-1)
 			served = true
 			innerCtx := context.WithValue(ctx, priorityAndFairnessKey, classification)
 			innerReq := r.Clone(innerCtx)
@@ -122,10 +136,15 @@ func WithPriorityAndFairness(
 			handler.ServeHTTP(w, innerReq)
 		}
 		digest := utilflowcontrol.RequestDigest{requestInfo, user}
-		fcIfc.Handle(ctx, digest, note, execute)
+		fcIfc.Handle(ctx, digest, note, func(inQueue bool) {
+			if inQueue {
+				noteWaitingDelta(1)
+			} else {
+				noteWaitingDelta(-1)
+			}
+		}, execute)
 		if !served {
 			tooManyRequests(r, w)
-			return
 		}
 
 	})

staging/src/k8s.io/apiserver/pkg/util/flowcontrol/BUILD

@@ -85,6 +85,7 @@ go_test(
         "//staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/testing:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/util/flowcontrol/format: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/fake:go_default_library",
         "//staging/src/k8s.io/client-go/kubernetes/typed/flowcontrol/v1alpha1:go_default_library",