server, core: support add learner node

pdctl/command/operator.go

@@ -225,7 +225,7 @@ func mergeRegionCommandFunc(cmd *cobra.Command, args []string) {
 	input := make(map[string]interface{})
 	input["name"] = cmd.Name()
 	input["source_region_id"] = ids[0]
-	input["merge_region_id"] = ids[1]
+	input["target_region_id"] = ids[1]
 	postJSON(cmd, operatorsPrefix, input)
 }
 

pdctl/command/region_command.go

@@ -191,7 +191,7 @@ func decodeProtobufText(text string) (string, error) {
 // NewRegionWithCheckCommand return a region with check subcommand of regionCmd
 func NewRegionWithCheckCommand() *cobra.Command {
 	r := &cobra.Command{
-		Use:   "check [miss-replica|extra-replica|down-replica|pending-replica|incorrect-ns]",
+		Use:   "check [miss-peer|extra-peer|down-peer|pending-peer|incorrect-ns]",
 		Short: "show the region with check specific status",
 		Run:   showRegionWithCheckCommandFunc,
 	}

server/api/region.go

@@ -48,12 +48,11 @@ func newRegionInfo(r *core.RegionInfo) *regionInfo {
 		return nil
 	}
 	return &regionInfo{
-		ID:          r.Id,
-		StartKey:    strings.Trim(fmt.Sprintf("%q", r.StartKey), "\""),
-		EndKey:      strings.Trim(fmt.Sprintf("%q", r.EndKey), "\""),
-		RegionEpoch: r.RegionEpoch,
-		Peers:       r.Peers,
-
+		ID:              r.Id,
+		StartKey:        strings.Trim(fmt.Sprintf("%q", r.StartKey), "\""),
+		EndKey:          strings.Trim(fmt.Sprintf("%q", r.EndKey), "\""),
+		RegionEpoch:     r.RegionEpoch,
+		Peers:           r.Peers,
 		Leader:          r.Leader,
 		DownPeers:       r.DownPeers,
 		PendingPeers:    r.PendingPeers,

server/api/router.go

@@ -84,11 +84,11 @@ func createRouter(prefix string, svr *server.Server) *mux.Router {
 	router.HandleFunc("/api/v1/regions", regionsHandler.GetAll).Methods("GET")
 	router.HandleFunc("/api/v1/regions/writeflow", regionsHandler.GetTopWriteFlow).Methods("GET")
 	router.HandleFunc("/api/v1/regions/readflow", regionsHandler.GetTopReadFlow).Methods("GET")
+	router.HandleFunc("/api/v1/regions/check/miss-peer", regionsHandler.GetMissPeerRegions).Methods("GET")
+	router.HandleFunc("/api/v1/regions/check/extra-peer", regionsHandler.GetExtraPeerRegions).Methods("GET")
+	router.HandleFunc("/api/v1/regions/check/pending-peer", regionsHandler.GetPendingPeerRegions).Methods("GET")
+	router.HandleFunc("/api/v1/regions/check/down-peer", regionsHandler.GetDownPeerRegions).Methods("GET")
 	router.HandleFunc("/api/v1/regions/sibling/{id}", regionsHandler.GetRegionSiblings).Methods("GET")
-	router.HandleFunc("/api/v1/regions/check/miss-replica", regionsHandler.GetMissPeerRegions).Methods("GET")
-	router.HandleFunc("/api/v1/regions/check/extra-replica", regionsHandler.GetExtraPeerRegions).Methods("GET")
-	router.HandleFunc("/api/v1/regions/check/pending-replica", regionsHandler.GetPendingPeerRegions).Methods("GET")
-	router.HandleFunc("/api/v1/regions/check/down-replica", regionsHandler.GetDownPeerRegions).Methods("GET")
 	router.HandleFunc("/api/v1/regions/check/incorrect-ns", regionsHandler.GetIncorrectNamespaceRegions).Methods("GET")
 
 	router.Handle("/api/v1/version", newVersionHandler(rd)).Methods("GET")

server/api/trend_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/pingcap/kvproto/pkg/metapb"
 	"github.com/pingcap/pd/server"
 	"github.com/pingcap/pd/server/core"
+	"github.com/pingcap/pd/server/schedule"
 )
 
 var _ = Suite(&testTrendSuite{})
@@ -48,11 +49,22 @@ func (s *testTrendSuite) TestTend(c *C) {
 
 	// Complete the operators.
 	mustRegionHeartbeat(c, svr, s.newRegionInfo(4, "", "a", 2, 2, []uint64{1, 2}, 2))
-	mustRegionHeartbeat(c, svr, s.newRegionInfo(5, "a", "b", 3, 2, []uint64{1, 3}, 1))
-	mustRegionHeartbeat(c, svr, s.newRegionInfo(6, "b", "", 3, 2, []uint64{2, 3}, 2))
+	region := s.newRegionInfo(5, "a", "b", 3, 2, []uint64{1, 3}, 1)
+	op, err := svr.GetHandler().GetOperator(5)
+	c.Assert(op, NotNil)
+	region.Peers[1].Id = op.Step(0).(schedule.AddPeer).PeerID
+	region.Voters[1].Id = op.Step(0).(schedule.AddPeer).PeerID
+	mustRegionHeartbeat(c, svr, region)
+
+	op, err = svr.GetHandler().GetOperator(6)
+	c.Assert(op, NotNil)
+	region = s.newRegionInfo(6, "b", "", 3, 2, []uint64{2, 3}, 2)
+	region.Peers[1].Id = op.Step(0).(schedule.AddPeer).PeerID
+	region.Voters[1].Id = op.Step(0).(schedule.AddPeer).PeerID
+	mustRegionHeartbeat(c, svr, region)
 
 	var trend Trend
-	err := readJSONWithURL(fmt.Sprintf("%s%s/api/v1/trend", svr.GetAddr(), apiPrefix), &trend)
+	err = readJSONWithURL(fmt.Sprintf("%s%s/api/v1/trend", svr.GetAddr(), apiPrefix), &trend)
 	c.Assert(err, IsNil)
 
 	// Check store states.
@@ -82,20 +94,20 @@ func (s *testTrendSuite) newRegionInfo(id uint64, startKey, endKey string, confV
 		leader *metapb.Peer
 	)
 	for _, id := range stores {
-		p := &metapb.Peer{Id: id, StoreId: id}
+		p := &metapb.Peer{Id: 10 + id, StoreId: id}
 		if id == leaderStore {
 			leader = p
 		}
 		peers = append(peers, p)
 	}
-	return &core.RegionInfo{
-		Region: &metapb.Region{
+	return core.NewRegionInfo(
+		&metapb.Region{
 			Id:          id,
 			StartKey:    []byte(startKey),
 			EndKey:      []byte(endKey),
 			RegionEpoch: &metapb.RegionEpoch{ConfVer: confVer, Version: ver},
 			Peers:       peers,
 		},
-		Leader: leader,
-	}
+		leader,
+	)
 }

server/cache.go

@@ -592,6 +592,10 @@ func (c *clusterInfo) GetHotRegionLowThreshold() int {
 	return c.opt.GetHotRegionLowThreshold()
 }
 
+func (c *clusterInfo) IsRaftLearnerEnabled() bool {
+	return c.opt.IsRaftLearnerEnabled()
+}
+
 func (c *clusterInfo) CheckLabelProperty(typ string, labels []*metapb.StoreLabel) bool {
 	return c.opt.CheckLabelProperty(typ, labels)
 }

server/config.go

@@ -352,6 +352,8 @@ type ScheduleConfig struct {
 	MergeScheduleLimit uint64 `toml:"merge-schedule-limit,omitempty" json:"merge-schedule-limit"`
 	// TolerantSizeRatio is the ratio of buffer size for balance scheduler.
 	TolerantSizeRatio float64 `toml:"tolerant-size-ratio,omitempty" json:"tolerant-size-ratio"`
+	// EnableRaftLearner is the option for using AddLearnerNode instead of AddNode
+	EnableRaftLearner bool `toml:"enable-raft-learner" json:"enable-raft-learner,string"`
 	// Schedulers support for loding customized schedulers
 	Schedulers SchedulerConfigs `toml:"schedulers,omitempty" json:"schedulers-v2"` // json v2 is for the sake of compatible upgrade
 }
@@ -369,6 +371,7 @@ func (c *ScheduleConfig) clone() *ScheduleConfig {
 		ReplicaScheduleLimit: c.ReplicaScheduleLimit,
 		MergeScheduleLimit:   c.MergeScheduleLimit,
 		TolerantSizeRatio:    c.TolerantSizeRatio,
+		EnableRaftLearner:    c.EnableRaftLearner,
 		Schedulers:           schedulers,
 	}
 }

server/coordinator.go

@@ -109,6 +109,21 @@ func (c *coordinator) dispatch(region *core.RegionInfo) {
 			c.removeOperator(op)
 		}
 	}
+	// If PD has restarted, it need to check learners added before and promote them.
+	if c.cluster.IsRaftLearnerEnabled() && c.getOperator(region.GetId()) == nil {
+		for _, p := range region.GetLearners() {
+			if region.GetPendingLearner(p.GetId()) != nil {
+				continue
+			}
+			step := schedule.PromoteLearner{
+				ToStore: p.GetStoreId(),
+				PeerID:  p.GetId(),
+			}
+			op := schedule.NewOperator("promoteLearner", region.GetId(), schedule.OpRegion, step)
+			c.addOperator(op)
+			break
+		}
+	}
 }
 
 func (c *coordinator) patrolRegions() {
@@ -139,20 +154,23 @@ func (c *coordinator) patrolRegions() {
 			key = region.GetEndKey()
 
 			if op := c.namespaceChecker.Check(region); op != nil {
-				c.addOperator(op)
-				break
+				if c.addOperator(op) {
+					break
+				}
 			}
 			if c.limiter.OperatorCount(schedule.OpReplica) < c.cluster.GetReplicaScheduleLimit() {
 				if op := c.replicaChecker.Check(region); op != nil {
-					c.addOperator(op)
-					break
+					if c.addOperator(op) {
+						break
+					}
 				}
 			}
 			if c.limiter.OperatorCount(schedule.OpMerge) < c.cluster.GetMergeScheduleLimit() {
 				if op1, op2 := c.mergeChecker.Check(region); op1 != nil && op2 != nil {
 					// make sure two operators can add successfully altogether
-					c.addOperators(op1, op2)
-					break
+					if c.addOperators(op1, op2) {
+						break
+					}
 				}
 			}
 		}
@@ -550,6 +568,34 @@ func (c *coordinator) sendScheduleCommand(region *core.RegionInfo, step schedule
 			},
 		}
 		c.hbStreams.sendMsg(region, cmd)
+	case schedule.AddLearner:
+		if region.GetStorePeer(s.ToStore) != nil {
+			// The newly added peer is pending.
+			return
+		}
+		cmd := &pdpb.RegionHeartbeatResponse{
+			ChangePeer: &pdpb.ChangePeer{
+				ChangeType: eraftpb.ConfChangeType_AddLearnerNode,
+				Peer: &metapb.Peer{
+					Id:        s.PeerID,
+					StoreId:   s.ToStore,
+					IsLearner: true,
+				},
+			},
+		}
+		c.hbStreams.sendMsg(region, cmd)
+	case schedule.PromoteLearner:
+		cmd := &pdpb.RegionHeartbeatResponse{
+			ChangePeer: &pdpb.ChangePeer{
+				// reuse AddNode type
+				ChangeType: eraftpb.ConfChangeType_AddNode,
+				Peer: &metapb.Peer{
+					Id:      s.PeerID,
+					StoreId: s.ToStore,
+				},
+			},
+		}
+		c.hbStreams.sendMsg(region, cmd)
 	case schedule.RemovePeer:
 		cmd := &pdpb.RegionHeartbeatResponse{
 			ChangePeer: &pdpb.ChangePeer{

server/coordinator_test.go

@@ -698,7 +698,7 @@ func waitAddPeer(c *C, stream *mockHeartbeatStream, region *core.RegionInfo, sto
 		}
 		return false
 	})
-	region.Peers = append(region.Peers, res.GetChangePeer().GetPeer())
+	region.AddPeer(res.GetChangePeer().GetPeer())
 	region.RegionEpoch = &metapb.RegionEpoch{
 		ConfVer: region.GetRegionEpoch().GetConfVer() + 1,
 		Version: region.GetRegionEpoch().GetVersion(),