metrics: clean up metrics when a store is removed from the cluster (#1751)

fix pingcap/tidb#63245

Signed-off-by: zyguan <[email protected]>

Author: zyguan

internal/client/client.go

@@ -743,3 +743,56 @@ func (m *rpcMetrics) get(cmd tikvrpc.CmdType, stale bool, internal bool) prometh
 	}
 	return lat.(prometheus.Observer)
 }
+
+type storeMetrics struct {
+	storeID           uint64
+	rpcLatHist        *rpcMetrics
+	rpcSrcLatSum      sync.Map
+	rpcNetLatExternal prometheus.Observer
+	rpcNetLatInternal prometheus.Observer
+}
+
+func newStoreMetrics(storeID uint64) *storeMetrics {
+	store := strconv.FormatUint(storeID, 10)
+	m := &storeMetrics{
+		storeID:           storeID,
+		rpcLatHist:        deriveRPCMetrics(metrics.TiKVSendReqHistogram.MustCurryWith(prometheus.Labels{metrics.LblStore: store})),
+		rpcNetLatExternal: metrics.TiKVRPCNetLatencyHistogram.WithLabelValues(store, "false"),
+		rpcNetLatInternal: metrics.TiKVRPCNetLatencyHistogram.WithLabelValues(store, "true"),
+	}
+	return m
+}
+
+func (m *storeMetrics) updateRPCMetrics(req *tikvrpc.Request, resp *tikvrpc.Response, latency time.Duration) {
+	seconds := latency.Seconds()
+	stale := req.GetStaleRead()
+	source := req.GetRequestSource()
+	internal := util.IsInternalRequest(req.GetRequestSource())
+
+	m.rpcLatHist.get(req.Type, stale, internal).Observe(seconds)
+
+	srcLatSum, ok := m.rpcSrcLatSum.Load(source)
+	if !ok {
+		srcLatSum = deriveRPCMetrics(metrics.TiKVSendReqBySourceSummary.MustCurryWith(
+			prometheus.Labels{metrics.LblStore: strconv.FormatUint(m.storeID, 10), metrics.LblSource: source}))
+		m.rpcSrcLatSum.Store(source, srcLatSum)
+	}
+	srcLatSum.(*rpcMetrics).get(req.Type, stale, internal).Observe(seconds)
+
+	if execDetail := resp.GetExecDetailsV2(); execDetail != nil {
+		var totalRpcWallTimeNs uint64
+		if execDetail.TimeDetailV2 != nil {
+			totalRpcWallTimeNs = execDetail.TimeDetailV2.TotalRpcWallTimeNs
+		} else if execDetail.TimeDetail != nil {
+			totalRpcWallTimeNs = execDetail.TimeDetail.TotalRpcWallTimeNs
+		}
+		if totalRpcWallTimeNs > 0 {
+			lat := latency - time.Duration(totalRpcWallTimeNs)
+			if internal {
+				m.rpcNetLatInternal.Observe(lat.Seconds())
+			} else {
+				m.rpcNetLatExternal.Observe(lat.Seconds())
+			}
+		}
+	}
+}

internal/client/conn_pool.go

@@ -23,12 +23,9 @@ import (
 
 	grpc_opentracing "github.com/grpc-ecosystem/go-grpc-middleware/tracing/opentracing"
 	"github.com/pkg/errors"
-	"github.com/prometheus/client_golang/prometheus"
 	"github.com/tikv/client-go/v2/config"
 	tikverr "github.com/tikv/client-go/v2/error"
-	"github.com/tikv/client-go/v2/metrics"
 	"github.com/tikv/client-go/v2/tikvrpc"
-	"github.com/tikv/client-go/v2/util"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/backoff"
 	"google.golang.org/grpc/credentials"
@@ -55,12 +52,7 @@ type connPool struct {
 
 	monitor *connMonitor
 
-	metrics struct {
-		rpcLatHist        *rpcMetrics
-		rpcSrcLatSum      sync.Map
-		rpcNetLatExternal prometheus.Observer
-		rpcNetLatInternal prometheus.Observer
-	}
+	metrics atomic.Pointer[storeMetrics]
 }
 
 func newConnPool(maxSize uint, addr string, ver uint64, security config.Security,
@@ -74,9 +66,6 @@ func newConnPool(maxSize uint, addr string, ver uint64, security config.Security
 		dialTimeout:   dialTimeout,
 		monitor:       m,
 	}
-	a.metrics.rpcLatHist = deriveRPCMetrics(metrics.TiKVSendReqHistogram.MustCurryWith(prometheus.Labels{metrics.LblStore: addr}))
-	a.metrics.rpcNetLatExternal = metrics.TiKVRPCNetLatencyHistogram.WithLabelValues(addr, "false")
-	a.metrics.rpcNetLatInternal = metrics.TiKVRPCNetLatencyHistogram.WithLabelValues(addr, "true")
 	if err := a.Init(addr, security, idleNotify, enableBatch, eventListener, opts...); err != nil {
 		return nil, err
 	}
@@ -221,35 +210,15 @@ func (a *connPool) Close() {
 }
 
 func (a *connPool) updateRPCMetrics(req *tikvrpc.Request, resp *tikvrpc.Response, latency time.Duration) {
-	seconds := latency.Seconds()
-	stale := req.GetStaleRead()
-	source := req.GetRequestSource()
-	internal := util.IsInternalRequest(req.GetRequestSource())
-
-	a.metrics.rpcLatHist.get(req.Type, stale, internal).Observe(seconds)
-
-	srcLatSum, ok := a.metrics.rpcSrcLatSum.Load(source)
-	if !ok {
-		srcLatSum = deriveRPCMetrics(metrics.TiKVSendReqSummary.MustCurryWith(
-			prometheus.Labels{metrics.LblStore: a.target, metrics.LblSource: source}))
-		a.metrics.rpcSrcLatSum.Store(source, srcLatSum)
-	}
-	srcLatSum.(*rpcMetrics).get(req.Type, stale, internal).Observe(seconds)
-
-	if execDetail := resp.GetExecDetailsV2(); execDetail != nil {
-		var totalRpcWallTimeNs uint64
-		if execDetail.TimeDetailV2 != nil {
-			totalRpcWallTimeNs = execDetail.TimeDetailV2.TotalRpcWallTimeNs
-		} else if execDetail.TimeDetail != nil {
-			totalRpcWallTimeNs = execDetail.TimeDetail.TotalRpcWallTimeNs
-		}
-		if totalRpcWallTimeNs > 0 {
-			lat := latency - time.Duration(totalRpcWallTimeNs)
-			if internal {
-				a.metrics.rpcNetLatInternal.Observe(lat.Seconds())
-			} else {
-				a.metrics.rpcNetLatExternal.Observe(lat.Seconds())
-			}
-		}
+	m := a.metrics.Load()
+	storeID := req.Context.GetPeer().GetStoreId()
+	if m == nil || m.storeID != storeID {
+		// The client selects a connPool by addr via RPCClient.getConnPool, so it's possible that the storeID of the
+		// selected connPool is not the same as the storeID in req.Context. We need to create a new storeMetrics for the
+		// new storeID. Note that connPool.metrics just works as a cache, the metric data is stored in corresponding
+		// MetricVec, so it's ok to overwrite it here.
+		m = newStoreMetrics(storeID)
+		a.metrics.Store(m)
 	}
+	m.updateRPCMetrics(req, resp, latency)
 }

internal/locate/region_cache.go

@@ -353,6 +353,11 @@ func newRegion(bo *retry.Backoffer, c *RegionCache, pdRegion *router.Region) (*R
 		if err != nil {
 			return nil, err
 		}
+
+		if !exists {
+			updateStoreLivenessGauge(store)
+		}
+
 		// Filter out the peer on a tombstone or down store.
 		if addr == "" || slices.ContainsFunc(pdRegion.DownPeers, func(dp *metapb.Peer) bool { return isSamePeer(dp, p) }) {
 			continue
@@ -749,47 +754,11 @@ func NewRegionCache(pdClient pd.Client, opt ...RegionCacheOpt) *RegionCache {
 		// cache GC is incompatible with cache refresh
 		c.bg.schedule(c.gcRoundFunc(cleanRegionNumPerRound), cleanCacheInterval)
 	}
-	c.bg.schedule(
-		func(ctx context.Context, _ time.Time) bool {
-			refreshFullStoreList(ctx, c.stores)
-			return false
-		}, refreshStoreListInterval,
-	)
+	updater := &storeCacheUpdater{stores: c.stores}
+	c.bg.schedule(updater.tick, refreshStoreListInterval)
 	return c
 }
 
-// Try to refresh full store list. Errors are ignored.
-func refreshFullStoreList(ctx context.Context, stores storeCache) {
-	storeList, err := stores.fetchAllStores(ctx)
-	if err != nil {
-		logutil.Logger(ctx).Info("refresh full store list failed", zap.Error(err))
-		return
-	}
-	for _, store := range storeList {
-		_, exist := stores.get(store.GetId())
-		if exist {
-			continue
-		}
-		// GetAllStores is supposed to return only Up and Offline stores.
-		// This check is being defensive and to make it consistent with store resolve code.
-		if store == nil || store.GetState() == metapb.StoreState_Tombstone {
-			continue
-		}
-		addr := store.GetAddress()
-		if addr == "" {
-			continue
-		}
-		s := stores.getOrInsertDefault(store.GetId())
-		// TODO: maybe refactor this, together with other places initializing Store
-		s.addr = addr
-		s.peerAddr = store.GetPeerAddress()
-		s.saddr = store.GetStatusAddress()
-		s.storeType = tikvrpc.GetStoreTypeByMeta(store)
-		s.labels = store.GetLabels()
-		s.changeResolveStateTo(unresolved, resolved)
-	}
-}
-
 // only used fot test.
 func newTestRegionCache() *RegionCache {
 	c := &RegionCache{}
@@ -2760,6 +2729,7 @@ func (c *RegionCache) UpdateBucketsIfNeeded(regionID RegionVerID, latestBucketsV
 const cleanCacheInterval = time.Second
 const cleanRegionNumPerRound = 50
 const refreshStoreListInterval = 10 * time.Second
+const cleanStoreMetricsInterval = time.Minute
 
 // gcScanItemHook is only used for testing
 var gcScanItemHook = new(atomic.Pointer[func(*btreeItem)])