diff --git a/client/go.mod b/client/go.mod index e99a4eca9a..14d38e300c 100644 --- a/client/go.mod +++ b/client/go.mod @@ -44,3 +44,5 @@ require ( gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) + +replace github.com/pingcap/kvproto v0.0.0-20260903054228-107095f1d250 => github.com/JmPotato/kvproto v0.0.0-20260920123146-2a45fb4cd2dc diff --git a/client/go.sum b/client/go.sum index b7e014b2e7..fe35a6ba36 100644 --- a/client/go.sum +++ b/client/go.sum @@ -1,5 +1,7 @@ github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= +github.com/JmPotato/kvproto v0.0.0-20260920123146-2a45fb4cd2dc h1:7QHq/GlSggwICab7ETDp9ike7ED5XKLkTrfuJMlZasU= +github.com/JmPotato/kvproto v0.0.0-20260920123146-2a45fb4cd2dc/go.mod h1:z6+aAHB7dBkA+LyinEX+48/ImRJ3jag0Hg0c7wkhEvE= github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA= github.com/benbjohnson/clock v1.3.0 h1:ip6w0uFQkncKQ979AypyG0ER7mqUSBdKLOgAle/AT8A= github.com/benbjohnson/clock v1.3.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA= @@ -53,8 +55,6 @@ github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c h1:xpW9bvK+HuuTm github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c/go.mod h1:X2r9ueLEUZgtx2cIogM0v4Zj5uvvzhuuiu7Pn8HzMPg= github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86 h1:tdMsjOqUR7YXHoBitzdebTvOjs/swniBTOLy5XiMtuE= github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86/go.mod h1:exzhVYca3WRtd6gclGNErRWb1qEgff3LYta0LvRmON4= -github.com/pingcap/kvproto v0.0.0-20260903054228-107095f1d250 h1:6yUryXKVbKpCNdZWL58/OcZj8NPLUA/xsJYXSbsD59w= -github.com/pingcap/kvproto v0.0.0-20260903054228-107095f1d250/go.mod h1:z6+aAHB7dBkA+LyinEX+48/ImRJ3jag0Hg0c7wkhEvE= github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 h1:HR/ylkkLmGdSSDaD8IDP+SZrdhV1Kibl9KrHxJ9eciw= github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3/go.mod h1:DWQW5jICDR7UJh4HtxXSM20Churx4CQL0fwL/SoOSA4= github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= diff --git a/go.mod b/go.mod index 827bda4a3c..1f72091a82 100644 --- a/go.mod +++ b/go.mod @@ -264,3 +264,5 @@ require ( // which will cause several different tests to fail. So this is a temporary workaround to use the old version of `testify`. // TODO: fix those flasky tests introduced by the behavior change of `Eventually` and `EventuallyWithT` assertions. replace github.com/stretchr/testify => github.com/stretchr/testify v1.10.0 + +replace github.com/pingcap/kvproto v0.0.0-20260903054228-107095f1d250 => github.com/JmPotato/kvproto v0.0.0-20260920123146-2a45fb4cd2dc diff --git a/go.sum b/go.sum index b74b2ade8c..3c4fbb4cc5 100644 --- a/go.sum +++ b/go.sum @@ -25,6 +25,8 @@ github.com/AzureAD/microsoft-authentication-library-for-go v1.6.0/go.mod h1:HKpQ github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= github.com/BurntSushi/toml v1.5.0 h1:W5quZX/G/csjUnuI8SUYlsHs9M38FC7znL0lIO+DvMg= github.com/BurntSushi/toml v1.5.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho= +github.com/JmPotato/kvproto v0.0.0-20260920123146-2a45fb4cd2dc h1:7QHq/GlSggwICab7ETDp9ike7ED5XKLkTrfuJMlZasU= +github.com/JmPotato/kvproto v0.0.0-20260920123146-2a45fb4cd2dc/go.mod h1:z6+aAHB7dBkA+LyinEX+48/ImRJ3jag0Hg0c7wkhEvE= github.com/KyleBanks/depth v1.2.1 h1:5h8fQADFrWtarTdtDudMmGsC7GPbOAu6RVB3ffsVFHc= github.com/KyleBanks/depth v1.2.1/go.mod h1:jzSb9d0L43HxTQfT+oSA1EEp2q+ne2uh6XgeJcm8brE= github.com/Masterminds/semver v1.5.0 h1:H65muMkzWKEuNDnfl9d70GUjFniHKHRbFPGBuZ3QEww= @@ -490,8 +492,6 @@ github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c/go.mod h1:X2r9ue github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86 h1:tdMsjOqUR7YXHoBitzdebTvOjs/swniBTOLy5XiMtuE= github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86/go.mod h1:exzhVYca3WRtd6gclGNErRWb1qEgff3LYta0LvRmON4= github.com/pingcap/kvproto v0.0.0-20191211054548-3c6b38ea5107/go.mod h1:WWLmULLO7l8IOcQG+t+ItJ3fEcrL5FxF0Wu+HrMy26w= -github.com/pingcap/kvproto v0.0.0-20260903054228-107095f1d250 h1:6yUryXKVbKpCNdZWL58/OcZj8NPLUA/xsJYXSbsD59w= -github.com/pingcap/kvproto v0.0.0-20260903054228-107095f1d250/go.mod h1:z6+aAHB7dBkA+LyinEX+48/ImRJ3jag0Hg0c7wkhEvE= github.com/pingcap/log v0.0.0-20210625125904-98ed8e2eb1c7/go.mod h1:8AanEdAHATuRurdGxZXBz0At+9avep+ub7U1AGYLIMM= github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 h1:HR/ylkkLmGdSSDaD8IDP+SZrdhV1Kibl9KrHxJ9eciw= github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3/go.mod h1:DWQW5jICDR7UJh4HtxXSM20Churx4CQL0fwL/SoOSA4= diff --git a/pkg/mcs/resourcemanager/server/grpc_service.go b/pkg/mcs/resourcemanager/server/grpc_service.go index e7eeab4feb..836d21adfa 100644 --- a/pkg/mcs/resourcemanager/server/grpc_service.go +++ b/pkg/mcs/resourcemanager/server/grpc_service.go @@ -244,7 +244,7 @@ func (s *Service) AcquireTokenBuckets(stream rmpb.ResourceManager_AcquireTokenBu continue } // Send the consumption to update the metrics. - err = s.manager.dispatchConsumption(req) + err = s.manager.dispatchConsumption(clientUniqueID, req) if err != nil { return err } diff --git a/pkg/mcs/resourcemanager/server/keyspace_manager.go b/pkg/mcs/resourcemanager/server/keyspace_manager.go index cdad2f2476..ce1dbc1bd8 100644 --- a/pkg/mcs/resourcemanager/server/keyspace_manager.go +++ b/pkg/mcs/resourcemanager/server/keyspace_manager.go @@ -58,6 +58,7 @@ const ( // consumptionItem is used to send the consumption info to the background metrics flusher. type consumptionItem struct { + clientUniqueID uint64 keyspaceID uint32 keyspaceName string resourceGroupName string diff --git a/pkg/mcs/resourcemanager/server/manager.go b/pkg/mcs/resourcemanager/server/manager.go index 6a2be0c2b4..1c422fc397 100644 --- a/pkg/mcs/resourcemanager/server/manager.go +++ b/pkg/mcs/resourcemanager/server/manager.go @@ -111,6 +111,9 @@ type Manager struct { keyspaceIDLookup map[string]uint32 // metrics is the collection of metrics. metrics *metrics + // metricsLoopMu serializes backgroundMetricsFlush, since a new leadership + // term can start before the previous flusher exits. + metricsLoopMu sync.Mutex // ruCollector is used to collect the RU metering data. ruCollector *ruCollector } @@ -659,13 +662,14 @@ func (m *Manager) getKeyspaceResourceGroupManagers() []*keyspaceResourceGroupMan return krgms } -func (m *Manager) dispatchConsumption(req *rmpb.TokenBucketRequest) error { +func (m *Manager) dispatchConsumption(clientUniqueID uint64, req *rmpb.TokenBucketRequest) error { isBackground := req.GetIsBackground() isTiFlash := req.GetIsTiflash() if isBackground && isTiFlash { return errors.New("background and tiflash cannot be true at the same time") } m.consumptionDispatcher <- &consumptionItem{ + clientUniqueID: clientUniqueID, keyspaceID: ExtractKeyspaceID(req.GetKeyspaceId()), resourceGroupName: req.GetResourceGroupName(), Consumption: req.GetConsumptionSinceLastRequest(), @@ -752,6 +756,10 @@ func (m *Manager) GetKeyspaceIDByName(ctx context.Context, name string) (*rmpb.K func (m *Manager) backgroundMetricsFlush(ctx context.Context) { defer logutil.LogPanic() defer m.wg.Done() + m.metricsLoopMu.Lock() + defer m.metricsLoopMu.Unlock() + // The RU timeline belongs to one leadership term. + defer m.metrics.ruTimeline.reset() cleanUpTicker := time.NewTicker(metricsCleanupInterval) defer cleanUpTicker.Stop() metricsTicker := time.NewTicker(tickPerSecond) @@ -832,6 +840,7 @@ func (m *Manager) backgroundMetricsFlush(ctx context.Context) { } } case <-metricsTicker.C: + m.metrics.ruTimeline.flush(time.Now()) // Prevent from holding the lock too long when there're many keyspaces and resource groups. for _, krgm := range m.getKeyspaceResourceGroupManagers() { // Conciliate the fill rates. diff --git a/pkg/mcs/resourcemanager/server/manager_test.go b/pkg/mcs/resourcemanager/server/manager_test.go index baa6fc6f0d..4a7f2229a1 100644 --- a/pkg/mcs/resourcemanager/server/manager_test.go +++ b/pkg/mcs/resourcemanager/server/manager_test.go @@ -460,7 +460,7 @@ func checkBackgroundMetricsFlush(ctx context.Context, re *require.Assertions, ma }, KeyspaceId: keyspaceIDValue, } - err = manager.dispatchConsumption(req) + err = manager.dispatchConsumption(1, req) re.NoError(err) keyspaceID := ExtractKeyspaceID(req.GetKeyspaceId()) @@ -487,10 +487,11 @@ func TestDispatchConsumptionIncludesOnlyConsumption(t *testing.T) { KeyspaceId: &rmpb.KeyspaceIDValue{Keyspace: &rmpb.KeyspaceIDValue_Value{Value: 42}}, } - err := m.dispatchConsumption(req) + err := m.dispatchConsumption(7, req) re.NoError(err) item := <-m.consumptionDispatcher + re.Equal(uint64(7), item.clientUniqueID) re.Equal(uint32(42), item.keyspaceID) re.Equal(req.GetResourceGroupName(), item.resourceGroupName) re.Equal(req.GetConsumptionSinceLastRequest(), item.Consumption) diff --git a/pkg/mcs/resourcemanager/server/metrics.go b/pkg/mcs/resourcemanager/server/metrics.go index b4891fa21f..2dbdc645e2 100644 --- a/pkg/mcs/resourcemanager/server/metrics.go +++ b/pkg/mcs/resourcemanager/server/metrics.go @@ -219,6 +219,7 @@ var ( ) type metrics struct { + ruTimeline *ruTimeline // record update time of each resource group consumptionRecordMap map[consumptionRecordKey]time.Time // max per sec trackers for each keyspace and resource group. @@ -273,6 +274,7 @@ func init() { func newMetrics() *metrics { return &metrics{ + ruTimeline: newRUTimeline(ruSummaryMetrics), consumptionRecordMap: make(map[consumptionRecordKey]time.Time), maxPerSecTrackerMap: make(map[trackerKey]*maxPerSecCostTracker), counterMetricsMap: make(map[metricsKey]*counterMetrics), @@ -348,6 +350,7 @@ func (m *metrics) recordConsumption( if consumption == nil { return } + m.ruTimeline.record(consumptionInfo, now) m.getMaxPerSecTracker(keyspaceID, keyspaceName, groupName).collect(consumption) m.getCounterMetrics(keyspaceID, keyspaceName, groupName, ruLabelType).add(consumption, controllerConfig, keyspaceID) m.insertConsumptionRecord(keyspaceID, groupName, ruLabelType, now) diff --git a/pkg/mcs/resourcemanager/server/metrics_test.go b/pkg/mcs/resourcemanager/server/metrics_test.go index 7efddacb1b..1204eab555 100644 --- a/pkg/mcs/resourcemanager/server/metrics_test.go +++ b/pkg/mcs/resourcemanager/server/metrics_test.go @@ -150,6 +150,19 @@ func TestRecordConsumptionUsesActualRequestUnitCounters(t *testing.T) { groupName: groupName, ruType: defaultTypeLabel, }) + + // Client RU timelines are merged without changing counter accounting. + now := time.Now() + report := timelineReport(1, now.Unix(), [][2]float64{{12, 8}}) + report.keyspaceID, report.keyspaceName = keyspaceID, keyspaceName + report.resourceGroupName = groupName + m.recordConsumption(report, &ControllerConfig{}, now.Add(time.Second)) + source := m.ruTimeline.groups[trackerKey{keyspaceID, groupName}].sources[ruSourceKey{client: 1}] + bucket := source.buckets[now.Unix()%int64(len(source.buckets))] + re.Equal(float64(12), bucket.rru) + re.Equal(float64(8), bucket.wru) + re.Equal(float64(12), testutil.ToFloat64(counter.RRUMetrics)) + re.Equal(float64(8), testutil.ToFloat64(counter.WRUMetrics)) } func TestRecordConsumptionKeepsRecordForEmptyConsumption(t *testing.T) { diff --git a/pkg/mcs/resourcemanager/server/ru_timeline.go b/pkg/mcs/resourcemanager/server/ru_timeline.go new file mode 100644 index 0000000000..903593314a --- /dev/null +++ b/pkg/mcs/resourcemanager/server/ru_timeline.go @@ -0,0 +1,341 @@ +// Copyright 2026 TiKV Project Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package server + +import ( + "maps" + "math" + "sync" + "time" + + "github.com/prometheus/client_golang/prometheus" +) + +const ( + // ruTimelineSeconds is the number of seconds each source can replay. + ruTimelineSeconds = 180 + // ruTimelineClockSkew is how far, in seconds, a client clock may run ahead + // of the resource manager before its reported seconds are rejected. + ruTimelineClockSkew = 5 + // ruWindowSeconds is the aligned window summarized from the timeline. + ruWindowSeconds = 60 + // ruWindowDelay is how long a closed window waits for regular reports. It + // covers both the report interval and any lag of the client clock. + ruWindowDelay = 30 + // ruTimelineMaxSources bounds the retained sources, which hold nearly all + // of the timeline's memory. + ruTimelineMaxSources = 10000 +) + +// ruWindowSummary describes the busiest second of a closed window. +type ruWindowSummary struct { + end int64 + peak, rru, wru float64 + peakAt int64 + available bool +} + +type publishedRUSummary struct { + keyspaceName string + summary ruWindowSummary +} + +// ruSummaryCollector exposes immutable, explicitly timestamped window summaries. +// Scrapes never advance windows, clear timelines, or perform aggregation. +type ruSummaryCollector struct { + mu sync.RWMutex + results map[trackerKey]publishedRUSummary + peak, available, peakAt, rru, wru *prometheus.Desc +} + +func newRUSummaryCollector() *ruSummaryCollector { + desc := func(name, help string) *prometheus.Desc { + return prometheus.NewDesc("resource_manager_resource_unit_"+name, help, []string{keyspaceNameLabel, newResourceGroupNameLabel}, nil) + } + return &ruSummaryCollector{ + results: make(map[trackerKey]publishedRUSummary), + peak: desc("peak_per_second", "Maximum cluster net RRU+WRU in a natural second of the completed minute, in RU/s."), + available: desc("peak_available", "Whether the completed minute passed observed source coverage and known data quality checks."), + peakAt: desc("peak_second_timestamp_seconds", "Unix timestamp of the earliest second attaining the minute peak."), + rru: desc("peak_rru_per_second", "RRU contribution in the second attaining the total minute peak."), + wru: desc("peak_wru_per_second", "WRU contribution in the second attaining the total minute peak."), + } +} + +// Describe implements prometheus.Collector. +func (c *ruSummaryCollector) Describe(ch chan<- *prometheus.Desc) { + for _, desc := range []*prometheus.Desc{c.peak, c.available, c.peakAt, c.rru, c.wru} { + ch <- desc + } +} + +// Collect implements prometheus.Collector. +func (c *ruSummaryCollector) Collect(ch chan<- prometheus.Metric) { + c.mu.RLock() + defer c.mu.RUnlock() + for key, p := range c.results { + emit := func(desc *prometheus.Desc, value float64) { + ch <- prometheus.NewMetricWithTimestamp(time.Unix(p.summary.end, 0), prometheus.MustNewConstMetric(desc, prometheus.GaugeValue, value, p.keyspaceName, key.groupName)) + } + if !p.summary.available { + emit(c.available, 0) + continue + } + emit(c.available, 1) + emit(c.peak, p.summary.peak) + emit(c.peakAt, float64(p.summary.peakAt)) + emit(c.rru, p.summary.rru) + emit(c.wru, p.summary.wru) + } +} + +var ( + ruSummaryMetrics = newRUSummaryCollector() + ruTimelineQualityEvents = prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: "resource_manager", Subsystem: "resource_unit", Name: "peak_quality_events_total", + Help: "RU timeline data quality events, including violations discovered after publication.", + }, []string{"reason"}) + ruTimelineMissing = ruTimelineQualityEvents.WithLabelValues("missing_payload") + ruTimelineInvalid = ruTimelineQualityEvents.WithLabelValues("invalid_payload") + ruTimelineConflict = ruTimelineQualityEvents.WithLabelValues("conflict") + ruTimelineLate = ruTimelineQualityEvents.WithLabelValues("late") + ruTimelineCapacity = ruTimelineQualityEvents.WithLabelValues("capacity") +) + +func init() { prometheus.MustRegister(ruSummaryMetrics, ruTimelineQualityEvents) } + +type ruSourceKey struct { + client uint64 + isBackground, isTiFlash bool +} + +type ruSecondBucket struct { + second int64 + rru, wru float64 +} + +type ruSource struct { + buckets [ruTimelineSeconds + 1]ruSecondBucket + first, lastSeen int64 +} + +type ruTimelineGroup struct { + keyspaceName string + sources map[ruSourceKey]*ruSource + nextWindow int64 + invalid map[int64]bool +} + +// ruTimeline merges the RU timelines replayed by clients into +// per-group timelines and publishes a summary for each closed window. +// All state belongs to backgroundMetricsFlush. Only immutable summaries cross +// into the scrape goroutines under the collector lock. +// The resource manager clock only schedules windows, while values come from +// client seconds. A server clock step therefore delays or withholds windows, +// and each group's windows still only move forward. +type ruTimeline struct { + collector *ruSummaryCollector + groups map[trackerKey]*ruTimelineGroup + sourceCount int + // since is the first second observed in this term, or an earlier one + // after the clock steps back. + since int64 +} + +func newRUTimeline(c *ruSummaryCollector) *ruTimeline { + t := &ruTimeline{collector: c} + t.reset() + return t +} + +func windowStart(second int64) int64 { + return second / ruWindowSeconds * ruWindowSeconds +} + +func (t *ruTimeline) reset() { + t.collector.mu.Lock() + clear(t.collector.results) + t.collector.mu.Unlock() + t.groups = make(map[trackerKey]*ruTimelineGroup) + t.sourceCount = 0 + t.since = 0 +} + +func (t *ruTimeline) remove(key trackerKey) { + if g := t.groups[key]; g != nil { + t.sourceCount -= len(g.sources) + delete(t.groups, key) + } + t.collector.mu.Lock() + delete(t.collector.results, key) + t.collector.mu.Unlock() +} + +func (g *ruTimelineGroup) invalidate(start, end int64) { + for w := max(windowStart(start), g.nextWindow); w <= windowStart(end); w += ruWindowSeconds { + g.invalid[w] = true + } +} + +func (t *ruTimeline) record(item *consumptionItem, now time.Time) { + sec := now.Unix() + if t.since == 0 || sec < t.since { + t.since = sec + } + key := trackerKey{item.keyspaceID, item.resourceGroupName} + g := t.groups[key] + if g == nil { + g = &ruTimelineGroup{keyspaceName: item.keyspaceName, sources: make(map[ruSourceKey]*ruSource), nextWindow: windowStart(sec), invalid: make(map[int64]bool)} + t.groups[key] = g + // A newly observed group cannot establish coverage before its first + // report. Early in a term, it also cannot know the sources that have + // not reported yet, so it withholds windows until any of them would + // have expired, as a source seen earlier would have. + g.invalidate(sec, max(sec, t.since+ruTimelineSeconds)) + } + sourceKey := ruSourceKey{item.clientUniqueID, item.isBackground, item.isTiFlash} + source := g.sources[sourceKey] + created := source == nil + if created { + if t.sourceCount >= ruTimelineMaxSources { + // A rejected source has no state to merge or expire, so withhold + // every window it may replay or leave unreported. + ruTimelineCapacity.Inc() + g.invalidate(sec-ruTimelineSeconds, sec+ruTimelineSeconds) + return + } + source = &ruSource{first: sec} + g.sources[sourceKey] = source + t.sourceCount++ + } + source.lastSeen = sec + payload := item.GetRuBySecond() + if payload == nil { + // A legacy or quarantined client cannot attribute its consumption. + ruTimelineMissing.Inc() + g.invalidate(sec-ruTimelineSeconds, sec) + return + } + // Validate atomically: a malformed suffix must not leave a usable prefix. + // Buckets must be closed seconds of the client, allowing for a small clock + // skew; those older than retention are skipped below. + valid := item.clientUniqueID != 0 && len(payload.Buckets) <= ruTimelineSeconds && + int64(len(payload.Buckets)) <= sec+ruTimelineClockSkew-payload.StartUnixSec + if valid { + for _, b := range payload.Buckets { + if b == nil || math.IsNaN(b.Rru) || math.IsInf(b.Rru, 0) || math.IsNaN(b.Wru) || math.IsInf(b.Wru, 0) { + valid = false + break + } + } + } + if !valid { + ruTimelineInvalid.Inc() + g.invalidate(sec-ruTimelineSeconds, sec) + return + } + for i, b := range payload.Buckets { + second := payload.StartUnixSec + int64(i) + if second < sec-ruTimelineSeconds { + ruTimelineLate.Inc() + continue + } + source.first = min(source.first, second) + old := &source.buckets[second%int64(len(source.buckets))] + if old.second == second { + if old.rru != b.Rru || old.wru != b.Wru { + ruTimelineConflict.Inc() + g.invalidate(second, second) + } + continue + } + if second < g.nextWindow { + ruTimelineLate.Inc() + continue + } + *old = ruSecondBucket{second: second, rru: b.Rru, wru: b.Wru} + } + if created { + // The first report may lack the source's earliest replayable seconds, + // for example after a client trims its backlog, so the minutes before + // its first second cannot be known to be complete. + g.invalidate(source.first-ruTimelineSeconds, source.first-1) + } +} + +// summarize sums every source by second and finds the busiest second of the +// window beginning at start. Every observed source must cover the entire +// window: new membership, legacy reports, crashes, and gaps all withhold it. +func (g *ruTimelineGroup) summarize(start int64) ruWindowSummary { + end := start + ruWindowSeconds + s := ruWindowSummary{end: end, peak: math.Inf(-1), peakAt: start, available: !g.invalid[start] && len(g.sources) > 0} + var totals [ruWindowSeconds]ruSecondBucket + for _, source := range g.sources { + if source.first >= end { + continue + } + for second := start; second < end; second++ { + b := source.buckets[second%int64(len(source.buckets))] + if b.second != second { + s.available = false + continue + } + totals[second-start].rru += b.rru + totals[second-start].wru += b.wru + } + } + for i, b := range totals { + total := b.rru + b.wru + if math.IsNaN(total) || math.IsInf(total, 0) { + s.available = false + } + if total > s.peak { + s.peak, s.peakAt, s.rru, s.wru = total, start+int64(i), b.rru, b.wru + } + } + return s +} + +func (t *ruTimeline) flush(now time.Time) { + sec := now.Unix() + for key, g := range t.groups { + // Windows beyond retention cannot be verified; skip them after a stall + // or a forward clock step. + if oldest := windowStart(sec - ruTimelineSeconds); g.nextWindow < oldest { + g.nextWindow = oldest + maps.DeleteFunc(g.invalid, func(w int64, _ bool) bool { return w < oldest }) + } + for ; g.nextWindow+ruWindowSeconds+ruWindowDelay <= sec; g.nextWindow += ruWindowSeconds { + s := g.summarize(g.nextWindow) + t.collector.mu.Lock() + t.collector.results[key] = publishedRUSummary{keyspaceName: g.keyspaceName, summary: s} + t.collector.mu.Unlock() + delete(g.invalid, g.nextWindow) + } + for k, source := range g.sources { + if sec-source.lastSeen > ruTimelineSeconds { + // The source can no longer replay its unreported tail. Later + // windows are judged by the remaining sources, as for a source + // that never reported. + g.invalidate(source.lastSeen, sec) + delete(g.sources, k) + t.sourceCount-- + } + } + if len(g.sources) == 0 { + t.remove(key) + } + } +} diff --git a/pkg/mcs/resourcemanager/server/ru_timeline_test.go b/pkg/mcs/resourcemanager/server/ru_timeline_test.go new file mode 100644 index 0000000000..e8a04fdf42 --- /dev/null +++ b/pkg/mcs/resourcemanager/server/ru_timeline_test.go @@ -0,0 +1,281 @@ +// Copyright 2026 TiKV Project Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package server + +import ( + "math" + "math/rand/v2" + "slices" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/stretchr/testify/require" + + rmpb "github.com/pingcap/kvproto/pkg/resource_manager" +) + +const testTimelineStart int64 = 1800000000 + +var testTimelineKey = trackerKey{0, "test"} + +func timelineReport(client uint64, start int64, values [][2]float64) *consumptionItem { + buckets := make([]*rmpb.RUConsumptionBucket, len(values)) + for i, v := range values { + buckets[i] = &rmpb.RUConsumptionBucket{Rru: v[0], Wru: v[1]} + } + return &consumptionItem{clientUniqueID: client, resourceGroupName: "test", Consumption: &rmpb.Consumption{RuBySecond: &rmpb.RUConsumptionBySecond{StartUnixSec: start, Buckets: buckets}}} +} + +// newWarmTimeline returns a timeline whose clients were already observed +// before the window starting at testTimelineStart. +func newWarmTimeline(clients ...uint64) (*ruSummaryCollector, *ruTimeline) { + const start = testTimelineStart + c := newRUSummaryCollector() + t := newRUTimeline(c) + // The term began early enough that no unseen source can still replay. + t.since = start - ruWindowSeconds - ruTimelineSeconds + for _, client := range clients { + t.record(timelineReport(client, start-60, make([][2]float64, 1)), time.Unix(start-59, 0)) + } + return c, t +} + +func TestRUTimelineMergesSources(t *testing.T) { + re := require.New(t) + const start = testTimelineStart + for _, shuffled := range []bool{false, true} { + c, timeline := newWarmTimeline(1, 2) + one, two := make([][2]float64, 60), make([][2]float64, 60) + one[10] = [2]float64{100, 0} + two[11] = [2]float64{0, 120} + one[20] = [2]float64{80, 0} + two[20] = [2]float64{0, 70} + one[21] = [2]float64{-500, 0} + two[21] = [2]float64{0, 200} + var reports []*consumptionItem + for i := 0; i < 60; i += 5 { + reports = append(reports, timelineReport(1, start+int64(i), one[i:i+5]), timelineReport(2, start+int64(i), two[i:i+5])) + } + if shuffled { + rand.Shuffle(len(reports), func(i, j int) { reports[i], reports[j] = reports[j], reports[i] }) + } + for _, r := range reports { + timeline.record(r, time.Unix(start+65, 0)) + timeline.record(r, time.Unix(start+65, 0)) + } + timeline.flush(time.Unix(start+89, 0)) + re.NotEqual(start+60, c.results[testTimelineKey].summary.end) + timeline.flush(time.Unix(start+90, 0)) + got := c.results[testTimelineKey].summary + re.Equal(ruWindowSummary{end: start + 60, peak: 150, peakAt: start + 20, rru: 80, wru: 70, available: true}, got) + + registry := prometheus.NewRegistry() + registry.MustRegister(c) + first, err := registry.Gather() + re.NoError(err) + second, err := registry.Gather() + re.NoError(err) + re.Equal(first, second) + for _, family := range first { + re.Equal((start+60)*1000, family.GetMetric()[0].GetTimestampMs()) + } + // A late conflict is observable but never rewrites the published summary. + one[20][0] = 999 + timeline.record(timelineReport(1, start, one), time.Unix(start+95, 0)) + re.Equal(got, c.results[testTimelineKey].summary) + timeline.remove(testTimelineKey) + empty, err := registry.Gather() + re.NoError(err) + re.Empty(empty) + } +} + +func TestRUTimelineUnavailable(t *testing.T) { + const start = testTimelineStart + for _, scenario := range []string{"gap", "overflow", "conflict", "legacy", "nan", "future-second", "oversized", "new-source", "late-joiner", "source-kind", "capacity", "capacity-tail"} { + t.Run(scenario, func(t *testing.T) { + re := require.New(t) + c, timeline := newWarmTimeline(1) + values := make([][2]float64, 60) + values[10] = [2]float64{10, 20} + report := timelineReport(1, start, values) + switch scenario { + case "gap": + report.RuBySecond.Buckets = report.RuBySecond.Buckets[:59] + case "overflow": + for _, b := range report.RuBySecond.Buckets { + b.Rru, b.Wru = math.MaxFloat64, math.MaxFloat64 + } + case "capacity-tail": + // A source rejected before the window may still be consuming in it. + timeline.sourceCount = ruTimelineMaxSources + timeline.record(timelineReport(2, start-1, nil), time.Unix(start-1, 0)) + } + timeline.record(report, time.Unix(start+65, 0)) + // Each follow-up would leave the window available if it were ignored. + var next *consumptionItem + switch scenario { + case "conflict": + next = timelineReport(1, start+10, [][2]float64{{11, 20}}) + case "legacy": + next = timelineReport(1, start, values) + next.RuBySecond = nil + case "nan": + next = timelineReport(1, start+60, [][2]float64{{math.NaN(), 0}}) + case "future-second": + // Beyond the tolerated clock skew of the client. + next = timelineReport(1, start, slices.Concat(values, make([][2]float64, 6+ruTimelineClockSkew+1))) + case "oversized": + next = timelineReport(1, start-121, slices.Concat(make([][2]float64, 121), values)) + case "new-source": + next = timelineReport(2, start+15, values[15:]) + case "late-joiner": + // A first report trimmed to the next minute hides this one. + next = timelineReport(2, start+60, make([][2]float64, 6)) + case "source-kind": + next = timelineReport(1, start+15, values[15:]) + next.isBackground = true + case "capacity": + timeline.sourceCount = ruTimelineMaxSources + next = timelineReport(2, start, values) + } + missing, invalid := testutil.ToFloat64(ruTimelineMissing), testutil.ToFloat64(ruTimelineInvalid) + if next != nil { + timeline.record(next, time.Unix(start+66, 0)) + } + if scenario == "legacy" { + re.Equal(missing+1, testutil.ToFloat64(ruTimelineMissing)) + re.Equal(invalid, testutil.ToFloat64(ruTimelineInvalid)) + } + timeline.flush(time.Unix(start+90, 0)) + + registry := prometheus.NewRegistry() + registry.MustRegister(c) + families, err := registry.Gather() + re.NoError(err) + re.Len(families, 1) + re.Equal("resource_manager_resource_unit_peak_available", families[0].GetName()) + re.Zero(families[0].GetMetric()[0].GetGauge().GetValue()) + re.Equal((start+60)*1000, families[0].GetMetric()[0].GetTimestampMs()) + }) + } +} + +func TestRUTimelineToleratesClockSkew(t *testing.T) { + re := require.New(t) + const start = testTimelineStart + c, timeline := newWarmTimeline(1) + // The client clock runs one second ahead: its last closed second is the + // resource manager's current second. + timeline.record(timelineReport(1, start, make([][2]float64, 66)), time.Unix(start+65, 0)) + timeline.flush(time.Unix(start+90, 0)) + re.True(c.results[testTimelineKey].summary.available) +} + +func TestRUTimelineServerClockStep(t *testing.T) { + re := require.New(t) + const start = testTimelineStart + c, timeline := newWarmTimeline(1) + values := make([][2]float64, 60) + values[10] = [2]float64{10, 20} + timeline.record(timelineReport(1, start, values), time.Unix(start+65, 0)) + timeline.flush(time.Unix(start+90, 0)) + published := c.results[testTimelineKey].summary + re.True(published.available) + + // After a backward step, the published window stays put, and seconds from + // the client's future are rejected rather than merged early. + timeline.record(timelineReport(1, start+60, make([][2]float64, 60)), time.Unix(start+30, 0)) + timeline.flush(time.Unix(start+30, 0)) + re.Equal(published, c.results[testTimelineKey].summary) + // Once the clock catches up, later windows are published without a reset. + timeline.record(timelineReport(1, start+60, make([][2]float64, 60)), time.Unix(start+125, 0)) + timeline.flush(time.Unix(start+150, 0)) + re.Equal(ruWindowSummary{end: start + 120, peakAt: start + 60, available: true}, c.results[testTimelineKey].summary) + + // After a forward step, a new source only withholds the windows it could + // replay, however far the clock jumped. + future := start + 365*86400 + timeline.record(timelineReport(2, future-5, make([][2]float64, 5)), time.Unix(future, 0)) + re.LessOrEqual(len(timeline.groups[testTimelineKey].invalid), ruTimelineSeconds/ruWindowSeconds+1) + // After a backward step, a new group withholds no more windows than early + // in a term. + past := start - 365*86400 + report := timelineReport(3, past-5, make([][2]float64, 5)) + report.resourceGroupName = "other" + timeline.record(report, time.Unix(past, 0)) + re.LessOrEqual(len(timeline.groups[trackerKey{0, "other"}].invalid), ruTimelineSeconds/ruWindowSeconds+1) +} + +func TestRUTimelineResetWarmup(t *testing.T) { + re := require.New(t) + const start = testTimelineStart + c, timeline := newWarmTimeline(1) + timeline.record(timelineReport(1, start, make([][2]float64, 60)), time.Unix(start+65, 0)) + timeline.flush(time.Unix(start+90, 0)) + re.NotEmpty(c.results) + + timeline.reset() + re.Empty(c.results) + // A new term does not know the sources that stayed silent across the + // reset, so even complete replay withholds windows until any such source + // would have expired. + timeline.record(timelineReport(1, start, make([][2]float64, 120)), time.Unix(start+120, 0)) + for w := start + 120; w <= start+360; w += ruWindowSeconds { + timeline.record(timelineReport(1, w, make([][2]float64, 60)), time.Unix(w+65, 0)) + timeline.flush(time.Unix(w+90, 0)) + re.Equal(ruWindowSummary{end: w + 60, peakAt: w, available: w > start+ruTimelineSeconds+120}, c.results[testTimelineKey].summary) + } +} + +func TestRUTimelineIdleAndRetention(t *testing.T) { + re := require.New(t) + const start = testTimelineStart + c, timeline := newWarmTimeline(1) + timeline.record(timelineReport(1, start, make([][2]float64, 60)), time.Unix(start+65, 0)) + timeline.flush(time.Unix(start+90, 0)) + re.True(c.results[testTimelineKey].summary.available) + re.Zero(c.results[testTimelineKey].summary.peak) + timeline.flush(time.Unix(start+600, 0)) + re.Empty(timeline.groups) + re.Zero(timeline.sourceCount) + re.Empty(c.results) +} + +func TestRUTimelineSourceExpiry(t *testing.T) { + re := require.New(t) + const start = testTimelineStart + c, timeline := newWarmTimeline(1, 2) + published := make(map[int64]bool) + for offset := int64(5); offset <= 405; offset += 20 { + now := start + offset + replay := max(start-60, now-ruTimelineSeconds) + timeline.record(timelineReport(1, replay, make([][2]float64, now-replay)), time.Unix(now, 0)) + // Client 2 stops reporting; its seconds from start+65 are never known. + if offset <= 65 { + timeline.record(timelineReport(2, replay, make([][2]float64, now-replay)), time.Unix(now, 0)) + } + timeline.flush(time.Unix(now, 0)) + if s, ok := c.results[testTimelineKey]; ok { + published[s.summary.end-start] = s.summary.available + } + } + re.Len(timeline.groups[testTimelineKey].sources, 1) + // Windows overlapping the unknown tail are withheld until expiry; later + // windows are judged by the remaining source alone. + re.Equal(map[int64]bool{0: false, 60: true, 120: false, 180: false, 240: false, 300: false, 360: true}, published) +} diff --git a/server/gc_service.go b/server/gc_service.go index 122ddc9d5d..4fc0b82306 100644 --- a/server/gc_service.go +++ b/server/gc_service.go @@ -405,6 +405,11 @@ func (*GrpcServer) WatchGCSafePointV2(_ *pdpb.WatchGCSafePointV2Request, _ pdpb. return status.Errorf(codes.Unimplemented, "WatchGCSafePointV2 is obsolete. Poll GetAllKeyspacesGCStates instead if necessary") } +// WatchGCStates is not supported yet. +func (*GrpcServer) WatchGCStates(_ *pdpb.WatchGCStatesRequest, _ pdpb.PD_WatchGCStatesServer) error { + return status.Errorf(codes.Unimplemented, "WatchGCStates is not supported yet") +} + // GetAllGCSafePointV2 return all gc safe point v2. // // Deprecated: Use GetAllKeyspacesGCStates instead. diff --git a/tests/integrations/go.mod b/tests/integrations/go.mod index 00b9e075b4..26c3c91b1e 100644 --- a/tests/integrations/go.mod +++ b/tests/integrations/go.mod @@ -253,3 +253,5 @@ require ( // which will cause several different tests to fail. So this is a temporary workaround to use the old version of `testify`. // TODO: fix those flasky tests introduced by the behavior change of `Eventually` and `EventuallyWithT` assertions. replace github.com/stretchr/testify => github.com/stretchr/testify v1.10.0 + +replace github.com/pingcap/kvproto v0.0.0-20260903054228-107095f1d250 => github.com/JmPotato/kvproto v0.0.0-20260920123146-2a45fb4cd2dc diff --git a/tests/integrations/go.sum b/tests/integrations/go.sum index 5db28b5723..1d773261b6 100644 --- a/tests/integrations/go.sum +++ b/tests/integrations/go.sum @@ -25,6 +25,8 @@ github.com/BurntSushi/toml v1.5.0 h1:W5quZX/G/csjUnuI8SUYlsHs9M38FC7znL0lIO+DvMg github.com/BurntSushi/toml v1.5.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho= github.com/DATA-DOG/go-sqlmock v1.5.0 h1:Shsta01QNfFxHCfpW6YH2STWB0MudeXXEWMr20OEh60= github.com/DATA-DOG/go-sqlmock v1.5.0/go.mod h1:f/Ixk793poVmq4qj/V1dPUg2JEAKC73Q5eFN3EC/SaM= +github.com/JmPotato/kvproto v0.0.0-20260920123146-2a45fb4cd2dc h1:7QHq/GlSggwICab7ETDp9ike7ED5XKLkTrfuJMlZasU= +github.com/JmPotato/kvproto v0.0.0-20260920123146-2a45fb4cd2dc/go.mod h1:z6+aAHB7dBkA+LyinEX+48/ImRJ3jag0Hg0c7wkhEvE= github.com/KyleBanks/depth v1.2.1 h1:5h8fQADFrWtarTdtDudMmGsC7GPbOAu6RVB3ffsVFHc= github.com/KyleBanks/depth v1.2.1/go.mod h1:jzSb9d0L43HxTQfT+oSA1EEp2q+ne2uh6XgeJcm8brE= github.com/Masterminds/semver v1.5.0 h1:H65muMkzWKEuNDnfl9d70GUjFniHKHRbFPGBuZ3QEww= @@ -483,8 +485,6 @@ github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c/go.mod h1:X2r9ue github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86 h1:tdMsjOqUR7YXHoBitzdebTvOjs/swniBTOLy5XiMtuE= github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86/go.mod h1:exzhVYca3WRtd6gclGNErRWb1qEgff3LYta0LvRmON4= github.com/pingcap/kvproto v0.0.0-20191211054548-3c6b38ea5107/go.mod h1:WWLmULLO7l8IOcQG+t+ItJ3fEcrL5FxF0Wu+HrMy26w= -github.com/pingcap/kvproto v0.0.0-20260903054228-107095f1d250 h1:6yUryXKVbKpCNdZWL58/OcZj8NPLUA/xsJYXSbsD59w= -github.com/pingcap/kvproto v0.0.0-20260903054228-107095f1d250/go.mod h1:z6+aAHB7dBkA+LyinEX+48/ImRJ3jag0Hg0c7wkhEvE= github.com/pingcap/log v0.0.0-20210625125904-98ed8e2eb1c7/go.mod h1:8AanEdAHATuRurdGxZXBz0At+9avep+ub7U1AGYLIMM= github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 h1:HR/ylkkLmGdSSDaD8IDP+SZrdhV1Kibl9KrHxJ9eciw= github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3/go.mod h1:DWQW5jICDR7UJh4HtxXSM20Churx4CQL0fwL/SoOSA4= diff --git a/tools/go.mod b/tools/go.mod index 747f919856..46f0f54bf9 100644 --- a/tools/go.mod +++ b/tools/go.mod @@ -255,3 +255,5 @@ require ( // which will cause several different tests to fail. So this is a temporary workaround to use the old version of `testify`. // TODO: fix those flasky tests introduced by the behavior change of `Eventually` and `EventuallyWithT` assertions. replace github.com/stretchr/testify => github.com/stretchr/testify v1.10.0 + +replace github.com/pingcap/kvproto v0.0.0-20260903054228-107095f1d250 => github.com/JmPotato/kvproto v0.0.0-20260920123146-2a45fb4cd2dc diff --git a/tools/go.sum b/tools/go.sum index 769f9c9e60..6545d12ed2 100644 --- a/tools/go.sum +++ b/tools/go.sum @@ -23,6 +23,8 @@ github.com/AzureAD/microsoft-authentication-library-for-go v1.6.0/go.mod h1:HKpQ github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= github.com/BurntSushi/toml v1.5.0 h1:W5quZX/G/csjUnuI8SUYlsHs9M38FC7znL0lIO+DvMg= github.com/BurntSushi/toml v1.5.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho= +github.com/JmPotato/kvproto v0.0.0-20260920123146-2a45fb4cd2dc h1:7QHq/GlSggwICab7ETDp9ike7ED5XKLkTrfuJMlZasU= +github.com/JmPotato/kvproto v0.0.0-20260920123146-2a45fb4cd2dc/go.mod h1:z6+aAHB7dBkA+LyinEX+48/ImRJ3jag0Hg0c7wkhEvE= github.com/KyleBanks/depth v1.2.1 h1:5h8fQADFrWtarTdtDudMmGsC7GPbOAu6RVB3ffsVFHc= github.com/KyleBanks/depth v1.2.1/go.mod h1:jzSb9d0L43HxTQfT+oSA1EEp2q+ne2uh6XgeJcm8brE= github.com/Masterminds/semver v1.5.0 h1:H65muMkzWKEuNDnfl9d70GUjFniHKHRbFPGBuZ3QEww= @@ -488,8 +490,6 @@ github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c/go.mod h1:X2r9ue github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86 h1:tdMsjOqUR7YXHoBitzdebTvOjs/swniBTOLy5XiMtuE= github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86/go.mod h1:exzhVYca3WRtd6gclGNErRWb1qEgff3LYta0LvRmON4= github.com/pingcap/kvproto v0.0.0-20191211054548-3c6b38ea5107/go.mod h1:WWLmULLO7l8IOcQG+t+ItJ3fEcrL5FxF0Wu+HrMy26w= -github.com/pingcap/kvproto v0.0.0-20260903054228-107095f1d250 h1:6yUryXKVbKpCNdZWL58/OcZj8NPLUA/xsJYXSbsD59w= -github.com/pingcap/kvproto v0.0.0-20260903054228-107095f1d250/go.mod h1:z6+aAHB7dBkA+LyinEX+48/ImRJ3jag0Hg0c7wkhEvE= github.com/pingcap/log v0.0.0-20210625125904-98ed8e2eb1c7/go.mod h1:8AanEdAHATuRurdGxZXBz0At+9avep+ub7U1AGYLIMM= github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 h1:HR/ylkkLmGdSSDaD8IDP+SZrdhV1Kibl9KrHxJ9eciw= github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3/go.mod h1:DWQW5jICDR7UJh4HtxXSM20Churx4CQL0fwL/SoOSA4=