server: existing statistics will not be deleted when putting store && check when store status changing (#3200)

server/cluster/cluster.go

@@ -289,7 +289,7 @@ func (c *RaftCluster) LoadClusterInfo() (*RaftCluster, error) {
 		zap.Duration("cost", time.Since(start)),
 	)
 	for _, store := range c.GetStores() {
-		c.storesStats.CreateRollingStoreStats(store.GetID())
+		c.storesStats.GetOrCreateRollingStoreStats(store.GetID())
 	}
 	return c, nil
 }
@@ -884,27 +884,31 @@ func (c *RaftCluster) UpdateStoreLabels(storeID uint64, labels []*metapb.StoreLa
 	newStore := proto.Clone(store.GetMeta()).(*metapb.Store)
 	newStore.Labels = labels
 	// PutStore will perform label merge.
-	err := c.PutStore(newStore, force)
-	return err
+	return c.putStoreImpl(newStore, force)
 }
 
 // PutStore puts a store.
+func (c *RaftCluster) PutStore(store *metapb.Store) error {
+	if err := c.putStoreImpl(store, false); err != nil {
+		return err
+	}
+	c.OnStoreVersionChange()
+	c.AddStoreLimit(store)
+	return nil
+}
+
+// putStoreImpl puts a store.
 // If 'force' is true, then overwrite the store's labels.
-func (c *RaftCluster) PutStore(store *metapb.Store, force bool) error {
+func (c *RaftCluster) putStoreImpl(store *metapb.Store, force bool) error {
 	c.Lock()
 	defer c.Unlock()
 
 	if store.GetId() == 0 {
 		return errors.Errorf("invalid put store %v", store)
 	}
 
-	v, err := ParseVersion(store.GetVersion())
-	if err != nil {
-		return errors.Errorf("invalid put store %v, error: %s", store, err)
-	}
-	clusterVersion := *c.opt.GetClusterVersion()
-	if !IsCompatible(clusterVersion, *v) {
-		return errors.Errorf("version should compatible with version  %s, got %s", clusterVersion, v)
+	if err := c.checkStoreVersion(store); err != nil {
+		return err
 	}
 
 	// Store address can not be the same as other stores.
@@ -938,12 +942,24 @@ func (c *RaftCluster) PutStore(store *metapb.Store, force bool) error {
 			core.SetStoreDeployPath(store.DeployPath),
 		)
 	}
-	if err = c.checkStoreLabels(s); err != nil {
+	if err := c.checkStoreLabels(s); err != nil {
 		return err
 	}
 	return c.putStoreLocked(s)
 }
 
+func (c *RaftCluster) checkStoreVersion(store *metapb.Store) error {
+	v, err := ParseVersion(store.GetVersion())
+	if err != nil {
+		return errors.Errorf("invalid put store %v, error: %s", store, err)
+	}
+	clusterVersion := *c.opt.GetClusterVersion()
+	if !IsCompatible(clusterVersion, *v) {
+		return errors.Errorf("version should compatible with version  %s, got %s", clusterVersion, v)
+	}
+	return nil
+}
+
 func (c *RaftCluster) checkStoreLabels(s *core.StoreInfo) error {
 	if c.opt.IsPlacementRulesEnabled() {
 		return nil
@@ -1066,6 +1082,12 @@ func (c *RaftCluster) SetStoreState(storeID uint64, state metapb.StoreState) err
 		return errs.ErrStoreNotFound.FastGenByArgs(storeID)
 	}
 
+	if store.GetState() == metapb.StoreState_Tombstone && state != metapb.StoreState_Tombstone {
+		if err := c.checkStoreVersion(store.GetMeta()); err != nil {
+			return err
+		}
+	}
+
 	newStore := store.Clone(core.SetStoreState(state))
 	log.Warn("store update state",
 		zap.Uint64("store-id", storeID),
@@ -1102,7 +1124,7 @@ func (c *RaftCluster) putStoreLocked(store *core.StoreInfo) error {
 		}
 	}
 	c.core.PutStore(store)
-	c.storesStats.CreateRollingStoreStats(store.GetID())
+	c.storesStats.GetOrCreateRollingStoreStats(store.GetID())
 	return nil
 }
 
@@ -1731,7 +1753,12 @@ func (c *RaftCluster) GetAllStoresLimit() map[uint64]config.StoreLimitConfig {
 
 // AddStoreLimit add a store limit for a given store ID.
 func (c *RaftCluster) AddStoreLimit(store *metapb.Store) {
+	storeID := store.GetId()
 	cfg := c.opt.GetScheduleConfig().Clone()
+	if _, ok := cfg.StoreLimit[storeID]; ok {
+		return
+	}
+
 	sc := config.StoreLimitConfig{
 		AddPeer:    config.DefaultStoreLimit.GetDefaultStoreLimit(storelimit.AddPeer),
 		RemovePeer: config.DefaultStoreLimit.GetDefaultStoreLimit(storelimit.RemovePeer),
@@ -1742,7 +1769,7 @@ func (c *RaftCluster) AddStoreLimit(store *metapb.Store) {
 			RemovePeer: config.DefaultTiFlashStoreLimit.GetDefaultStoreLimit(storelimit.RemovePeer),
 		}
 	}
-	storeID := store.GetId()
+
 	cfg.StoreLimit[storeID] = sc
 	c.opt.SetScheduleConfig(cfg)
 }

server/cluster/cluster_test.go

@@ -48,7 +48,7 @@ func (s *testClusterInfoSuite) TestStoreHeartbeat(c *C) {
 	cluster := newTestRaftCluster(mockid.NewIDAllocator(), opt, core.NewStorage(kv.NewMemoryKV()), core.NewBasicCluster())
 
 	n, np := uint64(3), uint64(3)
-	stores := newTestStores(n)
+	stores := newTestStores(n, "2.0.0")
 	storeMetasAfterHeartbeat := make([]*metapb.Store, 0, n)
 	regions := newTestRegions(n, np)
 
@@ -96,7 +96,7 @@ func (s *testClusterInfoSuite) TestFilterUnhealthyStore(c *C) {
 	c.Assert(err, IsNil)
 	cluster := newTestRaftCluster(mockid.NewIDAllocator(), opt, core.NewStorage(kv.NewMemoryKV()), core.NewBasicCluster())
 
-	stores := newTestStores(3)
+	stores := newTestStores(3, "2.0.0")
 	for _, store := range stores {
 		storeStats := &pdpb.StoreStats{
 			StoreId:     store.GetID(),
@@ -123,14 +123,38 @@ func (s *testClusterInfoSuite) TestFilterUnhealthyStore(c *C) {
 	}
 }
 
+func (s *testClusterInfoSuite) TestSetStoreState(c *C) {
+	_, opt, err := newTestScheduleConfig()
+	c.Assert(err, IsNil)
+	cluster := newTestRaftCluster(mockid.NewIDAllocator(), opt, core.NewStorage(kv.NewMemoryKV()), core.NewBasicCluster())
+
+	// Put 4 stores.
+	for _, store := range newTestStores(4, "2.0.0") {
+		c.Assert(cluster.PutStore(store.GetMeta()), IsNil)
+	}
+	// Store 3 and 4 offline normally.
+	for _, id := range []uint64{3, 4} {
+		c.Assert(cluster.RemoveStore(id), IsNil)
+		c.Assert(cluster.BuryStore(id, false), IsNil)
+	}
+	// Change the status of 3 directly back to Up.
+	c.Assert(cluster.SetStoreState(3, metapb.StoreState_Up), IsNil)
+	// Update store 1 2 3
+	for _, store := range newTestStores(3, "3.0.0") {
+		c.Assert(cluster.PutStore(store.GetMeta()), IsNil)
+	}
+	// Since the store 4 version is too low, setting it to Up should fail.
+	c.Assert(cluster.SetStoreState(4, metapb.StoreState_Up), NotNil)
+}
+
 func (s *testClusterInfoSuite) TestRegionHeartbeat(c *C) {
 	_, opt, err := newTestScheduleConfig()
 	c.Assert(err, IsNil)
 	cluster := newTestRaftCluster(mockid.NewIDAllocator(), opt, core.NewStorage(kv.NewMemoryKV()), core.NewBasicCluster())
 
 	n, np := uint64(3), uint64(3)
 
-	stores := newTestStores(3)
+	stores := newTestStores(3, "2.0.0")
 	regions := newTestRegions(n, np)
 
 	for _, store := range stores {
@@ -512,7 +536,7 @@ func (s *testClusterInfoSuite) TestUpdateStorePendingPeerCount(c *C) {
 	_, opt, err := newTestScheduleConfig()
 	c.Assert(err, IsNil)
 	tc := newTestCluster(opt)
-	stores := newTestStores(5)
+	stores := newTestStores(5, "2.0.0")
 	for _, s := range stores {
 		c.Assert(tc.putStoreLocked(s), IsNil)
 	}
@@ -549,7 +573,7 @@ type testStoresInfoSuite struct{}
 func (s *testStoresInfoSuite) TestStores(c *C) {
 	n := uint64(10)
 	cache := core.NewStoresInfo()
-	stores := newTestStores(n)
+	stores := newTestStores(n, "2.0.0")
 
 	for i, store := range stores {
 		id := store.GetID()
@@ -698,7 +722,7 @@ func (s *testGetStoresSuite) SetUpSuite(c *C) {
 	cluster := newTestRaftCluster(mockid.NewIDAllocator(), opt, core.NewStorage(kv.NewMemoryKV()), core.NewBasicCluster())
 	s.cluster = cluster
 
-	stores := newTestStores(200)
+	stores := newTestStores(200, "2.0.0")
 
 	for _, store := range stores {
 		c.Assert(s.cluster.putStoreLocked(store), IsNil)
@@ -739,11 +763,15 @@ func newTestRaftCluster(id id.Allocator, opt *config.PersistOptions, storage *co
 }
 
 // Create n stores (0..n).
-func newTestStores(n uint64) []*core.StoreInfo {
+func newTestStores(n uint64, version string) []*core.StoreInfo {
 	stores := make([]*core.StoreInfo, 0, n)
 	for i := uint64(1); i <= n; i++ {
 		store := &metapb.Store{
-			Id: i,
+			Id:         i,
+			Address:    fmt.Sprintf("127.0.0.1:%d", i),
+			State:      metapb.StoreState_Up,
+			Version:    version,
+			DeployPath: fmt.Sprintf("test/store%d", i),
 		}
 		stores = append(stores, core.NewStoreInfo(store))
 	}

server/grpc_service.go

@@ -230,14 +230,12 @@ func (s *Server) PutStore(ctx context.Context, request *pdpb.PutStoreRequest) (*
 		return nil, status.Errorf(codes.FailedPrecondition, "placement rules is disabled")
 	}
 
-	if err := rc.PutStore(store, false); err != nil {
+	if err := rc.PutStore(store); err != nil {
 		return nil, status.Errorf(codes.Unknown, err.Error())
 	}
 
 	log.Info("put store ok", zap.Stringer("store", store))
-	rc.OnStoreVersionChange()
 	CheckPDVersion(s.persistOptions)
-	rc.AddStoreLimit(store)
 
 	return &pdpb.PutStoreResponse{
 		Header:            s.header(),

server/statistics/store.go

@@ -40,13 +40,6 @@ func NewStoresStats() *StoresStats {
 	}
 }
 
-// CreateRollingStoreStats creates RollingStoreStats with a given store ID.
-func (s *StoresStats) CreateRollingStoreStats(storeID uint64) {
-	s.Lock()
-	defer s.Unlock()
-	s.rollingStoresStats[storeID] = newRollingStoreStats()
-}
-
 // RemoveRollingStoreStats removes RollingStoreStats with a given store ID.
 func (s *StoresStats) RemoveRollingStoreStats(storeID uint64) {
 	s.Lock()

server/statistics/store_collection_test.go

@@ -45,7 +45,7 @@ func (t *testStoreStatisticsSuite) TestStoreStatistics(c *C) {
 	stores := make([]*core.StoreInfo, 0, len(metaStores))
 	for _, m := range metaStores {
 		s := core.NewStoreInfo(m, core.SetLastHeartbeatTS(time.Now()))
-		storesStats.CreateRollingStoreStats(m.GetId())
+		storesStats.GetOrCreateRollingStoreStats(m.GetId())
 		stores = append(stores, s)
 	}
 

tests/server/cluster/cluster_test.go

@@ -367,7 +367,7 @@ func (s *clusterTestSuite) TestRaftClusterMultipleRestart(c *C) {
 	store := newMetaStore(storeID, "127.0.0.1:4", "2.1.0", metapb.StoreState_Offline, fmt.Sprintf("test/store%d", storeID))
 	rc := leaderServer.GetRaftCluster()
 	c.Assert(rc, NotNil)
-	err = rc.PutStore(store, false)
+	err = rc.PutStore(store)
 	c.Assert(err, IsNil)
 	c.Assert(tc, NotNil)