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/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 7b846a241a..c9b46bed0b 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" @@ -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. diff --git a/pkg/mcs/scheduling/server/grpc_service.go b/pkg/mcs/scheduling/server/grpc_service.go index 3cba680f03..a567547f41 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" @@ -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() diff --git a/pkg/mcs/scheduling/server/meta/watcher.go b/pkg/mcs/scheduling/server/meta/watcher.go index a1fd6defc7..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,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() diff --git a/pkg/mcs/scheduling/server/metrics.go b/pkg/mcs/scheduling/server/metrics.go index 04f94cb7c1..b54bb16573 100644 --- a/pkg/mcs/scheduling/server/metrics.go +++ b/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) +} 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/pkg/schedule/filter/metrics.go b/pkg/schedule/filter/metrics.go index 944b670133..6801c77ec0 100644 --- a/pkg/schedule/filter/metrics.go +++ b/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() +} 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/hbstream/metric.go b/pkg/schedule/hbstream/metric.go index 1620ddebce..532e207a1e 100644 --- a/pkg/schedule/hbstream/metric.go +++ b/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() +} 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/pkg/schedule/schedulers/metrics.go b/pkg/schedule/schedulers/metrics.go index 4f943571e2..cb412c922b 100644 --- a/pkg/schedule/schedulers/metrics.go +++ b/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) } diff --git a/pkg/schedule/schedulers/scheduler_controller.go b/pkg/schedule/schedulers/scheduler_controller.go index c5fec56868..401b80b207 100644 --- a/pkg/schedule/schedulers/scheduler_controller.go +++ b/pkg/schedule/schedulers/scheduler_controller.go @@ -145,6 +145,16 @@ func ResetSchedulerMetrics() { schedulerStatusGauge.Reset() ruleStatusGauge.Reset() regionLabelStatusGauge.Reset() + opInfluenceStatus.Reset() + hotSchedulerResultCounter.Reset() + balanceWitnessCounter.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_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/hot_peer_cache.go b/pkg/statistics/hot_peer_cache.go index 5edf8f505a..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 @@ -549,19 +558,38 @@ 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 + // 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) + 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/pkg/statistics/hot_peer_cache_test.go b/pkg/statistics/hot_peer_cache_test.go index 6aacae3aea..235ce5b4d6 100644 --- a/pkg/statistics/hot_peer_cache_test.go +++ b/pkg/statistics/hot_peer_cache_test.go @@ -23,6 +23,8 @@ 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" "github.com/pingcap/kvproto/pkg/metapb" @@ -822,16 +824,28 @@ 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. 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 aef38b70c3..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 @@ -290,38 +309,15 @@ 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) +// 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)) + for _, m := range storeStats { + clusterStatusGauge.DeletePartialMatch(prometheus.Labels{"type": m, "store": id}) } - clusterStatusGauge.DeletePartialMatch(utils.SingleLabel("store", id)) } type storeStatisticsMap struct { diff --git a/server/cluster/cluster.go b/server/cluster/cluster.go index 80a8a4b689..ced728f4a4 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,19 @@ 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. +func (c *RaftCluster) SetOnStoreBuried(fn func(storeID string)) { + c.onStoreBuried.Store(&fn) } // Status saves some state information. @@ -1284,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. @@ -1768,12 +1785,17 @@ 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) + schedule.DeleteStoreMetrics(storeIDStr) if !c.IsServiceIndependent(constant.SchedulingServiceName) { c.removeStoreStatistics(storeID) } + if fn := c.onStoreBuried.Load(); fn != nil { + (*fn)(storeIDStr) + } } return err } @@ -2176,7 +2198,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 f9819e8dc5..d5ea168f9d 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" @@ -28,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" @@ -131,8 +133,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 { @@ -176,6 +180,8 @@ func resetSchedulingMetrics() { statistics.ResetLabelStatsMetrics() // reset hot cache metrics statistics.ResetHotCacheStatusMetrics() + filter.ResetFilterMetrics() + hbstream.ResetHeartbeatStreamMetrics() } func (sc *schedulingController) collectSchedulingMetrics() { @@ -202,6 +208,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/grpc_service.go b/server/grpc_service.go index 22a2849c8e..0ab1fb4e4a 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() { + continue } else { storeLabel = strconv.FormatUint(store.GetID(), 10) storeAddress = store.GetAddress() @@ -1371,6 +1373,10 @@ 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() { + log.Debug("skip region heartbeat from tombstone store", zap.Uint64("store-id", storeID)) + continue + } storeAddress := store.GetAddress() regionHeartbeatCounter.WithLabelValues(storeAddress, storeLabel, "report", "recv").Inc() diff --git a/server/metrics.go b/server/metrics.go index f995cf5efa..71e06b9f8e 100644 --- a/server/metrics.go +++ b/server/metrics.go @@ -238,3 +238,19 @@ func init() { prometheus.MustRegister(forwardTsoDuration) prometheus.MustRegister(regionRequestCounter) } + +// DeleteStoreMetrics deletes the per-store heartbeat/bucket-report 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} + 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..4abc7d55f4 100644 --- a/server/server.go +++ b/server/server.go @@ -525,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, 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())