Merge pull request kubernetes#91177 from MikeSpreitzer/more-concurrency-details

Introduce more metrics on concurrency

Author: k8s-ci-robot

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