cluster: handle region after report split

server/cluster/cluster.go

@@ -1025,7 +1025,7 @@ func (c *RaftCluster) processRegionHeartbeat(region *core.RegionInfo) error {
 		// To prevent a concurrent heartbeat of another region from overriding the up-to-date region info by a stale one,
 		// check its validation again here.
 		//
-		// However it can't solve the race condition of concurrent heartbeats from the same region.
+		// However, it can't solve the race condition of concurrent heartbeats from the same region.
 		if overlaps, err = c.core.AtomicCheckAndPutRegion(region); err != nil {
 			return err
 		}
@@ -1082,6 +1082,72 @@ func (c *RaftCluster) processRegionHeartbeat(region *core.RegionInfo) error {
 	return nil
 }
 
+func (c *RaftCluster) processRegionSplit(regions []*metapb.Region) []error {
+	if err := c.checkSplitRegions(regions); err != nil {
+		return []error{err}
+	}
+	total := len(regions) - 1
+	regions[0], regions[total] = regions[total], regions[0]
+	leaderStoreID := uint64(0)
+	if r := c.core.GetRegion(regions[0].GetId()); r != nil {
+		leaderStoreID = r.GetLeader().GetStoreId()
+	}
+	errList := make([]error, 0, total)
+	for _, region := range regions {
+		// It should update rightmost because it can override by left.
+		if len(region.GetPeers()) == 0 {
+			errList = append(errList, errors.New(fmt.Sprintf("region:%d has no peer", region.GetId())))
+			continue
+		}
+		// region split initiator store will be leader with a high probability
+		leader := region.Peers[0]
+		if leaderStoreID > 0 {
+			for _, peer := range region.GetPeers() {
+				if peer.GetStoreId() == leaderStoreID {
+					leader = peer
+					break
+				}
+			}
+		}
+		region := core.NewRegionInfo(region, leader)
+		overlaps, perr := c.core.AtomicCheckAndPutRegion(region)
+		if perr != nil {
+			errList = append(errList, perr)
+			continue
+		}
+		for _, item := range overlaps {
+			if c.regionStats != nil {
+				c.regionStats.ClearDefunctRegion(item.GetID())
+			}
+			c.labelLevelStats.ClearDefunctRegion(item.GetID())
+			c.ruleManager.InvalidCache(item.GetID())
+		}
+		regionUpdateCacheEventCounter.Inc()
+		if c.storage != nil {
+			// If there are concurrent heartbeats from the same region, the last write will win even if
+			// writes to storage in the critical area. So don't use mutex to protect it.
+			// Not successfully saved to storage is not fatal, it only leads to longer warm-up
+			// after restart. Here we only log the error then go on updating cache.
+			for _, item := range overlaps {
+				if err := c.storage.DeleteRegion(item.GetMeta()); err != nil {
+					log.Error("failed to delete region from storage",
+						zap.Uint64("region-id", item.GetID()),
+						logutil.ZapRedactStringer("region-meta", core.RegionToHexMeta(item.GetMeta())),
+						errs.ZapError(err))
+				}
+			}
+			if err := c.storage.SaveRegion(region.GetMeta()); err != nil {
+				log.Error("failed to save region to storage",
+					zap.Uint64("region-id", region.GetID()),
+					logutil.ZapRedactStringer("region-meta", core.RegionToHexMeta(region.GetMeta())),
+					errs.ZapError(err))
+			}
+			regionUpdateKVEventCounter.Inc()
+		}
+	}
+	return errList
+}
+
 func (c *RaftCluster) putMetaLocked(meta *metapb.Cluster) error {
 	if c.storage != nil {
 		if err := c.storage.SaveMeta(meta); err != nil {

server/cluster/cluster_worker.go

@@ -26,7 +26,6 @@
 	"github.com/tikv/pd/pkg/schedule/operator"
 	"github.com/tikv/pd/pkg/statistics/buckets"
 	"github.com/tikv/pd/pkg/utils/logutil"
-	"github.com/tikv/pd/pkg/utils/typeutil"
 	"github.com/tikv/pd/pkg/versioninfo"
 	"go.uber.org/zap"
 )
@@ -165,20 +164,8 @@
 	return resp, nil
 }
 
 func (c *RaftCluster) checkSplitRegion(left *metapb.Region, right *metapb.Region) error {
-	if left == nil || right == nil {
-		return errors.New("invalid split region")
-	}
-
-	if !bytes.Equal(left.GetEndKey(), right.GetStartKey()) {
-		return errors.New("invalid split region")
-	}
-
-	if len(right.GetEndKey()) == 0 || bytes.Compare(left.GetStartKey(), right.GetEndKey()) < 0 {
-		return nil
-	}
-
-	return errors.New("invalid split region")
+	return c.checkSplitRegions([]*metapb.Region{left, right})
 }
 
 func (c *RaftCluster) checkSplitRegions(regions []*metapb.Region) error {
@@ -204,43 +191,38 @@
 	left := request.GetLeft()
 	right := request.GetRight()
 
-	err := c.checkSplitRegion(left, right)
-	if err != nil {
+	if errs := c.processRegionSplit([]*metapb.Region{left, right}); len(errs) > 0 {
 		log.Warn("report split region is invalid",
 			logutil.ZapRedactStringer("left-region", core.RegionToHexMeta(left)),
 			logutil.ZapRedactStringer("right-region", core.RegionToHexMeta(right)),
-			errs.ZapError(err))
-		return nil, err
+			zap.Errors("errs", errs),
+		)
+		// error[0] may be checker error, others are ignored.
+		return nil, errs[0]
 	}
 
-	// Build origin region by using left and right.
-	originRegion := typeutil.DeepClone(right, core.RegionFactory)
-	originRegion.RegionEpoch = nil
-	originRegion.StartKey = left.GetStartKey()
 	log.Info("region split, generate new region",
-		zap.Uint64("region-id", originRegion.GetId()),
+		zap.Uint64("region-id", right.GetId()),
 		logutil.ZapRedactStringer("region-meta", core.RegionToHexMeta(left)))
 	return &pdpb.ReportSplitResponse{}, nil
 }
 
 // HandleBatchReportSplit handles the batch report split request.
 func (c *RaftCluster) HandleBatchReportSplit(request *pdpb.ReportBatchSplitRequest) (*pdpb.ReportBatchSplitResponse, error) {
 	regions := request.GetRegions()
-
 	hrm := core.RegionsToHexMeta(regions)
-	err := c.checkSplitRegions(regions)
-	if err != nil {
+	if errs := c.processRegionSplit(regions); len(errs) > 0 {
 		log.Warn("report batch split region is invalid",
 			zap.Stringer("region-meta", hrm),
-			errs.ZapError(err))
-		return nil, err
+			zap.Errors("errs", errs))
+		// error[0] may be checker error, others are ignored.
+		return nil, errs[0]
 	}
 	last := len(regions) - 1
-	originRegion := typeutil.DeepClone(regions[last], core.RegionFactory)
-	hrm = core.RegionsToHexMeta(regions[:last])
+	originRegionID := regions[last].GetId()
 	log.Info("region batch split, generate new regions",
-		zap.Uint64("region-id", originRegion.GetId()),
-		zap.Stringer("origin", hrm),
+		zap.Uint64("region-id", originRegionID),
+		zap.Stringer("new_peer", hrm[:last]),
 		zap.Int("total", last))
 	return &pdpb.ReportBatchSplitResponse{}, nil
 }

server/cluster/cluster_worker_test.go

@@ -26,6 +26,19 @@ import (
 	"github.com/tikv/pd/pkg/storage"
 )
 
+func mockRegionPeer(cluster *RaftCluster, voters []uint64) []*metapb.Peer {
+	rst := make([]*metapb.Peer, len(voters))
+	for i, v := range voters {
+		id, _ := cluster.AllocID()
+		rst[i] = &metapb.Peer{
+			Id:      id,
+			StoreId: v,
+			Role:    metapb.PeerRole_Voter,
+		}
+	}
+	return rst
+}
+
 func TestReportSplit(t *testing.T) {
 	re := require.New(t)
 	ctx, cancel := context.WithCancel(context.Background())
@@ -34,12 +47,53 @@ func TestReportSplit(t *testing.T) {
 	_, opt, err := newTestScheduleConfig()
 	re.NoError(err)
 	cluster := newTestRaftCluster(ctx, mockid.NewIDAllocator(), opt, storage.NewStorageWithMemoryBackend(), core.NewBasicCluster())
-	left := &metapb.Region{Id: 1, StartKey: []byte("a"), EndKey: []byte("b")}
-	right := &metapb.Region{Id: 2, StartKey: []byte("b"), EndKey: []byte("c")}
-	_, err = cluster.HandleReportSplit(&pdpb.ReportSplitRequest{Left: left, Right: right})
-	re.NoError(err)
+	right := &metapb.Region{Id: 1, StartKey: []byte("a"), EndKey: []byte("c"), Peers: mockRegionPeer(cluster, []uint64{1, 2, 3}),
+		RegionEpoch: &metapb.RegionEpoch{ConfVer: 1, Version: 1}}
+	region := core.NewRegionInfo(right, right.Peers[0])
+	cluster.putRegion(region)
+
+	// split failed, split region keys must be continuous.
+	left := &metapb.Region{Id: 2, StartKey: []byte("a"), EndKey: []byte("b"), Peers: mockRegionPeer(cluster, []uint64{1, 2, 3}),
+		RegionEpoch: &metapb.RegionEpoch{ConfVer: 1, Version: 2}}
 	_, err = cluster.HandleReportSplit(&pdpb.ReportSplitRequest{Left: right, Right: left})
 	re.Error(err)
+
+	// split success with continuous region keys.
+	right = &metapb.Region{Id: 1, StartKey: []byte("b"), EndKey: []byte("c"), Peers: mockRegionPeer(cluster, []uint64{1, 2, 3}),
+		RegionEpoch: &metapb.RegionEpoch{ConfVer: 1, Version: 2}}
+	_, err = cluster.HandleReportSplit(&pdpb.ReportSplitRequest{Left: left, Right: right})
+	re.NoError(err)
+	// no range hole
+	storeID := region.GetLeader().GetStoreId()
+	re.Equal(storeID, cluster.GetRegionByKey([]byte("b")).GetLeader().GetStoreId())
+	re.Equal(storeID, cluster.GetRegionByKey([]byte("a")).GetLeader().GetStoreId())
+	re.Equal(uint64(1), cluster.GetRegionByKey([]byte("b")).GetID())
+	re.Equal(uint64(2), cluster.GetRegionByKey([]byte("a")).GetID())
+
+	testdata := []struct {
+		regionID uint64
+		startKey []byte
+		endKey   []byte
+	}{
+		{
+			regionID: 1,
+			startKey: []byte("b"),
+			endKey:   []byte("c"),
+		}, {
+			regionID: 2,
+			startKey: []byte("a"),
+			endKey:   []byte("b"),
+		},
+	}
+
+	for _, data := range testdata {
+		r := metapb.Region{}
+		ok, err := cluster.storage.LoadRegion(data.regionID, &r)
+		re.Nil(err)
+		re.True(ok)
+		re.Equal(data.startKey, r.GetStartKey())
+		re.Equal(data.endKey, r.GetEndKey())
+	}
 }
 
 func TestReportBatchSplit(t *testing.T) {
@@ -51,10 +105,10 @@ func TestReportBatchSplit(t *testing.T) {
 	re.NoError(err)
 	cluster := newTestRaftCluster(ctx, mockid.NewIDAllocator(), opt, storage.NewStorageWithMemoryBackend(), core.NewBasicCluster())
 	regions := []*metapb.Region{
-		{Id: 1, StartKey: []byte(""), EndKey: []byte("a")},
-		{Id: 2, StartKey: []byte("a"), EndKey: []byte("b")},
-		{Id: 3, StartKey: []byte("b"), EndKey: []byte("c")},
-		{Id: 3, StartKey: []byte("c"), EndKey: []byte("")},
+		{Id: 1, StartKey: []byte(""), EndKey: []byte("a"), Peers: mockRegionPeer(cluster, []uint64{1, 2, 3})},
+		{Id: 2, StartKey: []byte("a"), EndKey: []byte("b"), Peers: mockRegionPeer(cluster, []uint64{1, 2, 3})},
+		{Id: 3, StartKey: []byte("b"), EndKey: []byte("c"), Peers: mockRegionPeer(cluster, []uint64{1, 2, 3})},
+		{Id: 3, StartKey: []byte("c"), EndKey: []byte(""), Peers: mockRegionPeer(cluster, []uint64{1, 2, 3})},
 	}
 	_, err = cluster.HandleBatchReportSplit(&pdpb.ReportBatchSplitRequest{Regions: regions})
 	re.NoError(err)

server/grpc_service.go

@@ -1266,10 +1266,25 @@ func (s *GrpcServer) GetRegion(ctx context.Context, request *pdpb.GetRegionReque
 	if rc == nil {
 		return &pdpb.GetRegionResponse{Header: s.notBootstrappedHeader()}, nil
 	}
-	region := rc.GetRegionByKey(request.GetRegionKey())
+	var region *core.RegionInfo
+	// allow region miss temporarily if this key can't be found in the region tree.
+	for retry := 0; retry <= 10; retry++ {
+		region = rc.GetRegionByKey(request.GetRegionKey())
+		if region != nil {
+			break
+		}
+		select {
+		case <-ctx.Done():
+			break
+
+		case <-time.After(10 * time.Millisecond):
+			continue
+		}
+	}
 	if region == nil {
 		return &pdpb.GetRegionResponse{Header: s.header()}, nil
 	}
+
 	var buckets *metapb.Buckets
 	if rc.GetStoreConfig().IsEnableRegionBucket() && request.GetNeedBuckets() {
 		buckets = region.GetBuckets()
@@ -1300,7 +1315,22 @@ func (s *GrpcServer) GetPrevRegion(ctx context.Context, request *pdpb.GetRegionR
 		return &pdpb.GetRegionResponse{Header: s.notBootstrappedHeader()}, nil
 	}
 
-	region := rc.GetPrevRegionByKey(request.GetRegionKey())
+	var region *core.RegionInfo
+	// allow region miss temporarily if this key can't be found in the region tree.
+	for retry := 0; retry <= 10; retry++ {
+		region = rc.GetPrevRegionByKey(request.GetRegionKey())
+		if region != nil {
+			break
+		}
+		select {
+		case <-ctx.Done():
+			break
+
+		case <-time.After(10 * time.Millisecond):
+			continue
+		}
+	}
+
 	if region == nil {
 		return &pdpb.GetRegionResponse{Header: s.header()}, nil
 	}