server, member: check target PD readiness before leader transfer

errors.toml

@@ -746,6 +746,11 @@ error = '''
 etcd leader not found
 '''
 
+["PD:member:ErrEtcdLeaderTransferTargetCheck"]
+error = '''
+etcd leader transfer target check failed: %s
+'''
+
 ["PD:member:ErrMarshalMember"]
 error = '''
 marshal member failed

pkg/errs/errno.go

@@ -120,10 +120,11 @@ var (
 
 // member errors
 var (
-	ErrEtcdLeaderNotFound = errors.Normalize("etcd leader not found", errors.RFCCodeText("PD:member:ErrEtcdLeaderNotFound"))
-	ErrMarshalMember      = errors.Normalize("marshal member failed", errors.RFCCodeText("PD:member:ErrMarshalMember"))
-	ErrMarshalParticipant = errors.Normalize("marshal participant failed", errors.RFCCodeText("PD:member:ErrMarshalParticipant"))
-	ErrCheckCampaign      = errors.Normalize("check campaign failed", errors.RFCCodeText("PD:member:ErrCheckCampaign"))
+	ErrEtcdLeaderNotFound            = errors.Normalize("etcd leader not found", errors.RFCCodeText("PD:member:ErrEtcdLeaderNotFound"))
+	ErrMarshalMember                 = errors.Normalize("marshal member failed", errors.RFCCodeText("PD:member:ErrMarshalMember"))
+	ErrMarshalParticipant            = errors.Normalize("marshal participant failed", errors.RFCCodeText("PD:member:ErrMarshalParticipant"))
+	ErrCheckCampaign                 = errors.Normalize("check campaign failed", errors.RFCCodeText("PD:member:ErrCheckCampaign"))
+	ErrEtcdLeaderTransferTargetCheck = errors.Normalize("etcd leader transfer target check failed: %s", errors.RFCCodeText("PD:member:ErrEtcdLeaderTransferTargetCheck"))
 )
 
 // core errors

pkg/member/member.go

@@ -66,6 +66,28 @@ type Member struct {
 	lastLeaderUpdatedTime atomic.Value
 }
 
+// MoveEtcdLeaderOption customizes embedded etcd leader transfer.
+type MoveEtcdLeaderOption func(*moveEtcdLeaderOptions)
+
+type moveEtcdLeaderOptions struct {
+	targetChecker func(context.Context, uint64) error
+}
+
+// WithTargetChecker checks whether the target member can become leader before transfer.
+func WithTargetChecker(checker func(context.Context, uint64) error) MoveEtcdLeaderOption {
+	return func(opts *moveEtcdLeaderOptions) {
+		opts.targetChecker = checker
+	}
+}
+
+func newMoveEtcdLeaderOptions(opts ...MoveEtcdLeaderOption) *moveEtcdLeaderOptions {
+	options := &moveEtcdLeaderOptions{}
+	for _, opt := range opts {
+		opt(options)
+	}
+	return options
+}
+
 // NewMember create a new Member.
 func NewMember(etcd *embed.Etcd, client *clientv3.Client, id uint64) *Member {
 	return &Member{
@@ -280,7 +302,7 @@ func (m *Member) Resign() {
 }
 
 // CheckPriority checks whether the etcd leader should be moved according to the priority.
-func (m *Member) CheckPriority(ctx context.Context) {
+func (m *Member) CheckPriority(ctx context.Context, opts ...MoveEtcdLeaderOption) {
 	myPriority, err := m.GetMemberLeaderPriority(m.ID())
 	if err != nil {
 		log.Error("failed to load leader priority", errs.ZapError(err))
@@ -299,9 +321,19 @@ func (m *Member) CheckPriority(ctx context.Context) {
 		return
 	}
 	if myPriority > leaderPriority {
-		err := m.MoveEtcdLeader(ctx, etcdLeader, m.ID())
+		err := m.MoveEtcdLeader(ctx, etcdLeader, m.ID(), opts...)
 		if err != nil {
-			log.Error("failed to transfer etcd leader", errs.ZapError(err))
+			if errors.ErrorEqual(err, errs.ErrEtcdLeaderTransferTargetCheck) {
+				log.Warn("skip transferring etcd leader by priority because target check failed",
+					zap.Uint64("from", etcdLeader),
+					zap.Uint64("to", m.ID()),
+					errs.ZapError(err))
+			} else {
+				log.Error("failed to transfer etcd leader by priority",
+					zap.Uint64("from", etcdLeader),
+					zap.Uint64("to", m.ID()),
+					errs.ZapError(err))
+			}
 		} else {
 			log.Info("transfer etcd leader",
 				zap.Uint64("from", etcdLeader),
@@ -311,9 +343,15 @@ func (m *Member) CheckPriority(ctx context.Context) {
 }
 
 // MoveEtcdLeader tries to transfer etcd leader.
-func (m *Member) MoveEtcdLeader(ctx context.Context, old, new uint64) error {
+func (m *Member) MoveEtcdLeader(ctx context.Context, old, new uint64, opts ...MoveEtcdLeaderOption) error {
 	moveCtx, cancel := context.WithTimeout(ctx, moveLeaderTimeout)
 	defer cancel()
+	options := newMoveEtcdLeaderOptions(opts...)
+	if options.targetChecker != nil {
+		if err := options.targetChecker(moveCtx, new); err != nil {
+			return errs.ErrEtcdLeaderTransferTargetCheck.GenWithStackByArgs(err.Error())
+		}
+	}
 	err := m.etcd.Server.MoveLeader(moveCtx, old, new)
 	if err != nil {
 		return errs.ErrEtcdMoveLeader.Wrap(err).GenWithStackByCause()
@@ -353,7 +391,7 @@ func (m *Member) InitMemberInfo(advertiseClientUrls, advertisePeerUrls, name str
 
 // ResignEtcdLeader resigns current PD's etcd leadership. If nextLeader is empty, all
 // other pd-servers can campaign.
-func (m *Member) ResignEtcdLeader(ctx context.Context, from string, nextEtcdLeader string) error {
+func (m *Member) ResignEtcdLeader(ctx context.Context, from string, nextEtcdLeader string, opts ...MoveEtcdLeaderOption) error {
 	log.Info("try to resign etcd leader to next pd-server", zap.String("from", from), zap.String("to", nextEtcdLeader))
 	// Determine next etcd leader candidates.
 	var etcdLeaderIDs []uint64
@@ -375,6 +413,30 @@ func (m *Member) ResignEtcdLeader(ctx context.Context, from string, nextEtcdLead
 	if len(etcdLeaderIDs) == 0 {
 		return errors.New("no valid pd to transfer etcd leader")
 	}
+	options := newMoveEtcdLeaderOptions(opts...)
+	if options.targetChecker != nil {
+		readyEtcdLeaderIDs := make([]uint64, 0, len(etcdLeaderIDs))
+		var lastErr error
+		for _, id := range etcdLeaderIDs {
+			checkCtx, cancel := context.WithTimeout(ctx, moveLeaderTimeout)
+			err := options.targetChecker(checkCtx, id)
+			cancel()
+			if err != nil {
+				lastErr = err
+				log.Warn("skip unready pd when resigning etcd leader", zap.Uint64("target-member-id", id), errs.ZapError(err))
+				continue
+			}
+			readyEtcdLeaderIDs = append(readyEtcdLeaderIDs, id)
+		}
+		if len(readyEtcdLeaderIDs) == 0 {
+			errMsg := "no ready pd to transfer etcd leader"
+			if lastErr != nil {
+				errMsg += ": " + lastErr.Error()
+			}
+			return errs.ErrEtcdLeaderTransferTargetCheck.GenWithStackByArgs(errMsg)
+		}
+		etcdLeaderIDs = readyEtcdLeaderIDs
+	}
 	nextEtcdLeaderID := etcdLeaderIDs[rand.IntN(len(etcdLeaderIDs))]
 	return m.MoveEtcdLeader(ctx, m.ID(), nextEtcdLeaderID)
 }

pkg/utils/apiutil/apiutil.go

@@ -438,6 +438,8 @@ func ReadJSONRespondError(rd *render.Render, w http.ResponseWriter, body io.Read
 const (
 	// CorePath the core group, is at REST path `/pd/api/v1`.
 	CorePath = "/pd/api/v1"
+	// CoreV2Path is the core group REST path for API v2.
+	CoreV2Path = "/pd/api/v2"
 	// ExtensionsPath the named groups are REST at `/pd/apis/{GROUP_NAME}/{Version}`.
 	ExtensionsPath = "/pd/apis"
 )

pkg/versioninfo/feature.go

@@ -49,6 +49,8 @@ const (
 	HotScheduleWithQuery
 	// SwitchWithess supports switch between witness and non-witness.
 	SwitchWitness
+	// ReadyAPI supports checking whether PD has finished initial region loading.
+	ReadyAPI
 )
 
 var featuresDict = map[Feature]string{
@@ -62,6 +64,7 @@ var featuresDict = map[Feature]string{
 	ConfChangeV2:         "5.0.0",
 	HotScheduleWithQuery: "5.2.0",
 	SwitchWitness:        "6.6.0",
+	ReadyAPI:             "8.5.2",
 }
 
 // MinSupportedVersion returns the minimum support version for the specified feature.
@@ -73,3 +76,13 @@ func MinSupportedVersion(v Feature) *semver.Version {
 	version := MustParseVersion(target)
 	return version
 }
+
+// IsReadyAPISupported returns whether PD supports the ready API for initial region loading.
+func IsReadyAPISupported(pdVersion *semver.Version) bool {
+	if pdVersion == nil {
+		return false
+	}
+	// ReadyAPI was introduced in a patch release, so use strict semver comparison
+	// instead of IsFeatureSupported's same-minor compatibility semantics.
+	return !pdVersion.LessThan(*MinSupportedVersion(ReadyAPI))
+}

pkg/versioninfo/versioninfo_test.go

@@ -46,3 +46,23 @@ func TestIsHotScheduleWithCPUSupported(t *testing.T) {
 		re.Equal(test.expect, IsHotScheduleWithCPUSupported(MustParseVersion(test.version)), test.version)
 	}
 }
+
+func TestIsReadyAPISupported(t *testing.T) {
+	re := require.New(t)
+	re.False(IsReadyAPISupported(nil))
+
+	tests := []struct {
+		version string
+		expect  bool
+	}{
+		{"8.5.1", false},
+		{"v8.5.1", false},
+		{"8.5.2", true},
+		{"v8.5.2", true},
+		{"8.5.3", true},
+		{"9.0.0", true},
+	}
+	for _, test := range tests {
+		re.Equal(test.expect, IsReadyAPISupported(MustParseVersion(test.version)), test.version)
+	}
+}

server/api/member.go

@@ -29,6 +29,7 @@ import (
 	"github.com/pingcap/log"
 
 	"github.com/tikv/pd/pkg/errs"
+	"github.com/tikv/pd/pkg/member"
 	"github.com/tikv/pd/pkg/utils/apiutil"
 	"github.com/tikv/pd/pkg/utils/etcdutil"
 	"github.com/tikv/pd/pkg/utils/keypath"
@@ -281,7 +282,8 @@ func (h *leaderHandler) GetLeader(w http.ResponseWriter, _ *http.Request) {
 //	@Failure	500	{string}	string	"PD server failed to proceed the request."
 //	@Router		/leader/resign [post]
 func (h *leaderHandler) ResignLeader(w http.ResponseWriter, _ *http.Request) {
-	err := h.svr.GetMember().ResignEtcdLeader(h.svr.Context(), h.svr.Name(), "")
+	err := h.svr.GetMember().ResignEtcdLeader(h.svr.Context(), h.svr.Name(), "",
+		member.WithTargetChecker(h.svr.CheckMemberReadyForLeaderTransfer))
 	if err != nil {
 		h.rd.JSON(w, http.StatusInternalServerError, err.Error())
 		return
@@ -300,7 +302,8 @@ func (h *leaderHandler) ResignLeader(w http.ResponseWriter, _ *http.Request) {
 //	@Failure	500	{string}	string	"PD server failed to proceed the request."
 //	@Router		/leader/transfer/{nextLeader} [post]
 func (h *leaderHandler) TransferLeader(w http.ResponseWriter, r *http.Request) {
-	err := h.svr.GetMember().ResignEtcdLeader(h.svr.Context(), h.svr.Name(), mux.Vars(r)["next_leader"])
+	err := h.svr.GetMember().ResignEtcdLeader(h.svr.Context(), h.svr.Name(), mux.Vars(r)["next_leader"],
+		member.WithTargetChecker(h.svr.CheckMemberReadyForLeaderTransfer))
 	if err != nil {
 		h.rd.JSON(w, http.StatusInternalServerError, err.Error())
 		return

server/apiv2/middlewares/microservice_redirector.go

@@ -75,7 +75,7 @@ func MicroserviceRedirector(rules ...serverapi.RedirectRule) gin.HandlerFunc {
 
 // AffinityMicroserviceRedirector only forwards affinity GET requests to the scheduling service.
 func AffinityMicroserviceRedirector() gin.HandlerFunc {
-	pdAffinityPath := "/pd/api/v2/affinity-groups"
+	pdAffinityPath := apiutil.CoreV2Path + "/affinity-groups"
 	targetAffinityPath := scheapi.APIPathPrefix + "/affinity-groups"
 	return MicroserviceRedirector(
 		serverapi.RedirectRule{

server/apiv2/router.go

@@ -33,7 +33,7 @@ var group = apiutil.APIServiceGroup{
 	PathPrefix: apiV2Prefix,
 }
 
-const apiV2Prefix = "/pd/api/v2/"
+const apiV2Prefix = apiutil.CoreV2Path + "/"
 
 // NewV2Handler creates a HTTP handler for API.
 //