Merge branch 'master' of github.com:tikv/pd into feat/router_grpc_impl

Author: bufferflies

client/clients/tso/stream_test.go

@@ -29,6 +29,7 @@ import (
 	"github.com/pingcap/log"
 
 	"github.com/tikv/pd/client/errs"
+	"github.com/tikv/pd/client/pkg/utils/testutil"
 )
 
 const mockStreamURL = "mock:///"
@@ -356,10 +357,11 @@ func (s *testTSOStreamSuite) TestTSOStreamBasic() {
 
 	// After an error from the (simulated) RPC stream, the tsoStream should be in a broken status and can't accept
 	// new request anymore.
-	err := s.stream.processRequests(1, 2, 3, 1, time.Now(), func(_result tsoRequestResult, _reqKeyspaceGroupID uint32, _err error) {
-		panic("unreachable")
+	testutil.Eventually(s.re, func() bool {
+		return s.stream.processRequests(1, 2, 3, 1, time.Now(), func(_result tsoRequestResult, _reqKeyspaceGroupID uint32, err error) {
+			s.re.Error(err)
+		}) != nil
 	})
-	s.re.Error(err)
 }
 
 func (s *testTSOStreamSuite) testTSOStreamBrokenImpl(err error, pendingRequests int) {

pkg/core/store.go

@@ -63,6 +63,7 @@ type StoreInfo struct {
 	pauseLeaderTransferOut atomic.Int64 // not allow to be used as source of transfer leader
 	slowStoreEvicted       atomic.Int64 // this store has been evicted as a slow store, should not transfer leader to it
 	slowTrendEvicted       atomic.Int64 // this store has been evicted as a slow store by trend, should not transfer leader to it
+	stoppingStoreEvicted   atomic.Int64 // this store has been evicted as a stopping store, should not transfer leader to it
 	leaderCount            int
 	regionCount            int
 	learnerCount           int
@@ -137,6 +138,7 @@ func (s *StoreInfo) Clone(opts ...StoreCreateOption) *StoreInfo {
 	store.pauseLeaderTransferOut.Store(s.pauseLeaderTransferOut.Load())
 	store.slowStoreEvicted.Store(s.slowStoreEvicted.Load())
 	store.slowTrendEvicted.Store(s.slowTrendEvicted.Load())
+	store.stoppingStoreEvicted.Store(s.stoppingStoreEvicted.Load())
 	for _, opt := range opts {
 		if opt != nil {
 			opt(store)
@@ -183,6 +185,7 @@ func (s *StoreInfo) ShallowClone(opts ...StoreCreateOption) *StoreInfo {
 	store.pauseLeaderTransferOut.Store(s.pauseLeaderTransferOut.Load())
 	store.slowStoreEvicted.Store(s.slowStoreEvicted.Load())
 	store.slowTrendEvicted.Store(s.slowTrendEvicted.Load())
+	store.stoppingStoreEvicted.Store(s.stoppingStoreEvicted.Load())
 	for _, opt := range opts {
 		opt(store)
 	}
@@ -208,7 +211,7 @@ func (s *StoreInfo) EvictedAsSlowStore() bool {
 
 // EvictedAsStoppingStore returns if the store should be evicted as a stopping store.
 func (s *StoreInfo) EvictedAsStoppingStore() bool {
-	return s.rawStats.IsStopping
+	return s.stoppingStoreEvicted.Load() > 0
 }
 
 // IsEvictedAsSlowTrend returns if the store should be evicted as a slow store by trend.

pkg/core/store_option.go

@@ -146,7 +146,7 @@ func SlowStoreEvicted() StoreCreateOption {
 // leader to the store
 func StoppingStoreEvicted() StoreCreateOption {
 	return func(store *StoreInfo) {
-		store.rawStats.IsStopping = true
+		store.stoppingStoreEvicted.Add(1)
 	}
 }
 
@@ -175,7 +175,7 @@ func SlowStoreRecovered() StoreCreateOption {
 // StoppingStoreRecovered cleans the evicted state of a store.
 func StoppingStoreRecovered() StoreCreateOption {
 	return func(store *StoreInfo) {
-		store.rawStats.IsStopping = false
+		store.stoppingStoreEvicted.Add(-1)
 	}
 }
 

pkg/mcs/resourcemanager/server/keyspace_manager.go

@@ -278,6 +278,15 @@ func (krgm *keyspaceResourceGroupManager) getServiceLimiter() *serviceLimiter {
 	return krgm.sl
 }
 
+func (krgm *keyspaceResourceGroupManager) getServiceLimit() (float64, bool) {
+	krgm.RLock()
+	defer krgm.RUnlock()
+	if krgm.sl == nil || krgm.sl.ServiceLimit == 0 {
+		return 0, false
+	}
+	return krgm.sl.ServiceLimit, true
+}
+
 func (krgm *keyspaceResourceGroupManager) getOrCreateRUTracker(name string) *ruTracker {
 	rt := krgm.getRUTracker(name)
 	if rt == nil {
@@ -411,16 +420,16 @@ func (rt *ruTracker) getRUPerSec() float64 {
 // - No group exceeds its configured limits
 // - Inactive groups (no RU consumption) are skipped to avoid unnecessary computation
 func (krgm *keyspaceResourceGroupManager) conciliateFillRates() {
-	serviceLimiter := krgm.getServiceLimiter()
+	serviceLimit, isSet := krgm.getServiceLimit()
 	// No need to conciliate if the service limit is not set or is 0.
-	if serviceLimiter == nil || serviceLimiter.ServiceLimit == 0 {
+	if !isSet || serviceLimit == 0 {
 		return
 	}
 	priorityQueues := krgm.getPriorityQueues()
 	if len(priorityQueues) == 0 {
 		return
 	}
-	remainingServiceLimit := serviceLimiter.ServiceLimit
+	remainingServiceLimit := serviceLimit
 	for _, queue := range priorityQueues {
 		if len(queue) == 0 {
 			continue

server/cluster/cluster.go

@@ -1809,8 +1809,9 @@ func (c *RaftCluster) checkStores() {
 	)
 
 	for _, store := range stores {
-		isInOffline := c.checkStore(store, stores)
-		if store.IsUp() && store.IsLowSpace(c.opt.GetLowSpaceRatio()) {
+		storeID := store.GetID()
+		isInUp, isInOffline := c.checkStore(storeID)
+		if isInUp {
 			upStoreCount++
 		}
 		if isInOffline {
@@ -1826,21 +1827,27 @@ func (c *RaftCluster) checkStores() {
 	}
 }
 
-func (c *RaftCluster) checkStore(store *core.StoreInfo, stores []*core.StoreInfo) (isInOffline bool) {
+func (c *RaftCluster) checkStore(storeID uint64) (isInUp, isInOffline bool) {
+	c.storeStateLock.Lock(uint32(storeID))
+	defer c.storeStateLock.Unlock(uint32(storeID))
+
+	store := c.GetStore(storeID)
+	if store == nil {
+		log.Warn("store not found when checking, may be deleted", zap.Uint64("store-id", storeID))
+		return false, false
+	}
+
 	var (
 		regionSize = float64(store.GetRegionSize())
-		storeID    = store.GetID()
 		threshold  float64
 	)
-	c.storeStateLock.Lock(uint32(storeID))
-	defer c.storeStateLock.Unlock(uint32(storeID))
 	switch store.GetNodeState() {
 	case metapb.NodeState_Preparing:
 		readyToServe := store.GetUptime() >= c.opt.GetMaxStorePreparingTime() ||
 			c.GetTotalRegionCount() < core.InitClusterRegionThreshold
 		if !readyToServe && (c.IsPrepared() || (c.IsServiceIndependent(constant.SchedulingServiceName) && c.isStorePrepared())) {
 			kr := keyutil.NewKeyRange("", "")
-			threshold = c.getThreshold(stores, store, &kr)
+			threshold = c.getThreshold(c.GetStores(), store, &kr)
 			log.Debug("store preparing threshold", zap.Uint64("store-id", storeID),
 				zap.Float64("threshold", threshold),
 				zap.Float64("region-size", regionSize))
@@ -1886,7 +1893,11 @@ func (c *RaftCluster) checkStore(store *core.StoreInfo, stores []*core.StoreInfo
 		}
 	}
 	c.progressManager.UpdateProgress(store, regionSize, threshold)
-	return
+	// When store is preparing or serving, we think it is in up.
+	// `checkStore` only may change store state from preparing to serving.
+	// So we don't need to get store again.
+	isInUp = store.IsUp() && !store.IsLowSpace(c.opt.GetLowSpaceRatio())
+	return isInUp, isInOffline
 }
 
 func (c *RaftCluster) getThreshold(stores []*core.StoreInfo, store *core.StoreInfo, kr *keyutil.KeyRange) float64 {

server/cluster/cluster_test.go

@@ -3921,6 +3921,78 @@ func TestConcurrentStoreStats(t *testing.T) {
 	wg.Wait()
 }
 
+func TestCheckStoresUpCountWithLowSpace(t *testing.T) {
+	re := require.New(t)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	_, opt, err := newTestScheduleConfig()
+	re.NoError(err)
+	cluster := newTestRaftCluster(ctx, mockid.NewIDAllocator(), opt, storage.NewStorageWithMemoryBackend())
+	cluster.coordinator = schedule.NewCoordinator(ctx, cluster, nil)
+	cluster.ruleManager = placement.NewRuleManager(ctx, storage.NewStorageWithMemoryBackend(), cluster, cluster.GetOpts())
+	if opt.IsPlacementRulesEnabled() {
+		err := cluster.ruleManager.Initialize(opt.GetMaxReplicas(), opt.GetLocationLabels(), opt.GetIsolationLevel(), false)
+		if err != nil {
+			panic(err)
+		}
+	}
+
+	// Add stores with low space to the cluster.
+	storeCount := uint64(3)
+	for i := uint64(1); i <= storeCount; i++ {
+		store := &metapb.Store{
+			Id:            i,
+			Address:       fmt.Sprintf("mock://tikv-%d:%d", i, i),
+			StatusAddress: fmt.Sprintf("mock://tikv-%d:%d", i, i+1),
+			Version:       "2.0.0",
+			DeployPath:    getTestDeployPath(i),
+			NodeState:     metapb.NodeState_Preparing,
+		}
+		err = cluster.PutMetaStore(store)
+		re.NoError(err)
+		req := &pdpb.StoreHeartbeatRequest{}
+		resp := &pdpb.StoreHeartbeatResponse{}
+		req.Stats = &pdpb.StoreStats{
+			StoreId:     i,
+			Capacity:    100 * 1024, // 100 GiB
+			Available:   1 * 1024,   // 1 GiB
+			RegionCount: 1,
+		}
+		re.NoError(cluster.HandleStoreHeartbeat(req, resp))
+	}
+	re.Equal(int(storeCount), cluster.GetStoreCount())
+
+	upStoreCount := 0
+	for _, s := range cluster.GetStores() {
+		isUp, _ := cluster.checkStore(s.GetID())
+		if isUp {
+			upStoreCount++
+		}
+	}
+	re.Equal(0, upStoreCount, "upStoreCount should be 0 when store is low space")
+
+	// Add stores to the cluster.
+	for i := uint64(1); i <= storeCount; i++ {
+		req := &pdpb.StoreHeartbeatRequest{}
+		resp := &pdpb.StoreHeartbeatResponse{}
+		req.Stats = &pdpb.StoreStats{
+			StoreId:     i,
+			Capacity:    100 * 1024, // 100 GiB
+			Available:   50 * 1024,  // 50 GiB
+			RegionCount: 1,
+		}
+		re.NoError(cluster.HandleStoreHeartbeat(req, resp))
+	}
+	for _, s := range cluster.GetStores() {
+		isUp, _ := cluster.checkStore(s.GetID())
+		if isUp {
+			upStoreCount++
+		}
+	}
+	re.Equal(int(storeCount), upStoreCount, "upStoreCount should be 3 when store is not low space")
+}
+
 func waitAddLearner(re *require.Assertions, stream mockhbstream.HeartbeatStream, region *core.RegionInfo, storeID uint64) *core.RegionInfo {
 	var res *pdpb.RegionHeartbeatResponse
 	testutil.Eventually(re, func() bool {

tests/server/gc/gc_test.go

@@ -350,7 +350,9 @@ func TestGCOperations(t *testing.T) {
 		re.Nil(resp.Header.Error)
 		re.Equal("b1", resp.GetDeletedBarrierInfo().GetBarrierId())
 		re.Equal(uint64(20), resp.GetDeletedBarrierInfo().GetBarrierTs())
-		re.Equal(3600, int(resp.GetDeletedBarrierInfo().GetTtlSeconds()))
+		// The TTL value decreases over time
+		re.Greater(resp.GetDeletedBarrierInfo().GetTtlSeconds(), int64(3595))
+		re.LessOrEqual(resp.GetDeletedBarrierInfo().GetTtlSeconds(), int64(3600))
 	}
 
 	for _, keyspaceID := range []uint32{constant.NullKeyspaceID, ks1.Id} {