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..ed085ba1ed 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()) @@ -436,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{} diff --git a/client/resource_group/controller/group_controller.go b/client/resource_group/controller/group_controller.go index 49fc4fd55e..9a52502268 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), } } @@ -178,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. @@ -186,6 +212,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 +238,7 @@ type tokenCounter struct { func newGroupCostController( group *rmpb.ResourceGroup, + clientUniqueID uint64, mainCfg *RUConfig, lowRUNotifyChan chan notifyMsg, tokenBucketUpdateChan chan *groupCostController, @@ -219,11 +253,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 +307,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 +335,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() { @@ -320,10 +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 } - 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) } @@ -361,22 +413,47 @@ func (gc *groupCostController) handleTokenBucketUpdateEvent(ctx context.Context) } } -func (gc *groupCostController) calcAvg(counter *tokenCounter, new float64) bool { - deltaDuration := gc.run.now.Sub(counter.avgLastTime) +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(demandRUPerSec) + } + return ok +} + +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 }) - 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 +493,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 @@ -434,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) } @@ -469,6 +547,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) { @@ -553,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 @@ -561,6 +647,69 @@ 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 { + 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", 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 { + fields = append(fields, zap.Duration("wait_duration", waitDuration)) + } + if err != nil { + fields = append(fields, zap.Error(err)) + } + 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() @@ -568,6 +717,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 +728,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,6 +793,9 @@ 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()) @@ -652,6 +804,8 @@ func (gc *groupCostController) onRequestWaitImpl( 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 +851,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,6 +898,9 @@ 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()) @@ -757,6 +916,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 +932,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..53677eb7e8 100644 --- a/client/resource_group/controller/group_controller_test.go +++ b/client/resource_group/controller/group_controller_test.go @@ -17,10 +17,12 @@ package controller import ( "context" "fmt" + "sync" "testing" "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 +55,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 +77,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 +130,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 +237,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 +919,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 +950,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 +996,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 +1050,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 +1101,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 +1135,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 +1187,86 @@ 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 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()) + + 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) }