mcs: use separate participant (tikv#7032)

ref tikv#5839

Signed-off-by: Ryan Leung <rleungx@gmail.com>

Co-authored-by: ti-chi-bot[bot] <108142056+ti-chi-bot[bot]@users.noreply.github.com>

pd.code-workspace

@@ -23,6 +23,10 @@
 		{
 			"name": "pd-tso-bench",
 			"path": "tools/pd-tso-bench"
+		},
+		{
+			"name": "pd-api-bench",
+			"path": "tools/pd-api-bench"
 		}
 	],
 	"settings": {}

pkg/mcs/resourcemanager/server/server.go

@@ -16,11 +16,9 @@ package server
 
 import (
 	"context"
-	"fmt"
 	"net/http"
 	"os"
 	"os/signal"
-	"path"
 	"strconv"
 	"sync"
 	"sync/atomic"
@@ -30,6 +28,7 @@ import (
 	grpcprometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/kvproto/pkg/diagnosticspb"
+	"github.com/pingcap/kvproto/pkg/resource_manager"
 	"github.com/pingcap/log"
 	"github.com/pingcap/sysutil"
 	"github.com/spf13/cobra"
@@ -284,10 +283,13 @@ func (s *Server) startServer() (err error) {
 	uniqueName := s.cfg.ListenAddr
 	uniqueID := memberutil.GenerateUniqueID(uniqueName)
 	log.Info("joining primary election", zap.String("participant-name", uniqueName), zap.Uint64("participant-id", uniqueID))
-	resourceManagerPrimaryPrefix := endpoint.ResourceManagerSvcRootPath(s.clusterID)
-	s.participant = member.NewParticipant(s.GetClient())
-	s.participant.InitInfo(uniqueName, uniqueID, path.Join(resourceManagerPrimaryPrefix, fmt.Sprintf("%05d", 0)),
-		utils.PrimaryKey, "primary election", s.cfg.AdvertiseListenAddr)
+	s.participant = member.NewParticipant(s.GetClient(), utils.ResourceManagerServiceName)
+	p := &resource_manager.Participant{
+		Name:       uniqueName,
+		Id:         uniqueID, // id is unique among all participants
+		ListenUrls: []string{s.cfg.AdvertiseListenAddr},
+	}
+	s.participant.InitInfo(p, endpoint.ResourceManagerSvcRootPath(s.clusterID), utils.PrimaryKey, "primary election")
 
 	s.service = &Service{
 		ctx:     s.Context(),

pkg/mcs/scheduling/server/server.go

@@ -20,7 +20,6 @@ import (
 	"net/http"
 	"os"
 	"os/signal"
-	"path"
 	"strconv"
 	"sync"
 	"sync/atomic"
@@ -31,6 +30,7 @@ import (
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/kvproto/pkg/diagnosticspb"
 	"github.com/pingcap/kvproto/pkg/pdpb"
+	"github.com/pingcap/kvproto/pkg/schedulingpb"
 	"github.com/pingcap/log"
 	"github.com/pingcap/sysutil"
 	"github.com/spf13/cobra"
@@ -370,10 +370,13 @@ func (s *Server) startServer() (err error) {
 	uniqueName := s.cfg.ListenAddr
 	uniqueID := memberutil.GenerateUniqueID(uniqueName)
 	log.Info("joining primary election", zap.String("participant-name", uniqueName), zap.Uint64("participant-id", uniqueID))
-	schedulingPrimaryPrefix := endpoint.SchedulingSvcRootPath(s.clusterID)
-	s.participant = member.NewParticipant(s.GetClient())
-	s.participant.InitInfo(uniqueName, uniqueID, path.Join(schedulingPrimaryPrefix, fmt.Sprintf("%05d", 0)),
-		utils.PrimaryKey, "primary election", s.cfg.AdvertiseListenAddr)
+	s.participant = member.NewParticipant(s.GetClient(), utils.SchedulingServiceName)
+	p := &schedulingpb.Participant{
+		Name:       uniqueName,
+		Id:         uniqueID, // id is unique among all participants
+		ListenUrls: []string{s.cfg.AdvertiseListenAddr},
+	}
+	s.participant.InitInfo(p, endpoint.SchedulingSvcRootPath(s.clusterID), utils.PrimaryKey, "primary election")
 	s.basicCluster = core.NewBasicCluster()
 	err = s.startWatcher()
 	if err != nil {

pkg/member/election_leader.go

@@ -18,7 +18,6 @@ import (
 	"context"
 
 	"github.com/pingcap/kvproto/pkg/pdpb"
-	"github.com/pingcap/kvproto/pkg/tsopb"
 )
 
 // ElectionLeader defines the common interface of the leader, which is the pdpb.Member
@@ -64,14 +63,14 @@ func (l *EmbeddedEtcdLeader) Watch(ctx context.Context) {
 // EtcdLeader is the leader in the election group backed by the etcd, but it's
 // decoupled from the embedded etcd.
 type EtcdLeader struct {
-	wrapper      *Participant
-	pariticipant *tsopb.Participant
-	revision     int64
+	wrapper     *Participant
+	participant participant
+	revision    int64
 }
 
 // GetListenUrls returns current leader's client urls
 func (l *EtcdLeader) GetListenUrls() []string {
-	return l.pariticipant.GetListenUrls()
+	return l.participant.GetListenUrls()
 }
 
 // GetRevision the revision of the leader in etcd
@@ -81,10 +80,10 @@ func (l *EtcdLeader) GetRevision() int64 {
 
 // String declares fmt.Stringer
 func (l *EtcdLeader) String() string {
-	return l.pariticipant.String()
+	return l.participant.String()
 }
 
 // Watch on the leader
 func (l *EtcdLeader) Watch(ctx context.Context) {
-	l.wrapper.WatchLeader(ctx, l.pariticipant, l.revision)
+	l.wrapper.WatchLeader(ctx, l.participant, l.revision)
 }

pkg/member/participant.go

@@ -22,28 +22,42 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/pingcap/kvproto/pkg/resource_manager"
+	"github.com/pingcap/kvproto/pkg/schedulingpb"
 	"github.com/pingcap/kvproto/pkg/tsopb"
 	"github.com/pingcap/log"
 	"github.com/tikv/pd/pkg/election"
 	"github.com/tikv/pd/pkg/errs"
+	"github.com/tikv/pd/pkg/mcs/utils"
 	"github.com/tikv/pd/pkg/utils/etcdutil"
 	"go.etcd.io/etcd/clientv3"
 	"go.uber.org/zap"
 )
 
 type leadershipCheckFunc func(*election.Leadership) bool
 
+type participant interface {
+	GetName() string
+	GetId() uint64
+	GetListenUrls() []string
+	String() string
+	Marshal() ([]byte, error)
+	Reset()
+	ProtoMessage()
+}
+
 // Participant is used for the election related logic. Compared to its counterpart
 // EmbeddedEtcdMember, Participant relies on etcd for election, but it's decoupled
 // from the embedded etcd. It implements Member interface.
 type Participant struct {
 	leadership *election.Leadership
 	// stored as member type
-	leader     atomic.Value
-	client     *clientv3.Client
-	rootPath   string
-	leaderPath string
-	member     *tsopb.Participant
+	leader      atomic.Value
+	client      *clientv3.Client
+	rootPath    string
+	leaderPath  string
+	member      participant
+	serviceName string
 	// memberValue is the serialized string of `member`. It will be saved in the
 	// leader key when this participant is successfully elected as the leader of
 	// the group. Every write will use it to check the leadership.
@@ -56,42 +70,37 @@ type Participant struct {
 }
 
 // NewParticipant create a new Participant.
-func NewParticipant(client *clientv3.Client) *Participant {
+func NewParticipant(client *clientv3.Client, serviceName string) *Participant {
 	return &Participant{
-		client: client,
+		client:      client,
+		serviceName: serviceName,
 	}
 }
 
 // InitInfo initializes the member info. The leader key is path.Join(rootPath, leaderName)
-func (m *Participant) InitInfo(name string, id uint64, rootPath string, leaderName string, purpose string, advertiseListenAddr string) {
-	leader := &tsopb.Participant{
-		Name:       name,
-		Id:         id, // id is unique among all participants
-		ListenUrls: []string{advertiseListenAddr},
-	}
-
-	data, err := leader.Marshal()
+func (m *Participant) InitInfo(p participant, rootPath string, leaderName string, purpose string) {
+	data, err := p.Marshal()
 	if err != nil {
 		// can't fail, so panic here.
-		log.Fatal("marshal leader meet error", zap.Stringer("leader-name", leader), errs.ZapError(errs.ErrMarshalLeader, err))
+		log.Fatal("marshal leader meet error", zap.String("member-name", p.String()), errs.ZapError(errs.ErrMarshalLeader, err))
 	}
-	m.member = leader
+	m.member = p
 	m.memberValue = string(data)
 	m.rootPath = rootPath
 	m.leaderPath = path.Join(rootPath, leaderName)
 	m.leadership = election.NewLeadership(m.client, m.GetLeaderPath(), purpose)
 	m.lastLeaderUpdatedTime.Store(time.Now())
-	log.Info("participant joining election", zap.Stringer("participant-info", m.member), zap.String("leader-path", m.leaderPath))
+	log.Info("participant joining election", zap.String("participant-info", p.String()), zap.String("leader-path", m.leaderPath))
 }
 
 // ID returns the unique ID for this participant in the election group
 func (m *Participant) ID() uint64 {
-	return m.member.Id
+	return m.member.GetId()
 }
 
 // Name returns the unique name in the election group.
 func (m *Participant) Name() string {
-	return m.member.Name
+	return m.member.GetName()
 }
 
 // GetMember returns the member.
@@ -112,6 +121,9 @@ func (m *Participant) Client() *clientv3.Client {
 // IsLeader returns whether the participant is the leader or not by checking its leadership's
 // lease and leader info.
 func (m *Participant) IsLeader() bool {
+	if m.GetLeader() == nil {
+		return false
+	}
 	return m.leadership.Check() && m.GetLeader().GetId() == m.member.GetId() && m.campaignCheck()
 }
 
@@ -122,6 +134,9 @@ func (m *Participant) IsLeaderElected() bool {
 
 // GetLeaderListenUrls returns current leader's listen urls
 func (m *Participant) GetLeaderListenUrls() []string {
+	if m.GetLeader() == nil {
+		return nil
+	}
 	return m.GetLeader().GetListenUrls()
 }
 
@@ -131,27 +146,36 @@ func (m *Participant) GetLeaderID() uint64 {
 }
 
 // GetLeader returns current leader of the election group.
-func (m *Participant) GetLeader() *tsopb.Participant {
+func (m *Participant) GetLeader() participant {
 	leader := m.leader.Load()
 	if leader == nil {
 		return nil
 	}
-	member := leader.(*tsopb.Participant)
+	member := leader.(participant)
 	if member.GetId() == 0 {
 		return nil
 	}
 	return member
 }
 
 // setLeader sets the member's leader.
-func (m *Participant) setLeader(member *tsopb.Participant) {
+func (m *Participant) setLeader(member participant) {
 	m.leader.Store(member)
 	m.lastLeaderUpdatedTime.Store(time.Now())
 }
 
 // unsetLeader unsets the member's leader.
 func (m *Participant) unsetLeader() {
-	m.leader.Store(&tsopb.Participant{})
+	var leader participant
+	switch m.serviceName {
+	case utils.TSOServiceName:
+		leader = &tsopb.Participant{}
+	case utils.SchedulingServiceName:
+		leader = &schedulingpb.Participant{}
+	case utils.ResourceManagerServiceName:
+		leader = &resource_manager.Participant{}
+	}
+	m.leader.Store(leader)
 	m.lastLeaderUpdatedTime.Store(time.Now())
 }
 
@@ -200,8 +224,16 @@ func (m *Participant) PreCheckLeader() error {
 }
 
 // getPersistentLeader gets the corresponding leader from etcd by given leaderPath (as the key).
-func (m *Participant) getPersistentLeader() (*tsopb.Participant, int64, error) {
-	leader := &tsopb.Participant{}
+func (m *Participant) getPersistentLeader() (participant, int64, error) {
+	var leader participant
+	switch m.serviceName {
+	case utils.TSOServiceName:
+		leader = &tsopb.Participant{}
+	case utils.SchedulingServiceName:
+		leader = &schedulingpb.Participant{}
+	case utils.ResourceManagerServiceName:
+		leader = &resource_manager.Participant{}
+	}
 	ok, rev, err := etcdutil.GetProtoMsgWithModRev(m.client, m.GetLeaderPath(), leader)
 	if err != nil {
 		return nil, 0, err
@@ -248,14 +280,14 @@ func (m *Participant) CheckLeader() (ElectionLeader, bool) {
 	}
 
 	return &EtcdLeader{
-		wrapper:      m,
-		pariticipant: leader,
-		revision:     revision,
+		wrapper:     m,
+		participant: leader,
+		revision:    revision,
 	}, false
 }
 
 // WatchLeader is used to watch the changes of the leader.
-func (m *Participant) WatchLeader(ctx context.Context, leader *tsopb.Participant, revision int64) {
+func (m *Participant) WatchLeader(ctx context.Context, leader participant, revision int64) {
 	m.setLeader(leader)
 	m.leadership.Watch(ctx, revision)
 	m.unsetLeader()
@@ -269,7 +301,7 @@ func (m *Participant) ResetLeader() {
 }
 
 // IsSameLeader checks whether a server is the leader itself.
-func (m *Participant) IsSameLeader(leader *tsopb.Participant) bool {
+func (m *Participant) IsSameLeader(leader participant) bool {
 	return leader.GetId() == m.ID()
 }
 

pkg/tso/keyspace_group_manager.go

@@ -29,6 +29,7 @@ import (
 	perrors "github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/kvproto/pkg/pdpb"
+	"github.com/pingcap/kvproto/pkg/tsopb"
 	"github.com/pingcap/log"
 	"github.com/tikv/pd/pkg/election"
 	"github.com/tikv/pd/pkg/errs"
@@ -736,10 +737,13 @@ func (kgm *KeyspaceGroupManager) updateKeyspaceGroup(group *endpoint.KeyspaceGro
 		zap.String("participant-name", uniqueName),
 		zap.Uint64("participant-id", uniqueID))
 	// Initialize the participant info to join the primary election.
-	participant := member.NewParticipant(kgm.etcdClient)
-	participant.InitInfo(
-		uniqueName, uniqueID, endpoint.KeyspaceGroupsElectionPath(kgm.tsoSvcRootPath, group.ID),
-		mcsutils.PrimaryKey, "keyspace group primary election", kgm.cfg.GetAdvertiseListenAddr())
+	participant := member.NewParticipant(kgm.etcdClient, mcsutils.TSOServiceName)
+	p := &tsopb.Participant{
+		Name:       uniqueName,
+		Id:         uniqueID, // id is unique among all participants
+		ListenUrls: []string{kgm.cfg.GetAdvertiseListenAddr()},
+	}
+	participant.InitInfo(p, endpoint.KeyspaceGroupsElectionPath(kgm.tsoSvcRootPath, group.ID), mcsutils.PrimaryKey, "keyspace group primary election")
 	// If the keyspace group is in split, we should ensure that the primary elected by the new keyspace group
 	// is always on the same TSO Server node as the primary of the old keyspace group, and this constraint cannot
 	// be broken until the entire split process is completed.

tests/integrations/mcs/resourcemanager/resource_manager_test.go

@@ -675,7 +675,7 @@ func (suite *resourceManagerClientTestSuite) TestBasicResourceGroupCURD() {
 		},
 
 		{"test2", rmpb.GroupMode_RUMode, true, true,
-			`{"name":"test2","mode":1,"r_u_settings":{"r_u":{"settings":{"fill_rate":20000},"state":{"initialized":false}}},"priority":0}`,
+			`{"name":"test2","mode":1,"r_u_settings":{"r_u":{"settings":{"fill_rate":20000},"state":{"initialized":false}}},"priority":0,"runaway_settings":{"rule":{"exec_elapsed_time_ms":10000},"action":2},"background_settings":{"job_types":["test"]}}`,
 			func(gs *rmpb.ResourceGroup) {
 				gs.RUSettings = &rmpb.GroupRequestUnitSettings{
 					RU: &rmpb.TokenBucket{
@@ -687,7 +687,7 @@ func (suite *resourceManagerClientTestSuite) TestBasicResourceGroupCURD() {
 			},
 		},
 		{"test2", rmpb.GroupMode_RUMode, false, true,
-			`{"name":"test2","mode":1,"r_u_settings":{"r_u":{"settings":{"fill_rate":30000,"burst_limit":-1},"state":{"initialized":false}}},"priority":0}`,
+			`{"name":"test2","mode":1,"r_u_settings":{"r_u":{"settings":{"fill_rate":30000,"burst_limit":-1},"state":{"initialized":false}}},"priority":0,"runaway_settings":{"rule":{"exec_elapsed_time_ms":1000},"action":3,"watch":{"lasting_duration_ms":100000,"type":2}},"background_settings":{"job_types":["br","lightning"]}}`,
 			func(gs *rmpb.ResourceGroup) {
 				gs.RUSettings = &rmpb.GroupRequestUnitSettings{
 					RU: &rmpb.TokenBucket{

tests/integrations/tso/go.mod

@@ -12,9 +12,9 @@ replace (
 replace google.golang.org/grpc v1.54.0 => google.golang.org/grpc v1.26.0
 
 require (
-	github.com/pingcap/failpoint v0.0.0-20210918120811-547c13e3eb00
+	github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c
 	github.com/pingcap/kvproto v0.0.0-20230925123611-87bebcc0d071
-	github.com/stretchr/testify v1.8.2
+	github.com/stretchr/testify v1.8.4
 	github.com/tikv/pd v0.0.0-00010101000000-000000000000
 	github.com/tikv/pd/client v0.0.0-00010101000000-000000000000
 	github.com/tikv/pd/tests/integrations/mcs v0.0.0-00010101000000-000000000000