pdms: support primary/transfer api for scheduling and tso

pkg/election/leadership.go

@@ -93,7 +93,7 @@ func (ls *Leadership) getLease() *lease {
 	return l.(*lease)
 }
 
-func (ls *Leadership) setLease(lease *lease) {
+func (ls *Leadership) SetLease(lease *lease) {
 	ls.lease.Store(lease)
 }
 
@@ -156,7 +156,7 @@ func (ls *Leadership) Campaign(leaseTimeout int64, leaderData string, cmps ...cl
 		client:  ls.client,
 		lease:   clientv3.NewLease(ls.client),
 	}
-	ls.setLease(newLease)
+	ls.SetLease(newLease)
 
 	failpoint.Inject("skipGrantLeader", func(val failpoint.Value) {
 		var member pdpb.Member

pkg/election/lease.go

@@ -48,6 +48,14 @@
 	expireTime   atomic.Value
 }
 
+func NewLease(client *clientv3.Client, purpose string) *lease {
+	return &lease{
+		Purpose: purpose,
+		client:  client,
+		lease:   clientv3.NewLease(client),
+	}
+}
+
 // Grant uses `lease.Grant` to initialize the lease and expireTime.
 func (l *lease) Grant(leaseTimeout int64) error {
 	if l == nil {

pkg/mcs/discovery/discover.go

@@ -45,14 +45,14 @@
 }
 
 // GetMSMembers returns all the members of the specified service name.
-func GetMSMembers(name string, client *clientv3.Client) ([]ServiceRegistryEntry, error) {
-	switch name {
+func GetMSMembers(serviceName string, client *clientv3.Client) ([]ServiceRegistryEntry, error) {
+	switch serviceName {
 	case utils.TSOServiceName, utils.SchedulingServiceName, utils.ResourceManagerServiceName:
 		clusterID, err := etcdutil.GetClusterID(client, utils.ClusterIDPath)
 		if err != nil {
 			return nil, err
 		}
-		servicePath := ServicePath(strconv.FormatUint(clusterID, 10), name)
+		servicePath := ServicePath(strconv.FormatUint(clusterID, 10), serviceName)
 		resps, err := kv.NewSlowLogTxn(client).Then(clientv3.OpGet(servicePath, clientv3.WithPrefix())).Commit()
 		if err != nil {
 			return nil, errs.ErrEtcdKVGet.Wrap(err).GenWithStackByCause()
@@ -75,5 +75,5 @@
 		return entries, nil
 	}
 
-	return nil, errors.Errorf("unknown service name %s", name)
+	return nil, errors.Errorf("unknown service name %s", serviceName)
 }

pkg/mcs/scheduling/server/apis/v1/api.go

@@ -29,6 +29,7 @@
 	"github.com/gin-gonic/gin"
 	"github.com/pingcap/kvproto/pkg/metapb"
 	"github.com/pingcap/log"
+	"github.com/tikv/pd/pkg/election"
 	"github.com/tikv/pd/pkg/errs"
 	scheserver "github.com/tikv/pd/pkg/mcs/scheduling/server"
 	mcsutils "github.com/tikv/pd/pkg/mcs/utils"
@@ -39,11 +40,14 @@
 	"github.com/tikv/pd/pkg/schedule/schedulers"
 	"github.com/tikv/pd/pkg/statistics/utils"
 	"github.com/tikv/pd/pkg/storage"
+	"github.com/tikv/pd/pkg/storage/endpoint"
+	"github.com/tikv/pd/pkg/storage/kv"
 	"github.com/tikv/pd/pkg/utils/apiutil"
 	"github.com/tikv/pd/pkg/utils/apiutil/multiservicesapi"
 	"github.com/tikv/pd/pkg/utils/logutil"
 	"github.com/tikv/pd/pkg/utils/typeutil"
 	"github.com/unrolled/render"
+	"go.etcd.io/etcd/clientv3"
 )
 
 // APIPathPrefix is the prefix of the API path.
@@ -112,6 +116,7 @@
 		rd:               createIndentRender(),
 	}
 	s.RegisterAdminRouter()
+	s.RegisterMemberRouter()
 	s.RegisterConfigRouter()
 	s.RegisterOperatorsRouter()
 	s.RegisterSchedulersRouter()
@@ -130,7 +135,13 @@
 	router.DELETE("cache/regions/:id", deleteRegionCacheByID)
 }
 
-// RegisterSchedulersRouter registers the router of the schedulers handler.
+// RegisterMemberRouter registers the router of the member handler.
+func (s *Service) RegisterMemberRouter() {
+	router := s.root.Group("member")
+	router.POST("/primary/transfer", transferPrimary)
+}
+
+// RegisterSchedulersRouter registers the router of the schedulers' handler.
 func (s *Service) RegisterSchedulersRouter() {
 	router := s.root.Group("schedulers")
 	router.GET("", getSchedulers)
@@ -259,6 +270,47 @@
 	c.IndentedJSON(http.StatusOK, cfg)
 }
 
+func transferPrimary(c *gin.Context) {
+	if len(c.Request.Header.Get(multiservicesapi.ServiceAllowDirectHandle)) == 0 {
+		c.AbortWithStatusJSON(http.StatusInternalServerError, "please add `service-allow-direct-handle` in header")
+		return
+	}
+
+	svr := c.MustGet(multiservicesapi.ServiceContextKey).(*scheserver.Server)
+	if svr.IsServing() {
+		c.AbortWithStatusJSON(http.StatusInternalServerError, "now is primary")
+		return
+	}
+
+	newLease := election.NewLease(svr.GetClient(), "primary election")
+	if err := newLease.Grant(mcsutils.DefaultLeaderLease); err != nil {
+		c.AbortWithStatusJSON(http.StatusInternalServerError, "newLease grant error")
+	}
+
+	// delete previous primary firstly
+	primaryKey := endpoint.SchedulingPrimaryPath(svr.GetClusterID())
+	deleteResp, err := kv.NewSlowLogTxn(svr.GetClient()).
+		Then(
+			clientv3.OpDelete(primaryKey),
+		).Commit()
+	if err != nil || !deleteResp.Succeeded {
+		c.AbortWithStatusJSON(http.StatusInternalServerError, "delete resp error")
+	}
+
+	memberValue := svr.GetParticipant()
+	memberValue.GetLeadership().SetLease(newLease)
+	putResp, err := kv.NewSlowLogTxn(svr.GetClient()).
+		Then(
+			clientv3.OpPut(primaryKey, memberValue.MemberValue(), clientv3.WithLease(newLease.ID.Load().(clientv3.LeaseID))),
+		).
+		Commit()
+	if err != nil || !putResp.Succeeded {
+		c.AbortWithStatusJSON(http.StatusInternalServerError, "put resp error")
+	}
+
+	c.IndentedJSON(http.StatusOK, "transfer submitted!")
+}
+
 // @Tags     admin
 // @Summary  Drop all regions from cache.
 // @Produce  json

pkg/mcs/scheduling/server/server.go

@@ -128,6 +128,14 @@
 	return s.cfg.BackendEndpoints
 }
 
+func (s *Server) GetClusterID() uint64 {
+	return s.clusterID
+}
+
+func (s *Server) GetParticipant() *member.Participant {
+	return s.participant
+}
+
 // SetLogLevel sets log level.
 func (s *Server) SetLogLevel(level string) error {
 	if !logutil.IsLevelLegal(level) {
@@ -249,18 +257,35 @@
 
 func (s *Server) campaignLeader() {
 	log.Info("start to campaign the primary/leader", zap.String("campaign-scheduling-primary-name", s.participant.Name()))
-	if err := s.participant.CampaignLeader(s.Context(), s.cfg.LeaderLease); err != nil {
-		if err.Error() == errs.ErrEtcdTxnConflict.Error() {
-			log.Info("campaign scheduling primary meets error due to txn conflict, another server may campaign successfully",
-				zap.String("campaign-scheduling-primary-name", s.participant.Name()))
-		} else {
-			log.Error("campaign scheduling primary meets error due to etcd error",
-				zap.String("campaign-scheduling-primary-name", s.participant.Name()),
-				errs.ZapError(err))
+	leader, _, err := s.participant.GetPersistentLeader()
+	if err != nil {
+		log.Error("getting the leader meets error", errs.ZapError(err))
+		return
+	}
+	if leader != nil && !s.participant.IsSameLeader(leader) {
+		leader, ok := leader.(*schedulingpb.Participant)
+		if !ok {
+			log.Error("failed to get the leader", zap.Any("leader", leader))
+			return
 		}
+		log.Info("the scheduling primary/leader is already elected", zap.Stringer("scheduling-primary", leader))
 		return
 	}
 
+	if leader == nil {
+		if err := s.participant.CampaignLeader(s.Context(), s.cfg.LeaderLease); err != nil {
+			if err.Error() == errs.ErrEtcdTxnConflict.Error() {
+				log.Info("campaign scheduling primary meets error due to txn conflict, another server may campaign successfully",
+					zap.String("campaign-scheduling-primary-name", s.participant.Name()))
+			} else {
+				log.Error("campaign scheduling primary meets error due to etcd error",
+					zap.String("campaign-scheduling-primary-name", s.participant.Name()),
+					errs.ZapError(err))
+			}
+			return
+		}
+	}
+
 	// Start keepalive the leadership and enable Scheduling service.
 	ctx, cancel := context.WithCancel(s.serverLoopCtx)
 	var resetLeaderOnce sync.Once
@@ -290,6 +315,19 @@
 	member.ServiceMemberGauge.WithLabelValues(serviceName).Set(1)
 	log.Info("scheduling primary is ready to serve", zap.String("scheduling-primary-name", s.participant.Name()))
 
+	go func() {
+		log.Info("[primary] start to watch the primary", zap.Stringer("scheduling-primary", s.participant.GetLeader()))
+		_, revision, err := s.participant.GetPersistentLeader()
+		if err != nil {
+			log.Error("[primary] getting the leader meets error", errs.ZapError(err))
+			return
+		}
+		// Watch will keep looping and never return unless the primary/leader has changed.
+		s.participant.GetLeadership().Watch(s.serverLoopCtx, revision)
+		s.participant.UnsetLeader()
+		log.Info("[primary] the scheduling primary has changed, try to re-campaign a primary")
+	}()
+
 	leaderTicker := time.NewTicker(utils.LeaderTickInterval)
 	defer leaderTicker.Stop()
 

pkg/mcs/tso/server/apis/v1/api.go

@@ -25,14 +25,17 @@
 	"github.com/gin-gonic/gin"
 	"github.com/pingcap/kvproto/pkg/tsopb"
 	"github.com/pingcap/log"
+	"github.com/tikv/pd/pkg/election"
 	"github.com/tikv/pd/pkg/errs"
 	tsoserver "github.com/tikv/pd/pkg/mcs/tso/server"
 	"github.com/tikv/pd/pkg/mcs/utils"
 	"github.com/tikv/pd/pkg/storage/endpoint"
+	"github.com/tikv/pd/pkg/storage/kv"
 	"github.com/tikv/pd/pkg/utils/apiutil"
 	"github.com/tikv/pd/pkg/utils/apiutil/multiservicesapi"
 	"github.com/tikv/pd/pkg/utils/logutil"
 	"github.com/unrolled/render"
+	"go.etcd.io/etcd/clientv3"
 	"go.uber.org/zap"
 )
 
@@ -94,6 +97,7 @@
 		rd:               createIndentRender(),
 	}
 	s.RegisterAdminRouter()
+	s.RegisterMemberRouter()
 	s.RegisterKeyspaceGroupRouter()
 	s.RegisterHealthRouter()
 	s.RegisterConfigRouter()
@@ -107,6 +111,12 @@
 	router.PUT("/log", changeLogLevel)
 }
 
+// RegisterMemberRouter registers the router of the member handler.
+func (s *Service) RegisterMemberRouter() {
+	router := s.root.Group("member")
+	router.POST("/primary/transfer", transferPrimary)
+}
+
 // RegisterKeyspaceGroupRouter registers the router of the TSO keyspace group handler.
 func (s *Service) RegisterKeyspaceGroupRouter() {
 	router := s.root.Group("keyspace-groups")
@@ -141,6 +151,51 @@
 	c.String(http.StatusOK, "The log level is updated.")
 }
 
+func transferPrimary(c *gin.Context) {
+	if len(c.Request.Header.Get(multiservicesapi.ServiceAllowDirectHandle)) == 0 {
+		c.AbortWithStatusJSON(http.StatusInternalServerError, "please add `service-allow-direct-handle` in header")
+		return
+	}
+
+	svr := c.MustGet(multiservicesapi.ServiceContextKey).(*tsoserver.Service)
+	if svr.IsServing() {
+		c.AbortWithStatusJSON(http.StatusInternalServerError, "now is primary")
+		return
+	}
+
+	newLease := election.NewLease(svr.GetClient(), "transfer-primary")
+	if err := newLease.Grant(utils.DefaultLeaderLease); err != nil {
+		c.AbortWithStatusJSON(http.StatusInternalServerError, "newLease grant error")
+	}
+
+	// delete previous primary firstly
+	tsoRootPath := endpoint.TSOSvcRootPath(svr.GetClusterID())
+	primaryKey := endpoint.KeyspaceGroupPrimaryPath(tsoRootPath, utils.DefaultKeyspaceGroupID)
+	deleteResp, err := kv.NewSlowLogTxn(svr.GetClient()).
+		Then(
+			clientv3.OpDelete(primaryKey),
+		).Commit()
+	if err != nil || !deleteResp.Succeeded {
+		c.AbortWithStatusJSON(http.StatusInternalServerError, "delete resp error")
+	}
+
+	memberValue, err := svr.GetMember(utils.DefaultKeyspaceID, utils.DefaultKeyspaceGroupID)
+	memberValue.GetLeadership().SetLease(newLease)
+	if err != nil {
+		c.AbortWithStatusJSON(http.StatusInternalServerError, "get tso member")
+	}
+	putResp, err := kv.NewSlowLogTxn(svr.GetClient()).
+		Then(
+			clientv3.OpPut(primaryKey, memberValue.MemberValue(), clientv3.WithLease(newLease.ID.Load().(clientv3.LeaseID))),
+		).
+		Commit()
+	if err != nil || !putResp.Succeeded {
+		c.AbortWithStatusJSON(http.StatusInternalServerError, "put resp error")
+	}
+
+	c.IndentedJSON(http.StatusOK, "transfer submitted!")
+}
+
 // ResetTSParams is the input json body params of ResetTS
 type ResetTSParams struct {
 	TSO           string `json:"tso"`

pkg/mcs/tso/server/server.go

@@ -121,6 +121,10 @@
 	s.serverLoopWg.Add(n)
 }
 
+func (s *Server) GetClusterID() uint64 {
+	return s.clusterID
+}
+
 // SetUpRestHandler sets up the REST handler.
 func (s *Server) SetUpRestHandler() (http.Handler, apiutil.APIServiceGroup) {
 	return SetUpRestHandler(s.service)