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

Pick the pure Prometheus-metric-leak subset of tikv#11166 (still open on
master) onto this branch, on top of tikv#11127's backport:

- ObserveHotStat / ResetStoreStatistics / Reset(): stop
  storeStatusGauge from being republished by an in-flight
  StoreHeartbeat after bury, clean up placementStatusGauge, reset
  StoreLimitGauge on a full leader-election reset.
- collectHotMetrics: gate hasHotLeader/hasHotPeer on a single
  IsRemoved() read so a tombstoned store's stale HotPeerCache data
  can't republish hotSpotStatusGauge between HotPeerCache.gc() ticks.
- SetStoreLimit: reject setting a limit on an already-tombstoned
  store, closing the only other write path that could re-add a
  cleared StoreLimitGauge/config entry.
- summaryPendingInfluence: re-check each store fresh through the
  cluster instead of trusting the possibly-stale StoreSummaryInfo
  snapshot before writing HotPendingSum.

Left out (not applicable to this branch): tikv#11166's evict_slow_store.go
/ adjustNetworkSlowStore guards (network-slow-store eviction doesn't
exist here) and its memory-leak-only fixes (region rule fit cache,
storesOfRegion reverse index, StoreHistoryLoads GC), which are a
separate concern from metric leakage and tracked separately.

Signed-off-by: bufferflies <1045931706@qq.com>

Author: bufferflies

pkg/schedule/coordinator.go

@@ -507,7 +507,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)
@@ -521,6 +529,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)
@@ -546,7 +555,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/schedulers/hot_region.go

@@ -27,6 +27,8 @@ import (
 	"github.com/pingcap/kvproto/pkg/pdpb"
 	"github.com/pingcap/log"
 	"github.com/prometheus/client_golang/prometheus"
+	"go.uber.org/zap"
+
 	"github.com/tikv/pd/pkg/core"
 	"github.com/tikv/pd/pkg/core/constant"
 	"github.com/tikv/pd/pkg/errs"
@@ -41,7 +43,6 @@ import (
 	"github.com/tikv/pd/pkg/statistics/utils"
 	"github.com/tikv/pd/pkg/utils/keyutil"
 	"github.com/tikv/pd/pkg/utils/syncutil"
-	"go.uber.org/zap"
 )
 
 const (
@@ -111,7 +112,7 @@ func newBaseHotScheduler(
 // 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)
+	s.summaryPendingInfluence(cluster, storeInfos)
 	storesLoads := cluster.GetStoresLoads()
 	isTraceRegionFlow := cluster.GetSchedulerConfig().IsTraceRegionFlow()
 
@@ -156,7 +157,7 @@ func (s *baseHotScheduler) updateHistoryLoadConfig(sampleDuration, sampleInterva
 // 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) {
 	for id, p := range s.regionPendings {
 		for _, from := range p.froms {
 			from := storeInfos[from]
@@ -179,6 +180,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,9 +22,12 @@ import (
 	"time"
 
 	"github.com/docker/go-units"
+	promtestutil "github.com/prometheus/client_golang/prometheus/testutil"
+
 	"github.com/pingcap/kvproto/pkg/metapb"
 	"github.com/pingcap/kvproto/pkg/pdpb"
 	"github.com/stretchr/testify/require"
+
 	"github.com/tikv/pd/pkg/core"
 	"github.com/tikv/pd/pkg/mock/mockcluster"
 	"github.com/tikv/pd/pkg/schedule/operator"
@@ -180,7 +183,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 {
@@ -2089,7 +2092,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))
@@ -2114,7 +2117,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))
@@ -2982,6 +2985,43 @@ func TestEncodeConfig(t *testing.T) {
 	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)
+	// storeInfos holds a stale snapshot still showing the store as serving,
+	// simulating one taken before the store was buried -- this only passes
+	// if the check re-reads the store fresh through cluster instead of
+	// trusting info.IsRemoved() on the snapshot.
+	stale := core.NewStoreInfo(&metapb.Store{
+		Id:        storeID,
+		NodeState: metapb.NodeState_Serving,
+	})
+	loads := make([]float64, utils.RegionStatCount)
+	loads[utils.RegionWriteBytes] = 1
+	storeInfos := map[uint64]*statistics.StoreSummaryInfo{
+		storeID: {
+			StoreInfo:  stale,
+			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))
+}
+
 func TestBucketFirstStat(t *testing.T) {
 	re := require.New(t)
 	testdata := []struct {

pkg/statistics/store_collection.go

@@ -151,6 +151,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)
@@ -272,6 +281,11 @@ func (s *storeStatistics) collect() {
 // previous address.
 func ResetStoreStatistics(id string) {
 	storeStatusGauge.DeletePartialMatch(prometheus.Labels{"store": id})
+	// placementStatusGauge's "name" label is an arbitrary, unbounded label-rule
+	// name (unlike clusterStatusGauge's small, fixed engine set below), so an
+	// exact DeleteLabelValues per combination isn't feasible here either --
+	// same tradeoff as storeStatusGauge above.
+	placementStatusGauge.DeletePartialMatch(prometheus.Labels{"store": id})
 	// mcs never cleaned StoreLimitGauge on its own: unlike the classic path's
 	// RemoveStoreLimit, the mcs scheduling service has no store-limit config
 	// of its own to persist a removal for. Deleting it here, alongside
@@ -309,6 +323,7 @@ func Reset() {
 	storeStatusGauge.Reset()
 	clusterStatusGauge.Reset()
 	placementStatusGauge.Reset()
+	StoreLimitGauge.Reset()
 	ResetRegionStatsMetrics()
 	ResetLabelStatsMetrics()
 	ResetHotCacheStatusMetrics()

pkg/statistics/store_collection_test.go

@@ -20,9 +20,13 @@ import (
 	"time"
 
 	"github.com/docker/go-units"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+
 	"github.com/pingcap/kvproto/pkg/metapb"
 	"github.com/pingcap/kvproto/pkg/pdpb"
 	"github.com/stretchr/testify/require"
+
 	"github.com/tikv/pd/pkg/core"
 	"github.com/tikv/pd/pkg/core/constant"
 	"github.com/tikv/pd/pkg/mock/mockconfig"
@@ -95,6 +99,22 @@ func TestStoreStatistics(t *testing.T) {
 	re.Len(stats.LabelCounter["zone:unknown"], 2)
 }
 
+func TestResetStoreStatisticsClearsPlacementStatusGauge(t *testing.T) {
+	re := require.New(t)
+	defer placementStatusGauge.Reset()
+
+	metric := placementStatusGauge.WithLabelValues("label-type", "label-name", "1")
+	metric.Set(1)
+	re.NotZero(testutil.ToFloat64(metric))
+
+	ResetStoreStatistics("1")
+	// DeletePartialMatch returns how many series it found and removed, so a
+	// zero return here proves ResetStoreStatistics already deleted it --
+	// unlike checking WithLabelValues' value, which would recreate a fresh
+	// (zero-valued) series regardless of whether the old one was cleaned up.
+	re.Zero(placementStatusGauge.DeletePartialMatch(prometheus.Labels{"store": "1"}))
+}
+
 func TestSummaryStoreInfos(t *testing.T) {
 	re := require.New(t)
 	rw := utils.Read

server/cluster/cluster.go

@@ -2473,6 +2473,18 @@ func (c *RaftCluster) SetExternalTS(timestamp uint64) error {
 
 // SetStoreLimit sets a store limit for a given type and rate.
 func (c *RaftCluster) SetStoreLimit(storeID uint64, typ storelimit.Type, ratePerMin float64) error {
+	// A tombstoned store's config entry is only ever cleared once, at bury time
+	// (RemoveStoreLimit); nothing sweeps it again afterward. Without this check,
+	// setting a limit for an already-tombstoned store re-adds it, and
+	// StoreLimitGauge stays republished for it until final removal.
+	//
+	// GetStore returning nil is deliberately NOT treated the same as removed
+	// here: callers legitimately set a store's limit before the store itself
+	// is registered (see testCluster.addRegionStore), so nil just means
+	// "not created yet," not "already gone."
+	if store := c.GetStore(storeID); store != nil && store.IsRemoved() {
+		return errs.ErrStoreRemoved.FastGenByArgs(storeID)
+	}
 	old := c.opt.GetScheduleConfig().Clone()
 	c.opt.SetStoreLimit(storeID, typ, ratePerMin)
 	if err := c.opt.Persist(c.storage); err != nil {

tests/server/cluster/cluster_test.go

@@ -32,6 +32,9 @@ import (
 	"github.com/pingcap/kvproto/pkg/pdpb"
 	"github.com/pingcap/kvproto/pkg/replication_modepb"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
 	"github.com/tikv/pd/pkg/core"
 	"github.com/tikv/pd/pkg/core/storelimit"
 	"github.com/tikv/pd/pkg/dashboard"
@@ -55,8 +58,6 @@ import (
 	"github.com/tikv/pd/server/config"
 	"github.com/tikv/pd/tests"
 	"github.com/tikv/pd/tests/server/api"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
 )
 
 const (
@@ -474,8 +475,20 @@ func testStateAndLimit(re *require.Assertions, clusterID uint64, rc *cluster.Raf
 	// prepare
 	storeID := store.GetId()
 	oc := rc.GetOperatorController()
-	rc.SetStoreLimit(storeID, storelimit.AddPeer, 60)
-	rc.SetStoreLimit(storeID, storelimit.RemovePeer, 60)
+	// The store can be left tombstoned by a previous call to this helper (the
+	// tombstone beforeState block runs it twice on the same store). Production
+	// never un-tombstones a store, so don't fake that transition here either --
+	// these SetStoreLimit calls only exist to seed a limit before resetStoreState
+	// below establishes the state this specific case actually wants to test, and
+	// resetStoreState's own Tombstone branch clears any limit anyway, so seeding
+	// one is pointless (and rejected by SetStoreLimit) once the store is already
+	// tombstoned.
+	if store := rc.GetStore(storeID); store == nil || !store.IsRemoved() {
+		err := rc.SetStoreLimit(storeID, storelimit.AddPeer, 60)
+		re.NoError(err)
+		err = rc.SetStoreLimit(storeID, storelimit.RemovePeer, 60)
+		re.NoError(err)
+	}
 	op := operator.NewTestOperator(2, &metapb.RegionEpoch{}, operator.OpRegion, operator.AddPeer{ToStore: storeID, PeerID: 3})
 	oc.AddOperator(op)
 	op = operator.NewTestOperator(2, &metapb.RegionEpoch{}, operator.OpRegion, operator.RemovePeer{FromStore: storeID})