From 22cce1e34d8cbdf598e5c4195312c2438e7451c4 Mon Sep 17 00:00:00 2001 From: JmPotato Date: Wed, 23 Sep 2026 16:07:44 +0800 Subject: [PATCH 1/6] resource_manager: merge client RU timelines into minute peaks Clients can report RRU and WRU per UTC second in Consumption.ru_by_second. Merge the seconds reported by every source of a resource group and publish the busiest second of each completed minute as minute-end-timestamped gauges, with its RRU/WRU components and a coverage signal. Minutes with missing, conflicting or invalid seconds are withheld rather than published as lower values. A source that stops reporting expires after 180 seconds and withholds only the minutes overlapping its unreported tail. Source and group counts are capped, and an overflow withholds only the affected group. Metrics flushers are serialized across leadership terms, and the timeline is cleared when a term ends. The kvproto replacement is temporary until pingcap/kvproto#1539 is merged; that revision also adds WatchGCStates, answered as unimplemented. Signed-off-by: JmPotato --- client/go.mod | 2 + client/go.sum | 4 +- go.mod | 2 + go.sum | 4 +- .../resourcemanager/server/grpc_service.go | 2 +- .../server/keyspace_manager.go | 1 + pkg/mcs/resourcemanager/server/manager.go | 11 +- .../resourcemanager/server/manager_test.go | 5 +- pkg/mcs/resourcemanager/server/metrics.go | 3 + .../resourcemanager/server/metrics_test.go | 13 + pkg/mcs/resourcemanager/server/ru_timeline.go | 327 ++++++++++++++++++ .../server/ru_timeline_test.go | 222 ++++++++++++ server/gc_service.go | 5 + tests/integrations/go.mod | 2 + tests/integrations/go.sum | 4 +- tools/go.mod | 2 + tools/go.sum | 4 +- 17 files changed, 601 insertions(+), 12 deletions(-) create mode 100644 pkg/mcs/resourcemanager/server/ru_timeline.go create mode 100644 pkg/mcs/resourcemanager/server/ru_timeline_test.go diff --git a/client/go.mod b/client/go.mod index e99a4eca9af..14d38e300cf 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 b7e014b2e72..fe35a6ba362 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 827bda4a3c0..1f72091a82f 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 b74b2ade8cc..3c4fbb4cc50 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 e7eeab4feb5..836d21adfac 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 cdad2f24763..ce1dbc1bd80 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 6a2be0c2b4e..53b1d59b33a 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 func() { m.metrics.ruTimeline.reset(time.Now()) }() 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 baa6fc6f0d5..4a7f2229a10 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 b4891fa21f2..2a699a6403a 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, time.Now()), 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 7efddacb1bf..1204eab5555 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 00000000000..f883beaf91e --- /dev/null +++ b/pkg/mcs/resourcemanager/server/ru_timeline.go @@ -0,0 +1,327 @@ +// 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 + // ruWindowSeconds is the aligned window summarized from the timeline. + ruWindowSeconds = 60 + // ruWindowDelay is how long a closed window waits for regular reports. + ruWindowDelay = 30 + ruTimelineMaxSources = 10000 + ruTimelineMaxGroups = 4096 +) + +// 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"}) + ruTimelineInvalid = ruTimelineQualityEvents.WithLabelValues("invalid_payload") + ruTimelineConflict = ruTimelineQualityEvents.WithLabelValues("conflict") + ruTimelineLate = ruTimelineQualityEvents.WithLabelValues("late") + ruTimelineCapacity = ruTimelineQualityEvents.WithLabelValues("capacity") + ruTimelineClock = ruTimelineQualityEvents.WithLabelValues("clock") +) + +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. +type ruTimeline struct { + collector *ruSummaryCollector + groups map[trackerKey]*ruTimelineGroup + sourceCount int + last time.Time +} + +func newRUTimeline(c *ruSummaryCollector, now time.Time) *ruTimeline { + t := &ruTimeline{collector: c} + t.reset(now) + return t +} + +func windowStart(second int64) int64 { + return second / ruWindowSeconds * ruWindowSeconds +} + +func (t *ruTimeline) reset(now time.Time) { + t.collector.mu.Lock() + clear(t.collector.results) + t.collector.mu.Unlock() + t.groups = make(map[trackerKey]*ruTimelineGroup) + t.sourceCount = 0 + t.last = now +} + +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 + } +} + +// advance resets the timeline after a server clock discontinuity. +func (t *ruTimeline) advance(now time.Time) { + if !t.last.IsZero() { + wall := float64(now.UnixNano()-t.last.UnixNano()) / 1e9 + if now.Unix() < t.last.Unix() || math.Abs(wall-now.Sub(t.last).Seconds()) >= 1 { + ruTimelineClock.Inc() + t.reset(now) + } + } + t.last = now +} + +func (t *ruTimeline) record(item *consumptionItem, now time.Time) { + t.advance(now) + sec := now.Unix() + key := trackerKey{item.keyspaceID, item.resourceGroupName} + g := t.groups[key] + if g == nil { + if len(t.groups) >= ruTimelineMaxGroups { + ruTimelineCapacity.Inc() + return + } + 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. + g.invalid[g.nextWindow] = true + } + sourceKey := ruSourceKey{item.clientUniqueID, item.isBackground, item.isTiFlash} + source := g.sources[sourceKey] + if source == nil { + if t.sourceCount >= ruTimelineMaxSources { + // The rejected source replays up to a full retained window. + ruTimelineCapacity.Inc() + g.invalidate(sec-ruTimelineSeconds, sec) + return + } + source = &ruSource{first: sec} + g.sources[sourceKey] = source + t.sourceCount++ + } + source.lastSeen = sec + payload := item.GetRuBySecond() + // Validate atomically: a malformed suffix must not leave a usable prefix. + // Buckets must be closed seconds; those older than retention are skipped below. + valid := payload != nil && item.clientUniqueID != 0 && len(payload.Buckets) <= ruTimelineSeconds && + int64(len(payload.Buckets)) <= sec-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} + } +} + +// 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) { + t.advance(now) + sec := now.Unix() + for key, g := range t.groups { + // Windows beyond retention cannot be verified; skip them after a stall. + 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 00000000000..b99cd544fc0 --- /dev/null +++ b/pkg/mcs/resourcemanager/server/ru_timeline_test.go @@ -0,0 +1,222 @@ +// 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/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, time.Unix(start-60, 0)) + 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", "open-second", "oversized", "new-source", "source-kind", "capacity"} { + 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 + } + } + 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 "open-second": + next = timelineReport(1, start, slices.Concat(values, make([][2]float64, 7))) + 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 "source-kind": + next = timelineReport(1, start+15, values[15:]) + next.isBackground = true + case "capacity": + timeline.sourceCount = ruTimelineMaxSources + next = timelineReport(2, start, values) + } + if next != nil { + timeline.record(next, time.Unix(start+66, 0)) + } + 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 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(time.Unix(start+120, 0)) + re.Empty(c.results) + // Even complete replay cannot establish coverage before the first report + // after a reset, because other sources may not have reported yet. + timeline.record(timelineReport(1, start, make([][2]float64, 120)), time.Unix(start+120, 0)) + timeline.record(timelineReport(1, start+120, make([][2]float64, 60)), time.Unix(start+185, 0)) + timeline.flush(time.Unix(start+210, 0)) + got := c.results[testTimelineKey].summary + re.Equal(start+180, got.end) + re.False(got.available) + + timeline.record(timelineReport(1, start+180, make([][2]float64, 60)), time.Unix(start+245, 0)) + timeline.flush(time.Unix(start+270, 0)) + re.Equal(ruWindowSummary{end: start + 240, peakAt: start + 180, available: true}, 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 122ddc9d5d2..4fc0b82306f 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 00b9e075b4f..26c3c91b1e7 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 5db28b57234..1d773261b6c 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 747f9198561..46f0f54bf9d 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 769f9c9e60d..6545d12ed2f 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= From 1e8d556f2a62e9f2838923c94e68a5a1592e30c7 Mon Sep 17 00:00:00 2001 From: JmPotato Date: Wed, 23 Sep 2026 17:35:20 +0800 Subject: [PATCH 2/6] resource_manager: tighten RU timeline coverage and validation A new source's first report may lack its earliest seconds, for example after a client trims its backlog, so withhold the unpublished minutes before its first second instead of summarizing them without it. Accept seconds from a client clock that runs up to five seconds ahead of the resource manager, so that small cross-host skew no longer rejects every report, and count reports without per-second data as missing_payload rather than invalid_payload. Signed-off-by: JmPotato --- pkg/mcs/resourcemanager/server/ru_timeline.go | 26 +++++++++++++++--- .../server/ru_timeline_test.go | 27 ++++++++++++++++--- 2 files changed, 46 insertions(+), 7 deletions(-) diff --git a/pkg/mcs/resourcemanager/server/ru_timeline.go b/pkg/mcs/resourcemanager/server/ru_timeline.go index f883beaf91e..ae62a52a0f2 100644 --- a/pkg/mcs/resourcemanager/server/ru_timeline.go +++ b/pkg/mcs/resourcemanager/server/ru_timeline.go @@ -26,6 +26,9 @@ import ( 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. @@ -102,6 +105,7 @@ var ( 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") @@ -208,7 +212,8 @@ func (t *ruTimeline) record(item *consumptionItem, now time.Time) { } sourceKey := ruSourceKey{item.clientUniqueID, item.isBackground, item.isTiFlash} source := g.sources[sourceKey] - if source == nil { + created := source == nil + if created { if t.sourceCount >= ruTimelineMaxSources { // The rejected source replays up to a full retained window. ruTimelineCapacity.Inc() @@ -221,10 +226,17 @@ func (t *ruTimeline) record(item *consumptionItem, now time.Time) { } 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; those older than retention are skipped below. - valid := payload != nil && item.clientUniqueID != 0 && len(payload.Buckets) <= ruTimelineSeconds && - int64(len(payload.Buckets)) <= sec-payload.StartUnixSec + // 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) { @@ -259,6 +271,12 @@ func (t *ruTimeline) record(item *consumptionItem, now time.Time) { } *old = ruSecondBucket{second: second, rru: b.Rru, wru: b.Wru} } + if created { + // The first report may lack the source's earliest seconds, for example + // after a client trims its backlog, so the minutes before its first + // second cannot be known to be complete. + g.invalidate(g.nextWindow, source.first-1) + } } // summarize sums every source by second and finds the busiest second of the diff --git a/pkg/mcs/resourcemanager/server/ru_timeline_test.go b/pkg/mcs/resourcemanager/server/ru_timeline_test.go index b99cd544fc0..95e2b78a251 100644 --- a/pkg/mcs/resourcemanager/server/ru_timeline_test.go +++ b/pkg/mcs/resourcemanager/server/ru_timeline_test.go @@ -22,6 +22,7 @@ import ( "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" @@ -103,7 +104,7 @@ func TestRUTimelineMergesSources(t *testing.T) { func TestRUTimelineUnavailable(t *testing.T) { const start = testTimelineStart - for _, scenario := range []string{"gap", "overflow", "conflict", "legacy", "nan", "open-second", "oversized", "new-source", "source-kind", "capacity"} { + for _, scenario := range []string{"gap", "overflow", "conflict", "legacy", "nan", "future-second", "oversized", "new-source", "late-joiner", "source-kind", "capacity"} { t.Run(scenario, func(t *testing.T) { re := require.New(t) c, timeline := newWarmTimeline(1) @@ -129,12 +130,16 @@ func TestRUTimelineUnavailable(t *testing.T) { next.RuBySecond = nil case "nan": next = timelineReport(1, start+60, [][2]float64{{math.NaN(), 0}}) - case "open-second": - next = timelineReport(1, start, slices.Concat(values, make([][2]float64, 7))) + 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 @@ -142,9 +147,14 @@ func TestRUTimelineUnavailable(t *testing.T) { 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() @@ -159,6 +169,17 @@ func TestRUTimelineUnavailable(t *testing.T) { } } +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 TestRUTimelineResetWarmup(t *testing.T) { re := require.New(t) const start = testTimelineStart From 9c9d476791ba5cb0b42a157010033c280834ec96 Mon Sep 17 00:00:00 2001 From: JmPotato Date: Wed, 23 Sep 2026 23:48:54 +0800 Subject: [PATCH 3/6] resource_manager: withhold the tail of capacity-rejected sources A source rejected by the source limit keeps no state, so it neither withholds its unreported tail nor expires like an admitted source. Withhold every window it may still be consuming in until an admitted source would have expired, instead of only the windows it may replay. Signed-off-by: JmPotato --- pkg/mcs/resourcemanager/server/ru_timeline.go | 5 +++-- pkg/mcs/resourcemanager/server/ru_timeline_test.go | 6 +++++- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/pkg/mcs/resourcemanager/server/ru_timeline.go b/pkg/mcs/resourcemanager/server/ru_timeline.go index ae62a52a0f2..aa9fd554f19 100644 --- a/pkg/mcs/resourcemanager/server/ru_timeline.go +++ b/pkg/mcs/resourcemanager/server/ru_timeline.go @@ -215,9 +215,10 @@ func (t *ruTimeline) record(item *consumptionItem, now time.Time) { created := source == nil if created { if t.sourceCount >= ruTimelineMaxSources { - // The rejected source replays up to a full retained window. + // 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) + g.invalidate(sec-ruTimelineSeconds, sec+ruTimelineSeconds) return } source = &ruSource{first: sec} diff --git a/pkg/mcs/resourcemanager/server/ru_timeline_test.go b/pkg/mcs/resourcemanager/server/ru_timeline_test.go index 95e2b78a251..569087fa047 100644 --- a/pkg/mcs/resourcemanager/server/ru_timeline_test.go +++ b/pkg/mcs/resourcemanager/server/ru_timeline_test.go @@ -104,7 +104,7 @@ func TestRUTimelineMergesSources(t *testing.T) { 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"} { + 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) @@ -118,6 +118,10 @@ func TestRUTimelineUnavailable(t *testing.T) { 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. From b5cb8631bbc1938174f20289691c13db8e0755f7 Mon Sep 17 00:00:00 2001 From: JmPotato Date: Thu, 24 Sep 2026 00:03:18 +0800 Subject: [PATCH 4/6] resource_manager: drop the RU timeline group limit Only existing resource groups reach the RU timeline, and every retained group holds at least one source, so the source limit already bounds its memory. The separate group limit only duplicated that bound. Signed-off-by: JmPotato --- pkg/mcs/resourcemanager/server/ru_timeline.go | 5 ----- 1 file changed, 5 deletions(-) diff --git a/pkg/mcs/resourcemanager/server/ru_timeline.go b/pkg/mcs/resourcemanager/server/ru_timeline.go index aa9fd554f19..b7f5dcdc2c6 100644 --- a/pkg/mcs/resourcemanager/server/ru_timeline.go +++ b/pkg/mcs/resourcemanager/server/ru_timeline.go @@ -34,7 +34,6 @@ const ( // ruWindowDelay is how long a closed window waits for regular reports. ruWindowDelay = 30 ruTimelineMaxSources = 10000 - ruTimelineMaxGroups = 4096 ) // ruWindowSummary describes the busiest second of a closed window. @@ -201,10 +200,6 @@ func (t *ruTimeline) record(item *consumptionItem, now time.Time) { key := trackerKey{item.keyspaceID, item.resourceGroupName} g := t.groups[key] if g == nil { - if len(t.groups) >= ruTimelineMaxGroups { - ruTimelineCapacity.Inc() - return - } 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. From 610e2e06b892693d7c1c8556641ca9e92c6e8e22 Mon Sep 17 00:00:00 2001 From: JmPotato Date: Thu, 24 Sep 2026 00:03:18 +0800 Subject: [PATCH 5/6] resource_manager: let server clock steps only reschedule RU windows Published values come from client seconds; the resource manager clock only decides when a window is summarized. A backward step pauses publication and rejects the seconds that now look early, and a forward step withholds the windows whose seconds have not arrived yet, while each group's windows keep moving forward. Resetting the timeline on a discontinuity protected nothing, and after a backward step it republished minutes that had already been published. Drop it, together with the clock quality event, and start the invalidation of a new source's earlier windows at the earliest second it could replay, so a large forward step cannot make it walk every window since the last flush. Signed-off-by: JmPotato --- pkg/mcs/resourcemanager/server/manager.go | 2 +- pkg/mcs/resourcemanager/server/metrics.go | 2 +- pkg/mcs/resourcemanager/server/ru_timeline.go | 44 +++++++------------ .../server/ru_timeline_test.go | 32 +++++++++++++- 4 files changed, 49 insertions(+), 31 deletions(-) diff --git a/pkg/mcs/resourcemanager/server/manager.go b/pkg/mcs/resourcemanager/server/manager.go index 53b1d59b33a..1c422fc3974 100644 --- a/pkg/mcs/resourcemanager/server/manager.go +++ b/pkg/mcs/resourcemanager/server/manager.go @@ -759,7 +759,7 @@ func (m *Manager) backgroundMetricsFlush(ctx context.Context) { m.metricsLoopMu.Lock() defer m.metricsLoopMu.Unlock() // The RU timeline belongs to one leadership term. - defer func() { m.metrics.ruTimeline.reset(time.Now()) }() + defer m.metrics.ruTimeline.reset() cleanUpTicker := time.NewTicker(metricsCleanupInterval) defer cleanUpTicker.Stop() metricsTicker := time.NewTicker(tickPerSecond) diff --git a/pkg/mcs/resourcemanager/server/metrics.go b/pkg/mcs/resourcemanager/server/metrics.go index 2a699a6403a..2dbdc645e2d 100644 --- a/pkg/mcs/resourcemanager/server/metrics.go +++ b/pkg/mcs/resourcemanager/server/metrics.go @@ -274,7 +274,7 @@ func init() { func newMetrics() *metrics { return &metrics{ - ruTimeline: newRUTimeline(ruSummaryMetrics, time.Now()), + ruTimeline: newRUTimeline(ruSummaryMetrics), consumptionRecordMap: make(map[consumptionRecordKey]time.Time), maxPerSecTrackerMap: make(map[trackerKey]*maxPerSecCostTracker), counterMetricsMap: make(map[metricsKey]*counterMetrics), diff --git a/pkg/mcs/resourcemanager/server/ru_timeline.go b/pkg/mcs/resourcemanager/server/ru_timeline.go index b7f5dcdc2c6..4fc6f17af00 100644 --- a/pkg/mcs/resourcemanager/server/ru_timeline.go +++ b/pkg/mcs/resourcemanager/server/ru_timeline.go @@ -31,8 +31,11 @@ const ( ruTimelineClockSkew = 5 // ruWindowSeconds is the aligned window summarized from the timeline. ruWindowSeconds = 60 - // ruWindowDelay is how long a closed window waits for regular reports. - ruWindowDelay = 30 + // 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 ) @@ -109,7 +112,6 @@ var ( ruTimelineConflict = ruTimelineQualityEvents.WithLabelValues("conflict") ruTimelineLate = ruTimelineQualityEvents.WithLabelValues("late") ruTimelineCapacity = ruTimelineQualityEvents.WithLabelValues("capacity") - ruTimelineClock = ruTimelineQualityEvents.WithLabelValues("clock") ) func init() { prometheus.MustRegister(ruSummaryMetrics, ruTimelineQualityEvents) } @@ -140,16 +142,18 @@ type ruTimelineGroup struct { // 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 - last time.Time } -func newRUTimeline(c *ruSummaryCollector, now time.Time) *ruTimeline { +func newRUTimeline(c *ruSummaryCollector) *ruTimeline { t := &ruTimeline{collector: c} - t.reset(now) + t.reset() return t } @@ -157,13 +161,12 @@ func windowStart(second int64) int64 { return second / ruWindowSeconds * ruWindowSeconds } -func (t *ruTimeline) reset(now time.Time) { +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.last = now } func (t *ruTimeline) remove(key trackerKey) { @@ -182,20 +185,7 @@ func (g *ruTimelineGroup) invalidate(start, end int64) { } } -// advance resets the timeline after a server clock discontinuity. -func (t *ruTimeline) advance(now time.Time) { - if !t.last.IsZero() { - wall := float64(now.UnixNano()-t.last.UnixNano()) / 1e9 - if now.Unix() < t.last.Unix() || math.Abs(wall-now.Sub(t.last).Seconds()) >= 1 { - ruTimelineClock.Inc() - t.reset(now) - } - } - t.last = now -} - func (t *ruTimeline) record(item *consumptionItem, now time.Time) { - t.advance(now) sec := now.Unix() key := trackerKey{item.keyspaceID, item.resourceGroupName} g := t.groups[key] @@ -268,10 +258,10 @@ func (t *ruTimeline) record(item *consumptionItem, now time.Time) { *old = ruSecondBucket{second: second, rru: b.Rru, wru: b.Wru} } if created { - // The first report may lack the source's earliest seconds, for example - // after a client trims its backlog, so the minutes before its first - // second cannot be known to be complete. - g.invalidate(g.nextWindow, source.first-1) + // 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) } } @@ -309,10 +299,10 @@ func (g *ruTimelineGroup) summarize(start int64) ruWindowSummary { } func (t *ruTimeline) flush(now time.Time) { - t.advance(now) sec := now.Unix() for key, g := range t.groups { - // Windows beyond retention cannot be verified; skip them after a stall. + // 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 }) diff --git a/pkg/mcs/resourcemanager/server/ru_timeline_test.go b/pkg/mcs/resourcemanager/server/ru_timeline_test.go index 569087fa047..b253c998970 100644 --- a/pkg/mcs/resourcemanager/server/ru_timeline_test.go +++ b/pkg/mcs/resourcemanager/server/ru_timeline_test.go @@ -45,7 +45,7 @@ func timelineReport(client uint64, start int64, values [][2]float64) *consumptio func newWarmTimeline(clients ...uint64) (*ruSummaryCollector, *ruTimeline) { const start = testTimelineStart c := newRUSummaryCollector() - t := newRUTimeline(c, time.Unix(start-60, 0)) + t := newRUTimeline(c) for _, client := range clients { t.record(timelineReport(client, start-60, make([][2]float64, 1)), time.Unix(start-59, 0)) } @@ -184,6 +184,34 @@ func TestRUTimelineToleratesClockSkew(t *testing.T) { 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) +} + func TestRUTimelineResetWarmup(t *testing.T) { re := require.New(t) const start = testTimelineStart @@ -192,7 +220,7 @@ func TestRUTimelineResetWarmup(t *testing.T) { timeline.flush(time.Unix(start+90, 0)) re.NotEmpty(c.results) - timeline.reset(time.Unix(start+120, 0)) + timeline.reset() re.Empty(c.results) // Even complete replay cannot establish coverage before the first report // after a reset, because other sources may not have reported yet. From f9b9c4c0cbc2662cedc71979e4c56d531edf1373 Mon Sep 17 00:00:00 2001 From: JmPotato Date: Thu, 24 Sep 2026 00:30:08 +0800 Subject: [PATCH 6/6] resource_manager: withhold RU windows until a new term knows its sources A new leader or a restarted resource manager only knows the sources that have reported to it. A client that is slow to reconnect is unknown, so the first windows could be published as available without its seconds, while the previous term would have withheld them until that source expired. Withhold windows until any source silent since the term began would have expired, applying the same 180-second rule across terms. Signed-off-by: JmPotato --- pkg/mcs/resourcemanager/server/ru_timeline.go | 14 ++++++++-- .../server/ru_timeline_test.go | 28 +++++++++++-------- 2 files changed, 29 insertions(+), 13 deletions(-) diff --git a/pkg/mcs/resourcemanager/server/ru_timeline.go b/pkg/mcs/resourcemanager/server/ru_timeline.go index 4fc6f17af00..903593314a0 100644 --- a/pkg/mcs/resourcemanager/server/ru_timeline.go +++ b/pkg/mcs/resourcemanager/server/ru_timeline.go @@ -149,6 +149,9 @@ 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 { @@ -167,6 +170,7 @@ func (t *ruTimeline) reset() { t.collector.mu.Unlock() t.groups = make(map[trackerKey]*ruTimelineGroup) t.sourceCount = 0 + t.since = 0 } func (t *ruTimeline) remove(key trackerKey) { @@ -187,13 +191,19 @@ func (g *ruTimelineGroup) invalidate(start, end int64) { 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. - g.invalid[g.nextWindow] = true + // 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] diff --git a/pkg/mcs/resourcemanager/server/ru_timeline_test.go b/pkg/mcs/resourcemanager/server/ru_timeline_test.go index b253c998970..e8a04fdf42d 100644 --- a/pkg/mcs/resourcemanager/server/ru_timeline_test.go +++ b/pkg/mcs/resourcemanager/server/ru_timeline_test.go @@ -46,6 +46,8 @@ 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)) } @@ -210,6 +212,13 @@ func TestRUTimelineServerClockStep(t *testing.T) { 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) { @@ -222,18 +231,15 @@ func TestRUTimelineResetWarmup(t *testing.T) { timeline.reset() re.Empty(c.results) - // Even complete replay cannot establish coverage before the first report - // after a reset, because other sources may not have reported yet. + // 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)) - timeline.record(timelineReport(1, start+120, make([][2]float64, 60)), time.Unix(start+185, 0)) - timeline.flush(time.Unix(start+210, 0)) - got := c.results[testTimelineKey].summary - re.Equal(start+180, got.end) - re.False(got.available) - - timeline.record(timelineReport(1, start+180, make([][2]float64, 60)), time.Unix(start+245, 0)) - timeline.flush(time.Unix(start+270, 0)) - re.Equal(ruWindowSummary{end: start + 240, peakAt: start + 180, available: true}, c.results[testTimelineKey].summary) + 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) {