syncer: sync region leaders between PD leader and follower

go.mod

@@ -2,6 +2,8 @@ module github.com/pingcap/pd/v4
 
 go 1.13
 
+replace github.com/pingcap/kvproto v0.0.0-20200616092848-8037ca08f377 => github.com/JmPotato/kvproto v0.0.0-20200630040751-338a4ea3453f
+
 require (
 	github.com/BurntSushi/toml v0.3.1
 	github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751

go.sum

@@ -1,6 +1,8 @@
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
+github.com/JmPotato/kvproto v0.0.0-20200630040751-338a4ea3453f h1:PbkNiWB/mNug+7QWMFMW5Dr60nG5oT2tZaUY2/eZcLA=
+github.com/JmPotato/kvproto v0.0.0-20200630040751-338a4ea3453f/go.mod h1:IOdRDPLyda8GX2hE/jO7gqaCV/PNFh8BZQCQZXfIOqI=
 github.com/KyleBanks/depth v1.2.1 h1:5h8fQADFrWtarTdtDudMmGsC7GPbOAu6RVB3ffsVFHc=
 github.com/KyleBanks/depth v1.2.1/go.mod h1:jzSb9d0L43HxTQfT+oSA1EEp2q+ne2uh6XgeJcm8brE=
 github.com/PuerkitoBio/purell v1.1.0/go.mod h1:c11w/QuzBsJSee3cPx9rAFu61PvFxuPbtSwDGJws/X0=
@@ -292,8 +294,6 @@ github.com/pingcap/failpoint v0.0.0-20191029060244-12f4ac2fd11d h1:F8vp38kTAckN+
 github.com/pingcap/failpoint v0.0.0-20191029060244-12f4ac2fd11d/go.mod h1:DNS3Qg7bEDhU6EXNHF+XSv/PGznQaMJ5FWvctpm6pQI=
 github.com/pingcap/kvproto v0.0.0-20191211054548-3c6b38ea5107/go.mod h1:WWLmULLO7l8IOcQG+t+ItJ3fEcrL5FxF0Wu+HrMy26w=
 github.com/pingcap/kvproto v0.0.0-20200411081810-b85805c9476c/go.mod h1:IOdRDPLyda8GX2hE/jO7gqaCV/PNFh8BZQCQZXfIOqI=
-github.com/pingcap/kvproto v0.0.0-20200616092848-8037ca08f377 h1:KUUCELlkPNwYXdTu9kCcf1kMSFqz6I6i9F0nXfPO2vs=
-github.com/pingcap/kvproto v0.0.0-20200616092848-8037ca08f377/go.mod h1:IOdRDPLyda8GX2hE/jO7gqaCV/PNFh8BZQCQZXfIOqI=
 github.com/pingcap/log v0.0.0-20191012051959-b742a5d432e9 h1:AJD9pZYm72vMgPcQDww9rkZ1DnWfl0pXV3BOWlkYIjA=
 github.com/pingcap/log v0.0.0-20191012051959-b742a5d432e9/go.mod h1:4rbK1p9ILyIfb6hU7OG2CiWSqMXnp3JMbiaVJ6mvoY8=
 github.com/pingcap/log v0.0.0-20200117041106-d28c14d3b1cd h1:CV3VsP3Z02MVtdpTMfEgRJ4T9NGgGTxdHpJerent7rM=

server/cluster/cluster.go

@@ -496,7 +496,7 @@ func (c *RaftCluster) processRegionHeartbeat(region *core.RegionInfo) error {
 	// Save to storage if meta is updated.
 	// Save to cache if meta or leader is updated, or contains any down/pending peer.
 	// Mark isNew if the region in cache does not have leader.
-	var saveKV, saveCache, isNew, statsChange bool
+	var saveKV, saveCache, isNew, needSync bool
 	if origin == nil {
 		log.Debug("insert new region",
 			zap.Uint64("region-id", region.GetID()),
@@ -534,7 +534,7 @@ func (c *RaftCluster) processRegionHeartbeat(region *core.RegionInfo) error {
 					zap.Uint64("to", region.GetLeader().GetStoreId()),
 				)
 			}
-			saveCache = true
+			saveCache, needSync = true, true
 		}
 		if len(region.GetDownPeers()) > 0 || len(region.GetPendingPeers()) > 0 {
 			saveCache = true
@@ -555,7 +555,7 @@ func (c *RaftCluster) processRegionHeartbeat(region *core.RegionInfo) error {
 			region.GetBytesRead() != origin.GetBytesRead() ||
 			region.GetKeysWritten() != origin.GetKeysWritten() ||
 			region.GetKeysRead() != origin.GetKeysRead() {
-			saveCache, statsChange = true, true
+			saveCache, needSync = true, true
 		}
 
 		if region.GetReplicationStatus().GetState() != replication_modepb.RegionReplicationState_UNKNOWN &&
@@ -646,7 +646,7 @@ func (c *RaftCluster) processRegionHeartbeat(region *core.RegionInfo) error {
 		}
 		regionEventCounter.WithLabelValues("update_kv").Inc()
 	}
-	if saveKV || statsChange {
+	if saveKV || needSync {
 		select {
 		case c.changedRegions <- region:
 		default:

server/region_syncer/client.go

@@ -17,6 +17,7 @@ import (
 	"context"
 	"time"
 
+	"github.com/pingcap/kvproto/pkg/metapb"
 	"github.com/pingcap/kvproto/pkg/pdpb"
 	"github.com/pingcap/log"
 	"github.com/pingcap/pd/v4/pkg/grpcutil"
@@ -186,18 +187,25 @@ func (s *RegionSyncer) StartSyncWithLeader(addr string) {
 				}
 				stats := resp.GetRegionStats()
 				regions := resp.GetRegions()
+				regionLeaders := resp.GetRegionLeaders()
 				hasStats := len(stats) == len(regions)
 				for i, r := range regions {
-					var region *core.RegionInfo
+					var (
+						region       *core.RegionInfo
+						regionLeader *metapb.Peer
+					)
+					if len(regionLeaders) > 0 && regionLeaders[i].Id != 0 {
+						regionLeader = regionLeaders[i]
+					}
 					if hasStats {
-						region = core.NewRegionInfo(r, nil,
+						region = core.NewRegionInfo(r, regionLeader,
 							core.SetWrittenBytes(stats[i].BytesWritten),
 							core.SetWrittenKeys(stats[i].KeysWritten),
 							core.SetReadBytes(stats[i].BytesRead),
 							core.SetReadKeys(stats[i].KeysRead),
 						)
 					} else {
-						region = core.NewRegionInfo(r, nil)
+						region = core.NewRegionInfo(r, regionLeader)
 					}
 
 					s.server.GetBasicCluster().CheckAndPutRegion(region)

server/region_syncer/server.go

@@ -99,6 +99,7 @@ func NewRegionSyncer(s Server) *RegionSyncer {
 func (s *RegionSyncer) RunServer(regionNotifier <-chan *core.RegionInfo, quit chan struct{}) {
 	var requests []*metapb.Region
 	var stats []*pdpb.RegionStat
+	var leaders []*metapb.Peer
 	ticker := time.NewTicker(syncerKeepAliveInterval)
 	for {
 		select {
@@ -108,20 +109,23 @@ func (s *RegionSyncer) RunServer(regionNotifier <-chan *core.RegionInfo, quit ch
 		case first := <-regionNotifier:
 			requests = append(requests, first.GetMeta())
 			stats := append(stats, first.GetStat())
+			leaders := append(leaders, first.GetLeader())
 			startIndex := s.history.GetNextIndex()
 			s.history.Record(first)
 			pending := len(regionNotifier)
 			for i := 0; i < pending && i < maxSyncRegionBatchSize; i++ {
 				region := <-regionNotifier
 				requests = append(requests, region.GetMeta())
 				stats = append(stats, region.GetStat())
+				leaders = append(leaders, region.GetLeader())
 				s.history.Record(region)
 			}
 			regions := &pdpb.SyncRegionResponse{
-				Header:      &pdpb.ResponseHeader{ClusterId: s.server.ClusterID()},
-				Regions:     requests,
-				StartIndex:  startIndex,
-				RegionStats: stats,
+				Header:        &pdpb.ResponseHeader{ClusterId: s.server.ClusterID()},
+				Regions:       requests,
+				StartIndex:    startIndex,
+				RegionStats:   stats,
+				RegionLeaders: leaders,
 			}
 			s.broadcast(regions)
 		case <-ticker.C:
@@ -179,17 +183,24 @@ func (s *RegionSyncer) syncHistoryRegion(request *pdpb.SyncRegionRequest, stream
 			start := time.Now()
 			metas := make([]*metapb.Region, 0, maxSyncRegionBatchSize)
 			stats := make([]*pdpb.RegionStat, 0, maxSyncRegionBatchSize)
+			leaders := make([]*metapb.Peer, 0, maxSyncRegionBatchSize)
 			for syncedIndex, r := range regions {
 				metas = append(metas, r.GetMeta())
 				stats = append(stats, r.GetStat())
+				leader := &metapb.Peer{}
+				if r.GetLeader() != nil {
+					leader = r.GetLeader()
+				}
+				leaders = append(leaders, leader)
 				if len(metas) < maxSyncRegionBatchSize && syncedIndex < len(regions)-1 {
 					continue
 				}
 				resp := &pdpb.SyncRegionResponse{
-					Header:      &pdpb.ResponseHeader{ClusterId: s.server.ClusterID()},
-					Regions:     metas,
-					StartIndex:  uint64(lastIndex),
-					RegionStats: stats,
+					Header:        &pdpb.ResponseHeader{ClusterId: s.server.ClusterID()},
+					Regions:       metas,
+					StartIndex:    uint64(lastIndex),
+					RegionStats:   stats,
+					RegionLeaders: leaders,
 				}
 				s.limit.Wait(int64(resp.Size()))
 				lastIndex += len(metas)
@@ -213,15 +224,22 @@ func (s *RegionSyncer) syncHistoryRegion(request *pdpb.SyncRegionRequest, stream
 		zap.Int("records-length", len(records)))
 	regions := make([]*metapb.Region, len(records))
 	stats := make([]*pdpb.RegionStat, len(records))
+	leaders := make([]*metapb.Peer, len(records))
 	for i, r := range records {
 		regions[i] = r.GetMeta()
 		stats[i] = r.GetStat()
+		leader := &metapb.Peer{}
+		if r.GetLeader() != nil {
+			leader = r.GetLeader()
+		}
+		leaders[i] = leader
 	}
 	resp := &pdpb.SyncRegionResponse{
-		Header:      &pdpb.ResponseHeader{ClusterId: s.server.ClusterID()},
-		Regions:     regions,
-		StartIndex:  startIndex,
-		RegionStats: stats,
+		Header:        &pdpb.ResponseHeader{ClusterId: s.server.ClusterID()},
+		Regions:       regions,
+		StartIndex:    startIndex,
+		RegionStats:   stats,
+		RegionLeaders: leaders,
 	}
 	return stream.Send(resp)
 }

tests/server/region_syncer/region_syncer_test.go

@@ -86,7 +86,10 @@ func (s *serverTestSuite) TestRegionSyncer(c *C) {
 			},
 			StartKey: []byte{byte(i)},
 			EndKey:   []byte{byte(i + 1)},
-			Peers:    []*metapb.Peer{{Id: allocator.alloc(), StoreId: uint64(0)}},
+			Peers: []*metapb.Peer{
+				{Id: allocator.alloc(), StoreId: uint64(0)},
+				{Id: allocator.alloc(), StoreId: uint64(0)},
+			},
 		}
 		regions = append(regions, core.NewRegionInfo(r, r.Peers[0]))
 	}
@@ -129,6 +132,13 @@ func (s *serverTestSuite) TestRegionSyncer(c *C) {
 		c.Assert(err, IsNil)
 	}
 
+	// change the leader of region
+	for i := 0; i < len(regions); i++ {
+		regions[i] = regions[i].Clone(core.WithLeader(regions[i].GetPeers()[1]))
+		err = rc.HandleRegionHeartbeat(regions[i])
+		c.Assert(err, IsNil)
+	}
+
 	// ensure flush to region storage, we use a duration larger than the
 	// region storage flush rate limit (3s).
 	time.Sleep(4 * time.Second)
@@ -142,6 +152,7 @@ func (s *serverTestSuite) TestRegionSyncer(c *C) {
 		r := followerServer.GetServer().GetBasicCluster().GetRegion(region.GetID())
 		c.Assert(r.GetMeta(), DeepEquals, region.GetMeta())
 		c.Assert(r.GetStat(), DeepEquals, region.GetStat())
+		c.Assert(r.GetLeader(), DeepEquals, region.GetLeader())
 	}
 
 	err = leaderServer.Stop()
@@ -155,6 +166,7 @@ func (s *serverTestSuite) TestRegionSyncer(c *C) {
 		r := leaderServer.GetRegionInfoByID(region.GetID())
 		c.Assert(r.GetMeta(), DeepEquals, region.GetMeta())
 		c.Assert(r.GetStat(), DeepEquals, region.GetStat())
+		c.Assert(r.GetLeader(), DeepEquals, region.GetLeader())
 	}
 }