*: delete per-store metrics when a store is tombstoned

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)
 	}

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
 		}
 

pkg/mcs/scheduling/server/cluster.go

@@ -21,6 +21,7 @@ import (
 	"io"
 	"net/http"
 	"runtime"
+	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -47,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"
@@ -324,6 +326,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() {
@@ -745,6 +751,8 @@ func resetMetrics() {
 	statistics.Reset()
 	schedulers.ResetSchedulerMetrics()
 	schedule.ResetHotSpotMetrics()
+	filter.ResetFilterMetrics()
+	hbstream.ResetHeartbeatStreamMetrics()
 }
 
 // StartBackgroundJobs starts background jobs.
@@ -917,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.

pkg/mcs/scheduling/server/grpc_service.go

@@ -16,6 +16,7 @@ package server
 
 import (
 	"context"
+	"fmt"
 	"io"
 	"net/http"
 	"strconv"
@@ -149,6 +150,10 @@ func (s *Service) RegionHeartbeat(stream schedulingpb.Scheduling_RegionHeartbeat
 		if store == nil {
 			return errors.Errorf("invalid store ID %d, not found", storeID)
 		}
+		if store.IsRemoved() {
+			log.Debug("skip region heartbeat from tombstone store", zap.Uint64("store-id", storeID))
+			continue
+		}
 
 		storeAddress := store.GetAddress()
 		storeLabel := strconv.FormatUint(storeID, 10)
@@ -220,6 +225,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() {
+			continue
 		} else {
 			storeLabel = strconv.FormatUint(store.GetID(), 10)
 			storeAddress = store.GetAddress()
@@ -257,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()

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,10 @@ 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"
 	"github.com/tikv/pd/pkg/statistics"
 	"github.com/tikv/pd/pkg/utils/etcdutil"
 	"github.com/tikv/pd/pkg/utils/keypath"
@@ -42,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.
@@ -82,8 +94,15 @@ func (w *Watcher) initializeStoreWatcher() error {
 		}
 
 		if store.GetNodeState() == metapb.NodeState_Removed {
-			statistics.ResetStoreStatistics(store.GetAddress(), strconv.FormatUint(store.GetId(), 10))
-			// TODO: remove hot stats
+			storeIDStr := strconv.FormatUint(store.GetId(), 10)
+			statistics.ResetStoreStatistics(storeIDStr)
+			filter.DeleteStoreMetrics(storeIDStr)
+			hbstream.DeleteStoreMetrics(storeIDStr)
+			schedulers.DeleteStoreMetrics(storeIDStr)
+			schedule.DeleteStoreMetrics(storeIDStr)
+			if fn := w.onStoreTombstoned.Load(); fn != nil {
+				(*fn)(store.GetId())
+			}
 		}
 
 		return nil
@@ -96,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))
 		}
@@ -117,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()

pkg/mcs/scheduling/server/metrics.go

@@ -94,3 +94,19 @@ func init() {
 	prometheus.MustRegister(regionBucketsCounter)
 	prometheus.MustRegister(regionBucketsReportInterval)
 }
+
+// DeleteStoreMetrics deletes the per-store heartbeat/bucket metrics of a store.
+// 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)
+	regionHeartbeatCounter.DeletePartialMatch(labels)
+	regionBucketsHandleDuration.DeletePartialMatch(labels)
+	regionBucketsCounter.DeletePartialMatch(labels)
+	regionBucketsReportInterval.DeletePartialMatch(labels)
+}

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))

pkg/schedule/filter/metrics.go

@@ -37,3 +37,15 @@ 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})
+}
+
+// ResetFilterMetrics resets the filter metrics.
+func ResetFilterMetrics() {
+	filterSourceCounter.Reset()
+	filterTargetCounter.Reset()
+}

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 {

pkg/schedule/hbstream/metric.go

@@ -30,3 +30,13 @@ 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})
+}
+
+// ResetHeartbeatStreamMetrics resets the heartbeat stream metrics.
+func ResetHeartbeatStreamMetrics() {
+	heartbeatStreamCounter.Reset()
+}

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})
+}

pkg/schedule/schedulers/metrics.go

@@ -210,6 +210,22 @@ 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})
+	HotPendingSum.DeletePartialMatch(prometheus.Labels{"store": storeID})
+}
+
 func balanceLeaderCounterWithEvent(event string) prometheus.Counter {
 	return schedulerCounter.WithLabelValues(types.BalanceLeaderScheduler.String(), event)
 }