From 5052fa30b9c1e9af8d9f1c76d66c19a0c715de19 Mon Sep 17 00:00:00 2001 From: bufferflies <1045931706@qq.com> Date: Mon, 10 Aug 2026 14:23:50 +0200 Subject: [PATCH 01/16] *: delete per-store metrics when a store is tombstoned Several per-store Prometheus metrics were never cleaned up on store removal, leaking stale time series until the next PD leader election forced a full Reset(). Most visibly, pd_hotcache_status entries in HotPeerCache.gc() were only removed from the in-memory map, never from the GaugeVec itself. Add DeleteStoreMetrics helpers to pkg/schedule/filter, pkg/schedule/hbstream, and pkg/schedule/schedulers, and wire them into both the classic bury path (server/cluster.BuryStoreLocked / removeStoreStatistics) and the standalone scheduling microservice's store watcher (pkg/mcs/scheduling/server/meta), matching the existing pattern used by statistics.ResetStoreStatistics. Also switch storeStatusGauge cleanup in ResetStoreStatistics from a hand-maintained type-string list (which had already drifted out of sync with what observe()/ObserveHotStat() actually set) to DeletePartialMatch, so it cannot silently drift again. The per-store heartbeat/bucket-report histograms and counters defined directly in package server and pkg/mcs/scheduling/server cannot be reached from the bury path without a cross-package cycle, so they are swept periodically instead (serverMetricsLoop and Cluster.collectMetrics respectively). Close tikv/pd#11126. Signed-off-by: bufferflies Signed-off-by: bufferflies <1045931706@qq.com> --- pkg/mcs/scheduling/server/cluster.go | 4 +++ pkg/mcs/scheduling/server/meta/watcher.go | 9 ++++++- pkg/mcs/scheduling/server/metrics.go | 12 +++++++++ pkg/schedule/filter/metrics.go | 6 +++++ pkg/schedule/hbstream/metric.go | 5 ++++ pkg/schedule/schedulers/metrics.go | 15 +++++++++++ pkg/statistics/hot_peer_cache.go | 1 + pkg/statistics/hot_peer_cache_test.go | 17 +++++++++--- pkg/statistics/store_collection.go | 32 ++--------------------- server/cluster/cluster.go | 2 ++ server/cluster/scheduling_controller.go | 2 ++ server/metrics.go | 12 +++++++++ server/server.go | 18 +++++++++++++ 13 files changed, 100 insertions(+), 35 deletions(-) diff --git a/pkg/mcs/scheduling/server/cluster.go b/pkg/mcs/scheduling/server/cluster.go index 7b846a241a..8d298a20e0 100644 --- a/pkg/mcs/scheduling/server/cluster.go +++ b/pkg/mcs/scheduling/server/cluster.go @@ -21,6 +21,7 @@ import ( "io" "net/http" "runtime" + "strconv" "strings" "sync" "sync/atomic" @@ -725,6 +726,9 @@ func (c *Cluster) collectMetrics() { for _, s := range stores { statsMap.Observe(s) statistics.ObserveHotStat(s, c.hotStat.StoresStats) + if s.IsRemoved() { + DeleteStoreMetrics(s.GetAddress(), strconv.FormatUint(s.GetID(), 10)) + } } statsMap.Collect() diff --git a/pkg/mcs/scheduling/server/meta/watcher.go b/pkg/mcs/scheduling/server/meta/watcher.go index a1fd6defc7..82b11fbab5 100644 --- a/pkg/mcs/scheduling/server/meta/watcher.go +++ b/pkg/mcs/scheduling/server/meta/watcher.go @@ -28,6 +28,9 @@ import ( "github.com/pingcap/log" "github.com/tikv/pd/pkg/core" + "github.com/tikv/pd/pkg/schedule/filter" + "github.com/tikv/pd/pkg/schedule/hbstream" + "github.com/tikv/pd/pkg/schedule/schedulers" "github.com/tikv/pd/pkg/statistics" "github.com/tikv/pd/pkg/utils/etcdutil" "github.com/tikv/pd/pkg/utils/keypath" @@ -82,7 +85,11 @@ func (w *Watcher) initializeStoreWatcher() error { } if store.GetNodeState() == metapb.NodeState_Removed { - statistics.ResetStoreStatistics(store.GetAddress(), strconv.FormatUint(store.GetId(), 10)) + storeIDStr := strconv.FormatUint(store.GetId(), 10) + statistics.ResetStoreStatistics(store.GetAddress(), storeIDStr) + filter.DeleteStoreMetrics(storeIDStr) + hbstream.DeleteStoreMetrics(storeIDStr) + schedulers.DeleteStoreMetrics(storeIDStr) // TODO: remove hot stats } diff --git a/pkg/mcs/scheduling/server/metrics.go b/pkg/mcs/scheduling/server/metrics.go index 04f94cb7c1..d5d1cb0d74 100644 --- a/pkg/mcs/scheduling/server/metrics.go +++ b/pkg/mcs/scheduling/server/metrics.go @@ -94,3 +94,15 @@ func init() { prometheus.MustRegister(regionBucketsCounter) prometheus.MustRegister(regionBucketsReportInterval) } + +// DeleteStoreMetrics deletes the per-store heartbeat/bucket metrics of a store. +func DeleteStoreMetrics(storeAddress, id string) { + labels := prometheus.Labels{"address": storeAddress, "store": id} + storeHeartbeatHandleDuration.DeletePartialMatch(labels) + storeHeartbeatCounter.DeletePartialMatch(labels) + regionHeartbeatHandleDuration.DeletePartialMatch(labels) + regionHeartbeatCounter.DeletePartialMatch(labels) + regionBucketsHandleDuration.DeletePartialMatch(labels) + regionBucketsCounter.DeletePartialMatch(labels) + regionBucketsReportInterval.DeletePartialMatch(labels) +} diff --git a/pkg/schedule/filter/metrics.go b/pkg/schedule/filter/metrics.go index 944b670133..f68077e4f1 100644 --- a/pkg/schedule/filter/metrics.go +++ b/pkg/schedule/filter/metrics.go @@ -37,3 +37,9 @@ func init() { prometheus.MustRegister(filterSourceCounter) prometheus.MustRegister(filterTargetCounter) } + +// DeleteStoreMetrics deletes the filter metrics of a store. +func DeleteStoreMetrics(storeID string) { + filterSourceCounter.DeletePartialMatch(prometheus.Labels{"source": storeID}) + filterTargetCounter.DeletePartialMatch(prometheus.Labels{"target": storeID}) +} diff --git a/pkg/schedule/hbstream/metric.go b/pkg/schedule/hbstream/metric.go index 1620ddebce..c02b43034c 100644 --- a/pkg/schedule/hbstream/metric.go +++ b/pkg/schedule/hbstream/metric.go @@ -30,3 +30,8 @@ var ( func init() { prometheus.MustRegister(heartbeatStreamCounter) } + +// DeleteStoreMetrics deletes the heartbeat stream metrics of a store. +func DeleteStoreMetrics(storeID string) { + heartbeatStreamCounter.DeletePartialMatch(prometheus.Labels{"store": storeID}) +} diff --git a/pkg/schedule/schedulers/metrics.go b/pkg/schedule/schedulers/metrics.go index 4f943571e2..25ec8e72cb 100644 --- a/pkg/schedule/schedulers/metrics.go +++ b/pkg/schedule/schedulers/metrics.go @@ -210,6 +210,21 @@ func init() { prometheus.MustRegister(balanceRangeJobGauge) } +// DeleteStoreMetrics deletes the per-store scheduler metrics of a store. +func DeleteStoreMetrics(storeID string) { + opInfluenceStatus.DeletePartialMatch(prometheus.Labels{"store": storeID}) + balanceWitnessCounter.DeleteLabelValues("move-witness", storeID+"-out") + balanceWitnessCounter.DeleteLabelValues("move-witness", storeID+"-in") + hotSchedulerResultCounter.DeletePartialMatch(prometheus.Labels{"store": storeID}) + balanceDirectionCounter.DeletePartialMatch(prometheus.Labels{"store": storeID}) + hotDirectionCounter.DeletePartialMatch(prometheus.Labels{"store": storeID}) + evictedSlowStoreStatusGauge.DeletePartialMatch(prometheus.Labels{"store": storeID}) + evictedStoppingStoreStatusGauge.DeleteLabelValues(storeID) + slowStoreTriggerLimitGauge.DeletePartialMatch(prometheus.Labels{"store": storeID}) + storeSlowTrendEvictedStatusGauge.DeletePartialMatch(prometheus.Labels{"store": storeID}) + balanceRangeGauge.DeletePartialMatch(prometheus.Labels{"store": storeID}) +} + func balanceLeaderCounterWithEvent(event string) prometheus.Counter { return schedulerCounter.WithLabelValues(types.BalanceLeaderScheduler.String(), event) } diff --git a/pkg/statistics/hot_peer_cache.go b/pkg/statistics/hot_peer_cache.go index 5edf8f505a..cde5eb20b8 100644 --- a/pkg/statistics/hot_peer_cache.go +++ b/pkg/statistics/hot_peer_cache.go @@ -560,6 +560,7 @@ func (f *HotPeerCache) gc() { delete(f.regionsOfStore, storeID) delete(f.thresholdsOfStore, storeID) delete(f.metrics, storeID) + hotCacheStatusGauge.DeletePartialMatch(prometheus.Labels{"store": storeTag(storeID), "type": f.kind.String()}) } } // remove expired items diff --git a/pkg/statistics/hot_peer_cache_test.go b/pkg/statistics/hot_peer_cache_test.go index 6aacae3aea..c8c08c4735 100644 --- a/pkg/statistics/hot_peer_cache_test.go +++ b/pkg/statistics/hot_peer_cache_test.go @@ -23,6 +23,7 @@ import ( "time" "github.com/docker/go-units" + "github.com/prometheus/client_golang/prometheus/testutil" "github.com/stretchr/testify/require" "github.com/pingcap/kvproto/pkg/metapb" @@ -822,16 +823,24 @@ func TestRemoveExpireItems(t *testing.T) { re.NotEmpty(cache.storesOfRegion[region2.GetID()]) time.Sleep(cache.topNTTL) // case2: remove items when the store is not exist - re.NotNil(cache.peersOfStore[region1.GetLeader().GetStoreId()]) - re.NotNil(cache.peersOfStore[region2.GetLeader().GetStoreId()]) + store1ID := region1.GetLeader().GetStoreId() + store2ID := region2.GetLeader().GetStoreId() + re.NotNil(cache.peersOfStore[store1ID]) + re.NotNil(cache.peersOfStore[store2ID]) + // the hotcache status gauge should be populated for the removed stores before gc. + re.NotZero(testutil.ToFloat64(hotCacheStatusGauge.WithLabelValues("add_item", storeTag(store1ID), cache.kind.String()))) + re.NotZero(testutil.ToFloat64(hotCacheStatusGauge.WithLabelValues("add_item", storeTag(store2ID), cache.kind.String()))) cluster.ResetStores() re.Empty(cluster.GetStores()) region3, err := buildRegion(cluster, utils.Write, 3, 10) re.NoError(err) checkAndUpdate(re, cache, region3) - re.Nil(cache.peersOfStore[region1.GetLeader().GetStoreId()]) - re.Nil(cache.peersOfStore[region2.GetLeader().GetStoreId()]) + re.Nil(cache.peersOfStore[store1ID]) + re.Nil(cache.peersOfStore[store2ID]) re.NotEmpty(cache.regionsOfStore[region3.GetLeader().GetStoreId()]) + // gc should also delete the hotcache status gauge series of the removed stores, not just the in-memory map entries. + re.Zero(testutil.ToFloat64(hotCacheStatusGauge.WithLabelValues("add_item", storeTag(store1ID), cache.kind.String()))) + re.Zero(testutil.ToFloat64(hotCacheStatusGauge.WithLabelValues("add_item", storeTag(store2ID), cache.kind.String()))) } func TestDifferentReportInterval(t *testing.T) { diff --git a/pkg/statistics/store_collection.go b/pkg/statistics/store_collection.go index aef38b70c3..5e911a0d90 100644 --- a/pkg/statistics/store_collection.go +++ b/pkg/statistics/store_collection.go @@ -19,6 +19,7 @@ import ( "strconv" "github.com/pingcap/kvproto/pkg/metapb" + "github.com/prometheus/client_golang/prometheus" "github.com/tikv/pd/pkg/core" "github.com/tikv/pd/pkg/core/constant" @@ -291,36 +292,7 @@ func (s *storeStatistics) collect() { // ResetStoreStatistics resets the metrics of store. func ResetStoreStatistics(storeAddress string, id string) { - metrics := []string{ - "region_score", - "leader_score", - "region_size", - "region_count", - "leader_size", - "leader_count", - "witness_count", - "learner_count", - "store_available", - "store_used", - "store_capacity", - "store_write_rate_bytes", - "store_read_rate_bytes", - "store_write_rate_keys", - "store_read_rate_keys", - "store_write_query_rate", - "store_read_query_rate", - "store_read_cpu_usage", - "store_read_cpu_usage_instant", - "store_regions_write_rate_bytes", - "store_regions_write_rate_keys", - "store_slow_trend_cause_value", - "store_slow_trend_cause_rate", - "store_slow_trend_result_value", - "store_slow_trend_result_rate", - } - for _, m := range metrics { - storeStatusGauge.DeleteLabelValues(storeAddress, id, m) - } + storeStatusGauge.DeletePartialMatch(prometheus.Labels{"address": storeAddress, "store": id}) clusterStatusGauge.DeletePartialMatch(utils.SingleLabel("store", id)) } diff --git a/server/cluster/cluster.go b/server/cluster/cluster.go index 80a8a4b689..4d4373567a 100644 --- a/server/cluster/cluster.go +++ b/server/cluster/cluster.go @@ -1771,6 +1771,8 @@ func (c *RaftCluster) BuryStoreLocked(storeID uint64, forceBury bool) error { addr := store.GetAddress() storeIDStr := strconv.FormatUint(storeID, 10) statistics.ResetStoreStatistics(addr, storeIDStr) + filter.DeleteStoreMetrics(storeIDStr) + hbstream.DeleteStoreMetrics(storeIDStr) if !c.IsServiceIndependent(constant.SchedulingServiceName) { c.removeStoreStatistics(storeID) } diff --git a/server/cluster/scheduling_controller.go b/server/cluster/scheduling_controller.go index f9819e8dc5..6df6f502d9 100644 --- a/server/cluster/scheduling_controller.go +++ b/server/cluster/scheduling_controller.go @@ -17,6 +17,7 @@ package cluster import ( "context" "net/http" + "strconv" "sync" "time" @@ -202,6 +203,7 @@ func (sc *schedulingController) collectSchedulingMetrics() { func (sc *schedulingController) removeStoreStatistics(storeID uint64) { sc.hotStat.RemoveRollingStoreStats(storeID) sc.slowStat.RemoveSlowStoreStatus(storeID) + schedulers.DeleteStoreMetrics(strconv.FormatUint(storeID, 10)) } func (sc *schedulingController) updateStoreStatistics(storeID uint64, isSlow bool) { diff --git a/server/metrics.go b/server/metrics.go index f995cf5efa..a12ca491b4 100644 --- a/server/metrics.go +++ b/server/metrics.go @@ -238,3 +238,15 @@ func init() { prometheus.MustRegister(forwardTsoDuration) prometheus.MustRegister(regionRequestCounter) } + +// DeleteStoreMetrics deletes the per-store heartbeat/bucket-report metrics of a store. +func DeleteStoreMetrics(storeAddress, id string) { + labels := prometheus.Labels{"address": storeAddress, "store": id} + regionHeartbeatCounter.DeletePartialMatch(labels) + regionHeartbeatLatency.DeletePartialMatch(labels) + regionHeartbeatHandleDuration.DeletePartialMatch(labels) + storeHeartbeatHandleDuration.DeletePartialMatch(labels) + bucketReportCounter.DeletePartialMatch(labels) + bucketReportLatency.DeletePartialMatch(labels) + bucketReportInterval.DeletePartialMatch(labels) +} diff --git a/server/server.go b/server/server.go index 58e3caa31e..0be25c7733 100644 --- a/server/server.go +++ b/server/server.go @@ -744,6 +744,7 @@ func (s *Server) serverMetricsLoop() { select { case <-ticker.C: s.collectEtcdStateMetrics() + s.cleanupRemovedStoreMetrics() case <-ctx.Done(): log.Info("server is closed, exit metrics loop") return @@ -751,6 +752,23 @@ func (s *Server) serverMetricsLoop() { } } +// cleanupRemovedStoreMetrics deletes the per-store heartbeat/bucket-report metrics +// of stores that have been tombstoned. These metrics are recorded directly in this +// package (not in pkg/statistics or pkg/schedule), so they cannot be cleaned up from +// within RaftCluster's bury path and are instead swept periodically here. +func (s *Server) cleanupRemovedStoreMetrics() { + rc := s.GetRaftCluster() + if rc == nil { + return + } + for _, store := range rc.GetStores() { + if !store.IsRemoved() { + continue + } + DeleteStoreMetrics(store.GetAddress(), strconv.FormatUint(store.GetID(), 10)) + } +} + // encryptionKeyManagerLoop is used to start monitor encryption key changes. func (s *Server) encryptionKeyManagerLoop() { defer logutil.LogPanic() From 2ff912f687cfeb70050d868033cd072a86909243 Mon Sep 17 00:00:00 2001 From: bufferflies <1045931706@qq.com> Date: Mon, 10 Aug 2026 14:37:27 +0200 Subject: [PATCH 02/16] server, mcs/scheduling: avoid re-scanning metric vecs every tick for tombstoned stores collectMetrics/cleanupRemovedStoreMetrics run on a fixed ticker and, for every store still IsRemoved() this tick, called DeleteStoreMetrics unconditionally. DeletePartialMatch takes an exclusive lock and does a linear scan of the whole metric vector, so a store lingering in the tombstoned-but-not-yet-fully-removed state (up to the tombstone GC interval) triggered that scan on every tick for no reason after the first pass. Track which store IDs have already been cleaned in a per-ticker-owned map (no lock needed, since only that ticker's single goroutine touches it), skip stores already in it, and drop entries once a store is no longer tombstoned (fully removed), so the tracking map itself stays bounded. Signed-off-by: bufferflies Signed-off-by: bufferflies <1045931706@qq.com> --- pkg/mcs/scheduling/server/cluster.go | 58 ++++++++++++++++++++-------- server/server.go | 28 +++++++++++++- 2 files changed, 68 insertions(+), 18 deletions(-) diff --git a/pkg/mcs/scheduling/server/cluster.go b/pkg/mcs/scheduling/server/cluster.go index 8d298a20e0..a726b4239a 100644 --- a/pkg/mcs/scheduling/server/cluster.go +++ b/pkg/mcs/scheduling/server/cluster.go @@ -94,6 +94,16 @@ type Cluster struct { pdLeader atomic.Value running atomic.Bool + // cleanedRemovedStoreMetrics tracks store IDs whose per-store heartbeat/bucket + // metrics (defined in this package, see metrics.go) have already been deleted by + // collectMetrics after the store was tombstoned. Only collectMetrics's own + // goroutine (via runMetricsCollectionJob's ticker) touches this map, so it needs + // no lock. Without it, DeleteStoreMetrics would DeletePartialMatch-scan every + // per-store metric vector on every tick for as long as the store stays + // tombstoned-but-not-yet-removed (up to the tombstone GC interval), even though + // there is nothing left to delete after the first pass. + cleanedRemovedStoreMetrics map[uint64]struct{} + backendAddress string httpClient *http.Client @@ -146,22 +156,23 @@ func NewCluster( return nil, err } c := &Cluster{ - ctx: ctx, - cancel: cancel, - BasicCluster: basicCluster, - ruleManager: ruleManager, - keyRangeManager: keyrange.NewManager(), - labelerManager: labelerManager, - affinityManager: affinityManager, - persistConfig: persistConfig, - hotStat: statistics.NewHotStat(ctx, basicCluster), - labelStats: statistics.NewLabelStatistics(), - regionStats: statistics.NewRegionStatistics(basicCluster, persistConfig, ruleManager), - storage: storage, - hbStreams: hbStreams, - checkMembershipCh: checkMembershipCh, - httpClient: httpClient, - backendAddress: backendAddress, + ctx: ctx, + cancel: cancel, + BasicCluster: basicCluster, + ruleManager: ruleManager, + keyRangeManager: keyrange.NewManager(), + labelerManager: labelerManager, + affinityManager: affinityManager, + persistConfig: persistConfig, + hotStat: statistics.NewHotStat(ctx, basicCluster), + labelStats: statistics.NewLabelStatistics(), + regionStats: statistics.NewRegionStatistics(basicCluster, persistConfig, ruleManager), + storage: storage, + hbStreams: hbStreams, + checkMembershipCh: checkMembershipCh, + httpClient: httpClient, + backendAddress: backendAddress, + cleanedRemovedStoreMetrics: make(map[uint64]struct{}), heartbeatRunner: ratelimit.NewConcurrentRunner(heartbeatTaskRunner, ratelimit.NewConcurrencyLimiter(uint64(runtime.NumCPU()*2)), time.Minute), miscRunner: ratelimit.NewConcurrentRunner(miscTaskRunner, ratelimit.NewConcurrencyLimiter(uint64(runtime.NumCPU()*2)), time.Minute), @@ -723,11 +734,24 @@ func (c *Cluster) runMetricsCollectionJob() { func (c *Cluster) collectMetrics() { statsMap := statistics.NewStoreStatisticsMap(c.persistConfig) stores := c.GetStores() + // Stores still tombstoned this tick; anything in cleanedRemovedStoreMetrics but + // not in this set has been fully removed, so its tracking entry can be dropped. + removed := make(map[uint64]struct{}) for _, s := range stores { statsMap.Observe(s) statistics.ObserveHotStat(s, c.hotStat.StoresStats) if s.IsRemoved() { - DeleteStoreMetrics(s.GetAddress(), strconv.FormatUint(s.GetID(), 10)) + storeID := s.GetID() + removed[storeID] = struct{}{} + if _, cleaned := c.cleanedRemovedStoreMetrics[storeID]; !cleaned { + DeleteStoreMetrics(s.GetAddress(), strconv.FormatUint(storeID, 10)) + c.cleanedRemovedStoreMetrics[storeID] = struct{}{} + } + } + } + for storeID := range c.cleanedRemovedStoreMetrics { + if _, stillTombstoned := removed[storeID]; !stillTombstoned { + delete(c.cleanedRemovedStoreMetrics, storeID) } } statsMap.Collect() diff --git a/server/server.go b/server/server.go index 0be25c7733..1e4d72b0a0 100644 --- a/server/server.go +++ b/server/server.go @@ -246,6 +246,16 @@ type Server struct { // Cgroup Monitor cgMonitor cgroup.Monitor + + // cleanedRemovedStoreMetrics tracks store IDs whose per-store heartbeat/bucket + // report metrics (defined in server/metrics.go) have already been deleted by + // cleanupRemovedStoreMetrics after the store was tombstoned. Only that method, + // driven by serverMetricsLoop's own single-goroutine ticker, touches this map, + // so it needs no lock. Without it, DeleteStoreMetrics would DeletePartialMatch- + // scan every per-store metric vector on every tick for as long as the store + // stays tombstoned-but-not-yet-removed (up to the tombstone GC interval), even + // though there is nothing left to delete after the first pass. + cleanedRemovedStoreMetrics map[uint64]struct{} } // HandlerBuilder builds a server HTTP handler. @@ -761,11 +771,27 @@ func (s *Server) cleanupRemovedStoreMetrics() { if rc == nil { return } + if s.cleanedRemovedStoreMetrics == nil { + s.cleanedRemovedStoreMetrics = make(map[uint64]struct{}) + } + // Stores still tombstoned this tick; anything in cleanedRemovedStoreMetrics but + // not in this set has been fully removed, so its tracking entry can be dropped. + removed := make(map[uint64]struct{}) for _, store := range rc.GetStores() { if !store.IsRemoved() { continue } - DeleteStoreMetrics(store.GetAddress(), strconv.FormatUint(store.GetID(), 10)) + storeID := store.GetID() + removed[storeID] = struct{}{} + if _, cleaned := s.cleanedRemovedStoreMetrics[storeID]; !cleaned { + DeleteStoreMetrics(store.GetAddress(), strconv.FormatUint(storeID, 10)) + s.cleanedRemovedStoreMetrics[storeID] = struct{}{} + } + } + for storeID := range s.cleanedRemovedStoreMetrics { + if _, stillTombstoned := removed[storeID]; !stillTombstoned { + delete(s.cleanedRemovedStoreMetrics, storeID) + } } } From a52f9051d11629f0b11c76fe740515091fc51465 Mon Sep 17 00:00:00 2001 From: bufferflies <1045931706@qq.com> Date: Tue, 11 Aug 2026 11:54:56 +0200 Subject: [PATCH 03/16] address review feedback: fix gci import order, drop unsafe dedup, harden test - pkg/statistics/store_collection.go: fix import group ordering flagged by the statics CI check (gci) -- prometheus/client_golang belongs in the default group before the pingcap group, not merged into it. - server/server.go, pkg/mcs/scheduling/server/cluster.go: drop the cleanedRemovedStoreMetrics dedup tracking added earlier. It was unsafe: HandleRegionHeartbeat (both server/grpc_service.go and pkg/mcs/scheduling/server/grpc_service.go) only checks store == nil, not store.IsRemoved(), before recording regionHeartbeat*/ bucketReport* metrics, unlike HandleStoreHeartbeat which rejects tombstoned stores up front via checkStore(). A late region heartbeat for an already-cleaned, still-tombstoned store could therefore recreate a metric series that the dedup tracking would then never sweep again. Go back to deleting unconditionally every tick; a DeletePartialMatch on labels that no longer exist is a cheap no-op. - pkg/statistics/hot_peer_cache_test.go: assert on hotCacheStatusGauge.DeletePartialMatch's return count instead of WithLabelValues+ToFloat64. The latter recreates a fresh, zero-valued series on every call regardless of whether gc() actually deleted the old one, so it verified less than it looked like; the former proves no series remain. Signed-off-by: bufferflies Signed-off-by: bufferflies <1045931706@qq.com> --- pkg/mcs/scheduling/server/cluster.go | 66 ++++++++++----------------- pkg/statistics/hot_peer_cache_test.go | 11 +++-- pkg/statistics/store_collection.go | 3 +- server/server.go | 37 ++++----------- 4 files changed, 45 insertions(+), 72 deletions(-) diff --git a/pkg/mcs/scheduling/server/cluster.go b/pkg/mcs/scheduling/server/cluster.go index a726b4239a..4d270a0211 100644 --- a/pkg/mcs/scheduling/server/cluster.go +++ b/pkg/mcs/scheduling/server/cluster.go @@ -94,16 +94,6 @@ type Cluster struct { pdLeader atomic.Value running atomic.Bool - // cleanedRemovedStoreMetrics tracks store IDs whose per-store heartbeat/bucket - // metrics (defined in this package, see metrics.go) have already been deleted by - // collectMetrics after the store was tombstoned. Only collectMetrics's own - // goroutine (via runMetricsCollectionJob's ticker) touches this map, so it needs - // no lock. Without it, DeleteStoreMetrics would DeletePartialMatch-scan every - // per-store metric vector on every tick for as long as the store stays - // tombstoned-but-not-yet-removed (up to the tombstone GC interval), even though - // there is nothing left to delete after the first pass. - cleanedRemovedStoreMetrics map[uint64]struct{} - backendAddress string httpClient *http.Client @@ -156,23 +146,22 @@ func NewCluster( return nil, err } c := &Cluster{ - ctx: ctx, - cancel: cancel, - BasicCluster: basicCluster, - ruleManager: ruleManager, - keyRangeManager: keyrange.NewManager(), - labelerManager: labelerManager, - affinityManager: affinityManager, - persistConfig: persistConfig, - hotStat: statistics.NewHotStat(ctx, basicCluster), - labelStats: statistics.NewLabelStatistics(), - regionStats: statistics.NewRegionStatistics(basicCluster, persistConfig, ruleManager), - storage: storage, - hbStreams: hbStreams, - checkMembershipCh: checkMembershipCh, - httpClient: httpClient, - backendAddress: backendAddress, - cleanedRemovedStoreMetrics: make(map[uint64]struct{}), + ctx: ctx, + cancel: cancel, + BasicCluster: basicCluster, + ruleManager: ruleManager, + keyRangeManager: keyrange.NewManager(), + labelerManager: labelerManager, + affinityManager: affinityManager, + persistConfig: persistConfig, + hotStat: statistics.NewHotStat(ctx, basicCluster), + labelStats: statistics.NewLabelStatistics(), + regionStats: statistics.NewRegionStatistics(basicCluster, persistConfig, ruleManager), + storage: storage, + hbStreams: hbStreams, + checkMembershipCh: checkMembershipCh, + httpClient: httpClient, + backendAddress: backendAddress, heartbeatRunner: ratelimit.NewConcurrentRunner(heartbeatTaskRunner, ratelimit.NewConcurrencyLimiter(uint64(runtime.NumCPU()*2)), time.Minute), miscRunner: ratelimit.NewConcurrentRunner(miscTaskRunner, ratelimit.NewConcurrencyLimiter(uint64(runtime.NumCPU()*2)), time.Minute), @@ -734,24 +723,19 @@ func (c *Cluster) runMetricsCollectionJob() { func (c *Cluster) collectMetrics() { statsMap := statistics.NewStoreStatisticsMap(c.persistConfig) stores := c.GetStores() - // Stores still tombstoned this tick; anything in cleanedRemovedStoreMetrics but - // not in this set has been fully removed, so its tracking entry can be dropped. - removed := make(map[uint64]struct{}) + // Delete unconditionally on every tick rather than tracking which stores were + // already cleaned: a region heartbeat for an already-tombstoned store can still + // land here and recreate a series. RegionHeartbeat (grpc_service.go) only checks + // store == nil, not store.IsRemoved(), before recording regionHeartbeat* metrics. + // If cleanup only ran once per store, such a late write would never get swept + // again for as long as the store stays tombstoned-but-not-yet-removed. + // DeletePartialMatch on labels that no longer exist is a cheap no-op, so + // repeating it every tick is safe. for _, s := range stores { statsMap.Observe(s) statistics.ObserveHotStat(s, c.hotStat.StoresStats) if s.IsRemoved() { - storeID := s.GetID() - removed[storeID] = struct{}{} - if _, cleaned := c.cleanedRemovedStoreMetrics[storeID]; !cleaned { - DeleteStoreMetrics(s.GetAddress(), strconv.FormatUint(storeID, 10)) - c.cleanedRemovedStoreMetrics[storeID] = struct{}{} - } - } - } - for storeID := range c.cleanedRemovedStoreMetrics { - if _, stillTombstoned := removed[storeID]; !stillTombstoned { - delete(c.cleanedRemovedStoreMetrics, storeID) + DeleteStoreMetrics(s.GetAddress(), strconv.FormatUint(s.GetID(), 10)) } } statsMap.Collect() diff --git a/pkg/statistics/hot_peer_cache_test.go b/pkg/statistics/hot_peer_cache_test.go index c8c08c4735..235ce5b4d6 100644 --- a/pkg/statistics/hot_peer_cache_test.go +++ b/pkg/statistics/hot_peer_cache_test.go @@ -23,6 +23,7 @@ import ( "time" "github.com/docker/go-units" + "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/testutil" "github.com/stretchr/testify/require" @@ -838,9 +839,13 @@ func TestRemoveExpireItems(t *testing.T) { re.Nil(cache.peersOfStore[store1ID]) re.Nil(cache.peersOfStore[store2ID]) re.NotEmpty(cache.regionsOfStore[region3.GetLeader().GetStoreId()]) - // gc should also delete the hotcache status gauge series of the removed stores, not just the in-memory map entries. - re.Zero(testutil.ToFloat64(hotCacheStatusGauge.WithLabelValues("add_item", storeTag(store1ID), cache.kind.String()))) - re.Zero(testutil.ToFloat64(hotCacheStatusGauge.WithLabelValues("add_item", storeTag(store2ID), cache.kind.String()))) + // gc should also delete the hotcache status gauge series of the removed stores, not + // just the in-memory map entries. DeletePartialMatch returns how many series it + // found and removed, so a zero return proves nothing is left -- unlike checking + // WithLabelValues' value, which would recreate a fresh (zero-valued) series on + // every call regardless of whether gc() actually deleted the old one. + re.Zero(hotCacheStatusGauge.DeletePartialMatch(prometheus.Labels{"store": storeTag(store1ID), "type": cache.kind.String()})) + re.Zero(hotCacheStatusGauge.DeletePartialMatch(prometheus.Labels{"store": storeTag(store2ID), "type": cache.kind.String()})) } func TestDifferentReportInterval(t *testing.T) { diff --git a/pkg/statistics/store_collection.go b/pkg/statistics/store_collection.go index 5e911a0d90..7e88505d83 100644 --- a/pkg/statistics/store_collection.go +++ b/pkg/statistics/store_collection.go @@ -18,9 +18,10 @@ import ( "fmt" "strconv" - "github.com/pingcap/kvproto/pkg/metapb" "github.com/prometheus/client_golang/prometheus" + "github.com/pingcap/kvproto/pkg/metapb" + "github.com/tikv/pd/pkg/core" "github.com/tikv/pd/pkg/core/constant" "github.com/tikv/pd/pkg/core/storelimit" diff --git a/server/server.go b/server/server.go index 1e4d72b0a0..a3d1948c64 100644 --- a/server/server.go +++ b/server/server.go @@ -246,16 +246,6 @@ type Server struct { // Cgroup Monitor cgMonitor cgroup.Monitor - - // cleanedRemovedStoreMetrics tracks store IDs whose per-store heartbeat/bucket - // report metrics (defined in server/metrics.go) have already been deleted by - // cleanupRemovedStoreMetrics after the store was tombstoned. Only that method, - // driven by serverMetricsLoop's own single-goroutine ticker, touches this map, - // so it needs no lock. Without it, DeleteStoreMetrics would DeletePartialMatch- - // scan every per-store metric vector on every tick for as long as the store - // stays tombstoned-but-not-yet-removed (up to the tombstone GC interval), even - // though there is nothing left to delete after the first pass. - cleanedRemovedStoreMetrics map[uint64]struct{} } // HandlerBuilder builds a server HTTP handler. @@ -771,27 +761,20 @@ func (s *Server) cleanupRemovedStoreMetrics() { if rc == nil { return } - if s.cleanedRemovedStoreMetrics == nil { - s.cleanedRemovedStoreMetrics = make(map[uint64]struct{}) - } - // Stores still tombstoned this tick; anything in cleanedRemovedStoreMetrics but - // not in this set has been fully removed, so its tracking entry can be dropped. - removed := make(map[uint64]struct{}) + // Delete unconditionally on every tick rather than tracking which stores were + // already cleaned: a region heartbeat for an already-tombstoned store can still + // land here and recreate a series. HandleRegionHeartbeat only checks + // store == nil, not store.IsRemoved(), before recording regionHeartbeat*/ + // bucketReport* metrics -- unlike HandleStoreHeartbeat, which rejects tombstoned + // stores up front via checkStore(). If cleanup only ran once per store, such a + // late write would never get swept again for as long as the store stays + // tombstoned-but-not-yet-removed. DeletePartialMatch on labels that no longer + // exist is a cheap no-op, so repeating it every tick is safe. for _, store := range rc.GetStores() { if !store.IsRemoved() { continue } - storeID := store.GetID() - removed[storeID] = struct{}{} - if _, cleaned := s.cleanedRemovedStoreMetrics[storeID]; !cleaned { - DeleteStoreMetrics(store.GetAddress(), strconv.FormatUint(storeID, 10)) - s.cleanedRemovedStoreMetrics[storeID] = struct{}{} - } - } - for storeID := range s.cleanedRemovedStoreMetrics { - if _, stillTombstoned := removed[storeID]; !stillTombstoned { - delete(s.cleanedRemovedStoreMetrics, storeID) - } + DeleteStoreMetrics(store.GetAddress(), strconv.FormatUint(store.GetID(), 10)) } } From c7b71807537df940a635ad28399e1cd3ce1b1b93 Mon Sep 17 00:00:00 2001 From: bufferflies <1045931706@qq.com> Date: Tue, 11 Aug 2026 13:35:12 +0200 Subject: [PATCH 04/16] schedule/filter, schedule/hbstream: sweep tombstoned-store metrics every tick filter.DeleteStoreMetrics and hbstream.DeleteStoreMetrics were only called once, from BuryStoreLocked at tombstone time. That one-shot delete gets undone almost immediately for both packages: - Every scheduler's Schedule() still runs cluster.GetStores() (which includes tombstoned-but-not-fully-removed stores) through StoreStateFilter each cycle. The rejection itself is what increments filterSourceCounter/filterTargetCounter, so as long as the store stays known, the counters get recreated on the very next scheduling cycle. - HeartbeatStreams' keepalive ticker sends to every entry in s.streams as long as storeInformer.GetStore(storeID) != nil, without checking IsRemoved(). Nothing unbinds a store's stream on bury, only on a failed Send, so heartbeatStreamCounter keeps getting rewritten for as long as the tombstoned store's stream stays connected. Neither has any cleanup tied to full removal either (unlike clusterStatusGauge's DeleteClusterStatusMetrics), so once recreated they would never be swept again. Move both into the existing per-tick collectSchedulingMetrics (server/cluster) and collectMetrics (pkg/mcs/scheduling/server) loops, which already iterate every store every cycle for other metrics, mirroring the same unconditional-delete pattern already used there for regionHeartbeat*/bucketReport* metrics. Keep the one-shot calls in BuryStoreLocked for immediacy; the periodic sweep makes them safe against the ongoing rewrites. Found via a systematic post-merge review of PR #11127 after CI went green, using the github-pr-review skill. Signed-off-by: bufferflies Signed-off-by: bufferflies <1045931706@qq.com> --- pkg/mcs/scheduling/server/cluster.go | 18 +++++++++++++----- server/cluster/scheduling_controller.go | 14 ++++++++++++++ 2 files changed, 27 insertions(+), 5 deletions(-) diff --git a/pkg/mcs/scheduling/server/cluster.go b/pkg/mcs/scheduling/server/cluster.go index 4d270a0211..bd1721a6e6 100644 --- a/pkg/mcs/scheduling/server/cluster.go +++ b/pkg/mcs/scheduling/server/cluster.go @@ -48,6 +48,7 @@ import ( "github.com/tikv/pd/pkg/schedule" "github.com/tikv/pd/pkg/schedule/affinity" sc "github.com/tikv/pd/pkg/schedule/config" + "github.com/tikv/pd/pkg/schedule/filter" "github.com/tikv/pd/pkg/schedule/hbstream" "github.com/tikv/pd/pkg/schedule/keyrange" "github.com/tikv/pd/pkg/schedule/labeler" @@ -727,15 +728,22 @@ func (c *Cluster) collectMetrics() { // already cleaned: a region heartbeat for an already-tombstoned store can still // land here and recreate a series. RegionHeartbeat (grpc_service.go) only checks // store == nil, not store.IsRemoved(), before recording regionHeartbeat* metrics. - // If cleanup only ran once per store, such a late write would never get swept - // again for as long as the store stays tombstoned-but-not-yet-removed. - // DeletePartialMatch on labels that no longer exist is a cheap no-op, so - // repeating it every tick is safe. + // The same is true of filter and heartbeat-stream metrics: every scheduler's + // Schedule() still runs StoreStateFilter against every known store each cycle + // (that rejection is what increments the filter counters), and the heartbeat + // stream keepalive ticker keeps touching a tombstoned store as long as its + // stream is still bound. If cleanup only ran once per store, such late writes + // would never get swept again for as long as the store stays + // tombstoned-but-not-yet-removed. DeletePartialMatch on labels that no longer + // exist is a cheap no-op, so repeating it every tick is safe. for _, s := range stores { statsMap.Observe(s) statistics.ObserveHotStat(s, c.hotStat.StoresStats) if s.IsRemoved() { - DeleteStoreMetrics(s.GetAddress(), strconv.FormatUint(s.GetID(), 10)) + storeIDStr := strconv.FormatUint(s.GetID(), 10) + DeleteStoreMetrics(s.GetAddress(), storeIDStr) + filter.DeleteStoreMetrics(storeIDStr) + hbstream.DeleteStoreMetrics(storeIDStr) } } statsMap.Collect() diff --git a/server/cluster/scheduling_controller.go b/server/cluster/scheduling_controller.go index 6df6f502d9..287224fa68 100644 --- a/server/cluster/scheduling_controller.go +++ b/server/cluster/scheduling_controller.go @@ -29,6 +29,7 @@ import ( "github.com/tikv/pd/pkg/schedule/checker" sc "github.com/tikv/pd/pkg/schedule/config" sche "github.com/tikv/pd/pkg/schedule/core" + "github.com/tikv/pd/pkg/schedule/filter" "github.com/tikv/pd/pkg/schedule/hbstream" "github.com/tikv/pd/pkg/schedule/operator" "github.com/tikv/pd/pkg/schedule/placement" @@ -185,6 +186,19 @@ func (sc *schedulingController) collectSchedulingMetrics() { for _, s := range stores { statsMap.Observe(s) statistics.ObserveHotStat(s, sc.hotStat.StoresStats) + if s.IsRemoved() { + // Unlike the other per-store cleanup called once from BuryStoreLocked, filter + // and heartbeat-stream metrics for a tombstoned store keep getting rewritten by + // unrelated, ongoing activity for as long as the store stays known: every + // scheduler's Schedule() still runs StoreStateFilter against it every cycle + // (that rejection is what increments the filter counters), and the heartbeat + // stream keepalive ticker keeps touching it as long as its stream is still + // bound. A one-shot delete at bury time gets undone almost immediately, so + // delete unconditionally here on every tick instead. + storeIDStr := strconv.FormatUint(s.GetID(), 10) + filter.DeleteStoreMetrics(storeIDStr) + hbstream.DeleteStoreMetrics(storeIDStr) + } } statsMap.Collect() sc.coordinator.GetSchedulersController().CollectSchedulerMetrics() From 9cef4747b62deb31e2d44d1d542bd20f0a8aac9a Mon Sep 17 00:00:00 2001 From: bufferflies <1045931706@qq.com> Date: Wed, 12 Aug 2026 05:01:03 +0200 Subject: [PATCH 05/16] schedule/filter, schedule/hbstream: reset counters on scheduling stop resetSchedulingMetrics (classic) and resetMetrics (mcs scheduling service) already reset schedulerStatusGauge, hotSpotStatusGauge, and other scheduling-related metrics when the scheduling service/job stops (context cancellation or primary handoff), but filter and heartbeat-stream counters were left out, so a standby replica or a freshly-stopped scheduling job could keep showing stale values from when it was last active. Add ResetFilterMetrics and ResetHeartbeatStreamMetrics, following the existing Reset*Metrics naming convention, and call them from both reset functions for parity with the other metrics already reset there. Signed-off-by: bufferflies Signed-off-by: bufferflies <1045931706@qq.com> --- pkg/mcs/scheduling/server/cluster.go | 2 ++ pkg/schedule/filter/metrics.go | 6 ++++++ pkg/schedule/hbstream/metric.go | 5 +++++ server/cluster/scheduling_controller.go | 2 ++ 4 files changed, 15 insertions(+) diff --git a/pkg/mcs/scheduling/server/cluster.go b/pkg/mcs/scheduling/server/cluster.go index bd1721a6e6..57be27320b 100644 --- a/pkg/mcs/scheduling/server/cluster.go +++ b/pkg/mcs/scheduling/server/cluster.go @@ -765,6 +765,8 @@ func resetMetrics() { statistics.Reset() schedulers.ResetSchedulerMetrics() schedule.ResetHotSpotMetrics() + filter.ResetFilterMetrics() + hbstream.ResetHeartbeatStreamMetrics() } // StartBackgroundJobs starts background jobs. diff --git a/pkg/schedule/filter/metrics.go b/pkg/schedule/filter/metrics.go index f68077e4f1..6801c77ec0 100644 --- a/pkg/schedule/filter/metrics.go +++ b/pkg/schedule/filter/metrics.go @@ -43,3 +43,9 @@ func DeleteStoreMetrics(storeID string) { filterSourceCounter.DeletePartialMatch(prometheus.Labels{"source": storeID}) filterTargetCounter.DeletePartialMatch(prometheus.Labels{"target": storeID}) } + +// ResetFilterMetrics resets the filter metrics. +func ResetFilterMetrics() { + filterSourceCounter.Reset() + filterTargetCounter.Reset() +} diff --git a/pkg/schedule/hbstream/metric.go b/pkg/schedule/hbstream/metric.go index c02b43034c..532e207a1e 100644 --- a/pkg/schedule/hbstream/metric.go +++ b/pkg/schedule/hbstream/metric.go @@ -35,3 +35,8 @@ func init() { func DeleteStoreMetrics(storeID string) { heartbeatStreamCounter.DeletePartialMatch(prometheus.Labels{"store": storeID}) } + +// ResetHeartbeatStreamMetrics resets the heartbeat stream metrics. +func ResetHeartbeatStreamMetrics() { + heartbeatStreamCounter.Reset() +} diff --git a/server/cluster/scheduling_controller.go b/server/cluster/scheduling_controller.go index 287224fa68..4bb965c4f6 100644 --- a/server/cluster/scheduling_controller.go +++ b/server/cluster/scheduling_controller.go @@ -178,6 +178,8 @@ func resetSchedulingMetrics() { statistics.ResetLabelStatsMetrics() // reset hot cache metrics statistics.ResetHotCacheStatusMetrics() + filter.ResetFilterMetrics() + hbstream.ResetHeartbeatStreamMetrics() } func (sc *schedulingController) collectSchedulingMetrics() { From 2f87de597e9d37cc2fd21e757ae9d0a2ed3ac26b Mon Sep 17 00:00:00 2001 From: bufferflies <1045931706@qq.com> Date: Wed, 12 Aug 2026 10:34:15 +0200 Subject: [PATCH 06/16] address maintainer review: idle-cluster gc, address changes, final-removal race Five issues raised in review, all fixed: - pkg/statistics/hot_cache.go: HotPeerCache.gc() was only ever triggered from UpdateStat, so a store removed while the cluster is idle (or while it was the only store still receiving hot-peer updates) would never have its hotCacheStatusGauge series cleaned up -- the main remove-tombstone scenario in #11126. Call gc() from CollectMetrics's existing periodic, activity-independent tick instead; it already self-throttles via topNTTL, so calling it every tick is cheap. - pkg/statistics/store_collection.go, server/metrics.go, pkg/mcs/scheduling/server/metrics.go: ResetStoreStatistics and DeleteStoreMetrics matched on address as well as store ID, but PD allows an existing store ID to change address (e.g. after a TiKV restart with a new IP). Matching both permanently leaked any series recorded under a previous address. Match on the store label alone; it's the stable identifier. Updated all call sites accordingly, including pkg/mcs/router's watcher, which had the same pattern. - pkg/schedule/schedulers/metrics.go: HotPendingSum is keyed by store but wasn't deleted by DeleteStoreMetrics. - server/server.go, pkg/mcs/scheduling/server/cluster.go, server/cluster/scheduling_controller.go: the periodic sweeps can only discover tombstoned stores while they remain in GetStores(). A write landing after the last sweep but before the store's final metadata deletion recreates a series that no later sweep can reach, since the per-tick loop only considers stores GetStores() currently returns. Track the set of store IDs seen tombstoned on the previous tick; when one drops out of GetStores() entirely between ticks (i.e. it was fully removed), do one more delete for it. This now needs only the store ID, not a cached address, since the address match was dropped above. - pkg/mcs/scheduling/server/cluster.go: ObserveHotStat runs just before the per-store cleanup block and can recreate storeStatusGauge from rolling stats that were never retired for a tombstoned store (unlike classic mode, nothing here calls RemoveRollingStoreStats). The block never reset that vector, so the meta watcher's one-shot reset was undone on the very next collection tick. Call ResetStoreStatistics from the same per-tick sweep as the other per-store metrics, via a small deleteTombstonedStoreMetrics helper shared with the final-removal catch-up path above. Signed-off-by: bufferflies Signed-off-by: bufferflies <1045931706@qq.com> --- pkg/mcs/router/server/meta/watcher.go | 2 +- pkg/mcs/scheduling/server/cluster.go | 45 +++++++++++++++++++---- pkg/mcs/scheduling/server/meta/watcher.go | 2 +- pkg/mcs/scheduling/server/metrics.go | 8 +++- pkg/schedule/schedulers/metrics.go | 1 + pkg/statistics/hot_cache.go | 7 ++++ pkg/statistics/store_collection.go | 10 +++-- server/cluster/cluster.go | 3 +- server/cluster/scheduling_controller.go | 41 ++++++++++++++++----- server/metrics.go | 8 +++- server/server.go | 22 ++++++++++- 11 files changed, 119 insertions(+), 30 deletions(-) diff --git a/pkg/mcs/router/server/meta/watcher.go b/pkg/mcs/router/server/meta/watcher.go index abe4de65f1..1ac0b1f75c 100644 --- a/pkg/mcs/router/server/meta/watcher.go +++ b/pkg/mcs/router/server/meta/watcher.go @@ -82,7 +82,7 @@ func (w *Watcher) initializeStoreWatcher() error { } if store.GetNodeState() == metapb.NodeState_Removed { - statistics.ResetStoreStatistics(store.GetAddress(), strconv.FormatUint(store.GetId(), 10)) + statistics.ResetStoreStatistics(strconv.FormatUint(store.GetId(), 10)) // TODO: remove hot stats } diff --git a/pkg/mcs/scheduling/server/cluster.go b/pkg/mcs/scheduling/server/cluster.go index 57be27320b..afc0a1547e 100644 --- a/pkg/mcs/scheduling/server/cluster.go +++ b/pkg/mcs/scheduling/server/cluster.go @@ -95,6 +95,11 @@ type Cluster struct { pdLeader atomic.Value running atomic.Bool + // recentlyTombstonedStores is the set of store IDs collectMetrics saw tombstoned + // on its last tick. Only that method, driven by runMetricsCollectionJob's own + // single-goroutine ticker, touches it, so it needs no lock. + recentlyTombstonedStores map[uint64]struct{} + backendAddress string httpClient *http.Client @@ -721,6 +726,16 @@ func (c *Cluster) runMetricsCollectionJob() { } } +// deleteTombstonedStoreMetrics deletes every per-store metric this package knows +// about (heartbeat/bucket metrics, storeStatusGauge/clusterStatusGauge, filter +// counters, and heartbeat-stream counters) for a tombstoned store. +func deleteTombstonedStoreMetrics(storeID string) { + statistics.ResetStoreStatistics(storeID) + DeleteStoreMetrics(storeID) + filter.DeleteStoreMetrics(storeID) + hbstream.DeleteStoreMetrics(storeID) +} + func (c *Cluster) collectMetrics() { statsMap := statistics.NewStoreStatisticsMap(c.persistConfig) stores := c.GetStores() @@ -732,20 +747,34 @@ func (c *Cluster) collectMetrics() { // Schedule() still runs StoreStateFilter against every known store each cycle // (that rejection is what increments the filter counters), and the heartbeat // stream keepalive ticker keeps touching a tombstoned store as long as its - // stream is still bound. If cleanup only ran once per store, such late writes - // would never get swept again for as long as the store stays - // tombstoned-but-not-yet-removed. DeletePartialMatch on labels that no longer - // exist is a cheap no-op, so repeating it every tick is safe. + // stream is still bound. ObserveHotStat above can likewise recreate + // storeStatusGauge from rolling stats that were never retired. If cleanup only + // ran once per store, such late writes would never get swept again for as long + // as the store stays tombstoned-but-not-yet-removed. DeletePartialMatch on + // labels that no longer exist is a cheap no-op, so repeating it every tick is + // safe. + current := make(map[uint64]struct{}) for _, s := range stores { statsMap.Observe(s) statistics.ObserveHotStat(s, c.hotStat.StoresStats) if s.IsRemoved() { - storeIDStr := strconv.FormatUint(s.GetID(), 10) - DeleteStoreMetrics(s.GetAddress(), storeIDStr) - filter.DeleteStoreMetrics(storeIDStr) - hbstream.DeleteStoreMetrics(storeIDStr) + storeID := s.GetID() + current[storeID] = struct{}{} + deleteTombstonedStoreMetrics(strconv.FormatUint(storeID, 10)) + } + } + // A store that was tombstoned as of the last sweep but isn't known at all this + // tick was fully removed in between. The loop above only reaches stores + // GetStores() currently returns, so a write landing in that gap -- after the + // last sweep saw it tombstoned but before removal completed -- would otherwise + // be unreachable by any future sweep. One more delete call closes that window; + // it's a no-op if nothing was actually rewritten. + for storeID := range c.recentlyTombstonedStores { + if _, stillKnown := current[storeID]; !stillKnown { + deleteTombstonedStoreMetrics(strconv.FormatUint(storeID, 10)) } } + c.recentlyTombstonedStores = current statsMap.Collect() c.coordinator.GetSchedulersController().CollectSchedulerMetrics() diff --git a/pkg/mcs/scheduling/server/meta/watcher.go b/pkg/mcs/scheduling/server/meta/watcher.go index 82b11fbab5..38d8e5acf4 100644 --- a/pkg/mcs/scheduling/server/meta/watcher.go +++ b/pkg/mcs/scheduling/server/meta/watcher.go @@ -86,7 +86,7 @@ func (w *Watcher) initializeStoreWatcher() error { if store.GetNodeState() == metapb.NodeState_Removed { storeIDStr := strconv.FormatUint(store.GetId(), 10) - statistics.ResetStoreStatistics(store.GetAddress(), storeIDStr) + statistics.ResetStoreStatistics(storeIDStr) filter.DeleteStoreMetrics(storeIDStr) hbstream.DeleteStoreMetrics(storeIDStr) schedulers.DeleteStoreMetrics(storeIDStr) diff --git a/pkg/mcs/scheduling/server/metrics.go b/pkg/mcs/scheduling/server/metrics.go index d5d1cb0d74..b54bb16573 100644 --- a/pkg/mcs/scheduling/server/metrics.go +++ b/pkg/mcs/scheduling/server/metrics.go @@ -96,8 +96,12 @@ func init() { } // DeleteStoreMetrics deletes the per-store heartbeat/bucket metrics of a store. -func DeleteStoreMetrics(storeAddress, id string) { - labels := prometheus.Labels{"address": storeAddress, "store": id} +// Matches on the store label alone, not address: PD allows an existing store ID to +// change address (e.g. after a TiKV restart with a new IP), so requiring the current +// address to match as well would permanently leak any series recorded under a +// previous address. +func DeleteStoreMetrics(id string) { + labels := prometheus.Labels{"store": id} storeHeartbeatHandleDuration.DeletePartialMatch(labels) storeHeartbeatCounter.DeletePartialMatch(labels) regionHeartbeatHandleDuration.DeletePartialMatch(labels) diff --git a/pkg/schedule/schedulers/metrics.go b/pkg/schedule/schedulers/metrics.go index 25ec8e72cb..cb412c922b 100644 --- a/pkg/schedule/schedulers/metrics.go +++ b/pkg/schedule/schedulers/metrics.go @@ -223,6 +223,7 @@ func DeleteStoreMetrics(storeID string) { slowStoreTriggerLimitGauge.DeletePartialMatch(prometheus.Labels{"store": storeID}) storeSlowTrendEvictedStatusGauge.DeletePartialMatch(prometheus.Labels{"store": storeID}) balanceRangeGauge.DeletePartialMatch(prometheus.Labels{"store": storeID}) + HotPendingSum.DeletePartialMatch(prometheus.Labels{"store": storeID}) } func balanceLeaderCounterWithEvent(event string) prometheus.Counter { diff --git a/pkg/statistics/hot_cache.go b/pkg/statistics/hot_cache.go index 835399b13d..08b1f70e7a 100644 --- a/pkg/statistics/hot_cache.go +++ b/pkg/statistics/hot_cache.go @@ -154,9 +154,16 @@ func (w *HotCache) GetHotPeerStat(kind utils.RWType, regionID, storeID uint64) * func (w *HotCache) CollectMetrics() { w.CheckWriteAsync(func(cache *HotPeerCache) { cache.collectMetrics() + // gc() is otherwise only triggered from UpdateStat, so a store removed while + // the cluster is idle (or was the only store still receiving updates) would + // never have its hotCacheStatusGauge series cleaned up. Piggyback on this + // periodic, activity-independent tick instead; gc() already self-throttles + // via topNTTL, so calling it every tick is cheap. + cache.gc() }) w.CheckReadAsync(func(cache *HotPeerCache) { cache.collectMetrics() + cache.gc() }) } diff --git a/pkg/statistics/store_collection.go b/pkg/statistics/store_collection.go index 7e88505d83..ac45aa3fc0 100644 --- a/pkg/statistics/store_collection.go +++ b/pkg/statistics/store_collection.go @@ -18,8 +18,6 @@ import ( "fmt" "strconv" - "github.com/prometheus/client_golang/prometheus" - "github.com/pingcap/kvproto/pkg/metapb" "github.com/tikv/pd/pkg/core" @@ -292,8 +290,12 @@ func (s *storeStatistics) collect() { } // ResetStoreStatistics resets the metrics of store. -func ResetStoreStatistics(storeAddress string, id string) { - storeStatusGauge.DeletePartialMatch(prometheus.Labels{"address": storeAddress, "store": id}) +// Matches on the store label alone, not address: PD allows an existing store ID to +// change address (e.g. after a TiKV restart with a new IP), so requiring the current +// address to match as well would permanently leak any series recorded under a +// previous address. +func ResetStoreStatistics(id string) { + storeStatusGauge.DeletePartialMatch(utils.SingleLabel("store", id)) clusterStatusGauge.DeletePartialMatch(utils.SingleLabel("store", id)) } diff --git a/server/cluster/cluster.go b/server/cluster/cluster.go index 4d4373567a..64903efd82 100644 --- a/server/cluster/cluster.go +++ b/server/cluster/cluster.go @@ -1768,9 +1768,8 @@ func (c *RaftCluster) BuryStoreLocked(storeID uint64, forceBury bool) error { // clean up the residual information. c.prevStoreLimit.Delete(storeID) c.RemoveStoreLimit(storeID) - addr := store.GetAddress() storeIDStr := strconv.FormatUint(storeID, 10) - statistics.ResetStoreStatistics(addr, storeIDStr) + statistics.ResetStoreStatistics(storeIDStr) filter.DeleteStoreMetrics(storeIDStr) hbstream.DeleteStoreMetrics(storeIDStr) if !c.IsServiceIndependent(constant.SchedulingServiceName) { diff --git a/server/cluster/scheduling_controller.go b/server/cluster/scheduling_controller.go index 4bb965c4f6..b70a0be385 100644 --- a/server/cluster/scheduling_controller.go +++ b/server/cluster/scheduling_controller.go @@ -59,6 +59,12 @@ type schedulingController struct { hotStat *statistics.HotStat slowStat *statistics.SlowStat running bool + + // recentlyTombstonedStores is the set of store IDs collectSchedulingMetrics saw + // tombstoned on its last tick. Only that method, driven by + // runSchedulingMetricsCollectionJob's own single-goroutine ticker, touches it, so + // it needs no lock. + recentlyTombstonedStores map[uint64]struct{} } // newSchedulingController creates a new scheduling controller. @@ -185,23 +191,40 @@ func resetSchedulingMetrics() { func (sc *schedulingController) collectSchedulingMetrics() { statsMap := statistics.NewStoreStatisticsMap(sc.opt) stores := sc.GetStores() + // Unlike the other per-store cleanup called once from BuryStoreLocked, filter + // and heartbeat-stream metrics for a tombstoned store keep getting rewritten by + // unrelated, ongoing activity for as long as the store stays known: every + // scheduler's Schedule() still runs StoreStateFilter against it every cycle + // (that rejection is what increments the filter counters), and the heartbeat + // stream keepalive ticker keeps touching it as long as its stream is still + // bound. A one-shot delete at bury time gets undone almost immediately, so + // delete unconditionally here on every tick instead. + current := make(map[uint64]struct{}) for _, s := range stores { statsMap.Observe(s) statistics.ObserveHotStat(s, sc.hotStat.StoresStats) if s.IsRemoved() { - // Unlike the other per-store cleanup called once from BuryStoreLocked, filter - // and heartbeat-stream metrics for a tombstoned store keep getting rewritten by - // unrelated, ongoing activity for as long as the store stays known: every - // scheduler's Schedule() still runs StoreStateFilter against it every cycle - // (that rejection is what increments the filter counters), and the heartbeat - // stream keepalive ticker keeps touching it as long as its stream is still - // bound. A one-shot delete at bury time gets undone almost immediately, so - // delete unconditionally here on every tick instead. - storeIDStr := strconv.FormatUint(s.GetID(), 10) + storeID := s.GetID() + current[storeID] = struct{}{} + storeIDStr := strconv.FormatUint(storeID, 10) + filter.DeleteStoreMetrics(storeIDStr) + hbstream.DeleteStoreMetrics(storeIDStr) + } + } + // A store that was tombstoned as of the last sweep but isn't known at all this + // tick was fully removed in between. The loop above only reaches stores + // GetStores() currently returns, so a write landing in that gap -- after the + // last sweep saw it tombstoned but before removal completed -- would otherwise + // be unreachable by any future sweep. One more delete call closes that window; + // it's a no-op if nothing was actually rewritten. + for storeID := range sc.recentlyTombstonedStores { + if _, stillKnown := current[storeID]; !stillKnown { + storeIDStr := strconv.FormatUint(storeID, 10) filter.DeleteStoreMetrics(storeIDStr) hbstream.DeleteStoreMetrics(storeIDStr) } } + sc.recentlyTombstonedStores = current statsMap.Collect() sc.coordinator.GetSchedulersController().CollectSchedulerMetrics() sc.coordinator.CollectHotSpotMetrics() diff --git a/server/metrics.go b/server/metrics.go index a12ca491b4..71e06b9f8e 100644 --- a/server/metrics.go +++ b/server/metrics.go @@ -240,8 +240,12 @@ func init() { } // DeleteStoreMetrics deletes the per-store heartbeat/bucket-report metrics of a store. -func DeleteStoreMetrics(storeAddress, id string) { - labels := prometheus.Labels{"address": storeAddress, "store": id} +// Matches on the store label alone, not address: PD allows an existing store ID to +// change address (e.g. after a TiKV restart with a new IP), so requiring the current +// address to match as well would permanently leak any series recorded under a +// previous address. +func DeleteStoreMetrics(id string) { + labels := prometheus.Labels{"store": id} regionHeartbeatCounter.DeletePartialMatch(labels) regionHeartbeatLatency.DeletePartialMatch(labels) regionHeartbeatHandleDuration.DeletePartialMatch(labels) diff --git a/server/server.go b/server/server.go index a3d1948c64..de381d9651 100644 --- a/server/server.go +++ b/server/server.go @@ -246,6 +246,11 @@ type Server struct { // Cgroup Monitor cgMonitor cgroup.Monitor + + // recentlyTombstonedStores is the set of store IDs cleanupRemovedStoreMetrics saw + // tombstoned on its last tick. Only that method, driven by serverMetricsLoop's own + // single-goroutine ticker, touches it, so it needs no lock. + recentlyTombstonedStores map[uint64]struct{} } // HandlerBuilder builds a server HTTP handler. @@ -770,12 +775,27 @@ func (s *Server) cleanupRemovedStoreMetrics() { // late write would never get swept again for as long as the store stays // tombstoned-but-not-yet-removed. DeletePartialMatch on labels that no longer // exist is a cheap no-op, so repeating it every tick is safe. + current := make(map[uint64]struct{}) for _, store := range rc.GetStores() { if !store.IsRemoved() { continue } - DeleteStoreMetrics(store.GetAddress(), strconv.FormatUint(store.GetID(), 10)) + storeID := store.GetID() + current[storeID] = struct{}{} + DeleteStoreMetrics(strconv.FormatUint(storeID, 10)) + } + // A store that was tombstoned as of the last sweep but isn't known at all this + // tick was fully removed in between. This loop only reaches stores GetStores() + // currently returns, so a write landing in that gap -- after the last sweep saw + // it tombstoned but before removal completed -- would otherwise be unreachable + // by any future sweep. One more delete call closes that window; it's a no-op if + // nothing was actually rewritten. + for storeID := range s.recentlyTombstonedStores { + if _, stillKnown := current[storeID]; !stillKnown { + DeleteStoreMetrics(strconv.FormatUint(storeID, 10)) + } } + s.recentlyTombstonedStores = current } // encryptionKeyManagerLoop is used to start monitor encryption key changes. From 0c7a06604e52c1e1a454fbd67e1d88dcb44f9cef Mon Sep 17 00:00:00 2001 From: bufferflies <1045931706@qq.com> Date: Fri, 14 Aug 2026 08:36:37 +0200 Subject: [PATCH 07/16] mcs, server: replace periodic tombstone-metric sweeps with bury/removal event hooks Fix the write-side gaps that forced periodic sweeps in the first place (RegionHeartbeat/ReportBuckets/RegionBuckets and the heartbeat-stream loop now check IsRemoved()), then move heartbeat, heartbeat-stream, and store-status metric cleanup to one-shot bury-time/full-removal-time hooks. Filter counters keep their periodic sweep since StoreStateFilter legitimately keeps rejecting a tombstoned-but-known store every cycle. Co-Authored-By: Claude Sonnet 5 Signed-off-by: bufferflies <1045931706@qq.com> --- pkg/mcs/scheduling/server/cluster.go | 46 ++++++++----------- pkg/mcs/scheduling/server/grpc_service.go | 5 +++ pkg/mcs/scheduling/server/meta/watcher.go | 38 +++++++++++++++- pkg/schedule/hbstream/heartbeat_streams.go | 8 ++++ pkg/schedule/metrics.go | 5 +++ server/cluster/cluster.go | 27 ++++++++++++ server/cluster/scheduling_controller.go | 28 ++++++------ server/grpc_service.go | 5 +++ server/server.go | 51 ++-------------------- 9 files changed, 124 insertions(+), 89 deletions(-) diff --git a/pkg/mcs/scheduling/server/cluster.go b/pkg/mcs/scheduling/server/cluster.go index afc0a1547e..64eb1a65f8 100644 --- a/pkg/mcs/scheduling/server/cluster.go +++ b/pkg/mcs/scheduling/server/cluster.go @@ -97,7 +97,14 @@ type Cluster struct { // recentlyTombstonedStores is the set of store IDs collectMetrics saw tombstoned // on its last tick. Only that method, driven by runMetricsCollectionJob's own - // single-goroutine ticker, touches it, so it needs no lock. + // single-goroutine ticker, touches it, so it needs no lock. Filter counters are + // the only metrics that still need this: every scheduler's Schedule() keeps + // running StoreStateFilter against a tombstoned-but-known store every cycle, + // which is what increments them, so a one-shot delete at bury time gets undone + // almost immediately. Heartbeat, heartbeat-stream, and store-status metrics no + // longer need it -- writes to them stop at bury time (see the store-tombstoned + // callback wired in SetRuntimeResources and the IsRemoved checks in + // grpc_service.go/heartbeat_streams.go). recentlyTombstonedStores map[uint64]struct{} backendAddress string @@ -331,6 +338,10 @@ func (c *Cluster) SetRuntimeResources( c.configWatcher = configWatcher c.ruleWatcher = ruleWatcher c.affinityWatcher = affinityWatcher + metaWatcher.SetOnStoreTombstoned(func(storeID uint64) { + c.hotStat.RemoveRollingStoreStats(storeID) + DeleteStoreMetrics(strconv.FormatUint(storeID, 10)) + }) } func (c *Cluster) stopCluster() { @@ -726,33 +737,14 @@ func (c *Cluster) runMetricsCollectionJob() { } } -// deleteTombstonedStoreMetrics deletes every per-store metric this package knows -// about (heartbeat/bucket metrics, storeStatusGauge/clusterStatusGauge, filter -// counters, and heartbeat-stream counters) for a tombstoned store. -func deleteTombstonedStoreMetrics(storeID string) { - statistics.ResetStoreStatistics(storeID) - DeleteStoreMetrics(storeID) - filter.DeleteStoreMetrics(storeID) - hbstream.DeleteStoreMetrics(storeID) -} - func (c *Cluster) collectMetrics() { statsMap := statistics.NewStoreStatisticsMap(c.persistConfig) stores := c.GetStores() - // Delete unconditionally on every tick rather than tracking which stores were - // already cleaned: a region heartbeat for an already-tombstoned store can still - // land here and recreate a series. RegionHeartbeat (grpc_service.go) only checks - // store == nil, not store.IsRemoved(), before recording regionHeartbeat* metrics. - // The same is true of filter and heartbeat-stream metrics: every scheduler's - // Schedule() still runs StoreStateFilter against every known store each cycle - // (that rejection is what increments the filter counters), and the heartbeat - // stream keepalive ticker keeps touching a tombstoned store as long as its - // stream is still bound. ObserveHotStat above can likewise recreate - // storeStatusGauge from rolling stats that were never retired. If cleanup only - // ran once per store, such late writes would never get swept again for as long - // as the store stays tombstoned-but-not-yet-removed. DeletePartialMatch on - // labels that no longer exist is a cheap no-op, so repeating it every tick is - // safe. + // Filter counters are the only ones that need repeated cleanup here: every + // scheduler's Schedule() still runs StoreStateFilter against every known store + // each cycle, and that rejection is what increments them, so a one-shot delete + // at bury time gets undone almost immediately. DeletePartialMatch on labels that + // no longer exist is a cheap no-op, so repeating it every tick is safe. current := make(map[uint64]struct{}) for _, s := range stores { statsMap.Observe(s) @@ -760,7 +752,7 @@ func (c *Cluster) collectMetrics() { if s.IsRemoved() { storeID := s.GetID() current[storeID] = struct{}{} - deleteTombstonedStoreMetrics(strconv.FormatUint(storeID, 10)) + filter.DeleteStoreMetrics(strconv.FormatUint(storeID, 10)) } } // A store that was tombstoned as of the last sweep but isn't known at all this @@ -771,7 +763,7 @@ func (c *Cluster) collectMetrics() { // it's a no-op if nothing was actually rewritten. for storeID := range c.recentlyTombstonedStores { if _, stillKnown := current[storeID]; !stillKnown { - deleteTombstonedStoreMetrics(strconv.FormatUint(storeID, 10)) + filter.DeleteStoreMetrics(strconv.FormatUint(storeID, 10)) } } c.recentlyTombstonedStores = current diff --git a/pkg/mcs/scheduling/server/grpc_service.go b/pkg/mcs/scheduling/server/grpc_service.go index 3cba680f03..9a1b6d3aeb 100644 --- a/pkg/mcs/scheduling/server/grpc_service.go +++ b/pkg/mcs/scheduling/server/grpc_service.go @@ -149,6 +149,9 @@ func (s *Service) RegionHeartbeat(stream schedulingpb.Scheduling_RegionHeartbeat if store == nil { return errors.Errorf("invalid store ID %d, not found", storeID) } + if store.IsRemoved() { + return errors.Errorf("store ID %d is tombstone", storeID) + } storeAddress := store.GetAddress() storeLabel := strconv.FormatUint(storeID, 10) @@ -220,6 +223,8 @@ func (s *Service) RegionBuckets(stream schedulingpb.Scheduling_RegionBucketsServ // As TiKV report buckets just after the region heartbeat, for new created region, PD may receive buckets report before the first region heartbeat is handled. // So we should not return error here. log.Debug("the store of the bucket in region is not found", zap.Uint64("region-id", buckets.GetRegionId())) + } else if store.IsRemoved() { + log.Debug("the store of the bucket in region is tombstone", zap.Uint64("region-id", buckets.GetRegionId()), zap.Uint64("store-id", store.GetID())) } else { storeLabel = strconv.FormatUint(store.GetID(), 10) storeAddress = store.GetAddress() diff --git a/pkg/mcs/scheduling/server/meta/watcher.go b/pkg/mcs/scheduling/server/meta/watcher.go index 38d8e5acf4..6d79678e9f 100644 --- a/pkg/mcs/scheduling/server/meta/watcher.go +++ b/pkg/mcs/scheduling/server/meta/watcher.go @@ -18,6 +18,7 @@ import ( "context" "strconv" "sync" + "sync/atomic" "github.com/gogo/protobuf/proto" "go.etcd.io/etcd/api/v3/mvccpb" @@ -28,6 +29,7 @@ import ( "github.com/pingcap/log" "github.com/tikv/pd/pkg/core" + "github.com/tikv/pd/pkg/schedule" "github.com/tikv/pd/pkg/schedule/filter" "github.com/tikv/pd/pkg/schedule/hbstream" "github.com/tikv/pd/pkg/schedule/schedulers" @@ -45,6 +47,13 @@ type Watcher struct { etcdClient *clientv3.Client basicCluster *core.BasicCluster storeWatcher *etcdutil.LoopWatcher + + // onStoreTombstoned is late-bound via SetOnStoreTombstoned once the parent + // Cluster (which owns hotStat and the scheduling-server-only metrics that this + // package can't import without a cycle) finishes construction. The watcher + // itself starts watching before that point, so it's read through an atomic + // pointer to stay safe against a store event racing the setup. + onStoreTombstoned atomic.Pointer[func(storeID uint64)] } // NewWatcher creates a new watcher to watch the meta change from PD. @@ -90,7 +99,10 @@ func (w *Watcher) initializeStoreWatcher() error { filter.DeleteStoreMetrics(storeIDStr) hbstream.DeleteStoreMetrics(storeIDStr) schedulers.DeleteStoreMetrics(storeIDStr) - // TODO: remove hot stats + schedule.DeleteStoreMetrics(storeIDStr) + if fn := w.onStoreTombstoned.Load(); fn != nil { + (*fn)(store.GetId()) + } } return nil @@ -103,7 +115,15 @@ func (w *Watcher) initializeStoreWatcher() error { } origin := w.basicCluster.GetStore(storeID) if origin != nil { + storeIDStr := strconv.FormatUint(storeID, 10) statistics.DeleteClusterStatusMetrics(origin) + filter.DeleteStoreMetrics(storeIDStr) + hbstream.DeleteStoreMetrics(storeIDStr) + schedulers.DeleteStoreMetrics(storeIDStr) + schedule.DeleteStoreMetrics(storeIDStr) + if fn := w.onStoreTombstoned.Load(); fn != nil { + (*fn)(storeID) + } w.basicCluster.DeleteStore(origin) log.Info("delete store meta", zap.Uint64("store-id", storeID)) } @@ -124,6 +144,22 @@ func (w *Watcher) initializeStoreWatcher() error { return w.storeWatcher.WaitLoad() } +// SetOnStoreTombstoned sets the callback invoked (at least once) when a store +// transitions to tombstone, for cleanup that only the parent Cluster can do +// without an import cycle (removing rolling hot stats, resetting the +// scheduling-server-owned heartbeat metrics). NewWatcher's initial load runs +// before the caller has a chance to install this callback, so a store already +// tombstoned at startup would otherwise never get it invoked; reconcile against +// whatever the watcher has already loaded here to close that gap. +func (w *Watcher) SetOnStoreTombstoned(fn func(storeID uint64)) { + w.onStoreTombstoned.Store(&fn) + for _, store := range w.basicCluster.GetStores() { + if store.IsRemoved() { + fn(store.GetID()) + } + } +} + // Close closes the watcher. func (w *Watcher) Close() { w.cancel() diff --git a/pkg/schedule/hbstream/heartbeat_streams.go b/pkg/schedule/hbstream/heartbeat_streams.go index e9d067a54c..f4edd48e3d 100644 --- a/pkg/schedule/hbstream/heartbeat_streams.go +++ b/pkg/schedule/hbstream/heartbeat_streams.go @@ -148,6 +148,10 @@ func (s *HeartbeatStreams) run() { delete(s.streams, storeID) continue } + if store.IsRemoved() { + delete(s.streams, storeID) + continue + } storeAddress := store.GetAddress() if stream, ok := s.streams[storeID]; ok { if err := stream.Send(msg); err != nil { @@ -172,6 +176,10 @@ func (s *HeartbeatStreams) run() { delete(s.streams, storeID) continue } + if store.IsRemoved() { + delete(s.streams, storeID) + continue + } storeAddress := store.GetAddress() storeLabel := strconv.FormatUint(storeID, 10) if err := stream.Send(keepAlive); err != nil { diff --git a/pkg/schedule/metrics.go b/pkg/schedule/metrics.go index de65484641..010beeb1a2 100644 --- a/pkg/schedule/metrics.go +++ b/pkg/schedule/metrics.go @@ -29,3 +29,8 @@ var ( func init() { prometheus.MustRegister(hotSpotStatusGauge) } + +// DeleteStoreMetrics deletes the hotspot status metrics of a store. +func DeleteStoreMetrics(storeID string) { + hotSpotStatusGauge.DeletePartialMatch(prometheus.Labels{"store": storeID}) +} diff --git a/server/cluster/cluster.go b/server/cluster/cluster.go index 64903efd82..3e646f9de3 100644 --- a/server/cluster/cluster.go +++ b/server/cluster/cluster.go @@ -55,6 +55,7 @@ import ( "github.com/tikv/pd/pkg/progress" "github.com/tikv/pd/pkg/ratelimit" "github.com/tikv/pd/pkg/replication" + "github.com/tikv/pd/pkg/schedule" "github.com/tikv/pd/pkg/schedule/affinity" sc "github.com/tikv/pd/pkg/schedule/config" "github.com/tikv/pd/pkg/schedule/filter" @@ -207,6 +208,21 @@ type RaftCluster struct { syncRegionRunner ratelimit.Runner stopGCStateManager func() + + // onStoreBuried is an optional callback invoked (at least once) with a store's + // ID right after it's buried, for cleanup that only the owning package can do + // without an import cycle -- e.g. the server package's own heartbeat/bucket + // metrics, which server/cluster cannot import directly. Set via + // SetOnStoreBuried; read through an atomic pointer since BuryStoreLocked can run + // before the owner has had a chance to install it. + onStoreBuried atomic.Pointer[func(storeID string)] +} + +// SetOnStoreBuried sets the callback invoked when a store is buried +// (transitions to tombstone), for per-store cleanup owned by a package that +// server/cluster cannot import. +func (c *RaftCluster) SetOnStoreBuried(fn func(storeID string)) { + c.onStoreBuried.Store(&fn) } // Status saves some state information. @@ -1772,9 +1788,13 @@ func (c *RaftCluster) BuryStoreLocked(storeID uint64, forceBury bool) error { statistics.ResetStoreStatistics(storeIDStr) filter.DeleteStoreMetrics(storeIDStr) hbstream.DeleteStoreMetrics(storeIDStr) + schedule.DeleteStoreMetrics(storeIDStr) if !c.IsServiceIndependent(constant.SchedulingServiceName) { c.removeStoreStatistics(storeID) } + if fn := c.onStoreBuried.Load(); fn != nil { + (*fn)(storeIDStr) + } } return err } @@ -2177,7 +2197,14 @@ func (c *RaftCluster) deleteStore(store *core.StoreInfo) error { return err } } + storeIDStr := strconv.FormatUint(store.GetID(), 10) statistics.DeleteClusterStatusMetrics(store) + filter.DeleteStoreMetrics(storeIDStr) + hbstream.DeleteStoreMetrics(storeIDStr) + schedule.DeleteStoreMetrics(storeIDStr) + if fn := c.onStoreBuried.Load(); fn != nil { + (*fn)(storeIDStr) + } c.DeleteStore(store) return nil } diff --git a/server/cluster/scheduling_controller.go b/server/cluster/scheduling_controller.go index b70a0be385..8276dcb086 100644 --- a/server/cluster/scheduling_controller.go +++ b/server/cluster/scheduling_controller.go @@ -63,7 +63,12 @@ type schedulingController struct { // recentlyTombstonedStores is the set of store IDs collectSchedulingMetrics saw // tombstoned on its last tick. Only that method, driven by // runSchedulingMetricsCollectionJob's own single-goroutine ticker, touches it, so - // it needs no lock. + // it needs no lock. Filter counters are the only metrics that still need this: + // every scheduler's Schedule() keeps running StoreStateFilter against a + // tombstoned-but-known store every cycle, which is what increments them, so a + // one-shot delete at bury time gets undone almost immediately. Heartbeat-stream + // metrics no longer need it -- BuryStoreLocked's cleanup plus the IsRemoved + // check in heartbeat_streams.go's run loop stop those writes at bury time. recentlyTombstonedStores map[uint64]struct{} } @@ -192,13 +197,12 @@ func (sc *schedulingController) collectSchedulingMetrics() { statsMap := statistics.NewStoreStatisticsMap(sc.opt) stores := sc.GetStores() // Unlike the other per-store cleanup called once from BuryStoreLocked, filter - // and heartbeat-stream metrics for a tombstoned store keep getting rewritten by - // unrelated, ongoing activity for as long as the store stays known: every - // scheduler's Schedule() still runs StoreStateFilter against it every cycle - // (that rejection is what increments the filter counters), and the heartbeat - // stream keepalive ticker keeps touching it as long as its stream is still - // bound. A one-shot delete at bury time gets undone almost immediately, so - // delete unconditionally here on every tick instead. + // metrics for a tombstoned store keep getting rewritten by unrelated, ongoing + // activity for as long as the store stays known: every scheduler's Schedule() + // still runs StoreStateFilter against it every cycle, and that rejection is + // what increments the filter counters. A one-shot delete at bury time gets + // undone almost immediately, so delete unconditionally here on every tick + // instead. current := make(map[uint64]struct{}) for _, s := range stores { statsMap.Observe(s) @@ -206,9 +210,7 @@ func (sc *schedulingController) collectSchedulingMetrics() { if s.IsRemoved() { storeID := s.GetID() current[storeID] = struct{}{} - storeIDStr := strconv.FormatUint(storeID, 10) - filter.DeleteStoreMetrics(storeIDStr) - hbstream.DeleteStoreMetrics(storeIDStr) + filter.DeleteStoreMetrics(strconv.FormatUint(storeID, 10)) } } // A store that was tombstoned as of the last sweep but isn't known at all this @@ -219,9 +221,7 @@ func (sc *schedulingController) collectSchedulingMetrics() { // it's a no-op if nothing was actually rewritten. for storeID := range sc.recentlyTombstonedStores { if _, stillKnown := current[storeID]; !stillKnown { - storeIDStr := strconv.FormatUint(storeID, 10) - filter.DeleteStoreMetrics(storeIDStr) - hbstream.DeleteStoreMetrics(storeIDStr) + filter.DeleteStoreMetrics(strconv.FormatUint(storeID, 10)) } } sc.recentlyTombstonedStores = current diff --git a/server/grpc_service.go b/server/grpc_service.go index 22a2849c8e..5791a890e7 100644 --- a/server/grpc_service.go +++ b/server/grpc_service.go @@ -1189,6 +1189,8 @@ func (s *GrpcServer) ReportBuckets(stream pdpb.PD_ReportBucketsServer) error { // As TiKV report buckets just after the region heartbeat, for new created region, PD may receive buckets report before the first region heartbeat is handled. // So we should not return error here. log.Debug("the store of the bucket in region is not found", zap.Uint64("region-id", buckets.GetRegionId())) + } else if store.IsRemoved() { + log.Debug("the store of the bucket in region is tombstone", zap.Uint64("region-id", buckets.GetRegionId()), zap.Uint64("store-id", store.GetID())) } else { storeLabel = strconv.FormatUint(store.GetID(), 10) storeAddress = store.GetAddress() @@ -1371,6 +1373,9 @@ func (s *GrpcServer) RegionHeartbeat(stream pdpb.PD_RegionHeartbeatServer) error if store == nil { return errors.Errorf("invalid store ID %d, not found", storeID) } + if store.IsRemoved() { + return errors.Errorf("store ID %d is tombstone", storeID) + } storeAddress := store.GetAddress() regionHeartbeatCounter.WithLabelValues(storeAddress, storeLabel, "report", "recv").Inc() diff --git a/server/server.go b/server/server.go index de381d9651..4abc7d55f4 100644 --- a/server/server.go +++ b/server/server.go @@ -246,11 +246,6 @@ type Server struct { // Cgroup Monitor cgMonitor cgroup.Monitor - - // recentlyTombstonedStores is the set of store IDs cleanupRemovedStoreMetrics saw - // tombstoned on its last tick. Only that method, driven by serverMetricsLoop's own - // single-goroutine ticker, touches it, so it needs no lock. - recentlyTombstonedStores map[uint64]struct{} } // HandlerBuilder builds a server HTTP handler. @@ -530,6 +525,10 @@ func (s *Server) startServer(ctx context.Context) error { s.tsoAllocator = tso.NewAllocator(s.ctx, constant.DefaultKeyspaceGroupID, s.member, tsoStorage, s) s.basicCluster = core.NewBasicCluster() s.cluster = cluster.NewRaftCluster(ctx, s.GetMember(), s.GetBasicCluster(), s.GetStorage(), syncer.NewRegionSyncer(s), s.client, s.httpClient, s.tsoAllocator) + // This package's own heartbeat/bucket-report metrics can't be cleaned up from + // within RaftCluster's bury path without an import cycle, so RaftCluster invokes + // this callback instead. + s.cluster.SetOnStoreBuried(DeleteStoreMetrics) keyspaceIDAllocator := id.NewAllocator(&id.AllocatorParams{ Client: s.client, Label: id.KeyspaceLabel, @@ -749,7 +748,6 @@ func (s *Server) serverMetricsLoop() { select { case <-ticker.C: s.collectEtcdStateMetrics() - s.cleanupRemovedStoreMetrics() case <-ctx.Done(): log.Info("server is closed, exit metrics loop") return @@ -757,47 +755,6 @@ func (s *Server) serverMetricsLoop() { } } -// cleanupRemovedStoreMetrics deletes the per-store heartbeat/bucket-report metrics -// of stores that have been tombstoned. These metrics are recorded directly in this -// package (not in pkg/statistics or pkg/schedule), so they cannot be cleaned up from -// within RaftCluster's bury path and are instead swept periodically here. -func (s *Server) cleanupRemovedStoreMetrics() { - rc := s.GetRaftCluster() - if rc == nil { - return - } - // Delete unconditionally on every tick rather than tracking which stores were - // already cleaned: a region heartbeat for an already-tombstoned store can still - // land here and recreate a series. HandleRegionHeartbeat only checks - // store == nil, not store.IsRemoved(), before recording regionHeartbeat*/ - // bucketReport* metrics -- unlike HandleStoreHeartbeat, which rejects tombstoned - // stores up front via checkStore(). If cleanup only ran once per store, such a - // late write would never get swept again for as long as the store stays - // tombstoned-but-not-yet-removed. DeletePartialMatch on labels that no longer - // exist is a cheap no-op, so repeating it every tick is safe. - current := make(map[uint64]struct{}) - for _, store := range rc.GetStores() { - if !store.IsRemoved() { - continue - } - storeID := store.GetID() - current[storeID] = struct{}{} - DeleteStoreMetrics(strconv.FormatUint(storeID, 10)) - } - // A store that was tombstoned as of the last sweep but isn't known at all this - // tick was fully removed in between. This loop only reaches stores GetStores() - // currently returns, so a write landing in that gap -- after the last sweep saw - // it tombstoned but before removal completed -- would otherwise be unreachable - // by any future sweep. One more delete call closes that window; it's a no-op if - // nothing was actually rewritten. - for storeID := range s.recentlyTombstonedStores { - if _, stillKnown := current[storeID]; !stillKnown { - DeleteStoreMetrics(strconv.FormatUint(storeID, 10)) - } - } - s.recentlyTombstonedStores = current -} - // encryptionKeyManagerLoop is used to start monitor encryption key changes. func (s *Server) encryptionKeyManagerLoop() { defer logutil.LogPanic() From 759506189293d3c29863105ac250af951a71435d Mon Sep 17 00:00:00 2001 From: bufferflies <1045931706@qq.com> Date: Fri, 14 Aug 2026 09:26:54 +0200 Subject: [PATCH 08/16] mcs, server: don't tear down the RegionHeartbeat stream for a tombstoned store Skip metrics recording and HandleRegionHeartbeat processing for a tombstoned store's region heartbeat, but keep the stream alive instead of returning an error that closes the whole connection. Matches the silent skip already used on the heartbeat-stream push side. Co-Authored-By: Claude Sonnet 5 Signed-off-by: bufferflies <1045931706@qq.com> --- pkg/mcs/scheduling/server/grpc_service.go | 3 ++- server/grpc_service.go | 3 ++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/pkg/mcs/scheduling/server/grpc_service.go b/pkg/mcs/scheduling/server/grpc_service.go index 9a1b6d3aeb..515bcbeec6 100644 --- a/pkg/mcs/scheduling/server/grpc_service.go +++ b/pkg/mcs/scheduling/server/grpc_service.go @@ -150,7 +150,8 @@ func (s *Service) RegionHeartbeat(stream schedulingpb.Scheduling_RegionHeartbeat return errors.Errorf("invalid store ID %d, not found", storeID) } if store.IsRemoved() { - return errors.Errorf("store ID %d is tombstone", storeID) + log.Debug("skip region heartbeat from tombstone store", zap.Uint64("store-id", storeID)) + continue } storeAddress := store.GetAddress() diff --git a/server/grpc_service.go b/server/grpc_service.go index 5791a890e7..2949102ca9 100644 --- a/server/grpc_service.go +++ b/server/grpc_service.go @@ -1374,7 +1374,8 @@ func (s *GrpcServer) RegionHeartbeat(stream pdpb.PD_RegionHeartbeatServer) error return errors.Errorf("invalid store ID %d, not found", storeID) } if store.IsRemoved() { - return errors.Errorf("store ID %d is tombstone", storeID) + log.Debug("skip region heartbeat from tombstone store", zap.Uint64("store-id", storeID)) + continue } storeAddress := store.GetAddress() From 1f8cc07548de1cff6b1e63411e37efd203e07aab Mon Sep 17 00:00:00 2001 From: bufferflies <1045931706@qq.com> Date: Fri, 14 Aug 2026 16:33:16 +0200 Subject: [PATCH 09/16] mcs, server, statistics: close remaining tombstone-metric leaks Four independent gaps found via a fresh review pass: - mcs StoreHeartbeat had no store==nil/IsRemoved guard at all, unlike classic's checkStore(); it recorded metrics and called HandleStoreHeartbeat (PutStore) for unknown/tombstoned stores. - ResetSchedulerMetrics only reset 3 of 12 scheduler-owned vectors, leaving most per-store series stale after a scheduling-job stop. - HotPeerCache.gc() only swept peersOfStore, missing stores whose calcHotThresholds-created series never got hot enough to enter it. - runStatsBackgroundJobs's periodic ObserveRegionsStats call (and its startup seed loop) recreated rolling stats for tombstoned-but-known stores every 30s regardless of BuryStoreLocked's one-shot cleanup; filter getWriteRate and the seed loop to close this for good. Also trims a few over-long comments down to just the load-bearing why. Co-Authored-By: Claude Sonnet 5 Signed-off-by: bufferflies <1045931706@qq.com> --- pkg/core/basic_cluster.go | 15 +++++--- pkg/mcs/scheduling/server/cluster.go | 31 ++++++--------- pkg/mcs/scheduling/server/grpc_service.go | 21 +++++++--- .../schedulers/scheduler_controller.go | 9 +++++ pkg/statistics/hot_peer_cache.go | 22 ++++++++--- server/cluster/cluster.go | 4 +- server/cluster/scheduling_controller.go | 38 ++++++++----------- 7 files changed, 78 insertions(+), 62 deletions(-) diff --git a/pkg/core/basic_cluster.go b/pkg/core/basic_cluster.go index 4bc192c88c..4f136b82a5 100644 --- a/pkg/core/basic_cluster.go +++ b/pkg/core/basic_cluster.go @@ -54,12 +54,17 @@ func (bc *BasicCluster) GetLeaderStoreByRegionID(regionID uint64) *StoreInfo { func (bc *BasicCluster) getWriteRate( f func(storeID uint64) (bytesRate, keysRate float64), ) (storeIDs []uint64, bytesRates, keysRates []float64) { - storeIDs = bc.GetStoreIDs() - count := len(storeIDs) - bytesRates = make([]float64, 0, count) - keysRates = make([]float64, 0, count) - for _, id := range storeIDs { + stores := bc.GetStores() + storeIDs = make([]uint64, 0, len(stores)) + bytesRates = make([]float64, 0, len(stores)) + keysRates = make([]float64, 0, len(stores)) + for _, store := range stores { + if store.IsRemoved() { + continue + } + id := store.GetID() bytesRate, keysRate := f(id) + storeIDs = append(storeIDs, id) bytesRates = append(bytesRates, bytesRate) keysRates = append(keysRates, keysRate) } diff --git a/pkg/mcs/scheduling/server/cluster.go b/pkg/mcs/scheduling/server/cluster.go index 64eb1a65f8..66b8f385c8 100644 --- a/pkg/mcs/scheduling/server/cluster.go +++ b/pkg/mcs/scheduling/server/cluster.go @@ -96,15 +96,10 @@ type Cluster struct { running atomic.Bool // recentlyTombstonedStores is the set of store IDs collectMetrics saw tombstoned - // on its last tick. Only that method, driven by runMetricsCollectionJob's own - // single-goroutine ticker, touches it, so it needs no lock. Filter counters are - // the only metrics that still need this: every scheduler's Schedule() keeps - // running StoreStateFilter against a tombstoned-but-known store every cycle, - // which is what increments them, so a one-shot delete at bury time gets undone - // almost immediately. Heartbeat, heartbeat-stream, and store-status metrics no - // longer need it -- writes to them stop at bury time (see the store-tombstoned - // callback wired in SetRuntimeResources and the IsRemoved checks in - // grpc_service.go/heartbeat_streams.go). + // on its last tick, needed to catch a store that gets fully removed between two + // ticks so its filter counters still get one final delete. Only collectMetrics, + // via runMetricsCollectionJob's single-goroutine ticker, touches it, so it needs + // no lock. recentlyTombstonedStores map[uint64]struct{} backendAddress string @@ -740,11 +735,10 @@ func (c *Cluster) runMetricsCollectionJob() { func (c *Cluster) collectMetrics() { statsMap := statistics.NewStoreStatisticsMap(c.persistConfig) stores := c.GetStores() - // Filter counters are the only ones that need repeated cleanup here: every - // scheduler's Schedule() still runs StoreStateFilter against every known store - // each cycle, and that rejection is what increments them, so a one-shot delete - // at bury time gets undone almost immediately. DeletePartialMatch on labels that - // no longer exist is a cheap no-op, so repeating it every tick is safe. + // Filter counters need repeated cleanup: schedulers keep rejecting a + // tombstoned-but-known store via StoreStateFilter every cycle, so a one-shot + // delete at bury time doesn't stick. Deleting already-gone labels is a cheap + // no-op, so repeating it every tick is safe. current := make(map[uint64]struct{}) for _, s := range stores { statsMap.Observe(s) @@ -755,12 +749,9 @@ func (c *Cluster) collectMetrics() { filter.DeleteStoreMetrics(strconv.FormatUint(storeID, 10)) } } - // A store that was tombstoned as of the last sweep but isn't known at all this - // tick was fully removed in between. The loop above only reaches stores - // GetStores() currently returns, so a write landing in that gap -- after the - // last sweep saw it tombstoned but before removal completed -- would otherwise - // be unreachable by any future sweep. One more delete call closes that window; - // it's a no-op if nothing was actually rewritten. + // A store fully removed between two ticks drops out of GetStores() before this + // sweep can catch it there; delete once more for anything recentlyTombstonedStores + // still remembers but current no longer has. for storeID := range c.recentlyTombstonedStores { if _, stillKnown := current[storeID]; !stillKnown { filter.DeleteStoreMetrics(strconv.FormatUint(storeID, 10)) diff --git a/pkg/mcs/scheduling/server/grpc_service.go b/pkg/mcs/scheduling/server/grpc_service.go index 515bcbeec6..c2d9432d59 100644 --- a/pkg/mcs/scheduling/server/grpc_service.go +++ b/pkg/mcs/scheduling/server/grpc_service.go @@ -16,6 +16,7 @@ package server import ( "context" + "fmt" "io" "net/http" "strconv" @@ -263,17 +264,25 @@ func (s *Service) StoreHeartbeat(_ context.Context, request *schedulingpb.StoreH return &schedulingpb.StoreHeartbeatResponse{Header: notBootstrappedHeader()}, nil } - start := time.Now() - if c.GetStore(request.GetStats().GetStoreId()) == nil { + storeID := request.GetStats().GetStoreId() + if c.GetStore(storeID) == nil { metaWatcher.GetStoreWatcher().ForceLoad() } - storeID := request.GetStats().GetStoreId() store := c.GetStore(storeID) - storeAddress := "" - if store != nil { - storeAddress = store.GetAddress() + if store == nil { + return &schedulingpb.StoreHeartbeatResponse{ + Header: wrapErrorToHeader(schedulingpb.ErrorType_UNKNOWN, fmt.Sprintf("store %v not found", storeID)), + }, nil } + if store.IsRemoved() { + return &schedulingpb.StoreHeartbeatResponse{ + Header: wrapErrorToHeader(schedulingpb.ErrorType_UNKNOWN, fmt.Sprintf("store %v is tombstone", storeID)), + }, nil + } + + start := time.Now() + storeAddress := store.GetAddress() storeLabel := strconv.FormatUint(storeID, 10) if err := c.HandleStoreHeartbeat(request); err != nil { storeHeartbeatCounter.WithLabelValues(storeAddress, storeLabel, "error").Inc() diff --git a/pkg/schedule/schedulers/scheduler_controller.go b/pkg/schedule/schedulers/scheduler_controller.go index c5fec56868..b3c318c5fd 100644 --- a/pkg/schedule/schedulers/scheduler_controller.go +++ b/pkg/schedule/schedulers/scheduler_controller.go @@ -145,6 +145,15 @@ func ResetSchedulerMetrics() { schedulerStatusGauge.Reset() ruleStatusGauge.Reset() regionLabelStatusGauge.Reset() + opInfluenceStatus.Reset() + hotSchedulerResultCounter.Reset() + balanceDirectionCounter.Reset() + hotDirectionCounter.Reset() + evictedSlowStoreStatusGauge.Reset() + evictedStoppingStoreStatusGauge.Reset() + slowStoreTriggerLimitGauge.Reset() + storeSlowTrendEvictedStatusGauge.Reset() + balanceRangeGauge.Reset() } // AddSchedulerHandler adds the HTTP handler for a scheduler. diff --git a/pkg/statistics/hot_peer_cache.go b/pkg/statistics/hot_peer_cache.go index cde5eb20b8..e4e690064f 100644 --- a/pkg/statistics/hot_peer_cache.go +++ b/pkg/statistics/hot_peer_cache.go @@ -554,15 +554,27 @@ func (f *HotPeerCache) gc() { for _, storeID := range f.cluster.GetStores() { stores[storeID.GetID()] = struct{}{} } + // calcHotThresholds can populate thresholdsOfStore for a store that never becomes + // hot enough to enter peersOfStore, so peersOfStore alone can miss it; check the + // union of both. + removed := make(map[uint64]struct{}) for storeID := range f.peersOfStore { if _, ok := stores[storeID]; !ok { - delete(f.peersOfStore, storeID) - delete(f.regionsOfStore, storeID) - delete(f.thresholdsOfStore, storeID) - delete(f.metrics, storeID) - hotCacheStatusGauge.DeletePartialMatch(prometheus.Labels{"store": storeTag(storeID), "type": f.kind.String()}) + removed[storeID] = struct{}{} } } + for storeID := range f.thresholdsOfStore { + if _, ok := stores[storeID]; !ok { + removed[storeID] = struct{}{} + } + } + for storeID := range removed { + delete(f.peersOfStore, storeID) + delete(f.regionsOfStore, storeID) + delete(f.thresholdsOfStore, storeID) + delete(f.metrics, storeID) + hotCacheStatusGauge.DeletePartialMatch(prometheus.Labels{"store": storeTag(storeID), "type": f.kind.String()}) + } // remove expired items for _, peers := range f.peersOfStore { regions := peers.RemoveExpired() diff --git a/server/cluster/cluster.go b/server/cluster/cluster.go index 3e646f9de3..b1d2995759 100644 --- a/server/cluster/cluster.go +++ b/server/cluster/cluster.go @@ -218,9 +218,7 @@ type RaftCluster struct { onStoreBuried atomic.Pointer[func(storeID string)] } -// SetOnStoreBuried sets the callback invoked when a store is buried -// (transitions to tombstone), for per-store cleanup owned by a package that -// server/cluster cannot import. +// SetOnStoreBuried sets the callback invoked when a store is buried. func (c *RaftCluster) SetOnStoreBuried(fn func(storeID string)) { c.onStoreBuried.Store(&fn) } diff --git a/server/cluster/scheduling_controller.go b/server/cluster/scheduling_controller.go index 8276dcb086..17c348600f 100644 --- a/server/cluster/scheduling_controller.go +++ b/server/cluster/scheduling_controller.go @@ -61,14 +61,10 @@ type schedulingController struct { running bool // recentlyTombstonedStores is the set of store IDs collectSchedulingMetrics saw - // tombstoned on its last tick. Only that method, driven by - // runSchedulingMetricsCollectionJob's own single-goroutine ticker, touches it, so - // it needs no lock. Filter counters are the only metrics that still need this: - // every scheduler's Schedule() keeps running StoreStateFilter against a - // tombstoned-but-known store every cycle, which is what increments them, so a - // one-shot delete at bury time gets undone almost immediately. Heartbeat-stream - // metrics no longer need it -- BuryStoreLocked's cleanup plus the IsRemoved - // check in heartbeat_streams.go's run loop stop those writes at bury time. + // tombstoned on its last tick, needed to catch a store that gets fully removed + // between two ticks so its filter counters still get one final delete. Only + // collectSchedulingMetrics, via runSchedulingMetricsCollectionJob's + // single-goroutine ticker, touches it, so it needs no lock. recentlyTombstonedStores map[uint64]struct{} } @@ -144,8 +140,10 @@ func (sc *schedulingController) runStatsBackgroundJobs() { defer ticker.Stop() for _, store := range sc.GetStores() { - storeID := store.GetID() - sc.hotStat.GetOrCreateRollingStoreStats(storeID) + if store.IsRemoved() { + continue + } + sc.hotStat.GetOrCreateRollingStoreStats(store.GetID()) } for { select { @@ -196,13 +194,10 @@ func resetSchedulingMetrics() { func (sc *schedulingController) collectSchedulingMetrics() { statsMap := statistics.NewStoreStatisticsMap(sc.opt) stores := sc.GetStores() - // Unlike the other per-store cleanup called once from BuryStoreLocked, filter - // metrics for a tombstoned store keep getting rewritten by unrelated, ongoing - // activity for as long as the store stays known: every scheduler's Schedule() - // still runs StoreStateFilter against it every cycle, and that rejection is - // what increments the filter counters. A one-shot delete at bury time gets - // undone almost immediately, so delete unconditionally here on every tick - // instead. + // Filter counters need repeated cleanup: schedulers keep rejecting a + // tombstoned-but-known store via StoreStateFilter every cycle, so a one-shot + // delete at bury time doesn't stick. Deleting already-gone labels is a cheap + // no-op, so repeating it every tick is safe. current := make(map[uint64]struct{}) for _, s := range stores { statsMap.Observe(s) @@ -213,12 +208,9 @@ func (sc *schedulingController) collectSchedulingMetrics() { filter.DeleteStoreMetrics(strconv.FormatUint(storeID, 10)) } } - // A store that was tombstoned as of the last sweep but isn't known at all this - // tick was fully removed in between. The loop above only reaches stores - // GetStores() currently returns, so a write landing in that gap -- after the - // last sweep saw it tombstoned but before removal completed -- would otherwise - // be unreachable by any future sweep. One more delete call closes that window; - // it's a no-op if nothing was actually rewritten. + // A store fully removed between two ticks drops out of GetStores() before this + // sweep can catch it there; delete once more for anything recentlyTombstonedStores + // still remembers but current no longer has. for storeID := range sc.recentlyTombstonedStores { if _, stillKnown := current[storeID]; !stillKnown { filter.DeleteStoreMetrics(strconv.FormatUint(storeID, 10)) From 963392667f8ddaf296236ced3685cd7b7040ffef Mon Sep 17 00:00:00 2001 From: bufferflies <1045931706@qq.com> Date: Mon, 17 Aug 2026 04:29:41 +0200 Subject: [PATCH 10/16] statistics: keep tombstone-count metrics visible across bury ResetStoreStatistics used to blanket-delete every clusterStatusGauge series for a store at bury time, including store_tombstone_count and the other status-count fields that observe() already refreshes every tick regardless of tombstone state. That created a transient gap (up to one collection interval) where ops couldn't see how many stores were currently tombstoned. Narrow the deletion to just the six fields observe() stops touching once a store is removed (region/leader/witness/learner count, storage size/capacity), which is what actually needed the bury-time cleanup. Co-Authored-By: Claude Sonnet 5 Signed-off-by: bufferflies <1045931706@qq.com> --- pkg/statistics/store_collection.go | 23 ++++++++++++++++++++++- 1 file changed, 22 insertions(+), 1 deletion(-) diff --git a/pkg/statistics/store_collection.go b/pkg/statistics/store_collection.go index ac45aa3fc0..1456e9a47b 100644 --- a/pkg/statistics/store_collection.go +++ b/pkg/statistics/store_collection.go @@ -18,6 +18,8 @@ import ( "fmt" "strconv" + "github.com/prometheus/client_golang/prometheus" + "github.com/pingcap/kvproto/pkg/metapb" "github.com/tikv/pd/pkg/core" @@ -67,6 +69,23 @@ var storeStatuses = []string{ clusterStatusStoreRemovedCount, } +// storeStats are the clusterStatusGauge sub-metrics that observe() stops +// refreshing once a store is tombstoned (it returns before reaching them), so +// they'd otherwise keep showing their last pre-tombstone value indefinitely. +// storeStatuses above is deliberately not included here: those are refreshed +// unconditionally on every observe() regardless of tombstone state, and +// deleting store_tombstone_count/store_removed_count at bury time would only +// make them vanish from dashboards until the next collection tick -- ops +// needs to see how many stores are currently tombstoned without that gap. +var storeStats = []string{ + clusterStatusRegionCount, + clusterStatusLeaderCount, + clusterStatusWitnessCount, + clusterStatusLearnerCount, + clusterStatusStorageSize, + clusterStatusStorageCapacity, +} + type storeStatistics struct { opt config.ConfProvider LabelCounter map[string][]uint64 @@ -296,7 +315,9 @@ func (s *storeStatistics) collect() { // previous address. func ResetStoreStatistics(id string) { storeStatusGauge.DeletePartialMatch(utils.SingleLabel("store", id)) - clusterStatusGauge.DeletePartialMatch(utils.SingleLabel("store", id)) + for _, m := range storeStats { + clusterStatusGauge.DeletePartialMatch(prometheus.Labels{"type": m, "store": id}) + } } type storeStatisticsMap struct { From e1e18bb6776272430a272056f3cfeb63b12ebf10 Mon Sep 17 00:00:00 2001 From: bufferflies <1045931706@qq.com> Date: Mon, 17 Aug 2026 04:50:48 +0200 Subject: [PATCH 11/16] schedulers: reset balanceWitnessCounter alongside the other per-store vectors ResetSchedulerMetrics's D fix missed balanceWitnessCounter, which is also store-labeled (type, store) and left stale on an inactive replica after a scheduling-job stop, same as the vectors already covered. Co-Authored-By: Claude Sonnet 5 Signed-off-by: bufferflies <1045931706@qq.com> --- pkg/schedule/schedulers/scheduler_controller.go | 1 + 1 file changed, 1 insertion(+) diff --git a/pkg/schedule/schedulers/scheduler_controller.go b/pkg/schedule/schedulers/scheduler_controller.go index b3c318c5fd..401b80b207 100644 --- a/pkg/schedule/schedulers/scheduler_controller.go +++ b/pkg/schedule/schedulers/scheduler_controller.go @@ -147,6 +147,7 @@ func ResetSchedulerMetrics() { regionLabelStatusGauge.Reset() opInfluenceStatus.Reset() hotSchedulerResultCounter.Reset() + balanceWitnessCounter.Reset() balanceDirectionCounter.Reset() hotDirectionCounter.Reset() evictedSlowStoreStatusGauge.Reset() From be431db524ac47824cab787a7bd23b31a31f8baf Mon Sep 17 00:00:00 2001 From: bufferflies <1045931706@qq.com> Date: Mon, 17 Aug 2026 05:33:19 +0200 Subject: [PATCH 12/16] tests/integrations/client: don't let TestGetStore tombstone a shared fixture store stores[0] (via peers[0]) is heartbeated by TestScanRegions and TestScatterRegion, which expect it to stay live for the rest of the suite. TestGetStore destructively transitions it through offline -> tombstone with no cleanup, and SetupTest only resets the region cache. Once the periodic node-state-check job (10s) buries it, RegionHeartbeat's IsRemoved() guard silently drops those tests' heartbeats and they time out waiting for region state that never arrives -- a race, not the flakiness it looked like. stores[3] is never referenced by peers[] or any other test, so use that instead. Co-Authored-By: Claude Sonnet 5 Signed-off-by: bufferflies <1045931706@qq.com> --- tests/integrations/client/client_test.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/tests/integrations/client/client_test.go b/tests/integrations/client/client_test.go index ee90426bfe..3fe6bba994 100644 --- a/tests/integrations/client/client_test.go +++ b/tests/integrations/client/client_test.go @@ -1268,7 +1268,11 @@ func (suite *clientStatelessTestSuite) TestGetStore() { re := suite.Require() cluster := suite.srv.GetRaftCluster() re.NotNil(cluster) - store := stores[0] + // Use stores[3]: this test destructively transitions it through + // offline -> tombstone, and stores[0..2] are the ones other tests in + // this suite heartbeat regions through (peers[0..2]) and expect to + // stay live for the rest of the suite. + store := stores[3] // Get an up store should be OK. n, err := suite.client.GetStore(context.Background(), store.GetId()) From 2190b39057cdb1a7f5c2b37e2b31925ca8a0255e Mon Sep 17 00:00:00 2001 From: bufferflies <1045931706@qq.com> Date: Mon, 17 Aug 2026 05:33:25 +0200 Subject: [PATCH 13/16] statistics: gc tombstoned stores from HotPeerCache at bury time gc()'s "still exists" set included any store GetStores() returns, which keeps a tombstoned-but-not-yet-fully-removed store's entries alive until remove-tombstone runs. Nothing writes region heartbeats for a tombstoned store anymore, so there's no reason to wait: treat IsRemoved() the same as absent so gc() cleans it up on the next tick after bury instead of after full removal. Co-Authored-By: Claude Sonnet 5 Signed-off-by: bufferflies <1045931706@qq.com> --- pkg/statistics/hot_peer_cache.go | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/pkg/statistics/hot_peer_cache.go b/pkg/statistics/hot_peer_cache.go index e4e690064f..7d659fde3f 100644 --- a/pkg/statistics/hot_peer_cache.go +++ b/pkg/statistics/hot_peer_cache.go @@ -549,10 +549,16 @@ func (f *HotPeerCache) gc() { return } f.lastGCTime = time.Now() - // remove tombstone stores + // remove tombstone stores. GetStores() still returns a tombstoned store + // until it's fully removed, so treat IsRemoved() the same as absent here + // -- nothing writes region heartbeats for it anymore, so there's no + // reason to wait for full removal before cleaning it up. stores := make(map[uint64]struct{}) - for _, storeID := range f.cluster.GetStores() { - stores[storeID.GetID()] = struct{}{} + for _, store := range f.cluster.GetStores() { + if store.IsRemoved() { + continue + } + stores[store.GetID()] = struct{}{} } // calcHotThresholds can populate thresholdsOfStore for a store that never becomes // hot enough to enter peersOfStore, so peersOfStore alone can miss it; check the From f29419500a3269b9d31774cd9cb54f7934f59042 Mon Sep 17 00:00:00 2001 From: bufferflies <1045931706@qq.com> Date: Mon, 17 Aug 2026 09:07:53 +0200 Subject: [PATCH 14/16] schedule/filter: don't count filter rejections for tombstoned stores SelectSourceStores/SelectTargetStores/SelectUnavailableTargetStores counted every filter rejection, including StoreStateFilter's rejection of a tombstoned-but-known store -- an expected, uninteresting reason for an operator. Counter buffers rejections until Flush(), so a periodic delete of the resulting series could always be undone by a Flush() that ran later, turning "clean it up every tick" into a delete/recreate race instead of an actual fix. Skip counting a rejection caused by IsRemoved() instead: filter counters no longer get written for a tombstoned store at all, so the periodic sweep that existed only to fight this is no longer needed -- remove it from both collectSchedulingMetrics and collectMetrics, along with the recentlyTombstonedStores tracking it required. Bury-time and full-removal-time cleanup (already in place) are sufficient now. Co-Authored-By: Claude Sonnet 5 Signed-off-by: bufferflies <1045931706@qq.com> --- pkg/mcs/scheduling/server/cluster.go | 26 ---------------- pkg/schedule/filter/filters.go | 41 ++++++++++++++++--------- server/cluster/scheduling_controller.go | 26 ---------------- 3 files changed, 26 insertions(+), 67 deletions(-) diff --git a/pkg/mcs/scheduling/server/cluster.go b/pkg/mcs/scheduling/server/cluster.go index 66b8f385c8..b4ca4e5974 100644 --- a/pkg/mcs/scheduling/server/cluster.go +++ b/pkg/mcs/scheduling/server/cluster.go @@ -95,13 +95,6 @@ type Cluster struct { pdLeader atomic.Value running atomic.Bool - // recentlyTombstonedStores is the set of store IDs collectMetrics saw tombstoned - // on its last tick, needed to catch a store that gets fully removed between two - // ticks so its filter counters still get one final delete. Only collectMetrics, - // via runMetricsCollectionJob's single-goroutine ticker, touches it, so it needs - // no lock. - recentlyTombstonedStores map[uint64]struct{} - backendAddress string httpClient *http.Client @@ -735,29 +728,10 @@ func (c *Cluster) runMetricsCollectionJob() { func (c *Cluster) collectMetrics() { statsMap := statistics.NewStoreStatisticsMap(c.persistConfig) stores := c.GetStores() - // Filter counters need repeated cleanup: schedulers keep rejecting a - // tombstoned-but-known store via StoreStateFilter every cycle, so a one-shot - // delete at bury time doesn't stick. Deleting already-gone labels is a cheap - // no-op, so repeating it every tick is safe. - current := make(map[uint64]struct{}) for _, s := range stores { statsMap.Observe(s) statistics.ObserveHotStat(s, c.hotStat.StoresStats) - if s.IsRemoved() { - storeID := s.GetID() - current[storeID] = struct{}{} - filter.DeleteStoreMetrics(strconv.FormatUint(storeID, 10)) - } - } - // A store fully removed between two ticks drops out of GetStores() before this - // sweep can catch it there; delete once more for anything recentlyTombstonedStores - // still remembers but current no longer has. - for storeID := range c.recentlyTombstonedStores { - if _, stillKnown := current[storeID]; !stillKnown { - filter.DeleteStoreMetrics(strconv.FormatUint(storeID, 10)) - } } - c.recentlyTombstonedStores = current statsMap.Collect() c.coordinator.GetSchedulersController().CollectSchedulerMetrics() diff --git a/pkg/schedule/filter/filters.go b/pkg/schedule/filter/filters.go index 550cadfee8..1baad57203 100644 --- a/pkg/schedule/filter/filters.go +++ b/pkg/schedule/filter/filters.go @@ -39,12 +39,19 @@ func SelectSourceStores(stores []*core.StoreInfo, filters []Filter, conf config. return slice.AllOf(filters, func(i int) bool { status := filters[i].Source(conf, s) if !status.IsOK() { - if counter != nil { - counter.inc(source, filters[i].Type(), s.GetID()) - } else { - sourceID := strconv.FormatUint(s.GetID(), 10) - // TODO: pre-allocate gauge metrics - filterSourceCounter.WithLabelValues(filters[i].Scope(), filters[i].Type().String(), sourceID).Inc() + // A tombstoned store is rejected here on every scheduling cycle for as + // long as it stays known, so counting it would either leak (nothing + // ever calls Counter.Flush again to zero it) or fight with tombstone + // cleanup deleting the series between Flush calls. It's not an + // actionable rejection reason for an operator either way. + if !s.IsRemoved() { + if counter != nil { + counter.inc(source, filters[i].Type(), s.GetID()) + } else { + sourceID := strconv.FormatUint(s.GetID(), 10) + // TODO: pre-allocate gauge metrics + filterSourceCounter.WithLabelValues(filters[i].Scope(), filters[i].Type().String(), sourceID).Inc() + } } if collector != nil { collector.Collect(plan.SetResource(s), plan.SetStatus(status)) @@ -64,10 +71,12 @@ func SelectUnavailableTargetStores(stores []*core.StoreInfo, filters []Filter, c return slice.AnyOf(filters, func(i int) bool { status := filters[i].Target(conf, s) if !status.IsOK() { - if counter != nil { - counter.inc(target, filters[i].Type(), s.GetID()) - } else { - filterTargetCounter.WithLabelValues(filters[i].Scope(), filters[i].Type().String(), targetID).Inc() + if !s.IsRemoved() { + if counter != nil { + counter.inc(target, filters[i].Type(), s.GetID()) + } else { + filterTargetCounter.WithLabelValues(filters[i].Scope(), filters[i].Type().String(), targetID).Inc() + } } if collector != nil { @@ -92,11 +101,13 @@ func SelectTargetStores(stores []*core.StoreInfo, filters []Filter, conf config. filter := filters[i] status := filter.Target(conf, s) if !status.IsOK() { - if counter != nil { - counter.inc(target, filter.Type(), s.GetID()) - } else { - targetIDStr := strconv.FormatUint(s.GetID(), 10) - filterTargetCounter.WithLabelValues(filter.Scope(), filter.Type().String(), targetIDStr).Inc() + if !s.IsRemoved() { + if counter != nil { + counter.inc(target, filter.Type(), s.GetID()) + } else { + targetIDStr := strconv.FormatUint(s.GetID(), 10) + filterTargetCounter.WithLabelValues(filter.Scope(), filter.Type().String(), targetIDStr).Inc() + } } if collector != nil { collector.Collect(plan.SetResource(s), plan.SetStatus(status)) diff --git a/server/cluster/scheduling_controller.go b/server/cluster/scheduling_controller.go index 17c348600f..d5ea168f9d 100644 --- a/server/cluster/scheduling_controller.go +++ b/server/cluster/scheduling_controller.go @@ -59,13 +59,6 @@ type schedulingController struct { hotStat *statistics.HotStat slowStat *statistics.SlowStat running bool - - // recentlyTombstonedStores is the set of store IDs collectSchedulingMetrics saw - // tombstoned on its last tick, needed to catch a store that gets fully removed - // between two ticks so its filter counters still get one final delete. Only - // collectSchedulingMetrics, via runSchedulingMetricsCollectionJob's - // single-goroutine ticker, touches it, so it needs no lock. - recentlyTombstonedStores map[uint64]struct{} } // newSchedulingController creates a new scheduling controller. @@ -194,29 +187,10 @@ func resetSchedulingMetrics() { func (sc *schedulingController) collectSchedulingMetrics() { statsMap := statistics.NewStoreStatisticsMap(sc.opt) stores := sc.GetStores() - // Filter counters need repeated cleanup: schedulers keep rejecting a - // tombstoned-but-known store via StoreStateFilter every cycle, so a one-shot - // delete at bury time doesn't stick. Deleting already-gone labels is a cheap - // no-op, so repeating it every tick is safe. - current := make(map[uint64]struct{}) for _, s := range stores { statsMap.Observe(s) statistics.ObserveHotStat(s, sc.hotStat.StoresStats) - if s.IsRemoved() { - storeID := s.GetID() - current[storeID] = struct{}{} - filter.DeleteStoreMetrics(strconv.FormatUint(storeID, 10)) - } - } - // A store fully removed between two ticks drops out of GetStores() before this - // sweep can catch it there; delete once more for anything recentlyTombstonedStores - // still remembers but current no longer has. - for storeID := range sc.recentlyTombstonedStores { - if _, stillKnown := current[storeID]; !stillKnown { - filter.DeleteStoreMetrics(strconv.FormatUint(storeID, 10)) - } } - sc.recentlyTombstonedStores = current statsMap.Collect() sc.coordinator.GetSchedulersController().CollectSchedulerMetrics() sc.coordinator.CollectHotSpotMetrics() From 87635524b0c3106b09d55e956c1df4149742abbc Mon Sep 17 00:00:00 2001 From: bufferflies <1045931706@qq.com> Date: Mon, 17 Aug 2026 09:08:00 +0200 Subject: [PATCH 15/16] statistics: skip tombstoned peers in HotPeerCache.CheckPeerFlow CheckPeerFlow iterates every peer of a region, not just the leader, calling calcHotThresholds for each one regardless of that peer's store state. A store doesn't disappear from a region's peer list the instant it's tombstoned -- that requires a separate raft config change -- so a live leader's next heartbeat for a region that still lists a just-tombstoned follower recreates the thresholdsOfStore entry and hotCacheStatusGauge series gc() just cleaned up. Skip a peer whose store is known and IsRemoved(); a store the cluster doesn't know about yet (e.g. an in-flight add-peer target) is a different case and isn't skipped. Co-Authored-By: Claude Sonnet 5 Signed-off-by: bufferflies <1045931706@qq.com> --- pkg/statistics/hot_peer_cache.go | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/pkg/statistics/hot_peer_cache.go b/pkg/statistics/hot_peer_cache.go index 7d659fde3f..e1ee927344 100644 --- a/pkg/statistics/hot_peer_cache.go +++ b/pkg/statistics/hot_peer_cache.go @@ -177,6 +177,15 @@ func (f *HotPeerCache) CheckPeerFlow(region *core.RegionInfo, peers []*metapb.Pe stats := make([]*HotPeerStat, 0, len(peers)) for _, peer := range peers { storeID := peer.GetStoreId() + // A tombstoned store can still show up as a peer here: the leader reporting + // this region may not have caught up with a raft config change removing it + // yet. Skip it so gc() cleaning up its entries at bury time doesn't get + // undone by the very next heartbeat from this region's (live) leader. A + // store the cluster doesn't know about yet is a different case (e.g. a + // target store for an in-flight add-peer) and isn't skipped here. + if store := f.cluster.GetStore(storeID); store != nil && store.IsRemoved() { + continue + } oldItem := f.getOldHotPeerStat(regionID, storeID) // check whether the peer is allowed to be inherited From 8ceab2c825d288f99899a8252b1d61d682c07523 Mon Sep 17 00:00:00 2001 From: bufferflies <1045931706@qq.com> Date: Mon, 17 Aug 2026 09:08:18 +0200 Subject: [PATCH 16/16] server, mcs: reject region-buckets reports whose leader is tombstoned processRegionBuckets updated the region's bucket data unconditionally, and the gRPC handlers still recorded a "success" series with an empty store label for a tombstoned leader instead of skipping the report entirely. Reject at the source: processRegionBuckets now no-ops when the region's current leader store is known and tombstoned, and the handler's IsRemoved() branch skips metric recording and processing for that message instead of falling through with empty labels. Co-Authored-By: Claude Sonnet 5 Signed-off-by: bufferflies <1045931706@qq.com> --- pkg/mcs/scheduling/server/cluster.go | 3 +++ pkg/mcs/scheduling/server/grpc_service.go | 2 +- server/cluster/cluster.go | 3 +++ server/grpc_service.go | 2 +- 4 files changed, 8 insertions(+), 2 deletions(-) diff --git a/pkg/mcs/scheduling/server/cluster.go b/pkg/mcs/scheduling/server/cluster.go index b4ca4e5974..c9b46bed0b 100644 --- a/pkg/mcs/scheduling/server/cluster.go +++ b/pkg/mcs/scheduling/server/cluster.go @@ -925,6 +925,9 @@ func (c *Cluster) processRegionBuckets(buckets *metapb.Buckets) error { if region == nil { return errors.Errorf("region %v not found", buckets.GetRegionId()) } + if store := c.GetStore(region.GetLeader().GetStoreId()); store != nil && store.IsRemoved() { + return nil + } // use CAS to update the bucket information. // the two request(A:3,B:2) get the same region and need to update the buckets. // the A will pass the check and set the version to 3, the B will fail because the region.bucket has changed. diff --git a/pkg/mcs/scheduling/server/grpc_service.go b/pkg/mcs/scheduling/server/grpc_service.go index c2d9432d59..a567547f41 100644 --- a/pkg/mcs/scheduling/server/grpc_service.go +++ b/pkg/mcs/scheduling/server/grpc_service.go @@ -226,7 +226,7 @@ func (s *Service) RegionBuckets(stream schedulingpb.Scheduling_RegionBucketsServ // So we should not return error here. log.Debug("the store of the bucket in region is not found", zap.Uint64("region-id", buckets.GetRegionId())) } else if store.IsRemoved() { - log.Debug("the store of the bucket in region is tombstone", zap.Uint64("region-id", buckets.GetRegionId()), zap.Uint64("store-id", store.GetID())) + continue } else { storeLabel = strconv.FormatUint(store.GetID(), 10) storeAddress = store.GetAddress() diff --git a/server/cluster/cluster.go b/server/cluster/cluster.go index b1d2995759..ced728f4a4 100644 --- a/server/cluster/cluster.go +++ b/server/cluster/cluster.go @@ -1298,6 +1298,9 @@ func (c *RaftCluster) processRegionBuckets(buckets *metapb.Buckets) error { core.RegionCacheMissCounter.Inc() return errors.Errorf("region %v not found", buckets.GetRegionId()) } + if store := c.GetStore(region.GetLeader().GetStoreId()); store != nil && store.IsRemoved() { + return nil + } // use CAS to update the bucket information. // the two request(A:3,B:2) get the same region and need to update the buckets. // the A will pass the check and set the version to 3, the B will fail because the region.bucket has changed. diff --git a/server/grpc_service.go b/server/grpc_service.go index 2949102ca9..0ab1fb4e4a 100644 --- a/server/grpc_service.go +++ b/server/grpc_service.go @@ -1190,7 +1190,7 @@ func (s *GrpcServer) ReportBuckets(stream pdpb.PD_ReportBucketsServer) error { // So we should not return error here. log.Debug("the store of the bucket in region is not found", zap.Uint64("region-id", buckets.GetRegionId())) } else if store.IsRemoved() { - log.Debug("the store of the bucket in region is tombstone", zap.Uint64("region-id", buckets.GetRegionId()), zap.Uint64("store-id", store.GetID())) + continue } else { storeLabel = strconv.FormatUint(store.GetID(), 10) storeAddress = store.GetAddress()