statistics, schedule: stop republishing metrics for a known-tombstoned store

pkg/mcs/scheduling/server/cluster.go

@@ -328,6 +328,7 @@ func (c *Cluster) SetRuntimeResources(
 	c.affinityWatcher = affinityWatcher
 	metaWatcher.SetOnStoreTombstoned(func(storeID uint64) {
 		c.hotStat.RemoveRollingStoreStats(storeID)
+		c.ruleManager.RemoveStoreCache(storeID)
 		DeleteStoreMetrics(strconv.FormatUint(storeID, 10))
 	})
 }

pkg/schedule/checker/rule_checker.go

@@ -132,7 +132,7 @@ func (c *RuleChecker) CheckWithFit(region *core.RegionInfo, fit *placement.Regio
 	if c.cluster.GetCheckerConfig().IsPlacementRulesCacheEnabled() {
 		if placement.ValidateFit(fit) && placement.ValidateRegion(region) && placement.ValidateStores(fit.GetRegionStores()) {
 			// If there is no need to fix, we will cache the fit
-			c.ruleManager.SetRegionFitCache(region, fit)
+			c.ruleManager.SetRegionFitCache(c.cluster, region, fit)
 			ruleCheckerSetCacheCounter.Inc()
 		}
 	}

pkg/schedule/coordinator.go

@@ -535,7 +535,15 @@ func collectHotMetrics(cluster sche.ClusterInformer, stores []*core.StoreInfo, t
 		storeAddress := s.GetAddress()
 		storeID := s.GetID()
 		storeLabel := strconv.FormatUint(storeID, 10)
+		// HotPeerCache.gc() only removes a tombstoned store from status.AsLeader/
+		// AsPeer's source data on its own TTL-throttled schedule, not every tick,
+		// so a known-tombstoned store here can still have stale hot-peer data.
+		// Treat it as not hot regardless, so the delete branches below run
+		// instead of republishing hotSpotStatusGauge every tick until
+		// HotPeerCache.gc() eventually catches up.
+		removed := s.IsRemoved()
 		stat, hasHotLeader := status.AsLeader[storeID]
+		hasHotLeader = hasHotLeader && !removed
 		if hasHotLeader {
 			hotSpotStatusGauge.WithLabelValues(storeAddress, storeLabel, "total_"+kind+"_bytes_as_leader").Set(stat.TotalBytesRate)
 			hotSpotStatusGauge.WithLabelValues(storeAddress, storeLabel, "total_"+kind+"_keys_as_leader").Set(stat.TotalKeysRate)
@@ -551,6 +559,7 @@ func collectHotMetrics(cluster sche.ClusterInformer, stores []*core.StoreInfo, t
 		}
 
 		stat, hasHotPeer := status.AsPeer[storeID]
+		hasHotPeer = hasHotPeer && !removed
 		if hasHotPeer {
 			hotSpotStatusGauge.WithLabelValues(storeAddress, storeLabel, "total_"+kind+"_bytes_as_peer").Set(stat.TotalBytesRate)
 			hotSpotStatusGauge.WithLabelValues(storeAddress, storeLabel, "total_"+kind+"_keys_as_peer").Set(stat.TotalKeysRate)
@@ -578,7 +587,7 @@ func collectHotMetrics(cluster sche.ClusterInformer, stores []*core.StoreInfo, t
 		// iteration's own s was still live: once a snapshot correctly shows
 		// IsRemoved(), a tombstoned store sitting in GetStores() for up to 30
 		// days doesn't cost a scan on every tick.
-		if !s.IsRemoved() {
+		if !removed {
 			if store := cluster.GetStore(storeID); store == nil || store.IsRemoved() {
 				DeleteStoreMetrics(storeLabel)
 			}

pkg/schedule/placement/region_rule_cache.go

@@ -53,6 +53,13 @@ func NewRegionRuleFitCacheManager() *RegionRuleFitCacheManager {
 	}
 }
 
+// RemoveStoreCache removes the store cache with a given store ID.
+func (manager *RegionRuleFitCacheManager) RemoveStoreCache(storeID uint64) {
+	manager.mu.Lock()
+	defer manager.mu.Unlock()
+	delete(manager.storeCaches, storeID)
+}
+
 // Invalid cache by regionID
 func (manager *RegionRuleFitCacheManager) Invalid(regionID uint64) {
 	manager.mu.Lock()
@@ -79,7 +86,7 @@ func (manager *RegionRuleFitCacheManager) CheckAndGetCache(region *core.RegionIn
 }
 
 // SetCache stores RegionFit cache
-func (manager *RegionRuleFitCacheManager) SetCache(region *core.RegionInfo, fit *RegionFit) {
+func (manager *RegionRuleFitCacheManager) SetCache(storeSet StoreSet, region *core.RegionInfo, fit *RegionFit) {
 	if !ValidateRegion(region) || !ValidateFit(fit) || !ValidateStores(fit.regionStores) {
 		return
 	}
@@ -92,7 +99,7 @@ func (manager *RegionRuleFitCacheManager) SetCache(region *core.RegionInfo, fit
 		}
 		return
 	}
-	manager.regionCaches[region.GetID()] = manager.toRegionRuleFitCache(region, fit)
+	manager.regionCaches[region.GetID()] = manager.toRegionRuleFitCache(storeSet, region, fit)
 }
 
 // regionRuleFitCache stores regions RegionFit result and involving variables
@@ -139,10 +146,10 @@ func storesEqual(a []*storeCache, b []*core.StoreInfo) bool {
 	})
 }
 
-func (manager *RegionRuleFitCacheManager) toRegionRuleFitCache(region *core.RegionInfo, fit *RegionFit) *regionRuleFitCache {
+func (manager *RegionRuleFitCacheManager) toRegionRuleFitCache(storeSet StoreSet, region *core.RegionInfo, fit *RegionFit) *regionRuleFitCache {
 	return &regionRuleFitCache{
 		region:       toRegionCache(region),
-		regionStores: manager.toStoreCacheList(fit.regionStores),
+		regionStores: manager.toStoreCacheList(storeSet, fit.regionStores),
 		rules:        toRuleCacheList(fit.rules),
 		bestFit:      nil,
 		hitCount:     0,
@@ -190,7 +197,7 @@ func (s storeCache) storeEqual(store *core.StoreInfo) bool {
 		labelEqual(s.labels, store.GetLabels())
 }
 
-func (manager *RegionRuleFitCacheManager) toStoreCacheList(stores []*core.StoreInfo) (c []*storeCache) {
+func (manager *RegionRuleFitCacheManager) toStoreCacheList(storeSet StoreSet, stores []*core.StoreInfo) (c []*storeCache) {
 	for _, s := range stores {
 		sCache, ok := manager.storeCaches[s.GetID()]
 		if !ok || !sCache.storeEqual(s) {
@@ -203,7 +210,16 @@ func (manager *RegionRuleFitCacheManager) toStoreCacheList(stores []*core.StoreI
 				labels:  m,
 				state:   s.GetState(),
 			}
-			manager.storeCaches[s.GetID()] = sCache
+			// A removed store's entry is only ever cleared once, when it's
+			// buried or finally removed; nothing sweeps it again after that.
+			// The stores slice can be a stale RegionInfo snapshot taken before
+			// a store was buried, so re-check the store fresh through storeSet
+			// here rather than trusting s.IsRemoved() -- otherwise a store that
+			// was buried while this same FitRegion call was still computing
+			// could get re-added and linger in storeCaches for good.
+			if current := storeSet.GetStore(s.GetID()); current != nil && !current.IsRemoved() {
+				manager.storeCaches[s.GetID()] = sCache
+			}
 		}
 		c = append(c, sCache)
 	}

pkg/schedule/placement/region_rule_cache_test.go

@@ -217,9 +217,13 @@ func TestPublicStoreCaches(t *testing.T) {
 }
 
 func (manager *RegionRuleFitCacheManager) mockRegionRuleFitCache(region *core.RegionInfo, rules []*Rule, regionStores []*core.StoreInfo) *regionRuleFitCache {
+	storeSet := core.NewStoresInfo()
+	for _, s := range regionStores {
+		storeSet.PutStore(s)
+	}
 	return &regionRuleFitCache{
 		region:       toRegionCache(region),
-		regionStores: manager.toStoreCacheList(regionStores),
+		regionStores: manager.toStoreCacheList(storeSet, regionStores),
 		rules:        toRuleCacheList(rules),
 		bestFit: &RegionFit{
 			regionStores: regionStores,

pkg/schedule/placement/rule_manager.go

@@ -453,21 +453,26 @@ func (m *RuleManager) FitRegion(storeSet StoreSet, region *core.RegionInfo) (fit
 	fit.regionStores = regionStores
 	fit.rules = rules
 	if isCached {
-		m.SetRegionFitCache(region, fit)
+		m.SetRegionFitCache(storeSet, region, fit)
 	}
 	return fit
 }
 
 // SetRegionFitCache sets RegionFitCache
-func (m *RuleManager) SetRegionFitCache(region *core.RegionInfo, fit *RegionFit) {
-	m.cache.SetCache(region, fit)
+func (m *RuleManager) SetRegionFitCache(storeSet StoreSet, region *core.RegionInfo, fit *RegionFit) {
+	m.cache.SetCache(storeSet, region, fit)
 }
 
 // InvalidCache invalids the cache.
 func (m *RuleManager) InvalidCache(regionID uint64) {
 	m.cache.Invalid(regionID)
 }
 
+// RemoveStoreCache removes the store cache with a given store ID.
+func (m *RuleManager) RemoveStoreCache(storeID uint64) {
+	m.cache.RemoveStoreCache(storeID)
+}
+
 // SetPlaceholderRegionFitCache sets a placeholder region fit cache information
 // Only used for testing
 func (m *RuleManager) SetPlaceholderRegionFitCache(region *core.RegionInfo) {

pkg/schedule/placement/rule_manager_test.go

@@ -482,7 +482,7 @@ func TestCacheManager(t *testing.T) {
 	}
 	region := core.NewRegionInfo(regionMeta, regionMeta.Peers[0])
 	fit := manager.FitRegion(stores, region)
-	manager.SetRegionFitCache(region, fit)
+	manager.SetRegionFitCache(stores, region, fit)
 	// bestFit is not stored when the total number of hits is insufficient.
 	for i := 1; i < minHitCountToCacheHit/2; i++ {
 		manager.FitRegion(stores, region)

pkg/schedule/schedulers/hot_region.go

@@ -100,8 +100,10 @@ func newBaseHotScheduler(
 // prepareForBalance calculate the summary of pending Influence for each store and prepare the load detail for
 // each store, only update read or write load detail
 func (s *baseHotScheduler) prepareForBalance(typ resourceType, cluster sche.SchedulerCluster) {
-	storeInfos := statistics.SummaryStoreInfos(cluster.GetStores())
+	stores := cluster.GetStores()
+	storeInfos := statistics.SummaryStoreInfos(stores)
 	s.summaryPendingInfluence(storeInfos)
+	s.stHistoryLoads.GC(stores)
 	storesLoads := cluster.GetStoresLoads()
 	isTraceRegionFlow := cluster.GetSchedulerConfig().IsTraceRegionFlow()
 
@@ -185,6 +187,14 @@ func (s *baseHotScheduler) summaryPendingInfluence(storeInfos map[uint64]*statis
 	}
 	// for metrics
 	for storeID, info := range storeInfos {
+		// storeInfos comes from SummaryStoreInfos(cluster.GetStores()), which
+		// includes tombstoned stores; without this check a pending influence
+		// entry still in its zombie period keeps republishing HotPendingSum
+		// for a removed store every Schedule() round, even after bury/final
+		// -removal or collectHotMetrics deletes the series.
+		if info.IsRemoved() {
+			continue
+		}
 		storeLabel := strconv.FormatUint(storeID, 10)
 		if infl := info.PendingSum; infl != nil && len(infl.Loads) != 0 {
 			utils.ForeachRegionStats(func(rwTy utils.RWType, dim int, kind utils.RegionStatKind) {

pkg/schedule/schedulers/hot_region_test.go

@@ -22,6 +22,7 @@ import (
 	"time"
 
 	"github.com/docker/go-units"
+	promtestutil "github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/stretchr/testify/require"
 
 	"github.com/pingcap/kvproto/pkg/metapb"
@@ -2649,3 +2650,28 @@ func TestEncodeConfig(t *testing.T) {
 	re.NoError(err)
 	re.NotEqual("null", string(data))
 }
+
+func TestSummaryPendingInfluenceSkipsRemovedStoreMetric(t *testing.T) {
+	re := require.New(t)
+	defer HotPendingSum.Reset()
+
+	hb := newBaseHotScheduler(nil, 0, 0, initHotRegionScheduleConfig())
+	storeID := uint64(1)
+	loads := make([]float64, utils.RegionStatCount)
+	loads[utils.RegionWriteBytes] = 1
+	storeInfos := map[uint64]*statistics.StoreSummaryInfo{
+		storeID: {
+			StoreInfo: core.NewStoreInfo(&metapb.Store{
+				Id:        storeID,
+				NodeState: metapb.NodeState_Removed,
+			}),
+			PendingSum: &statistics.Influence{Loads: loads},
+		},
+	}
+
+	metric := HotPendingSum.WithLabelValues("1", utils.Write.String(), utils.DimToString(utils.ByteDim))
+	metric.Set(42)
+	hb.summaryPendingInfluence(storeInfos)
+
+	re.Equal(float64(42), promtestutil.ToFloat64(metric))
+}

pkg/statistics/hot_peer_cache.go

@@ -584,6 +584,17 @@ func (f *HotPeerCache) gc() {
 		}
 	}
 	for storeID := range removed {
+		// regionsOfStore[storeID] is the exact set of regions this store is still
+		// referenced from in storesOfRegion; read it before deleting so the reverse
+		// index doesn't keep a stale storeID around for regions that are still active.
+		for regionID := range f.regionsOfStore[storeID] {
+			if stores, ok := f.storesOfRegion[regionID]; ok {
+				delete(stores, storeID)
+				if len(stores) == 0 {
+					delete(f.storesOfRegion, regionID)
+				}
+			}
+		}
 		delete(f.peersOfStore, storeID)
 		delete(f.regionsOfStore, storeID)
 		delete(f.thresholdsOfStore, storeID)

pkg/statistics/store_collection.go

@@ -211,6 +211,15 @@ func (s *storeStatistics) observe(store *core.StoreInfo) {
 
 // ObserveHotStat records the hot region metrics for the store.
 func ObserveHotStat(store *core.StoreInfo, stats *StoresStats) {
+	// A store's RollingStoreStats can be recreated after bury by a StoreHeartbeat
+	// that was already in flight (HandleStoreHeartbeat only rejects a fully
+	// unknown store, not a tombstoned one). Without this check, that would make
+	// this function keep republishing storeStatusGauge every collection tick for
+	// as long as the entry exists, up to 30 days until final removal, instead of
+	// stopping once the store is known tombstoned like observe() already does.
+	if store.IsRemoved() {
+		return
+	}
 	// Store flows.
 	storeAddress := store.GetAddress()
 	id := strconv.FormatUint(store.GetID(), 10)
@@ -359,6 +368,7 @@ func Reset() {
 	storeStatusGauge.Reset()
 	placementStatusGauge.Reset()
 	clusterStatusGauge.Reset()
+	StoreLimitGauge.Reset()
 	ResetRegionStatsMetrics()
 	ResetLabelStatsMetrics()
 	ResetHotCacheStatusMetrics()

pkg/statistics/store_hot_peers_infos.go

@@ -195,7 +195,14 @@ func summaryStoresLoadByEngine(
 					allStoreHistoryLoadSum[i][j] += historyLoad
 				}
 			}
-			storesHistoryLoads.Add(id, rwTy, kind, currentLoads)
+			// GC(stores) runs earlier in the same prepareForBalance call and
+			// clears a removed store's entry; without this check, this write
+			// -- unconditional on the collector's own filter, which doesn't
+			// reject removed stores for RegionKind -- recreates it in the
+			// same call, making that GC a no-op for this store.
+			if !store.IsRemoved() {
+				storesHistoryLoads.Add(id, rwTy, kind, currentLoads)
+			}
 		}
 
 		for i := range allStoreLoadSum {

pkg/statistics/store_load.go

@@ -303,6 +303,27 @@ func (s *StoreHistoryLoads) Add(storeID uint64, rwTp utils.RWType, kind constant
 	load.add(pointLoad)
 }
 
+// GC removes history load entries for stores that are no longer alive, so a
+// store that's gone doesn't keep its entry for the scheduler's entire
+// lifetime (there's otherwise no periodic sweep of this cache at all).
+func (s *StoreHistoryLoads) GC(stores []*core.StoreInfo) {
+	alive := make(map[uint64]struct{}, len(stores))
+	for _, store := range stores {
+		if !store.IsRemoved() {
+			alive[store.GetID()] = struct{}{}
+		}
+	}
+	for i := range s.loads {
+		for j := range s.loads[i] {
+			for storeID := range s.loads[i][j] {
+				if _, ok := alive[storeID]; !ok {
+					delete(s.loads[i][j], storeID)
+				}
+			}
+		}
+	}
+}
+
 // Get returns the store loads from the history, not one time point.
 // In another word, the result is [dim][time].
 func (s *StoreHistoryLoads) Get(storeID uint64, rwTp utils.RWType, kind constant.ResourceKind) HistoryLoads {

pkg/statistics/store_load_test.go

@@ -20,6 +20,9 @@ import (
 
 	"github.com/stretchr/testify/require"
 
+	"github.com/pingcap/kvproto/pkg/metapb"
+
+	"github.com/tikv/pd/pkg/core"
 	"github.com/tikv/pd/pkg/core/constant"
 	"github.com/tikv/pd/pkg/statistics/utils"
 )
@@ -61,3 +64,35 @@ func TestHistoryLoads(t *testing.T) {
 	historyLoads.Add(1, rwTp, kind, loads)
 	re.Empty(historyLoads.Get(1, rwTp, kind)[0])
 }
+
+func TestHistoryGCDoesNotReaddRemovedStore(t *testing.T) {
+	re := require.New(t)
+	history := NewStoreHistoryLoads(time.Minute, 0)
+	rw := utils.Read
+	kind := constant.RegionKind
+	storeID := uint64(1)
+
+	live := core.NewStoreInfo(&metapb.Store{
+		Id:        storeID,
+		NodeState: metapb.NodeState_Serving,
+	})
+	infos := map[uint64]*StoreSummaryInfo{
+		storeID: {StoreInfo: live},
+	}
+	loads := map[uint64]StoreKindLoads{
+		storeID: {1, 0, 0, 0, 0},
+	}
+
+	SummaryStoresLoad(infos, loads, history, nil, false, rw, kind)
+	re.NotEmpty(history.Get(storeID, rw, kind)[0])
+
+	removed := live.Clone(core.SetStoreState(metapb.StoreState_Tombstone))
+	history.GC([]*core.StoreInfo{removed})
+	re.Empty(history.Get(storeID, rw, kind)[0])
+
+	SummaryStoresLoad(
+		map[uint64]*StoreSummaryInfo{storeID: {StoreInfo: removed}},
+		loads, history, nil, false, rw, kind,
+	)
+	re.Empty(history.Get(storeID, rw, kind)[0])
+}