pdms: support primary/transfer api for scheduling and tso

client/http/api.go

@@ -205,6 +205,11 @@ func MicroServicePrimary(service string) string {
 	return fmt.Sprintf("%s/primary/%s", microServicePrefix, service)
 }
 
+// MicroServicePrimaryTransfer returns the path of PD HTTP API to transfer the primary of microservice.
+func MicroServicePrimaryTransfer(service string) string {
+	return fmt.Sprintf("%s/primary/transfer/%s", microServicePrefix, service)
+}
+
 // GetUpdateKeyspaceConfigURL returns the path of PD HTTP API to update keyspace config.
 func GetUpdateKeyspaceConfigURL(keyspaceName string) string {
 	return fmt.Sprintf(KeyspaceConfig, keyspaceName)

client/http/interface.go

@@ -95,6 +95,7 @@
 	/* Micro Service interfaces */
 	GetMicroServiceMembers(context.Context, string) ([]MicroServiceMember, error)
 	GetMicroServicePrimary(context.Context, string) (string, error)
+	TransferMicroServicePrimary(context.Context, string, string) error
 	DeleteOperators(context.Context) error
 
 	/* Keyspace interface */
@@ -908,6 +909,22 @@
 	return primary, err
 }
 
+func (c *client) TransferMicroServicePrimary(ctx context.Context, service, newPrimary string) error {
+	reqData, err := json.Marshal(struct {
+		NewPrimary string `json:"new_primary"`
+	}{
+		NewPrimary: newPrimary,
+	})
+	if err != nil {
+		return errors.Trace(err)
+	}
+	return c.request(ctx, newRequestInfo().
+		WithName(transferMicroServicePrimaryName).
+		WithURI(MicroServicePrimaryTransfer(service)).
+		WithMethod(http.MethodPost).
+		WithBody(reqData))
+}
+
 // GetPDVersion gets the release version of the PD binary.
 func (c *client) GetPDVersion(ctx context.Context) (string, error) {
 	var ver struct {

client/http/request_info.go

@@ -72,6 +72,7 @@ const (
 	getMinResolvedTSByStoresIDsName         = "GetMinResolvedTSByStoresIDs"
 	getMicroServiceMembersName              = "GetMicroServiceMembers"
 	getMicroServicePrimaryName              = "GetMicroServicePrimary"
+	transferMicroServicePrimaryName         = "TransferMicroServicePrimary"
 	getPDVersionName                        = "GetPDVersion"
 	resetTSName                             = "ResetTS"
 	resetBaseAllocIDName                    = "ResetBaseAllocID"

cmd/pd-server/main.go

@@ -94,6 +94,7 @@ func NewTSOServiceCommand() *cobra.Command {
 		Short: "Run the TSO service",
 		Run:   tso.CreateServerWrapper,
 	}
+	cmd.Flags().StringP("name", "", "", "human-readable name for this TSO member")
 	cmd.Flags().BoolP("version", "V", false, "print version information and exit")
 	cmd.Flags().StringP("config", "", "", "config file")
 	cmd.Flags().StringP("backend-endpoints", "", "", "url for etcd client")
@@ -114,6 +115,7 @@ func NewSchedulingServiceCommand() *cobra.Command {
 		Short: "Run the scheduling service",
 		Run:   scheduling.CreateServerWrapper,
 	}
+	cmd.Flags().StringP("name", "", "", "human-readable name for this Scheduling member")
 	cmd.Flags().BoolP("version", "V", false, "print version information and exit")
 	cmd.Flags().StringP("config", "", "", "config file")
 	cmd.Flags().StringP("backend-endpoints", "", "", "url for etcd client")
@@ -134,6 +136,7 @@ func NewResourceManagerServiceCommand() *cobra.Command {
 		Short: "Run the resource manager service",
 		Run:   resource_manager.CreateServerWrapper,
 	}
+	cmd.Flags().StringP("name", "", "", "human-readable name for this resource management member")
 	cmd.Flags().BoolP("version", "V", false, "print version information and exit")
 	cmd.Flags().StringP("config", "", "", "config file")
 	cmd.Flags().StringP("backend-endpoints", "", "", "url for etcd client")

pkg/election/leadership.go

@@ -64,6 +64,8 @@
 	leaderKey   string
 	leaderValue string
 
+	leaderWatch atomic.Bool
+
 	keepAliveCtx            context.Context
 	keepAliveCancelFunc     context.CancelFunc
 	keepAliveCancelFuncLock syncutil.Mutex
@@ -72,6 +74,10 @@
 	campaignTimes []time.Time
 }
 
+func (ls *Leadership) GetLeaderValue() string {
+	return ls.leaderValue
+}
+
 // NewLeadership creates a new Leadership.
 func NewLeadership(client *clientv3.Client, leaderKey, purpose string) *Leadership {
 	leadership := &Leadership{
@@ -113,6 +119,16 @@
 	return ls.leaderKey
 }
 
+// SetLeaderWatch sets the leader watch flag.
+func (ls *Leadership) SetLeaderWatch(val bool) {
+	ls.leaderWatch.Store(val)
+}
+
+// IsLeader gets the leader watch flag.
+func (ls *Leadership) IsLeader() bool {
+	return ls.leaderWatch.Load()
+}
+
 // GetCampaignTimesNum is used to get the campaign times of the leader within `campaignTimesRecordTimeout`.
 func (ls *Leadership) GetCampaignTimesNum() int {
 	if ls == nil {
@@ -375,6 +391,12 @@
 						zap.Int64("revision", wresp.Header.Revision), zap.String("leader-key", ls.leaderKey), zap.String("purpose", ls.purpose))
 					return
 				}
+				// only API update the leader key to transfer the primary will meet
+				if ev.Type == mvccpb.PUT && ls.IsLeader() {
+					log.Info("[PrimaryWatch] current leadership is updated",
+						zap.Int64("revision", wresp.Header.Revision), zap.String("leader-key", ls.leaderKey), zap.String("purpose", ls.purpose))
+					return
+				}
 			}
 			revision = wresp.Header.Revision + 1
 		}
@@ -393,4 +415,5 @@
 	}
 	ls.keepAliveCancelFuncLock.Unlock()
 	ls.getLease().Close()
+	ls.SetLeaderWatch(false)
 }

pkg/mcs/discovery/discover.go

@@ -15,12 +15,15 @@
 package discovery
 
 import (
+	"math/rand"
 	"strconv"
+	"time"
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	"github.com/tikv/pd/pkg/errs"
 	"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/etcdutil"
 	"go.etcd.io/etcd/clientv3"
@@ -45,14 +48,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 +78,60 @@
 		return entries, nil
 	}
 
-	return nil, errors.Errorf("unknown service name %s", name)
+	return nil, errors.Errorf("unknown service name %s", serviceName)
+}
+
+// TransferPrimary transfers the primary of the specified service.
+func TransferPrimary(client *clientv3.Client, serviceName, oldPrimary, newPrimary string, keyspaceGroupID uint32) error {
+	log.Info("transfer primary", zap.String("service", serviceName), zap.String("from", oldPrimary), zap.String("to", newPrimary))
+	entries, err := GetMSMembers(serviceName, client)
+	if err != nil {
+		return err
+	}
+
+	// Do nothing when I am the only member of cluster.
+	if len(entries) == 1 {
+		return errors.New("no valid secondary to transfer primary")
+	}
+
+	var primaryIDs []string
+	var secondaryValues []string
+	for _, member := range entries {
+		if (newPrimary == "" && member.ServiceAddr != oldPrimary) || (newPrimary != "" && member.Name == newPrimary) {
+			primaryIDs = append(primaryIDs, member.ServiceAddr)
+			if string(member.MemberValue) == "" {
+				return errors.New("member value is empty")
+			}
+			secondaryValues = append(secondaryValues, string(member.MemberValue))
+		}
+	}
+	if len(primaryIDs) == 0 {
+		return errors.New("no valid secondary to transfer primary")
+	}
+
+	r := rand.New(rand.NewSource(time.Now().UnixNano()))
+	nextPrimaryID := r.Intn(len(primaryIDs))
+
+	clusterID, err := etcdutil.GetClusterID(client, utils.ClusterIDPath)
+	if err != nil {
+		return errors.Errorf("failed to get cluster ID: %v", err)
+	}
+
+	var primaryKey string
+	switch serviceName {
+	case utils.SchedulingServiceName:
+		primaryKey = endpoint.SchedulingPrimaryPath(clusterID)
+	case utils.TSOServiceName:
+		tsoRootPath := endpoint.TSOSvcRootPath(clusterID)
+		primaryKey = endpoint.KeyspaceGroupPrimaryPath(tsoRootPath, keyspaceGroupID)
+	}
+
+	// update primary key to notify old primary server.
+	putResp, err := kv.NewSlowLogTxn(client).
+		Then(clientv3.OpPut(primaryKey, secondaryValues[nextPrimaryID])).
+		Commit()
+	if err != nil || !putResp.Succeeded {
+		return errors.Errorf("failed to write primary flag for %s", serviceName)
+	}
+	return nil
 }

pkg/mcs/discovery/registry_entry.go

@@ -23,11 +23,13 @@ import (
 
 // ServiceRegistryEntry is the registry entry of a service
 type ServiceRegistryEntry struct {
+	Name           string `json:"name"`
 	ServiceAddr    string `json:"service-addr"`
 	Version        string `json:"version"`
 	GitHash        string `json:"git-hash"`
 	DeployPath     string `json:"deploy-path"`
 	StartTimestamp int64  `json:"start-timestamp"`
+	MemberValue    []byte `json:"member-value"`
 }
 
 // Serialize this service registry entry

pkg/mcs/resourcemanager/server/config.go

@@ -194,6 +194,7 @@ func (c *Config) Parse(flagSet *pflag.FlagSet) error {
 	}
 
 	// Ignore the error check here
+	configutil.AdjustCommandLineString(flagSet, &c.Name, "name")
 	configutil.AdjustCommandLineString(flagSet, &c.Log.Level, "log-level")
 	configutil.AdjustCommandLineString(flagSet, &c.Log.File.Filename, "log-file")
 	configutil.AdjustCommandLineString(flagSet, &c.Metric.PushAddress, "metrics-addr")

pkg/mcs/resourcemanager/server/testutil.go

@@ -49,16 +49,18 @@ func NewTestServer(ctx context.Context, re *require.Assertions, cfg *Config) (*S
 // GenerateConfig generates a new config with the given options.
 func GenerateConfig(c *Config) (*Config, error) {
 	arguments := []string{
+		"--name=" + c.Name,
 		"--listen-addr=" + c.ListenAddr,
 		"--advertise-listen-addr=" + c.AdvertiseListenAddr,
 		"--backend-endpoints=" + c.BackendEndpoints,
 	}
 
 	flagSet := pflag.NewFlagSet("test", pflag.ContinueOnError)
+	flagSet.StringP("name", "", "", "human-readable name for this resource manager member")
 	flagSet.BoolP("version", "V", false, "print version information and exit")
 	flagSet.StringP("config", "", "", "config file")
 	flagSet.StringP("backend-endpoints", "", "", "url for etcd client")
-	flagSet.StringP("listen-addr", "", "", "listen address for tso service")
+	flagSet.StringP("listen-addr", "", "", "listen address for resource manager service")
 	flagSet.StringP("advertise-listen-addr", "", "", "advertise urls for listen address (default '${listen-addr}')")
 	flagSet.StringP("cacert", "", "", "path of file that contains list of trusted TLS CAs")
 	flagSet.StringP("cert", "", "", "path of file that contains X509 certificate in PEM format")

pkg/mcs/scheduling/server/config/config.go

@@ -104,6 +104,7 @@ func (c *Config) Parse(flagSet *pflag.FlagSet) error {
 	}
 
 	// Ignore the error check here
+	configutil.AdjustCommandLineString(flagSet, &c.Name, "name")
 	configutil.AdjustCommandLineString(flagSet, &c.Log.Level, "log-level")
 	configutil.AdjustCommandLineString(flagSet, &c.Log.File.Filename, "log-file")
 	configutil.AdjustCommandLineString(flagSet, &c.Metric.PushAddress, "metrics-addr")

pkg/mcs/scheduling/server/server.go

@@ -55,6 +55,7 @@
 	"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/etcdutil"
 	"github.com/tikv/pd/pkg/utils/grpcutil"
 	"github.com/tikv/pd/pkg/utils/logutil"
 	"github.com/tikv/pd/pkg/utils/memberutil"
@@ -128,6 +129,10 @@
 	return s.cfg.BackendEndpoints
 }
 
+func (s *Server) GetParticipant() *member.Participant {
+	return s.participant
+}
+
 // SetLogLevel sets log level.
 func (s *Server) SetLogLevel(level string) error {
 	if !logutil.IsLevelLegal(level) {
@@ -243,6 +248,18 @@
 			log.Info("the scheduling primary has changed, try to re-campaign a primary")
 		}
 
+		// To make sure the expected leader(if exist) and primary are on the same server.
+		expectedPrimary := utils.GetExpectedPrimary(s.GetClient(), s.participant.GetLeaderPath())
+		if expectedPrimary != "" && expectedPrimary != s.participant.MemberValue() {
+			log.Info("skip campaigning of scheduling primary and check later",
+				zap.String("server-name", s.Name()),
+				zap.String("target-primary-id", expectedPrimary),
+				zap.Uint64("member-id", s.participant.ID()),
+				zap.String("cur-memberValue", s.participant.MemberValue()))
+			time.Sleep(200 * time.Millisecond)
+			continue
+		}
+
 		s.campaignLeader()
 	}
 }
@@ -290,6 +307,9 @@
 	member.ServiceMemberGauge.WithLabelValues(serviceName).Set(1)
 	log.Info("scheduling primary is ready to serve", zap.String("scheduling-primary-name", s.participant.Name()))
 
+	exitPrimary := make(chan struct{})
+	go s.primaryWatch(ctx, exitPrimary)
+
 	leaderTicker := time.NewTicker(utils.LeaderTickInterval)
 	defer leaderTicker.Stop()
 
@@ -304,6 +324,42 @@
 			// Server is closed and it should return nil.
 			log.Info("server is closed")
 			return
+		case <-exitPrimary:
+			log.Info("no longer a primary because primary have been updated, the scheduling primary will step down")
+			return
+		}
+	}
+}
+
+func (s *Server) primaryWatch(ctx context.Context, exitPrimary chan struct{}) {
+	resp, err := etcdutil.EtcdKVGet(s.participant.GetLeadership().GetClient(), s.participant.GetLeaderPath())
+	if err != nil || resp == nil || len(resp.Kvs) == 0 {
+		log.Error("scheduling primary getting the primary meets error", errs.ZapError(err))
+		return
+	}
+	log.Info("scheduling primary start to watch the primary", zap.Stringer("scheduling-primary", s.participant.GetLeader()))
+	// Watch will keep looping and never return unless the primary has changed.
+	s.participant.GetLeadership().SetLeaderWatch(true)
+	s.participant.GetLeadership().Watch(s.serverLoopCtx, resp.Kvs[0].ModRevision+1)
+	s.participant.GetLeadership().SetLeaderWatch(false)
+
+	// only API update primary will set the expected leader
+	curPrimary, err := etcdutil.GetValue(s.participant.Client(), s.participant.GetLeaderPath())
+	if err != nil {
+		log.Error("scheduling primary getting the leader meets error", errs.ZapError(err))
+		return
+	}
+	// only trigger by updating primary
+	if curPrimary != nil && resp.Kvs[0].Value != nil && string(curPrimary) != string(resp.Kvs[0].Value) {
+		utils.SetExpectedPrimary(s.participant.Client(), s.participant.GetLeaderPath())
+
+		s.participant.UnsetLeader()
+		defer log.Info("scheduling primary exit the primary watch loop")
+		select {
+		case <-ctx.Done():
+			return
+		case exitPrimary <- struct{}{}:
+			return
 		}
 	}
 }
@@ -425,6 +481,7 @@
 		GitHash:        versioninfo.PDGitHash,
 		DeployPath:     deployPath,
 		StartTimestamp: s.StartTimestamp(),
+		Name:           s.Name(),
 	}
 	uniqueName := s.cfg.GetAdvertiseListenAddr()
 	uniqueID := memberutil.GenerateUniqueID(uniqueName)
@@ -436,6 +493,7 @@
 		ListenUrls: []string{s.cfg.GetAdvertiseListenAddr()},
 	}
 	s.participant.InitInfo(p, endpoint.SchedulingSvcRootPath(s.clusterID), utils.PrimaryKey, "primary election")
+	s.serviceID.MemberValue = []byte(s.participant.MemberValue())
 
 	s.service = &Service{Server: s}
 	s.AddServiceReadyCallback(s.startCluster)

pkg/mcs/scheduling/server/testutil.go

@@ -50,12 +50,14 @@ func NewTestServer(ctx context.Context, re *require.Assertions, cfg *config.Conf
 // GenerateConfig generates a new config with the given options.
 func GenerateConfig(c *config.Config) (*config.Config, error) {
 	arguments := []string{
+		"--name=" + c.Name,
 		"--listen-addr=" + c.ListenAddr,
 		"--advertise-listen-addr=" + c.AdvertiseListenAddr,
 		"--backend-endpoints=" + c.BackendEndpoints,
 	}
 
 	flagSet := pflag.NewFlagSet("test", pflag.ContinueOnError)
+	flagSet.StringP("name", "", "", "human-readable name for this scheduling member")
 	flagSet.BoolP("version", "V", false, "print version information and exit")
 	flagSet.StringP("config", "", "", "config file")
 	flagSet.StringP("backend-endpoints", "", "", "url for etcd client")