From fb9646fd60d4b8e79223b729c1cb54fc6818fdd1 Mon Sep 17 00:00:00 2001 From: Andrew Sy Kim Date: Mon, 24 Jul 2023 19:40:05 +0000 Subject: [PATCH] apiserver: add flow control metric current_inqueue_seats Signed-off-by: Andrew Sy Kim --- .../fairqueuing/queueset/queueset.go | 4 +++ .../fairqueuing/queueset/queueset_test.go | 30 +++++++++++++------ .../pkg/util/flowcontrol/metrics/metrics.go | 16 ++++++++++ 3 files changed, 41 insertions(+), 9 deletions(-) diff --git a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/queueset/queueset.go b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/queueset/queueset.go index aa54a9ccf1d..d955be765f5 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/queueset/queueset.go +++ b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/queueset/queueset.go @@ -448,6 +448,7 @@ func (req *request) wait() (bool, bool) { qs.totRequestsCancelled++ metrics.AddReject(req.ctx, qs.qCfg.Name, req.fsName, "cancelled") metrics.AddRequestsInQueues(req.ctx, qs.qCfg.Name, req.fsName, -1) + metrics.AddSeatsInQueues(req.ctx, qs.qCfg.Name, req.fsName, -req.MaxSeats()) req.NoteQueued(false) qs.reqsGaugePair.RequestsWaiting.Add(-1) qs.seatDemandIntegrator.Set(float64(qs.totSeatsInUse + qs.totSeatsWaiting)) @@ -652,6 +653,7 @@ func (qs *queueSet) removeTimedOutRequestsFromQueueToBoundLocked(queue *queue, f disqueueSeats += req.MaxSeats() req.NoteQueued(false) metrics.AddRequestsInQueues(req.ctx, qs.qCfg.Name, req.fsName, -1) + metrics.AddSeatsInQueues(req.ctx, qs.qCfg.Name, req.fsName, -req.MaxSeats()) } // we need to check if the next request has timed out. return true @@ -702,6 +704,7 @@ func (qs *queueSet) enqueueToBoundLocked(request *request) { qs.totRequestsWaiting++ qs.totSeatsWaiting += request.MaxSeats() metrics.AddRequestsInQueues(request.ctx, qs.qCfg.Name, request.fsName, 1) + metrics.AddSeatsInQueues(request.ctx, qs.qCfg.Name, request.fsName, request.MaxSeats()) request.NoteQueued(true) qs.reqsGaugePair.RequestsWaiting.Add(1) qs.seatDemandIntegrator.Set(float64(qs.totSeatsInUse + qs.totSeatsWaiting)) @@ -760,6 +763,7 @@ func (qs *queueSet) dispatchLocked() bool { qs.totRequestsWaiting-- qs.totSeatsWaiting -= request.MaxSeats() metrics.AddRequestsInQueues(request.ctx, qs.qCfg.Name, request.fsName, -1) + metrics.AddSeatsInQueues(request.ctx, qs.qCfg.Name, request.fsName, -request.MaxSeats()) request.NoteQueued(false) qs.reqsGaugePair.RequestsWaiting.Add(-1) defer qs.boundNextDispatchLocked(queue) diff --git a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/queueset/queueset_test.go b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/queueset/queueset_test.go index 89ed063d53b..1d3d5e5b215 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/queueset/queueset_test.go +++ b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/queueset/queueset_test.go @@ -209,13 +209,13 @@ func (us uniformScenario) exercise(t *testing.T) { type uniformScenarioState struct { t *testing.T uniformScenario - startTime time.Time - doSplit bool - execSeatsIntegrators []fq.Integrator - seatDemandIntegratorCheck fq.Integrator - failedCount uint64 - expectedInqueue, expectedExecuting, expectedConcurrencyInUse string - executions, rejects []int32 + startTime time.Time + doSplit bool + execSeatsIntegrators []fq.Integrator + seatDemandIntegratorCheck fq.Integrator + failedCount uint64 + expectedInqueueReqs, expectedInqueueSeats, expectedExecuting, expectedConcurrencyInUse string + executions, rejects []int32 } func (uss *uniformScenarioState) exercise() { @@ -226,7 +226,8 @@ func (uss *uniformScenarioState) exercise() { for i, uc := range uss.clients { uss.execSeatsIntegrators[i] = fq.NewNamedIntegrator(uss.clk, fmt.Sprintf("%s client %d execSeats", uss.name, i)) fsName := fmt.Sprintf("client%d", i) - uss.expectedInqueue = uss.expectedInqueue + fmt.Sprintf(` apiserver_flowcontrol_current_inqueue_requests{flow_schema=%q,priority_level=%q} 0%s`, fsName, uss.name, "\n") + uss.expectedInqueueReqs = uss.expectedInqueueReqs + fmt.Sprintf(` apiserver_flowcontrol_current_inqueue_requests{flow_schema=%q,priority_level=%q} 0%s`, fsName, uss.name, "\n") + uss.expectedInqueueSeats = uss.expectedInqueueSeats + fmt.Sprintf(` apiserver_flowcontrol_current_inqueue_seats{flow_schema=%q,priority_level=%q} 0%s`, fsName, uss.name, "\n") for j := 0; j < uc.nThreads; j++ { ust := uniformScenarioThread{ uss: uss, @@ -412,13 +413,24 @@ func (uss *uniformScenarioState) finalReview() { e := ` # HELP apiserver_flowcontrol_current_inqueue_requests [BETA] Number of requests currently pending in queues of the API Priority and Fairness subsystem # TYPE apiserver_flowcontrol_current_inqueue_requests gauge -` + uss.expectedInqueue +` + uss.expectedInqueueReqs err := metrics.GatherAndCompare(e, "apiserver_flowcontrol_current_inqueue_requests") if err != nil { uss.t.Error(err) } else { uss.t.Log("Success with" + e) } + + e = ` + # HELP apiserver_flowcontrol_current_inqueue_seats [ALPHA] Number of seats currently pending in queues of the API Priority and Fairness subsystem + # TYPE apiserver_flowcontrol_current_inqueue_seats gauge +` + uss.expectedInqueueSeats + err = metrics.GatherAndCompare(e, "apiserver_flowcontrol_current_inqueue_seats") + if err != nil { + uss.t.Error(err) + } else { + uss.t.Log("Success with" + e) + } } expectedRejects := "" for i := range uss.clients { 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 index 54af4415cd0..9fe7b15a050 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/metrics/metrics.go +++ b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/metrics/metrics.go @@ -210,6 +210,16 @@ var ( }, []string{priorityLevel, flowSchema}, ) + apiserverCurrentInqueueSeats = compbasemetrics.NewGaugeVec( + &compbasemetrics.GaugeOpts{ + Namespace: namespace, + Subsystem: subsystem, + Name: "current_inqueue_seats", + Help: "Number of seats currently pending in queues of the API Priority and Fairness subsystem", + StabilityLevel: compbasemetrics.ALPHA, + }, + []string{priorityLevel, flowSchema}, + ) apiserverRequestQueueLength = compbasemetrics.NewHistogramVec( &compbasemetrics.HistogramOpts{ Namespace: namespace, @@ -455,6 +465,7 @@ var ( apiserverNextSBounds, apiserverNextDiscountedSBounds, apiserverCurrentInqueueRequests, + apiserverCurrentInqueueSeats, apiserverRequestQueueLength, apiserverRequestConcurrencyLimit, apiserverRequestConcurrencyInUse, @@ -518,6 +529,11 @@ func AddRequestsInQueues(ctx context.Context, priorityLevel, flowSchema string, apiserverCurrentInqueueRequests.WithLabelValues(priorityLevel, flowSchema).Add(float64(delta)) } +// AddSeatsInQueues adds the given delta to the gauge of the # of seats in the queues of the specified flowSchema and priorityLevel +func AddSeatsInQueues(ctx context.Context, priorityLevel, flowSchema string, delta int) { + apiserverCurrentInqueueSeats.WithLabelValues(priorityLevel, flowSchema).Add(float64(delta)) +} + // AddRequestsExecuting adds the given delta to the gauge of executing requests of the given flowSchema and priorityLevel func AddRequestsExecuting(ctx context.Context, priorityLevel, flowSchema string, delta int) { apiserverCurrentExecutingRequests.WithLabelValues(priorityLevel, flowSchema).Add(float64(delta))