Skip to content

server, member, election: document and test the election client pinning (#11110) #11188

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ti-chi-bot wants to merge 1 commit into
base: release-8.5
Choose a base branch
from
Open

server, member, election: document and test the election client pinning (#11110) #11188

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 6 additions & 3 deletions pkg/election/leadership.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,8 @@ func GetLeader(c *clientv3.Client, leaderPath string) (*pdpb.Member, int64, erro
type Leadership struct {
// purpose is used to show what this election for
purpose string
// name scopes test failpoints to one member.
name string
// The lease which is used to get this leadership
lease atomic.Value // stored as *lease
client *clientv3.Client
Expand All @@ -76,10 +78,11 @@ type Leadership struct {
primaryWatch atomic.Bool
}

// NewLeadership creates a new Leadership.
func NewLeadership(client *clientv3.Client, leaderKey, purpose string) *Leadership {
// NewLeadership creates a new Leadership for the named member.
func NewLeadership(client *clientv3.Client, leaderKey, purpose, name string) *Leadership {
leadership := &Leadership{
purpose: purpose,
name: name,
client: client,
leaderKey: leaderKey,
campaignTimes: make([]time.Time, 0, defaultCampaignTimesSlot),
Expand Down Expand Up @@ -166,7 +169,7 @@ func (ls *Leadership) AddCampaignTimes() {
func (ls *Leadership) Campaign(leaseTimeout int64, leaderData string, cmps ...clientv3.Cmp) error {
ls.leaderValue = leaderData
// Create a new lease to campaign
newLease := NewLease(ls.client, ls.purpose)
newLease := NewLease(ls.client, ls.purpose, ls.name)
ls.SetLease(newLease)

failpoint.Inject("skipGrantLeader", func(val failpoint.Value) {
Expand Down
67 changes: 60 additions & 7 deletions pkg/election/leadership_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,8 +37,8 @@ func TestLeadership(t *testing.T) {
defer clean()

// Campaign the same leadership
leadership1 := NewLeadership(client, "/test_leader", "test_leader_1")
leadership2 := NewLeadership(client, "/test_leader", "test_leader_2")
leadership1 := NewLeadership(client, "/test_leader", "test_leader_1", "test_leader_1")
leadership2 := NewLeadership(client, "/test_leader", "test_leader_2", "test_leader_2")

// leadership1 starts first and get the leadership
err := leadership1.Campaign(defaultLeaseTimeout, "test_leader_1")
Expand Down Expand Up @@ -111,6 +111,59 @@ func TestLeadership(t *testing.T) {
re.NoError(lease2.Close())
}

<<<<<<< HEAD
=======
func TestDeleteLeaderKeyByRevisionDoesNotDeleteChangedLeader(t *testing.T) {
re := require.New(t)
_, client, clean := etcdutil.NewTestEtcdCluster(t, 1, nil)
defer clean()

leaderKey := "/test_leader"
leadership1 := NewLeadership(client, leaderKey, "test_leader_1", "test_leader_1")
leadership2 := NewLeadership(client, leaderKey, "test_leader_2", "test_leader_2")

err := leadership1.Campaign(defaultLeaseTimeout, "test_leader_1")
re.NoError(err)
resp, err := client.Get(context.Background(), leaderKey)
re.NoError(err)
re.Len(resp.Kvs, 1)
oldRevision := resp.Kvs[0].ModRevision

_, err = client.Delete(context.Background(), leaderKey)
re.NoError(err)
err = leadership2.Campaign(defaultLeaseTimeout, "test_leader_2")
re.NoError(err)

err = leadership1.DeleteLeaderKeyByRevision(oldRevision)
re.Error(err)
resp, err = client.Get(context.Background(), leaderKey)
re.NoError(err)
re.Len(resp.Kvs, 1)
re.Equal("test_leader_2", string(resp.Kvs[0].Value))

resp, err = client.Get(context.Background(), leaderKey)
re.NoError(err)
err = leadership2.DeleteLeaderKeyByRevision(resp.Kvs[0].ModRevision)
re.NoError(err)
resp, err = client.Get(context.Background(), leaderKey)
re.NoError(err)
re.Empty(resp.Kvs)

err = leadership2.DeleteLeaderKeyByRevision(resp.Header.Revision)
re.NoError(err)
}

func deleteLeaderKeyByCurrentRevision(t *testing.T, leadership *Leadership, client *clientv3.Client, leaderKey string) {
resp, err := client.Get(context.Background(), leaderKey)
require.NoError(t, err)
revision := resp.Header.Revision
if len(resp.Kvs) > 0 {
revision = resp.Kvs[0].ModRevision
}
require.NoError(t, leadership.DeleteLeaderKeyByRevision(revision))
}

>>>>>>> 703f4bb25d (server, member, election: document and test the election client pinning (#11110))
func TestExitWatch(t *testing.T) {
re := require.New(t)
leaderKey := "/test_leader"
Expand Down Expand Up @@ -193,8 +246,8 @@ func checkExitWatch(t *testing.T, leaderKey string, injectFunc func(server *embe
re.NoError(err)
defer client2.Close()

leadership1 := NewLeadership(client1, leaderKey, "test_leader_1")
leadership2 := NewLeadership(client2, leaderKey, "test_leader_2")
leadership1 := NewLeadership(client1, leaderKey, "test_leader_1", "test_leader_1")
leadership2 := NewLeadership(client2, leaderKey, "test_leader_2", "test_leader_2")
err = leadership1.Campaign(defaultLeaseTimeout, "test_leader_1")
re.NoError(err)
resp, err := client2.Get(context.Background(), leaderKey)
Expand Down Expand Up @@ -230,8 +283,8 @@ func TestRequestProgress(t *testing.T) {
defer client2.Close()

leaderKey := "/test_leader"
leadership1 := NewLeadership(client1, leaderKey, "test_leader_1")
leadership2 := NewLeadership(client2, leaderKey, "test_leader_2")
leadership1 := NewLeadership(client1, leaderKey, "test_leader_1", "test_leader_1")
leadership2 := NewLeadership(client2, leaderKey, "test_leader_2", "test_leader_2")
err = leadership1.Campaign(defaultLeaseTimeout, "test_leader_1")
re.NoError(err)

Expand Down Expand Up @@ -267,7 +320,7 @@ func TestCampaignTimes(t *testing.T) {
re := require.New(t)
_, client, clean := etcdutil.NewTestEtcdCluster(t, 1)
defer clean()
leadership := NewLeadership(client, "test_leader", "test_leader")
leadership := NewLeadership(client, "test_leader", "test_leader", "test_leader")

// all the campaign times are within the timeout.
campaignTimesRecordTimeout = 10 * time.Second
Expand Down
86 changes: 85 additions & 1 deletion pkg/election/lease.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,9 +16,18 @@

import (
"context"
"strings"
"sync/atomic"
"time"

<<<<<<< HEAD

Check failure on line 23 in pkg/election/lease.go

View workflow job for this annotation

GitHub Actions / tso-function-test 

missing import path

Check failure on line 23 in pkg/election/lease.go

View workflow job for this annotation

GitHub Actions / statics 

missing import path

Check failure on line 23 in pkg/election/lease.go

View workflow job for this annotation

GitHub Actions / chunks (6, Tools Test) 

missing import path

Check failure on line 23 in pkg/election/lease.go

View workflow job for this annotation

GitHub Actions / chunks (5, Tests(2)) 

missing import path

Check failure on line 23 in pkg/election/lease.go

View workflow job for this annotation

GitHub Actions / chunks (1, Unit Test(1)) 

missing import path

Check failure on line 23 in pkg/election/lease.go

View workflow job for this annotation

GitHub Actions / chunks (4, Tests(1)) 

missing import path

Check failure on line 23 in pkg/election/lease.go

View workflow job for this annotation

GitHub Actions / chunks (8, TSO Integration Test) 

missing import path

Check failure on line 23 in pkg/election/lease.go

View workflow job for this annotation

GitHub Actions / chunks (7, Client Integration Test) 

missing import path

Check failure on line 23 in pkg/election/lease.go

View workflow job for this annotation

GitHub Actions / chunks (3, Unit Test(3)) 

missing import path

Check failure on line 23 in pkg/election/lease.go

View workflow job for this annotation

GitHub Actions / chunks (2, Unit Test(2)) 

missing import path

Check failure on line 23 in pkg/election/lease.go

View workflow job for this annotation

GitHub Actions / chunks (10, MicroService Integration(TSO)) 

missing import path

Check failure on line 23 in pkg/election/lease.go

View workflow job for this annotation

GitHub Actions / chunks (9, MicroService Integration(!TSO)) 

missing import path
=======
clientv3 "go.etcd.io/etcd/client/v3"
"go.uber.org/zap"

"github.com/pingcap/errors"
"github.com/pingcap/failpoint"
>>>>>>> 703f4bb25d (server, member, election: document and test the election client pinning (#11110))
"github.com/pingcap/log"
"github.com/tikv/pd/pkg/errs"
"github.com/tikv/pd/pkg/utils/etcdutil"
Expand All @@ -38,7 +47,13 @@
// The way to gain and maintain leadership is to update and keep the lease alive continuously.
type Lease struct {
// purpose is used to show what this election for
<<<<<<< HEAD
Purpose string
=======
purpose string
// name scopes test failpoints to one member.
name string
>>>>>>> 703f4bb25d (server, member, election: document and test the election client pinning (#11110))
// etcd client and lease
client *clientv3.Client
lease clientv3.Lease
Expand All @@ -49,14 +64,55 @@
}

// NewLease creates a new Lease instance.
func NewLease(client *clientv3.Client, purpose string) *Lease {
func NewLease(client *clientv3.Client, purpose, name string) *Lease {
return &Lease{
<<<<<<< HEAD
Purpose: purpose,
=======
purpose: purpose,
name: name,
>>>>>>> 703f4bb25d (server, member, election: document and test the election client pinning (#11110))
client: client,
lease: clientv3.NewLease(client),
}
}

<<<<<<< HEAD
=======
// matchesFailpointTarget matches "<purpose>@<name>" for member-scoped failpoints.
func (l *Lease) matchesFailpointTarget(val failpoint.Value) bool {
target, ok := val.(string)
if !ok {
return false
}
purpose, name, found := strings.Cut(target, "@")
if !found {
return false
}
return purpose == l.purpose && name == l.name
}

func (l *Lease) setID(id clientv3.LeaseID) {
if l == nil {
return
}
l.id.Store(id)
}

// GetID returns the underlying etcd lease ID. It returns 0 if the lease has not
// been granted yet.
func (l *Lease) GetID() clientv3.LeaseID {
if l == nil {
return 0
}
loaded := l.id.Load()
if loaded == nil {
return 0
}
return loaded.(clientv3.LeaseID)
}

>>>>>>> 703f4bb25d (server, member, election: document and test the election client pinning (#11110))
// Grant uses `lease.Grant` to initialize the lease and expireTime.
func (l *Lease) Grant(leaseTimeout int64) error {
if l == nil {
Expand Down Expand Up @@ -167,9 +223,37 @@
defer logutil.LogPanic()
ctx1, cancel := context.WithTimeout(ctx, l.leaseTimeout)
defer cancel()
<<<<<<< HEAD
var leaseID clientv3.LeaseID
if l.ID.Load() != nil {
leaseID = l.ID.Load().(clientv3.LeaseID)
=======
// Record the start time of the `KeepAliveOnce` request to track the request duration
// and calculate the tick interval between consecutive `KeepAliveOnce` requests later.
requestStart := time.Now()
lastRequestStart, _ := lastTime.Swap(requestStart).(time.Time)
res, err := l.lease.KeepAliveOnce(ctx1, l.GetID())
failpoint.Inject("keepAliveFailed", func(val failpoint.Value) {
// Inject after the request so etcd keeps the lease alive while
// the caller observes renewal failure.
if l.matchesFailpointTarget(val) {
res, err = nil, errors.New("keep alive failed")
}
})

// Record the duration of the `KeepAliveOnce` request.
l.metrics.observeKeepAliveRequestDurationMetrics(time.Since(requestStart), err)
// Record the interval between the consecutive `KeepAliveOnce` requests.
tickInterval := requestStart.Sub(lastRequestStart)
l.metrics.tickInterval.Observe(tickInterval.Seconds())
// If the interval is too long, log a warning to indicate the potential runtime schedule delay.
if tickInterval > interval*2 {
logger.Warn("the interval between keeping alive lease is too long",
zap.Time("start", start),
zap.Time("current-time", requestStart),
zap.Time("last-time", lastRequestStart),
zap.Duration("tick-interval", tickInterval))
>>>>>>> 703f4bb25d (server, member, election: document and test the election client pinning (#11110))
}
res, err := l.lease.KeepAliveOnce(ctx1, leaseID)
if err != nil {
Expand Down
6 changes: 3 additions & 3 deletions pkg/election/lease_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,8 +29,8 @@ func TestLease(t *testing.T) {
defer clean()

// Create the lease.
lease1 := NewLease(client, "test_lease_1")
lease2 := NewLease(client, "test_lease_2")
lease1 := NewLease(client, "test_lease_1", "test_lease_1")
lease2 := NewLease(client, "test_lease_2", "test_lease_2")
re.True(lease1.IsExpired())
re.True(lease2.IsExpired())
re.NoError(lease1.Close())
Expand Down Expand Up @@ -86,7 +86,7 @@ func TestLeaseKeepAlive(t *testing.T) {
defer clean()

// Create the lease.
lease := NewLease(client, "test_lease")
lease := NewLease(client, "test_lease", "test_lease")

re.NoError(lease.Grant(defaultLeaseTimeout))
ch := lease.keepAliveWorker(context.Background(), 2*time.Second)
Expand Down
2 changes: 1 addition & 1 deletion pkg/encryption/key_manager_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,7 @@ func newTestKeyFile(t *testing.T, re *require.Assertions, key ...string) (keyFil
}

func newTestLeader(re *require.Assertions, client *clientv3.Client) *election.Leadership {
leader := election.NewLeadership(client, "test_leader", "test")
leader := election.NewLeadership(client, "test_leader", "test", "test_member")
timeout := int64(30000000) // about a year.
err := leader.Campaign(timeout, "")
re.NoError(err)
Expand Down
18 changes: 18 additions & 0 deletions pkg/member/member.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -321,8 +321,21 @@ func (m *EmbeddedEtcdMember) MoveEtcdLeader(ctx context.Context, old, new uint64
return nil
}

<<<<<<< HEAD
// GetEtcdLeader returns the etcd leader ID.
func (m *EmbeddedEtcdMember) GetEtcdLeader() uint64 {
=======
// GetEtcdLeader returns the embedded etcd server's cached leader ID, or 0.
// The value can remain stale while the Ready loop is blocked on storage, so it
// must not be used alone to decide whether this member may serve (tikv/pd#7780).
func (m *Member) GetEtcdLeader() uint64 {
failpoint.Inject("staleEtcdLeaderView", func(val failpoint.Value) {
// Simulate a stale local leader view.
if name, ok := val.(string); ok && name == m.Name() {
failpoint.Return(m.ID())
}
})
>>>>>>> 703f4bb25d (server, member, election: document and test the election client pinning (#11110))
return m.etcd.Server.Lead()
}

Expand All @@ -347,9 +360,14 @@ func (m *EmbeddedEtcdMember) InitMemberInfo(advertiseClientUrls, advertisePeerUr
}
m.member = leader
m.memberValue = string(data)
<<<<<<< HEAD
m.rootPath = rootPath
m.leadership = election.NewLeadership(m.client, m.GetLeaderPath(), "leader election")
log.Info("member joining election", zap.Stringer("member-info", m.member), zap.String("root-path", m.rootPath))
=======
m.leadership = election.NewLeadership(m.client, m.GetElectionPath(), "leader election", member.GetName())
log.Info("member joining election", zap.Stringer("member-info", m.member))
>>>>>>> 703f4bb25d (server, member, election: document and test the election client pinning (#11110))
}

// ResignEtcdLeader resigns current PD's etcd leadership. If nextLeader is empty, all
Expand Down
7 changes: 7 additions & 0 deletions pkg/member/participant.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,13 +86,20 @@ func (m *Participant) InitInfo(p participant, rootPath string, leaderName string
// can't fail, so panic here.
log.Fatal("marshal leader meet error", zap.String("member-name", p.String()), errs.ZapError(errs.ErrMarshalLeader, err))
}
<<<<<<< HEAD
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.String("participant-info", p.String()), zap.String("leader-path", m.leaderPath))
=======
p.participant = participant
p.participantValue = string(data)
p.leadership = election.NewLeadership(p.client, p.GetElectionPath(), purpose, participant.GetName())
log.Info("participant joining election", zap.String("participant-info", participant.String()), zap.String("primary-path", p.GetElectionPath()))
>>>>>>> 703f4bb25d (server, member, election: document and test the election client pinning (#11110))
}

// ID returns the unique ID for this participant in the election group
Expand Down
26 changes: 26 additions & 0 deletions pkg/storage/storage_tso_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,32 @@ import (

var defaultContext = context.Background()

<<<<<<< HEAD
=======
func prepare(t *testing.T) (storage Storage, clean func(), leadership *election.Leadership) {
_, client, clean := etcdutil.NewTestEtcdCluster(t, 1, nil)
storage = NewStorageWithEtcdBackend(client)
leadership = election.NewLeadership(client, testLeaderKey, "storage_tso_test", "test_member")
err := leadership.Campaign(60, testLeaderValue)
require.NoError(t, err)
return storage, clean, leadership
}

func TestSaveTimestampWithTimeout(t *testing.T) {
re := require.New(t)
re.NoError(failpoint.Enable("github.com/tikv/pd/pkg/storage/kv/slowTxn", "return(true)"))
defer func() {
re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/storage/kv/slowTxn"))
}()
storage, clean, leadership := prepare(t)
defer clean()
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
err := storage.SaveTimestamp(ctx, testGroupID, time.Now().Round(0), leadership)
re.ErrorIs(err, context.DeadlineExceeded)
}

>>>>>>> 703f4bb25d (server, member, election: document and test the election client pinning (#11110))
func TestSaveLoadTimestamp(t *testing.T) {
re := require.New(t)
storage, clean := newTestStorage(t)
Expand Down
4 changes: 3 additions & 1 deletion server/server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -394,7 +394,9 @@ func (s *Server) startClient() error {
if err != nil {
return errs.ErrNewEtcdClient.Wrap(err).GenWithStackByCause()
}
// This etcd client will only be used to read and write the election-related data, such as leader key.
// Keep this client pinned to local etcd. Local lease renewal verifies this
// member's etcd leadership, preventing a stalled member from renewing through
// a healthy peer (tikv/pd#10671).
s.electionClient, err = etcdutil.CreateEtcdClient(tlsConfig, etcdCfg.AdvertiseClientUrls, etcdutil.ElectionEtcdClientPurpose, false)
if err != nil {
return errs.ErrNewEtcdClient.Wrap(err).GenWithStackByCause()
Expand Down
Loading
Loading