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,11 @@ func (manager *RegionRuleFitCacheManager) SetCache(region *core.RegionInfo, fit
 		}
 		return
 	}
-	manager.regionCaches[region.GetID()] = manager.toRegionRuleFitCache(region, fit)
+	newCache, cacheable := manager.toRegionRuleFitCache(storeSet, region, fit)
+	if !cacheable {
+		return
+	}
+	manager.regionCaches[region.GetID()] = newCache
 }
 
 // regionRuleFitCache stores regions RegionFit result and involving variables
@@ -139,14 +150,15 @@ 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, bool) {
+	storeCacheList, cacheable := manager.toStoreCacheList(storeSet, fit.regionStores)
 	return &regionRuleFitCache{
 		region:       toRegionCache(region),
-		regionStores: manager.toStoreCacheList(fit.regionStores),
+		regionStores: storeCacheList,
 		rules:        toRuleCacheList(fit.rules),
 		bestFit:      nil,
 		hitCount:     0,
-	}
+	}, cacheable
 }
 
 type ruleCache struct {
@@ -190,8 +202,21 @@ 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, cacheable bool) {
+	cacheable = true
 	for _, s := range stores {
+		// The stores slice can be a stale RegionInfo/FitRegion snapshot taken
+		// before a store was buried, so re-check the store fresh through
+		// storeSet rather than trusting s.IsRemoved(). If any current store is
+		// missing or removed, the caller must not persist a region-level cache
+		// entry either -- one built from this stale snapshot would otherwise
+		// look valid (region/rule/store-state comparisons all pass) and never
+		// get re-evaluated until an unrelated region-level change invalidates
+		// it.
+		current := storeSet.GetStore(s.GetID())
+		if current == nil || current.IsRemoved() {
+			cacheable = false
+		}
 		sCache, ok := manager.storeCaches[s.GetID()]
 		if !ok || !sCache.storeEqual(s) {
 			m := make(map[string]string)
@@ -203,11 +228,15 @@ 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.
+			if current != nil && !current.IsRemoved() {
+				manager.storeCaches[s.GetID()] = sCache
+			}
 		}
 		c = append(c, sCache)
 	}
-	return c
+	return c, cacheable
 }
 
 func labelEqual(label1 map[string]string, label2 []*metapb.StoreLabel) bool {

pkg/schedule/placement/region_rule_cache_test.go

@@ -216,10 +216,64 @@ func TestPublicStoreCaches(t *testing.T) {
 	}
 }
 
+func TestToStoreCacheListDoesNotReinsertStaleStore(t *testing.T) {
+	re := require.New(t)
+	manager := NewRegionRuleFitCacheManager()
+	live := core.NewStoreInfo(&metapb.Store{
+		Id:        1,
+		NodeState: metapb.NodeState_Serving,
+	})
+	// storeSet reflects the live, current cluster state; the stores slice
+	// passed to toStoreCacheList simulates a FitRegion call still holding an
+	// older StoreInfo snapshot taken before the store was buried.
+	storeSet := core.NewStoresInfo()
+	storeSet.PutStore(live)
+
+	manager.toStoreCacheList(storeSet, []*core.StoreInfo{live})
+	manager.RemoveStoreCache(1)
+
+	removed := live.Clone(core.SetStoreState(metapb.StoreState_Tombstone))
+	storeSet.PutStore(removed)
+	manager.toStoreCacheList(storeSet, []*core.StoreInfo{live}) // stale FitRegion snapshot
+
+	_, ok := manager.storeCaches[1]
+	re.False(ok)
+}
+
+func TestToStoreCacheListReturnsUncacheableForRemovedStore(t *testing.T) {
+	re := require.New(t)
+	manager := NewRegionRuleFitCacheManager()
+	live := core.NewStoreInfo(&metapb.Store{
+		Id:        1,
+		NodeState: metapb.NodeState_Serving,
+	})
+	storeSet := core.NewStoresInfo()
+	storeSet.PutStore(live)
+
+	_, cacheable := manager.toStoreCacheList(storeSet, []*core.StoreInfo{live})
+	re.True(cacheable)
+
+	// storeSet now shows the store removed while the stores slice still holds
+	// the stale (pre-bury) snapshot -- SetCache must not persist a
+	// region-level cache built from this call, or it would look valid
+	// (region/rule/store-state comparisons all pass against the same stale
+	// snapshot) and never get re-evaluated until an unrelated region-level
+	// change invalidates it.
+	removed := live.Clone(core.SetStoreState(metapb.StoreState_Tombstone))
+	storeSet.PutStore(removed)
+	_, cacheable = manager.toStoreCacheList(storeSet, []*core.StoreInfo{live})
+	re.False(cacheable)
+}
+
 func (manager *RegionRuleFitCacheManager) mockRegionRuleFitCache(region *core.RegionInfo, rules []*Rule, regionStores []*core.StoreInfo) *regionRuleFitCache {
+	storeSet := core.NewStoresInfo()
+	for _, s := range regionStores {
+		storeSet.PutStore(s)
+	}
+	storeCacheList, _ := manager.toStoreCacheList(storeSet, regionStores)
 	return &regionRuleFitCache{
 		region:       toRegionCache(region),
-		regionStores: manager.toStoreCacheList(regionStores),
+		regionStores: storeCacheList,
 		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())
-	s.summaryPendingInfluence(storeInfos)
+	stores := cluster.GetStores()
+	storeInfos := statistics.SummaryStoreInfos(stores)
+	s.summaryPendingInfluence(cluster, storeInfos)
+	s.stHistoryLoads.GC(stores)
 	storesLoads := cluster.GetStoresLoads()
 	isTraceRegionFlow := cluster.GetSchedulerConfig().IsTraceRegionFlow()
 
@@ -160,7 +162,7 @@ func (s *baseHotScheduler) getEffectivePendingWeight() float64 {
 // summaryPendingInfluence calculate the summary of pending Influence for each store
 // and clean the region from regionInfluence if they have ended operator.
 // It makes each dim rate or count become `weight` times to the origin value.
-func (s *baseHotScheduler) summaryPendingInfluence(storeInfos map[uint64]*statistics.StoreSummaryInfo) {
+func (s *baseHotScheduler) summaryPendingInfluence(cluster sche.SchedulerCluster, storeInfos map[uint64]*statistics.StoreSummaryInfo) {
 	pendingWeight := s.getEffectivePendingWeight()
 	for id, p := range s.regionPendings {
 		for _, from := range p.froms {
@@ -185,6 +187,14 @@ func (s *baseHotScheduler) summaryPendingInfluence(storeInfos map[uint64]*statis
 	}
 	// for metrics
 	for storeID, info := range storeInfos {
+		// storeInfos is built from a snapshot taken at the top of
+		// prepareForBalance, so info.IsRemoved() can be stale by the time
+		// this loop runs; re-check the store fresh through cluster instead,
+		// otherwise a store buried after the snapshot was taken but before
+		// this write can still recreate HotPendingSum for it.
+		if store := cluster.GetStore(storeID); store == nil || store.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"
@@ -242,7 +243,7 @@ func checkGCPendingOpInfos(re *require.Assertions, enablePlacementRules bool) {
 	}
 
 	storeInfos := statistics.SummaryStoreInfos(tc.GetStores())
-	hb.summaryPendingInfluence(storeInfos) // Calling this function will GC.
+	hb.summaryPendingInfluence(tc, storeInfos) // Calling this function will GC.
 
 	for i := range opInfluenceCreators {
 		for j, typ := range typs {
@@ -2092,7 +2093,7 @@ func TestInfluenceByRWType(t *testing.T) {
 	re.NotNil(op)
 
 	storeInfos := statistics.SummaryStoreInfos(tc.GetStores())
-	hb.(*hotScheduler).summaryPendingInfluence(storeInfos)
+	hb.(*hotScheduler).summaryPendingInfluence(tc, storeInfos)
 	re.True(nearlyAbout(storeInfos[1].PendingSum.Loads[utils.RegionWriteKeys], -0.5*units.MiB))
 	re.True(nearlyAbout(storeInfos[1].PendingSum.Loads[utils.RegionWriteBytes], -0.5*units.MiB))
 	re.True(nearlyAbout(storeInfos[4].PendingSum.Loads[utils.RegionWriteKeys], 0.5*units.MiB))
@@ -2117,7 +2118,7 @@ func TestInfluenceByRWType(t *testing.T) {
 	re.NotNil(op)
 
 	storeInfos = statistics.SummaryStoreInfos(tc.GetStores())
-	hb.(*hotScheduler).summaryPendingInfluence(storeInfos)
+	hb.(*hotScheduler).summaryPendingInfluence(tc, storeInfos)
 	// assert read/write influence is the sum of write peer and write leader
 	re.True(nearlyAbout(storeInfos[1].PendingSum.Loads[utils.RegionWriteKeys], -1.2*units.MiB))
 	re.True(nearlyAbout(storeInfos[1].PendingSum.Loads[utils.RegionWriteBytes], -1.2*units.MiB))
@@ -2649,3 +2650,32 @@ 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()
+
+	cancel, _, tc, _ := prepareSchedulersTest()
+	defer cancel()
+	hb := newBaseHotScheduler(nil, 0, 0, initHotRegionScheduleConfig())
+	storeID := uint64(1)
+	removed := core.NewStoreInfo(&metapb.Store{
+		Id:        storeID,
+		NodeState: metapb.NodeState_Removed,
+	})
+	tc.PutStore(removed)
+	loads := make([]float64, utils.RegionStatCount)
+	loads[utils.RegionWriteBytes] = 1
+	storeInfos := map[uint64]*statistics.StoreSummaryInfo{
+		storeID: {
+			StoreInfo:  removed,
+			PendingSum: &statistics.Influence{Loads: loads},
+		},
+	}
+
+	metric := HotPendingSum.WithLabelValues("1", utils.Write.String(), utils.DimToString(utils.ByteDim))
+	metric.Set(42)
+	hb.summaryPendingInfluence(tc, 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()