From 645236a1c761dbac432d69c40e7f5853d1912c44 Mon Sep 17 00:00:00 2001 From: okjiang <819421878@qq.com> Date: Mon, 20 Apr 2026 17:44:09 +0800 Subject: [PATCH 1/6] client/resource_group: add client-side demand observability Signed-off-by: okjiang <819421878@qq.com> --- client/go.mod | 1 + client/go.sum | 2 + .../controller/global_controller.go | 31 ++- .../controller/global_controller_test.go | 59 ++++++ .../controller/group_controller.go | 196 +++++++++++++++--- .../controller/group_controller_test.go | 81 ++++++-- client/resource_group/controller/limiter.go | 23 +- .../controller/metrics/metrics.go | 145 ++++++++++++- 8 files changed, 484 insertions(+), 54 deletions(-) diff --git a/client/go.mod b/client/go.mod index 862feac2a3..82fabb6aa5 100644 --- a/client/go.mod +++ b/client/go.mod @@ -28,6 +28,7 @@ require ( github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/davecgh/go-spew v1.1.1 // indirect github.com/golang/protobuf v1.5.4 // indirect + github.com/kylelemons/godebug v1.1.0 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/pkg/errors v0.9.1 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect diff --git a/client/go.sum b/client/go.sum index d7c318da5c..7ddd7438ed 100644 --- a/client/go.sum +++ b/client/go.sum @@ -38,6 +38,8 @@ github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3x github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= +github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= +github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE= diff --git a/client/resource_group/controller/global_controller.go b/client/resource_group/controller/global_controller.go index 87e4f2cb10..949dff1b55 100644 --- a/client/resource_group/controller/global_controller.go +++ b/client/resource_group/controller/global_controller.go @@ -435,7 +435,7 @@ func (c *ResourceGroupsController) Start(ctx context.Context) { continue } // If the resource group is marked as tombstone before, re-create the resource group controller. - newGC, err := newGroupCostController(group, c.ruConfig, c.lowTokenNotifyChan, c.tokenBucketUpdateChan) + newGC, err := newGroupCostController(group, c.clientUniqueID, c.ruConfig, c.lowTokenNotifyChan, c.tokenBucketUpdateChan) if err != nil { log.Warn("[resource group controller] re-create resource group cost controller for tombstone failed", zap.String("name", name), zap.Error(err)) @@ -574,7 +574,7 @@ func (c *ResourceGroupsController) tryGetResourceGroupController( return gc, nil } // Initialize the resource group controller. - gc, err = newGroupCostController(group, c.ruConfig, c.lowTokenNotifyChan, c.tokenBucketUpdateChan) + gc, err = newGroupCostController(group, c.clientUniqueID, c.ruConfig, c.lowTokenNotifyChan, c.tokenBucketUpdateChan) if err != nil { return nil, err } @@ -610,7 +610,7 @@ func (c *ResourceGroupsController) tombstoneGroupCostController(name string) { return } // Create a default resource group controller for the tombstone resource group independently. - gc, err := newGroupCostController(defaultGC.getMeta(), c.ruConfig, c.lowTokenNotifyChan, c.tokenBucketUpdateChan) + gc, err := newGroupCostController(defaultGC.getMeta(), c.clientUniqueID, c.ruConfig, c.lowTokenNotifyChan, c.tokenBucketUpdateChan) if err != nil { log.Warn("[resource group controller] create default resource group cost controller for tombstone failed", zap.String("name", name), zap.Error(err)) @@ -639,6 +639,19 @@ func (c *ResourceGroupsController) cleanUpResourceGroup() { c.groupsController.Delete(resourceGroupName) metrics.ResourceGroupStatusGauge.DeleteLabelValues(resourceGroupName, resourceGroupName) gc.metrics.deletePagingLabels(resourceGroupName) + metrics.TokenConsumedHistogram.DeleteLabelValues(resourceGroupName) + metrics.TokenConsumedByTypeCounter.DeleteLabelValues(resourceGroupName, "rru") + metrics.TokenConsumedByTypeCounter.DeleteLabelValues(resourceGroupName, "wru") + metrics.TokenBalanceGauge.DeleteLabelValues(resourceGroupName) + metrics.FillRateGauge.DeleteLabelValues(resourceGroupName) + metrics.BurstLimitGauge.DeleteLabelValues(resourceGroupName) + metrics.AvgRUPerSecGauge.DeleteLabelValues(resourceGroupName) + metrics.DemandRUPerSecGauge.DeleteLabelValues(resourceGroupName) + metrics.ThrottledGauge.DeleteLabelValues(resourceGroupName) + metrics.ReadByteCost.DeleteLabelValues(resourceGroupName) + metrics.WriteByteCost.DeleteLabelValues(resourceGroupName) + metrics.KVCPUCost.DeleteLabelValues(resourceGroupName) + metrics.SQLCPUCost.DeleteLabelValues(resourceGroupName) return true } gc.inactive = true @@ -723,7 +736,17 @@ func (c *ResourceGroupsController) sendTokenBucketRequests(ctx context.Context, metrics.SuccessfulTokenRequestDuration.Observe(latency.Seconds()) } if !notifyMsg.startTime.IsZero() && time.Since(notifyMsg.startTime) > slowNotifyFilterDuration { - log.Warn("[resource group controller] slow token bucket request", zap.String("source", source), zap.Duration("cost", time.Since(notifyMsg.startTime))) + groupNames := make([]string, 0, len(req.Requests)) + for _, request := range req.Requests { + groupNames = append(groupNames, request.GetResourceGroupName()) + } + log.Warn( + "[resource group controller] slow token bucket request", + zap.String("source", source), + zap.Duration("cost", time.Since(notifyMsg.startTime)), + zap.Uint64("client_unique_id", c.clientUniqueID), + zap.Strings("resource_groups", groupNames), + ) } logControllerTrace("[resource group controller] token bucket response", zap.Time("now", time.Now()), zap.Any("resp", resp), zap.String("source", source), zap.Duration("latency", latency)) c.tokenResponseChan <- resp diff --git a/client/resource_group/controller/global_controller_test.go b/client/resource_group/controller/global_controller_test.go index 2526680998..029dfbf8d5 100644 --- a/client/resource_group/controller/global_controller_test.go +++ b/client/resource_group/controller/global_controller_test.go @@ -40,6 +40,7 @@ import ( "github.com/tikv/pd/client/errs" "github.com/tikv/pd/client/opt" "github.com/tikv/pd/client/pkg/utils/testutil" + "github.com/tikv/pd/client/resource_group/controller/metrics" ) func TestMain(m *testing.M) { @@ -304,6 +305,64 @@ func TestTryGetController(t *testing.T) { re.NotEmpty(consumption) } +func TestCleanupResourceGroupMetrics(t *testing.T) { + re := require.New(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + mockProvider := newMockResourceGroupProvider() + controller, err := NewResourceGroupController(ctx, 1, mockProvider, nil, constants.NullKeyspaceID) + re.NoError(err) + + testResourceGroup := &rmpb.ResourceGroup{ + Name: "cleanup-group", + Mode: rmpb.GroupMode_RUMode, + RUSettings: &rmpb.GroupRequestUnitSettings{ + RU: &rmpb.TokenBucket{ + Settings: &rmpb.TokenLimitSettings{FillRate: 1000}, + }, + }, + } + mockProvider.On("GetResourceGroup", mock.Anything, "cleanup-group", mock.Anything).Return(testResourceGroup, nil) + + gc, err := controller.tryGetResourceGroupController(ctx, "cleanup-group", false) + re.NoError(err) + re.NotNil(gc) + + metrics.ResourceGroupStatusGauge.WithLabelValues("cleanup-group", "cleanup-group").Set(1) + metrics.TokenConsumedHistogram.WithLabelValues("cleanup-group").Observe(1) + metrics.TokenConsumedByTypeCounter.WithLabelValues("cleanup-group", "rru").Add(1) + metrics.TokenConsumedByTypeCounter.WithLabelValues("cleanup-group", "wru").Add(1) + metrics.TokenBalanceGauge.WithLabelValues("cleanup-group").Set(1) + metrics.FillRateGauge.WithLabelValues("cleanup-group").Set(1) + metrics.BurstLimitGauge.WithLabelValues("cleanup-group").Set(1) + metrics.AvgRUPerSecGauge.WithLabelValues("cleanup-group").Set(1) + metrics.DemandRUPerSecGauge.WithLabelValues("cleanup-group").Set(1) + metrics.ThrottledGauge.WithLabelValues("cleanup-group").Set(1) + metrics.ReadByteCost.WithLabelValues("cleanup-group").Add(1) + metrics.WriteByteCost.WithLabelValues("cleanup-group").Add(1) + metrics.KVCPUCost.WithLabelValues("cleanup-group").Add(1) + metrics.SQLCPUCost.WithLabelValues("cleanup-group").Add(1) + + gc.inactive = true + controller.cleanUpResourceGroup() + + re.False(metrics.ResourceGroupStatusGauge.DeleteLabelValues("cleanup-group", "cleanup-group")) + re.False(metrics.TokenConsumedHistogram.DeleteLabelValues("cleanup-group")) + re.False(metrics.TokenConsumedByTypeCounter.DeleteLabelValues("cleanup-group", "rru")) + re.False(metrics.TokenConsumedByTypeCounter.DeleteLabelValues("cleanup-group", "wru")) + re.False(metrics.TokenBalanceGauge.DeleteLabelValues("cleanup-group")) + re.False(metrics.FillRateGauge.DeleteLabelValues("cleanup-group")) + re.False(metrics.BurstLimitGauge.DeleteLabelValues("cleanup-group")) + re.False(metrics.AvgRUPerSecGauge.DeleteLabelValues("cleanup-group")) + re.False(metrics.DemandRUPerSecGauge.DeleteLabelValues("cleanup-group")) + re.False(metrics.ThrottledGauge.DeleteLabelValues("cleanup-group")) + re.False(metrics.ReadByteCost.DeleteLabelValues("cleanup-group")) + re.False(metrics.WriteByteCost.DeleteLabelValues("cleanup-group")) + re.False(metrics.KVCPUCost.DeleteLabelValues("cleanup-group")) + re.False(metrics.SQLCPUCost.DeleteLabelValues("cleanup-group")) +} + func TestGetResourceGroup(t *testing.T) { re := require.New(t) ctx, cancel := context.WithCancel(context.Background()) diff --git a/client/resource_group/controller/group_controller.go b/client/resource_group/controller/group_controller.go index 49fc4fd55e..5192eaee2d 100644 --- a/client/resource_group/controller/group_controller.go +++ b/client/resource_group/controller/group_controller.go @@ -34,9 +34,10 @@ import ( type groupCostController struct { // invariant attributes - name string - mode rmpb.GroupMode - mainCfg *RUConfig + name string + clientUniqueID uint64 + mode rmpb.GroupMode + mainCfg *RUConfig // meta info meta *rmpb.ResourceGroup metaLock sync.RWMutex @@ -113,6 +114,18 @@ type groupMetricsCollection struct { actualBytesCounter prometheus.Counter predictionResidualBytes prometheus.Observer noPrechargeCounter prometheus.Counter + consumeTokenByTypeRRU prometheus.Counter + consumeTokenByTypeWRU prometheus.Counter + tokenBalanceGauge prometheus.Gauge + fillRateGauge prometheus.Gauge + burstLimitGauge prometheus.Gauge + avgRUPerSecGauge prometheus.Gauge + demandRUPerSecGauge prometheus.Gauge + throttledGauge prometheus.Gauge + readByteCostCounter prometheus.Counter + writeByteCostCounter prometheus.Counter + kvCPUCostCounter prometheus.Counter + sqlCPUCostCounter prometheus.Counter } func initMetrics(oldName, name string) *groupMetricsCollection { @@ -135,7 +148,19 @@ func initMetrics(oldName, name string) *groupMetricsCollection { actualBytesCounter: metrics.PagingActualBytesCounter.WithLabelValues(name), predictionResidualBytes: metrics.PagingPredictionResidualBytes.WithLabelValues(name), - noPrechargeCounter: metrics.CopReadNoPrechargeCounter.WithLabelValues(name), + noPrechargeCounter: metrics.CopReadNoPrechargeCounter.WithLabelValues(name), + consumeTokenByTypeRRU: metrics.TokenConsumedByTypeCounter.WithLabelValues(name, "rru"), + consumeTokenByTypeWRU: metrics.TokenConsumedByTypeCounter.WithLabelValues(name, "wru"), + tokenBalanceGauge: metrics.TokenBalanceGauge.WithLabelValues(name), + fillRateGauge: metrics.FillRateGauge.WithLabelValues(name), + burstLimitGauge: metrics.BurstLimitGauge.WithLabelValues(name), + avgRUPerSecGauge: metrics.AvgRUPerSecGauge.WithLabelValues(name), + demandRUPerSecGauge: metrics.DemandRUPerSecGauge.WithLabelValues(name), + throttledGauge: metrics.ThrottledGauge.WithLabelValues(name), + readByteCostCounter: metrics.ReadByteCost.WithLabelValues(name), + writeByteCostCounter: metrics.WriteByteCost.WithLabelValues(name), + kvCPUCostCounter: metrics.KVCPUCost.WithLabelValues(name), + sqlCPUCostCounter: metrics.SQLCPUCost.WithLabelValues(name), } } @@ -186,6 +211,13 @@ type tokenCounter struct { avgRUPerSecLastRU float64 avgLastTime time.Time + // demandRUPerSec is an exponentially-weighted moving average of the RU + // demand per second sampled before Reserve() throttling takes effect. + demandRUPerSec float64 + demandRUPerSecLastRU float64 + demandTotalRU float64 + demandAvgLastTime time.Time + notify struct { mu sync.Mutex setupNotificationCh <-chan time.Time @@ -205,6 +237,7 @@ type tokenCounter struct { func newGroupCostController( group *rmpb.ResourceGroup, + clientUniqueID uint64, mainCfg *RUConfig, lowRUNotifyChan chan notifyMsg, tokenBucketUpdateChan chan *groupCostController, @@ -219,11 +252,12 @@ func newGroupCostController( } ms := initMetrics(group.Name, group.Name) gc := &groupCostController{ - meta: group, - name: group.Name, - mainCfg: mainCfg, - mode: group.GetMode(), - metrics: ms, + meta: group, + name: group.Name, + clientUniqueID: clientUniqueID, + mainCfg: mainCfg, + mode: group.GetMode(), + metrics: ms, calculators: []ResourceCalculator{ newKVCalculator(mainCfg), newSQLCalculator(mainCfg), @@ -272,10 +306,11 @@ func (gc *groupCostController) initRunState() { defer gc.metaLock.RUnlock() limiter := NewLimiterWithCfg(gc.name, now, cfgFunc(gc.meta.RUSettings.RU), gc.lowRUNotifyChan) counter := &tokenCounter{ - limiter: limiter, - avgRUPerSec: 0, - avgLastTime: now, - fillRate: gc.meta.RUSettings.RU.Settings.FillRate, + limiter: limiter, + avgRUPerSec: 0, + avgLastTime: now, + demandAvgLastTime: now, + fillRate: gc.meta.RUSettings.RU.Settings.FillRate, } gc.run.requestUnitTokens = counter gc.burstable.Store(isBurstable) @@ -299,19 +334,33 @@ func (gc *groupCostController) applyDegradedMode() { cfg.newFillRate = 99999999 }) counter.limiter.Reconfigure(gc.run.now, cfg, resetLowProcess()) - log.Info("[resource group controller] resource token bucket enter degraded mode", zap.String("name", gc.name)) + log.Info("[resource group controller] resource token bucket enter degraded mode", gc.logFields(0, nil)...) } func (gc *groupCostController) updateRunState() { newTime := time.Now() gc.mu.Lock() + before := *gc.mu.consumption for _, calc := range gc.calculators { calc.Trickle(gc.mu.consumption) } + trickleDelta := *gc.mu.consumption + sub(&trickleDelta, &before) *gc.run.consumption = *gc.mu.consumption gc.mu.Unlock() logControllerTrace("[resource group controller] update run state", zap.String("name", gc.name), zap.Any("request-unit-consumption", gc.run.consumption), zap.Bool("is-throttled", gc.isThrottled.Load())) gc.run.now = newTime + + gc.observeComponentConsumption(&trickleDelta) + counter := gc.run.requestUnitTokens + gc.metrics.tokenBalanceGauge.Set(counter.limiter.AvailableTokens(gc.run.now)) + gc.metrics.fillRateGauge.Set(counter.limiter.GetFillRate()) + gc.metrics.burstLimitGauge.Set(float64(counter.limiter.GetBurst())) + if gc.isThrottled.Load() { + gc.metrics.throttledGauge.Set(1) + } else { + gc.metrics.throttledGauge.Set(0) + } } func (gc *groupCostController) updateAvgRequestResourcePerSec() { @@ -323,6 +372,7 @@ func (gc *groupCostController) updateAvgRequestResourcePerSec() { if !gc.calcAvg(counter, getRUValueFromConsumption(gc.run.consumption)) { return } + gc.metrics.avgRUPerSecGauge.Set(counter.avgRUPerSec) logControllerTrace("[resource group controller] update avg ru per sec", zap.String("name", gc.name), zap.Float64("avg-ru-per-sec", counter.avgRUPerSec), zap.Bool("is-throttled", gc.isThrottled.Load())) gc.burstable.Store(isBurstable) } @@ -362,21 +412,40 @@ func (gc *groupCostController) handleTokenBucketUpdateEvent(ctx context.Context) } func (gc *groupCostController) calcAvg(counter *tokenCounter, new float64) bool { - deltaDuration := gc.run.now.Sub(counter.avgLastTime) + return gc.calcMovingAvgAt(gc.run.now, &counter.avgRUPerSec, &counter.avgRUPerSecLastRU, &counter.avgLastTime, new) +} + +func (gc *groupCostController) calcDemandAvg(counter *tokenCounter, delta float64) bool { + if delta <= 0 { + return false + } + counter.demandTotalRU += delta + ok := gc.calcMovingAvgAt(time.Now(), &counter.demandRUPerSec, &counter.demandRUPerSecLastRU, &counter.demandAvgLastTime, counter.demandTotalRU) + if ok { + gc.metrics.demandRUPerSecGauge.Set(counter.demandRUPerSec) + } + return ok +} + +func (gc *groupCostController) calcMovingAvgAt(now time.Time, avg *float64, lastRU *float64, lastTime *time.Time, new float64) bool { + deltaDuration := now.Sub(*lastTime) failpoint.Inject("acceleratedReportingPeriod", func() { deltaDuration = 100 * time.Millisecond }) - delta := (new - counter.avgRUPerSecLastRU) / deltaDuration.Seconds() - counter.avgRUPerSec = movingAvgFactor*counter.avgRUPerSec + (1-movingAvgFactor)*delta + if deltaDuration <= 0 { + return false + } + delta := (new - *lastRU) / deltaDuration.Seconds() + *avg = movingAvgFactor**avg + (1-movingAvgFactor)*delta failpoint.Inject("acceleratedSpeedTrend", func() { if delta > 0 { - counter.avgRUPerSec = 1000 + *avg = 1000 } else { - counter.avgRUPerSec = 0 + *avg = 0 } }) - counter.avgLastTime = gc.run.now - counter.avgRUPerSecLastRU = new + *lastTime = now + *lastRU = new return true } @@ -416,6 +485,7 @@ func (gc *groupCostController) handleRUTokenResponse(resp *rmpb.TokenBucketRespo func (gc *groupCostController) modifyTokenCounter(counter *tokenCounter, bucket *rmpb.TokenBucket, trickleTimeMs int64) { granted := bucket.GetTokens() + wasThrottled := gc.isThrottled.Load() if !counter.lastDeadline.IsZero() { // If last request came with a trickle duration, we may have RUs that were // not made available to the bucket yet; throw them together with the newly @@ -469,6 +539,14 @@ func (gc *groupCostController) modifyTokenCounter(counter *tokenCounter, bucket counter.lastRate = cfg.newFillRate counter.limiter.Reconfigure(gc.run.now, cfg, resetLowProcess()) + isThrottled := gc.isThrottled.Load() + if wasThrottled != isThrottled { + message := "[resource group controller] throttled state changed" + if isThrottled { + message = "[resource group controller] resource group entered throttled mode" + } + log.Info(message, gc.logFields(0, nil)...) + } } func initCounterNotify(counter *tokenCounter) { @@ -561,6 +639,55 @@ func (gc *groupCostController) calcRequest(counter *tokenCounter) float64 { return value } +func (gc *groupCostController) observeConsumption(delta *rmpb.Consumption) { + if v := getRUValueFromConsumption(delta); v > 0 { + gc.metrics.consumeTokenHistogram.Observe(v) + } + if delta.RRU > 0 { + gc.metrics.consumeTokenByTypeRRU.Add(delta.RRU) + } + if delta.WRU > 0 { + gc.metrics.consumeTokenByTypeWRU.Add(delta.WRU) + } +} + +func (gc *groupCostController) observeComponentConsumption(delta *rmpb.Consumption) { + if delta == nil { + return + } + if delta.ReadBytes > 0 { + gc.metrics.readByteCostCounter.Add(float64(delta.ReadBytes)) + } + if delta.WriteBytes > 0 { + gc.metrics.writeByteCostCounter.Add(float64(delta.WriteBytes)) + } + if delta.SqlLayerCpuTimeMs > 0 { + gc.metrics.sqlCPUCostCounter.Add(delta.SqlLayerCpuTimeMs) + } + if kvCPUMs := delta.TotalCpuTimeMs - delta.SqlLayerCpuTimeMs; kvCPUMs > 0 { + gc.metrics.kvCPUCostCounter.Add(kvCPUMs) + } +} + +func (gc *groupCostController) logFields(waitDuration time.Duration, err error) []zap.Field { + fields := []zap.Field{ + zap.String("resource_group", gc.name), + zap.Uint64("client_unique_id", gc.clientUniqueID), + zap.Float64("available_tokens", gc.run.requestUnitTokens.limiter.AvailableTokens(time.Now())), + zap.Float64("fill_rate", gc.run.requestUnitTokens.limiter.GetFillRate()), + zap.Float64("avg_ru_per_sec", gc.run.requestUnitTokens.avgRUPerSec), + zap.Float64("demand_ru_per_sec", gc.run.requestUnitTokens.demandRUPerSec), + zap.Bool("throttled", gc.isThrottled.Load()), + } + if waitDuration > 0 { + fields = append(fields, zap.Duration("wait_duration", waitDuration)) + } + if err != nil { + fields = append(fields, zap.Error(err)) + } + return fields +} + func (gc *groupCostController) acquireTokens(ctx context.Context, delta *rmpb.Consumption, waitDuration *time.Duration, allowDebt bool) (time.Duration, error) { gc.metrics.runningKVRequestCounter.Inc() defer gc.metrics.runningKVRequestCounter.Dec() @@ -568,6 +695,10 @@ func (gc *groupCostController) acquireTokens(ctx context.Context, delta *rmpb.Co err error d time.Duration ) + counter := gc.run.requestUnitTokens + if v := getRUValueFromConsumption(delta); v > 0 { + gc.calcDemandAvg(counter, v) + } retryLoop: for range gc.mainCfg.WaitRetryTimes { counter := gc.run.requestUnitTokens @@ -575,10 +706,6 @@ retryLoop: now := time.Now() var res *Reservation if v := getRUValueFromConsumption(delta); v > 0 { - // record the consume token histogram if enable controller debug mode. - if enableControllerTraceLog.Load() { - gc.metrics.consumeTokenHistogram.Observe(v) - } // allow debt for small request or not in throttled. remove tokens directly. if allowDebt { counter.limiter.RemoveTokens(now, v) @@ -644,14 +771,22 @@ func (gc *groupCostController) onRequestWaitImpl( failpoint.Inject("triggerUpdate", func() { gc.lowRUNotifyChan <- notifyMsg{} }) + if waitDuration > slowNotifyFilterDuration { + log.Warn("[resource group controller] wait for tokens failed", gc.logFields(waitDuration, err)...) + } return nil, nil, waitDuration, 0, err } gc.metrics.successfulRequestDuration.Observe(d.Seconds()) waitDuration += d + if waitDuration > slowNotifyFilterDuration { + log.Warn("[resource group controller] waited for tokens", gc.logFields(waitDuration, nil)...) + } } if bytesForEst, ok := pagingReadEstimate(info); ok { gc.metrics.observePagingRequest(bytesForEst) } + gc.observeConsumption(delta) + gc.observeComponentConsumption(delta) gc.mu.Lock() // Calculate the penalty of the store @@ -697,6 +832,8 @@ func (gc *groupCostController) onResponseImpl( counter.limiter.RefundTokens(time.Now(), -v) } } + gc.observeConsumption(delta) + gc.observeComponentConsumption(delta) gc.mu.Lock() add(gc.mu.consumption, reportedDelta) @@ -742,10 +879,16 @@ func (gc *groupCostController) onResponseWaitImpl( } else { gc.metrics.failedRequestCounterWithOthers.Inc() } + if waitDuration > slowNotifyFilterDuration { + log.Warn("[resource group controller] response wait for tokens failed", gc.logFields(waitDuration, err)...) + } return nil, waitDuration, err } gc.metrics.successfulRequestDuration.Observe(d.Seconds()) waitDuration += d + if waitDuration > slowNotifyFilterDuration { + log.Warn("[resource group controller] response waited for tokens", gc.logFields(waitDuration, nil)...) + } } else if v < 0 && isPagingRead && bytesForEst > 0 { // Paging over-estimate: refund the excess pre-charge. gc.run.requestUnitTokens.limiter.RefundTokens(time.Now(), -v) @@ -757,6 +900,8 @@ func (gc *groupCostController) onResponseWaitImpl( if isPagingRead { gc.metrics.observePagingResponse(bytesForEst, resp.ReadBytes()) } + gc.observeConsumption(delta) + gc.observeComponentConsumption(delta) gc.mu.Lock() add(gc.mu.consumption, reportedDelta) @@ -771,6 +916,7 @@ func (gc *groupCostController) addRUConsumption(consumption *rmpb.Consumption) { gc.mu.Lock() add(gc.mu.consumption, consumption) gc.mu.Unlock() + gc.observeComponentConsumption(consumption) } func (gc *groupCostController) addRUV2Consumption(tikvRUV2, tidbRUV2, tiflashRUV2 float64) { diff --git a/client/resource_group/controller/group_controller_test.go b/client/resource_group/controller/group_controller_test.go index 88e20c1753..09f7143149 100644 --- a/client/resource_group/controller/group_controller_test.go +++ b/client/resource_group/controller/group_controller_test.go @@ -21,6 +21,7 @@ import ( "time" "github.com/prometheus/client_golang/prometheus" + promtestutil "github.com/prometheus/client_golang/prometheus/testutil" dto "github.com/prometheus/client_model/go" "github.com/stretchr/testify/require" @@ -53,9 +54,13 @@ func histogramSampleCount(re *require.Assertions, h prometheus.Observer) uint64 return histogram.GetSampleCount() } -func createTestGroupCostController(re *require.Assertions) *groupCostController { +func createTestGroupCostController(re *require.Assertions, names ...string) *groupCostController { + name := "test" + if len(names) > 0 { + name = names[0] + } group := &rmpb.ResourceGroup{ - Name: "test", + Name: name, Mode: rmpb.GroupMode_RUMode, Priority: 1, RUSettings: &rmpb.GroupRequestUnitSettings{ @@ -71,19 +76,21 @@ func createTestGroupCostController(re *require.Assertions) *groupCostController } ch1 := make(chan notifyMsg) ch2 := make(chan *groupCostController) - gc, err := newGroupCostController(group, DefaultRUConfig(), ch1, ch2) + gc, err := newGroupCostController(group, 1001, DefaultRUConfig(), ch1, ch2) re.NoError(err) return gc } func TestGroupControlBurstable(t *testing.T) { re := require.New(t) - gc := createTestGroupCostController(re) + gc := createTestGroupCostController(re, t.Name()) args := tokenBucketReconfigureArgs{ newFillRate: 1000, newBurst: -1, } gc.run.requestUnitTokens.limiter.Reconfigure(time.Now(), args) + gc.run.now = time.Now() + gc.run.requestUnitTokens.avgLastTime = gc.run.now.Add(-time.Second) gc.updateAvgRequestResourcePerSec() re.True(gc.burstable.Load()) } @@ -122,7 +129,7 @@ func TestDirectReportConsumptionReportsOnlyConsumption(t *testing.T) { func TestRequestAndResponseConsumption(t *testing.T) { re := require.New(t) - gc := createTestGroupCostController(re) + gc := createTestGroupCostController(re, t.Name()) testCases := []struct { req *TestRequestInfo resp *TestResponseInfo @@ -229,7 +236,7 @@ func TestRequestAndResponseConsumption(t *testing.T) { func TestOnResponseWaitConsumption(t *testing.T) { re := require.New(t) - gc := createTestGroupCostController(re) + gc := createTestGroupCostController(re, t.Name()) req := &TestRequestInfo{ isWrite: false, @@ -911,7 +918,7 @@ func TestNoPreChargeWithoutPredictedReadBytes(t *testing.T) { func TestHandleTokenBucketUpdateEventCanceledByInitCounterNotify(t *testing.T) { re := require.New(t) - gc := createTestGroupCostController(re) + gc := createTestGroupCostController(re, t.Name()) counter := gc.run.requestUnitTokens counter.notify.mu.Lock() @@ -942,7 +949,7 @@ func TestHandleTokenBucketUpdateEventCanceledByInitCounterNotify(t *testing.T) { func TestHandleTokenBucketUpdateEventCleansNotifyOnTimer(t *testing.T) { re := require.New(t) - gc := createTestGroupCostController(re) + gc := createTestGroupCostController(re, t.Name()) counter := gc.run.requestUnitTokens threshold := 42.0 @@ -988,7 +995,7 @@ func isClosed(ch <-chan struct{}) bool { func TestResourceGroupThrottledError(t *testing.T) { re := require.New(t) - gc := createTestGroupCostController(re) + gc := createTestGroupCostController(re, t.Name()) req := &TestRequestInfo{ isWrite: true, writeBytes: 10000000, @@ -1042,7 +1049,7 @@ func TestAcquireTokensSignalAwareWait(t *testing.T) { cfg := DefaultRUConfig() cfg.WaitRetryInterval = 5 * time.Second cfg.WaitRetryTimes = 3 - gc, err := newGroupCostController(group, cfg, notifyCh, make(chan *groupCostController, 1)) + gc, err := newGroupCostController(group, 1001, cfg, notifyCh, make(chan *groupCostController, 1)) re.NoError(err) // Set fillRate=0 so reservation always fails with InfDuration, @@ -1093,7 +1100,7 @@ func TestAcquireTokensSignalAwareWait(t *testing.T) { func TestAcquireTokensFallbackToTimer(t *testing.T) { re := require.New(t) - gc := createTestGroupCostController(re) + gc := createTestGroupCostController(re, t.Name()) // Short retry interval so the test runs fast. gc.mainCfg.WaitRetryInterval = 50 * time.Millisecond gc.mainCfg.WaitRetryTimes = 3 @@ -1127,7 +1134,7 @@ func TestAcquireTokensFallbackToTimer(t *testing.T) { // now_B, not rewind to now_A. func TestAcquireTokensCancelKeepsLastMonotonic(t *testing.T) { re := require.New(t) - gc := createTestGroupCostController(re) + gc := createTestGroupCostController(re, t.Name()) gc.mainCfg.LTBMaxWaitDuration = 30 * time.Second gc.mainCfg.WaitRetryTimes = 1 counter := gc.run.requestUnitTokens @@ -1179,3 +1186,53 @@ func TestAcquireTokensCancelKeepsLastMonotonic(t *testing.T) { re.False(counter.limiter.last.Before(tMid), "stale CancelAt rewound lim.last") } + +func TestAcquireTokensUpdatesDemandMetricOnThrottle(t *testing.T) { + re := require.New(t) + gc := createTestGroupCostController(re, t.Name()) + gc.mainCfg.LTBMaxWaitDuration = 0 + gc.mainCfg.WaitRetryTimes = 1 + gc.mainCfg.WaitRetryInterval = 0 + + counter := gc.run.requestUnitTokens + counter.limiter.Reconfigure(time.Now(), tokenBucketReconfigureArgs{ + newTokens: 0, + newFillRate: 1, + newBurst: 1, + }) + counter.demandAvgLastTime = time.Now().Add(-time.Second) + + before := promtestutil.ToFloat64(metrics.DemandRUPerSecGauge.WithLabelValues(gc.name)) + waitDuration := time.Duration(0) + _, err := gc.acquireTokens(context.Background(), &rmpb.Consumption{RRU: 100}, &waitDuration, false) + re.Error(err) + re.True(errs.ErrClientResourceGroupThrottled.Equal(err)) + re.Greater(promtestutil.ToFloat64(metrics.DemandRUPerSecGauge.WithLabelValues(gc.name)), before) +} + +func TestObserveConsumptionAndComponentMetrics(t *testing.T) { + re := require.New(t) + gc := createTestGroupCostController(re, t.Name()) + + rruBefore := promtestutil.ToFloat64(metrics.TokenConsumedByTypeCounter.WithLabelValues(gc.name, "rru")) + wruBefore := promtestutil.ToFloat64(metrics.TokenConsumedByTypeCounter.WithLabelValues(gc.name, "wru")) + readBefore := promtestutil.ToFloat64(metrics.ReadByteCost.WithLabelValues(gc.name)) + writeBefore := promtestutil.ToFloat64(metrics.WriteByteCost.WithLabelValues(gc.name)) + kvBefore := promtestutil.ToFloat64(metrics.KVCPUCost.WithLabelValues(gc.name)) + sqlBefore := promtestutil.ToFloat64(metrics.SQLCPUCost.WithLabelValues(gc.name)) + + gc.observeConsumption(&rmpb.Consumption{RRU: 3, WRU: 5}) + gc.addRUConsumption(&rmpb.Consumption{ + ReadBytes: 13, + WriteBytes: 17, + TotalCpuTimeMs: 11, + SqlLayerCpuTimeMs: 7, + }) + + re.Equal(rruBefore+3, promtestutil.ToFloat64(metrics.TokenConsumedByTypeCounter.WithLabelValues(gc.name, "rru"))) + re.Equal(wruBefore+5, promtestutil.ToFloat64(metrics.TokenConsumedByTypeCounter.WithLabelValues(gc.name, "wru"))) + re.Equal(readBefore+13, promtestutil.ToFloat64(metrics.ReadByteCost.WithLabelValues(gc.name))) + re.Equal(writeBefore+17, promtestutil.ToFloat64(metrics.WriteByteCost.WithLabelValues(gc.name))) + re.Equal(kvBefore+4, promtestutil.ToFloat64(metrics.KVCPUCost.WithLabelValues(gc.name))) + re.Equal(sqlBefore+7, promtestutil.ToFloat64(metrics.SQLCPUCost.WithLabelValues(gc.name))) +} diff --git a/client/resource_group/controller/limiter.go b/client/resource_group/controller/limiter.go index 12fce0bbb3..1907cde7cf 100644 --- a/client/resource_group/controller/limiter.go +++ b/client/resource_group/controller/limiter.go @@ -72,7 +72,7 @@ func Every(interval time.Duration) fillRate { // can be seen as r == Inf (burst within an unlimited capacity). // - If b > 0, that means the limiter is limited capacity. type Limiter struct { - mu sync.Mutex + mu sync.RWMutex fillRate fillRate tokens float64 burst int64 @@ -301,18 +301,25 @@ func (lim *Limiter) isLowTokensLocked() bool { // IsLowTokens returns whether the limiter is in low tokens func (lim *Limiter) IsLowTokens() bool { - lim.mu.Lock() - defer lim.mu.Unlock() + lim.mu.RLock() + defer lim.mu.RUnlock() return lim.isLowTokensLocked() } // GetBurst returns the burst size of the limiter func (lim *Limiter) GetBurst() int64 { - lim.mu.Lock() - defer lim.mu.Unlock() + lim.mu.RLock() + defer lim.mu.RUnlock() return lim.burst } +// GetFillRate returns the current fill rate of the limiter. +func (lim *Limiter) GetFillRate() float64 { + lim.mu.RLock() + defer lim.mu.RUnlock() + return float64(lim.fillRate) +} + // RemoveTokens decreases the amount of tokens currently available. func (lim *Limiter) RemoveTokens(now time.Time, amount float64) { lim.mu.Lock() @@ -399,10 +406,10 @@ func (lim *Limiter) Reconfigure(now time.Time, zap.Int64("burst", lim.burst)) } -// AvailableTokens decreases the amount of tokens currently available. +// AvailableTokens returns the current number of available tokens. func (lim *Limiter) AvailableTokens(now time.Time) float64 { - lim.mu.Lock() - defer lim.mu.Unlock() + lim.mu.RLock() + defer lim.mu.RUnlock() _, tokens := lim.getTokens(now) return tokens } diff --git a/client/resource_group/controller/metrics/metrics.go b/client/resource_group/controller/metrics/metrics.go index 871bc14ad9..320656436d 100644 --- a/client/resource_group/controller/metrics/metrics.go +++ b/client/resource_group/controller/metrics/metrics.go @@ -17,15 +17,18 @@ package metrics import "github.com/prometheus/client_golang/prometheus" const ( - namespace = "resource_manager_client" - requestSubsystem = "request" - tokenRequestSubsystem = "token_request" + namespace = "resource_manager_client" + requestSubsystem = "request" + tokenRequestSubsystem = "token_request" + resourceGroupSubsystem = "resource_group" + resourceUnitSubsystem = "resource_unit" + resourceSubsystem = "resource" // TODO: remove old label in 8.x resourceGroupNameLabel = "name" newResourceGroupNameLabel = "resource_group" - errType = "type" + typeLabel = "type" ) var ( @@ -69,6 +72,28 @@ var ( // PagingPredictionResidualBytes records the distribution of (actual - predicted) read // bytes for pre-charged coprocessor RPCs. PagingPredictionResidualBytes *prometheus.HistogramVec + // TokenConsumedByTypeCounter tracks RU consumption broken down by RU type (rru/wru). + TokenConsumedByTypeCounter *prometheus.CounterVec + // TokenBalanceGauge exposes the current available token balance per resource group. + TokenBalanceGauge *prometheus.GaugeVec + // FillRateGauge exposes the current effective fill rate (RU/s) per resource group. + FillRateGauge *prometheus.GaugeVec + // BurstLimitGauge exposes the current burst limit per resource group. + BurstLimitGauge *prometheus.GaugeVec + // AvgRUPerSecGauge exposes the estimated average RU consumption rate per resource group. + AvgRUPerSecGauge *prometheus.GaugeVec + // DemandRUPerSecGauge exposes the estimated pre-throttling demand RU/s per resource group. + DemandRUPerSecGauge *prometheus.GaugeVec + // ThrottledGauge exposes whether each resource group is currently in throttled (trickle) mode. + ThrottledGauge *prometheus.GaugeVec + // ReadByteCost tracks client-side read bytes per resource group. + ReadByteCost *prometheus.CounterVec + // WriteByteCost tracks client-side write bytes per resource group. + WriteByteCost *prometheus.CounterVec + // KVCPUCost tracks client-side KV CPU time in milliseconds per resource group. + KVCPUCost *prometheus.CounterVec + // SQLCPUCost tracks client-side SQL CPU time in milliseconds per resource group. + SQLCPUCost *prometheus.CounterVec ) func init() { @@ -112,7 +137,7 @@ func initMetrics(constLabels prometheus.Labels) { Name: "fail", Help: "Counter of failed request.", ConstLabels: constLabels, - }, []string{resourceGroupNameLabel, newResourceGroupNameLabel, errType}) + }, []string{resourceGroupNameLabel, newResourceGroupNameLabel, typeLabel}) GroupRunningKVRequestCounter = prometheus.NewGaugeVec( prometheus.GaugeOpts{ @@ -222,6 +247,105 @@ func initMetrics(constLabels prometheus.Labels) { Help: "Histogram of signed (actual_read_bytes - predicted_read_bytes) for pre-charged coprocessor RPCs. Use bucket series for distribution/quantiles; _sum is non-monotonic because observations can be negative.", ConstLabels: constLabels, }, []string{newResourceGroupNameLabel}) + + TokenConsumedByTypeCounter = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: namespace, + Subsystem: resourceGroupSubsystem, + Name: "consume_by_type", + Help: "Counter of token consumption broken down by RU type.", + ConstLabels: constLabels, + }, []string{newResourceGroupNameLabel, typeLabel}) + + TokenBalanceGauge = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: namespace, + Subsystem: resourceGroupSubsystem, + Name: "token_balance", + Help: "Current available token balance in the limiter per resource group.", + ConstLabels: constLabels, + }, []string{newResourceGroupNameLabel}) + + FillRateGauge = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: namespace, + Subsystem: resourceGroupSubsystem, + Name: "fill_rate", + Help: "Current effective fill rate (RU/s) of the limiter per resource group.", + ConstLabels: constLabels, + }, []string{newResourceGroupNameLabel}) + + BurstLimitGauge = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: namespace, + Subsystem: resourceGroupSubsystem, + Name: "burst_limit", + Help: "Current burst limit of the limiter per resource group.", + ConstLabels: constLabels, + }, []string{newResourceGroupNameLabel}) + + AvgRUPerSecGauge = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: namespace, + Subsystem: resourceGroupSubsystem, + Name: "avg_ru_per_sec", + Help: "Estimated average RU consumption rate per second (EMA) per resource group.", + ConstLabels: constLabels, + }, []string{newResourceGroupNameLabel}) + + DemandRUPerSecGauge = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: namespace, + Subsystem: resourceGroupSubsystem, + Name: "demand_ru_per_sec", + Help: "Estimated pre-throttling demand RU/s per resource group.", + ConstLabels: constLabels, + }, []string{newResourceGroupNameLabel}) + + ThrottledGauge = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: namespace, + Subsystem: resourceGroupSubsystem, + Name: "throttled", + Help: "Whether the resource group is currently in throttled (trickle) mode. 1 = throttled, 0 = normal.", + ConstLabels: constLabels, + }, []string{newResourceGroupNameLabel}) + + ReadByteCost = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: namespace, + Subsystem: resourceSubsystem, + Name: "read_byte_sum", + Help: "Counter of the read byte cost for all resource groups on the client side.", + ConstLabels: constLabels, + }, []string{newResourceGroupNameLabel}) + + WriteByteCost = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: namespace, + Subsystem: resourceSubsystem, + Name: "write_byte_sum", + Help: "Counter of the write byte cost for all resource groups on the client side.", + ConstLabels: constLabels, + }, []string{newResourceGroupNameLabel}) + + KVCPUCost = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: namespace, + Subsystem: resourceSubsystem, + Name: "kv_cpu_time_ms_sum", + Help: "Counter of the KV CPU time cost in milliseconds for all resource groups on the client side.", + ConstLabels: constLabels, + }, []string{newResourceGroupNameLabel}) + + SQLCPUCost = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: namespace, + Subsystem: resourceSubsystem, + Name: "sql_cpu_time_ms_sum", + Help: "Counter of the SQL CPU time cost in milliseconds for all resource groups on the client side.", + ConstLabels: constLabels, + }, []string{newResourceGroupNameLabel}) } // InitAndRegisterMetrics initializes and register metrics. @@ -242,4 +366,15 @@ func InitAndRegisterMetrics(constLabels prometheus.Labels) { prometheus.MustRegister(PagingPrechargeBytesCounter) prometheus.MustRegister(PagingActualBytesCounter) prometheus.MustRegister(PagingPredictionResidualBytes) + prometheus.MustRegister(TokenConsumedByTypeCounter) + prometheus.MustRegister(TokenBalanceGauge) + prometheus.MustRegister(FillRateGauge) + prometheus.MustRegister(BurstLimitGauge) + prometheus.MustRegister(AvgRUPerSecGauge) + prometheus.MustRegister(DemandRUPerSecGauge) + prometheus.MustRegister(ThrottledGauge) + prometheus.MustRegister(ReadByteCost) + prometheus.MustRegister(WriteByteCost) + prometheus.MustRegister(KVCPUCost) + prometheus.MustRegister(SQLCPUCost) } From cede109259b089f04b08cb89ee669a559cf253a0 Mon Sep 17 00:00:00 2001 From: okjiang <819421878@qq.com> Date: Tue, 21 Apr 2026 12:07:01 +0800 Subject: [PATCH 2/6] client/resource_group: fix static lint Signed-off-by: okjiang <819421878@qq.com> --- client/resource_group/controller/group_controller.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/client/resource_group/controller/group_controller.go b/client/resource_group/controller/group_controller.go index 5192eaee2d..a7c0d21990 100644 --- a/client/resource_group/controller/group_controller.go +++ b/client/resource_group/controller/group_controller.go @@ -412,7 +412,7 @@ func (gc *groupCostController) handleTokenBucketUpdateEvent(ctx context.Context) } func (gc *groupCostController) calcAvg(counter *tokenCounter, new float64) bool { - return gc.calcMovingAvgAt(gc.run.now, &counter.avgRUPerSec, &counter.avgRUPerSecLastRU, &counter.avgLastTime, new) + return calcMovingAvgAt(gc.run.now, &counter.avgRUPerSec, &counter.avgRUPerSecLastRU, &counter.avgLastTime, new) } func (gc *groupCostController) calcDemandAvg(counter *tokenCounter, delta float64) bool { @@ -420,14 +420,14 @@ func (gc *groupCostController) calcDemandAvg(counter *tokenCounter, delta float6 return false } counter.demandTotalRU += delta - ok := gc.calcMovingAvgAt(time.Now(), &counter.demandRUPerSec, &counter.demandRUPerSecLastRU, &counter.demandAvgLastTime, counter.demandTotalRU) + ok := calcMovingAvgAt(time.Now(), &counter.demandRUPerSec, &counter.demandRUPerSecLastRU, &counter.demandAvgLastTime, counter.demandTotalRU) if ok { gc.metrics.demandRUPerSecGauge.Set(counter.demandRUPerSec) } return ok } -func (gc *groupCostController) calcMovingAvgAt(now time.Time, avg *float64, lastRU *float64, lastTime *time.Time, new float64) bool { +func calcMovingAvgAt(now time.Time, avg *float64, lastRU *float64, lastTime *time.Time, new float64) bool { deltaDuration := now.Sub(*lastTime) failpoint.Inject("acceleratedReportingPeriod", func() { deltaDuration = 100 * time.Millisecond From ff892b991bb6529288c029a28d4bcabb9b7b98d6 Mon Sep 17 00:00:00 2001 From: okjiang <819421878@qq.com> Date: Thu, 4 Jun 2026 12:27:22 +0800 Subject: [PATCH 3/6] client/resource_group: guard demand stats Signed-off-by: okjiang <819421878@qq.com> --- .../controller/group_controller.go | 46 ++++++++++++++----- .../controller/group_controller_test.go | 34 ++++++++++++++ 2 files changed, 68 insertions(+), 12 deletions(-) diff --git a/client/resource_group/controller/group_controller.go b/client/resource_group/controller/group_controller.go index a7c0d21990..1ac82a7180 100644 --- a/client/resource_group/controller/group_controller.go +++ b/client/resource_group/controller/group_controller.go @@ -203,6 +203,7 @@ func (gmc *groupMetricsCollection) observePagingResponse(bytesForEst, actual uin type tokenCounter struct { fillRate uint64 + avgMu sync.Mutex // avgRUPerSec is an exponentially-weighted moving average of the RU // consumption per second; used to estimate the RU requirements for the next // request. @@ -369,11 +370,12 @@ func (gc *groupCostController) updateAvgRequestResourcePerSec() { if counter.limiter.GetBurst() >= 0 { isBurstable = false } - if !gc.calcAvg(counter, getRUValueFromConsumption(gc.run.consumption)) { + avgRUPerSec, ok := gc.calcAvg(counter, getRUValueFromConsumption(gc.run.consumption)) + if !ok { return } - gc.metrics.avgRUPerSecGauge.Set(counter.avgRUPerSec) - logControllerTrace("[resource group controller] update avg ru per sec", zap.String("name", gc.name), zap.Float64("avg-ru-per-sec", counter.avgRUPerSec), zap.Bool("is-throttled", gc.isThrottled.Load())) + gc.metrics.avgRUPerSecGauge.Set(avgRUPerSec) + logControllerTrace("[resource group controller] update avg ru per sec", zap.String("name", gc.name), zap.Float64("avg-ru-per-sec", avgRUPerSec), zap.Bool("is-throttled", gc.isThrottled.Load())) gc.burstable.Store(isBurstable) } @@ -411,18 +413,24 @@ func (gc *groupCostController) handleTokenBucketUpdateEvent(ctx context.Context) } } -func (gc *groupCostController) calcAvg(counter *tokenCounter, new float64) bool { - return calcMovingAvgAt(gc.run.now, &counter.avgRUPerSec, &counter.avgRUPerSecLastRU, &counter.avgLastTime, new) +func (gc *groupCostController) calcAvg(counter *tokenCounter, new float64) (float64, bool) { + counter.avgMu.Lock() + defer counter.avgMu.Unlock() + ok := calcMovingAvgAt(gc.run.now, &counter.avgRUPerSec, &counter.avgRUPerSecLastRU, &counter.avgLastTime, new) + return counter.avgRUPerSec, ok } func (gc *groupCostController) calcDemandAvg(counter *tokenCounter, delta float64) bool { if delta <= 0 { return false } + counter.avgMu.Lock() counter.demandTotalRU += delta ok := calcMovingAvgAt(time.Now(), &counter.demandRUPerSec, &counter.demandRUPerSecLastRU, &counter.demandAvgLastTime, counter.demandTotalRU) + demandRUPerSec := counter.demandRUPerSec + counter.avgMu.Unlock() if ok { - gc.metrics.demandRUPerSecGauge.Set(counter.demandRUPerSec) + gc.metrics.demandRUPerSecGauge.Set(demandRUPerSec) } return ok } @@ -504,7 +512,7 @@ func (gc *groupCostController) modifyTokenCounter(counter *tokenCounter, bucket cfg.newTokens = granted cfg.newFillRate = float64(bucket.GetSettings().FillRate) counter.lastDeadline = time.Time{} - cfg.newNotifyThreshold = math.Min(granted+counter.limiter.AvailableTokens(gc.run.now), counter.avgRUPerSec*defaultTargetPeriod.Seconds()) * notifyFraction + cfg.newNotifyThreshold = math.Min(granted+counter.limiter.AvailableTokens(gc.run.now), counter.getAvgRUPerSec()*defaultTargetPeriod.Seconds()) * notifyFraction if cfg.newBurst < 0 { cfg.newTokens = float64(counter.fillRate) } @@ -631,7 +639,7 @@ func (gc *groupCostController) calcRequest(counter *tokenCounter) float64 { // `needTokensAmplification` is used to properly amplify a need. The reason is that in the current implementation, // the token returned from the server determines the average consumption speed. // Therefore, when the fillrate of resource group increases, `needTokensAmplification` can enable the client to obtain more tokens. - value := counter.avgRUPerSec * gc.run.targetPeriod.Seconds() * needTokensAmplification + value := counter.getAvgRUPerSec() * gc.run.targetPeriod.Seconds() * needTokensAmplification value -= counter.limiter.AvailableTokens(gc.run.now) if value < 0 { value = 0 @@ -670,13 +678,15 @@ func (gc *groupCostController) observeComponentConsumption(delta *rmpb.Consumpti } func (gc *groupCostController) logFields(waitDuration time.Duration, err error) []zap.Field { + counter := gc.run.requestUnitTokens + avgRUPerSec, demandRUPerSec := counter.getAvgStats() fields := []zap.Field{ zap.String("resource_group", gc.name), zap.Uint64("client_unique_id", gc.clientUniqueID), - zap.Float64("available_tokens", gc.run.requestUnitTokens.limiter.AvailableTokens(time.Now())), - zap.Float64("fill_rate", gc.run.requestUnitTokens.limiter.GetFillRate()), - zap.Float64("avg_ru_per_sec", gc.run.requestUnitTokens.avgRUPerSec), - zap.Float64("demand_ru_per_sec", gc.run.requestUnitTokens.demandRUPerSec), + zap.Float64("available_tokens", counter.limiter.AvailableTokens(time.Now())), + zap.Float64("fill_rate", counter.limiter.GetFillRate()), + zap.Float64("avg_ru_per_sec", avgRUPerSec), + zap.Float64("demand_ru_per_sec", demandRUPerSec), zap.Bool("throttled", gc.isThrottled.Load()), } if waitDuration > 0 { @@ -688,6 +698,18 @@ func (gc *groupCostController) logFields(waitDuration time.Duration, err error) return fields } +func (counter *tokenCounter) getAvgRUPerSec() float64 { + counter.avgMu.Lock() + defer counter.avgMu.Unlock() + return counter.avgRUPerSec +} + +func (counter *tokenCounter) getAvgStats() (avgRUPerSec, demandRUPerSec float64) { + counter.avgMu.Lock() + defer counter.avgMu.Unlock() + return counter.avgRUPerSec, counter.demandRUPerSec +} + func (gc *groupCostController) acquireTokens(ctx context.Context, delta *rmpb.Consumption, waitDuration *time.Duration, allowDebt bool) (time.Duration, error) { gc.metrics.runningKVRequestCounter.Inc() defer gc.metrics.runningKVRequestCounter.Dec() diff --git a/client/resource_group/controller/group_controller_test.go b/client/resource_group/controller/group_controller_test.go index 09f7143149..53677eb7e8 100644 --- a/client/resource_group/controller/group_controller_test.go +++ b/client/resource_group/controller/group_controller_test.go @@ -17,6 +17,7 @@ package controller import ( "context" "fmt" + "sync" "testing" "time" @@ -1210,6 +1211,39 @@ func TestAcquireTokensUpdatesDemandMetricOnThrottle(t *testing.T) { re.Greater(promtestutil.ToFloat64(metrics.DemandRUPerSecGauge.WithLabelValues(gc.name)), before) } +func TestAcquireTokensDemandStatsConcurrentLogFields(t *testing.T) { + re := require.New(t) + gc := createTestGroupCostController(re, t.Name()) + gc.mainCfg.LTBMaxWaitDuration = 0 + gc.mainCfg.WaitRetryTimes = 1 + gc.mainCfg.WaitRetryInterval = 0 + + counter := gc.run.requestUnitTokens + counter.limiter.Reconfigure(time.Now(), tokenBucketReconfigureArgs{ + newTokens: 0, + newFillRate: 1, + newBurst: 1, + }) + counter.demandAvgLastTime = time.Now().Add(-time.Second) + + start := make(chan struct{}) + var wg sync.WaitGroup + for range 8 { + wg.Add(1) + go func() { + defer wg.Done() + <-start + for range 100 { + waitDuration := time.Duration(0) + _, _ = gc.acquireTokens(context.Background(), &rmpb.Consumption{RRU: 1}, &waitDuration, false) + _ = gc.logFields(waitDuration, nil) + } + }() + } + close(start) + wg.Wait() +} + func TestObserveConsumptionAndComponentMetrics(t *testing.T) { re := require.New(t) gc := createTestGroupCostController(re, t.Name()) From 369e6aacfe4f862f2bdb55776d006a69db1bad8e Mon Sep 17 00:00:00 2001 From: okjiang <819421878@qq.com> Date: Fri, 10 Jul 2026 10:54:41 +0800 Subject: [PATCH 4/6] client/resource_group: fix rebased test call Signed-off-by: okjiang <819421878@qq.com> --- client/resource_group/controller/global_controller_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/client/resource_group/controller/global_controller_test.go b/client/resource_group/controller/global_controller_test.go index 029dfbf8d5..ed085ba1ed 100644 --- a/client/resource_group/controller/global_controller_test.go +++ b/client/resource_group/controller/global_controller_test.go @@ -495,7 +495,7 @@ func TestGetResourceGroupRuntimeState(t *testing.T) { }, }, } - gc, err := newGroupCostController(group, DefaultRUConfig(), make(chan notifyMsg), make(chan *groupCostController)) + gc, err := newGroupCostController(group, 1001, DefaultRUConfig(), make(chan notifyMsg), make(chan *groupCostController)) re.NoError(err) controller := &ResourceGroupsController{} From b8a835cd398b30eb7fc95a95babd0349522e44ae Mon Sep 17 00:00:00 2001 From: lhy1024 <19542290+lhy1024@users.noreply.github.com> Date: Mon, 13 Jul 2026 15:43:25 +0800 Subject: [PATCH 5/6] client/resource_group: remove successful token wait logs Successful token waits are expected while resource control throttles requests. Logging every wait over 10ms at WARN can flood TiDB logs under high QPS. Keep the existing histogram for observability and record the accumulated retry and reservation wait duration. Signed-off-by: lhy1024 <19542290+lhy1024@users.noreply.github.com> --- .../controller/group_controller.go | 20 ++++----- .../controller/group_controller_test.go | 41 +++++++++++++++++++ 2 files changed, 51 insertions(+), 10 deletions(-) diff --git a/client/resource_group/controller/group_controller.go b/client/resource_group/controller/group_controller.go index 1ac82a7180..36788bb1dc 100644 --- a/client/resource_group/controller/group_controller.go +++ b/client/resource_group/controller/group_controller.go @@ -200,6 +200,14 @@ func (gmc *groupMetricsCollection) observePagingResponse(bytesForEst, actual uin gmc.predictionResidualBytes.Observe(float64(actual) - float64(bytesForEst)) } +func (gmc *groupMetricsCollection) recordSuccessfulRequestWaitMetrics( + accumulatedWaitDuration, reservationWaitDuration time.Duration, +) time.Duration { + totalWaitDuration := accumulatedWaitDuration + reservationWaitDuration + gmc.successfulRequestDuration.Observe(totalWaitDuration.Seconds()) + return totalWaitDuration +} + type tokenCounter struct { fillRate uint64 @@ -798,11 +806,7 @@ func (gc *groupCostController) onRequestWaitImpl( } return nil, nil, waitDuration, 0, err } - gc.metrics.successfulRequestDuration.Observe(d.Seconds()) - waitDuration += d - if waitDuration > slowNotifyFilterDuration { - log.Warn("[resource group controller] waited for tokens", gc.logFields(waitDuration, nil)...) - } + waitDuration = gc.metrics.recordSuccessfulRequestWaitMetrics(waitDuration, d) } if bytesForEst, ok := pagingReadEstimate(info); ok { gc.metrics.observePagingRequest(bytesForEst) @@ -906,11 +910,7 @@ func (gc *groupCostController) onResponseWaitImpl( } return nil, waitDuration, err } - gc.metrics.successfulRequestDuration.Observe(d.Seconds()) - waitDuration += d - if waitDuration > slowNotifyFilterDuration { - log.Warn("[resource group controller] response waited for tokens", gc.logFields(waitDuration, nil)...) - } + waitDuration = gc.metrics.recordSuccessfulRequestWaitMetrics(waitDuration, d) } else if v < 0 && isPagingRead && bytesForEst > 0 { // Paging over-estimate: refund the excess pre-charge. gc.run.requestUnitTokens.limiter.RefundTokens(time.Now(), -v) diff --git a/client/resource_group/controller/group_controller_test.go b/client/resource_group/controller/group_controller_test.go index 53677eb7e8..f273c5a359 100644 --- a/client/resource_group/controller/group_controller_test.go +++ b/client/resource_group/controller/group_controller_test.go @@ -55,6 +55,47 @@ func histogramSampleCount(re *require.Assertions, h prometheus.Observer) uint64 return histogram.GetSampleCount() } +func TestRecordSuccessfulRequestWaitMetrics(t *testing.T) { + testCases := []struct { + name string + accumulatedWaitDuration time.Duration + reservationWaitDuration time.Duration + expectedWaitDuration time.Duration + }{ + { + name: "reservation wait only", + reservationWaitDuration: 25 * time.Millisecond, + expectedWaitDuration: 25 * time.Millisecond, + }, + { + name: "retry and reservation wait", + accumulatedWaitDuration: 20 * time.Millisecond, + reservationWaitDuration: 30 * time.Millisecond, + expectedWaitDuration: 50 * time.Millisecond, + }, + } + + for _, testCase := range testCases { + t.Run(testCase.name, func(t *testing.T) { + re := require.New(t) + var observedSeconds float64 + metrics := &groupMetricsCollection{ + successfulRequestDuration: prometheus.ObserverFunc(func(value float64) { + observedSeconds = value + }), + } + + waitDuration := metrics.recordSuccessfulRequestWaitMetrics( + testCase.accumulatedWaitDuration, + testCase.reservationWaitDuration, + ) + + re.Equal(testCase.expectedWaitDuration, waitDuration) + re.InDelta(testCase.expectedWaitDuration.Seconds(), observedSeconds, 1e-9) + }) + } +} + func createTestGroupCostController(re *require.Assertions, names ...string) *groupCostController { name := "test" if len(names) > 0 { From 916fcbad5f8b1e645bb01fdbae9f9fd8f2117c6c Mon Sep 17 00:00:00 2001 From: lhy1024 <19542290+lhy1024@users.noreply.github.com> Date: Wed, 15 Jul 2026 15:36:21 +0800 Subject: [PATCH 6/6] client/resource_group: preserve successful wait metric semantics Signed-off-by: lhy1024 <19542290+lhy1024@users.noreply.github.com> --- .../controller/group_controller.go | 14 ++----- .../controller/group_controller_test.go | 41 ------------------- 2 files changed, 4 insertions(+), 51 deletions(-) diff --git a/client/resource_group/controller/group_controller.go b/client/resource_group/controller/group_controller.go index 36788bb1dc..9a52502268 100644 --- a/client/resource_group/controller/group_controller.go +++ b/client/resource_group/controller/group_controller.go @@ -200,14 +200,6 @@ func (gmc *groupMetricsCollection) observePagingResponse(bytesForEst, actual uin gmc.predictionResidualBytes.Observe(float64(actual) - float64(bytesForEst)) } -func (gmc *groupMetricsCollection) recordSuccessfulRequestWaitMetrics( - accumulatedWaitDuration, reservationWaitDuration time.Duration, -) time.Duration { - totalWaitDuration := accumulatedWaitDuration + reservationWaitDuration - gmc.successfulRequestDuration.Observe(totalWaitDuration.Seconds()) - return totalWaitDuration -} - type tokenCounter struct { fillRate uint64 @@ -806,7 +798,8 @@ func (gc *groupCostController) onRequestWaitImpl( } return nil, nil, waitDuration, 0, err } - waitDuration = gc.metrics.recordSuccessfulRequestWaitMetrics(waitDuration, d) + gc.metrics.successfulRequestDuration.Observe(d.Seconds()) + waitDuration += d } if bytesForEst, ok := pagingReadEstimate(info); ok { gc.metrics.observePagingRequest(bytesForEst) @@ -910,7 +903,8 @@ func (gc *groupCostController) onResponseWaitImpl( } return nil, waitDuration, err } - waitDuration = gc.metrics.recordSuccessfulRequestWaitMetrics(waitDuration, d) + gc.metrics.successfulRequestDuration.Observe(d.Seconds()) + waitDuration += d } else if v < 0 && isPagingRead && bytesForEst > 0 { // Paging over-estimate: refund the excess pre-charge. gc.run.requestUnitTokens.limiter.RefundTokens(time.Now(), -v) diff --git a/client/resource_group/controller/group_controller_test.go b/client/resource_group/controller/group_controller_test.go index f273c5a359..53677eb7e8 100644 --- a/client/resource_group/controller/group_controller_test.go +++ b/client/resource_group/controller/group_controller_test.go @@ -55,47 +55,6 @@ func histogramSampleCount(re *require.Assertions, h prometheus.Observer) uint64 return histogram.GetSampleCount() } -func TestRecordSuccessfulRequestWaitMetrics(t *testing.T) { - testCases := []struct { - name string - accumulatedWaitDuration time.Duration - reservationWaitDuration time.Duration - expectedWaitDuration time.Duration - }{ - { - name: "reservation wait only", - reservationWaitDuration: 25 * time.Millisecond, - expectedWaitDuration: 25 * time.Millisecond, - }, - { - name: "retry and reservation wait", - accumulatedWaitDuration: 20 * time.Millisecond, - reservationWaitDuration: 30 * time.Millisecond, - expectedWaitDuration: 50 * time.Millisecond, - }, - } - - for _, testCase := range testCases { - t.Run(testCase.name, func(t *testing.T) { - re := require.New(t) - var observedSeconds float64 - metrics := &groupMetricsCollection{ - successfulRequestDuration: prometheus.ObserverFunc(func(value float64) { - observedSeconds = value - }), - } - - waitDuration := metrics.recordSuccessfulRequestWaitMetrics( - testCase.accumulatedWaitDuration, - testCase.reservationWaitDuration, - ) - - re.Equal(testCase.expectedWaitDuration, waitDuration) - re.InDelta(testCase.expectedWaitDuration.Seconds(), observedSeconds, 1e-9) - }) - } -} - func createTestGroupCostController(re *require.Assertions, names ...string) *groupCostController { name := "test" if len(names) > 0 {