diff --git a/pkg/election/leadership.go b/pkg/election/leadership.go index 005b00c2321..9ad5bed9b14 100644 --- a/pkg/election/leadership.go +++ b/pkg/election/leadership.go @@ -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 @@ -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), @@ -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) { diff --git a/pkg/election/leadership_test.go b/pkg/election/leadership_test.go index d6a44c39e10..6348812416d 100644 --- a/pkg/election/leadership_test.go +++ b/pkg/election/leadership_test.go @@ -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") @@ -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" @@ -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) @@ -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) @@ -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 diff --git a/pkg/election/lease.go b/pkg/election/lease.go index f0d10b9581f..dd103911b3c 100644 --- a/pkg/election/lease.go +++ b/pkg/election/lease.go @@ -16,9 +16,18 @@ package election import ( "context" + "strings" "sync/atomic" "time" +<<<<<<< HEAD +======= + 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" @@ -38,7 +47,13 @@ const ( // 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 @@ -49,14 +64,55 @@ type Lease struct { } // 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 "@" 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 { @@ -167,9 +223,37 @@ func (l *Lease) keepAliveWorker(ctx context.Context, interval time.Duration) <-c 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 { diff --git a/pkg/election/lease_test.go b/pkg/election/lease_test.go index 3a02de97239..3c9dd43c1b3 100644 --- a/pkg/election/lease_test.go +++ b/pkg/election/lease_test.go @@ -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()) @@ -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) diff --git a/pkg/encryption/key_manager_test.go b/pkg/encryption/key_manager_test.go index e827a714a3a..c265e9560db 100644 --- a/pkg/encryption/key_manager_test.go +++ b/pkg/encryption/key_manager_test.go @@ -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) diff --git a/pkg/member/member.go b/pkg/member/member.go index e4246ab6bd9..69c296ff741 100644 --- a/pkg/member/member.go +++ b/pkg/member/member.go @@ -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() } @@ -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 diff --git a/pkg/member/participant.go b/pkg/member/participant.go index 599e56387d6..9b486653d82 100644 --- a/pkg/member/participant.go +++ b/pkg/member/participant.go @@ -86,6 +86,7 @@ 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 @@ -93,6 +94,12 @@ func (m *Participant) InitInfo(p participant, rootPath string, leaderName string 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 diff --git a/pkg/storage/storage_tso_test.go b/pkg/storage/storage_tso_test.go index 8005302c34e..ad07428cf8a 100644 --- a/pkg/storage/storage_tso_test.go +++ b/pkg/storage/storage_tso_test.go @@ -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) diff --git a/server/server.go b/server/server.go index f51181dc1ca..9df2c69f15b 100644 --- a/server/server.go +++ b/server/server.go @@ -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() diff --git a/tests/server/member/member_test.go b/tests/server/member/member_test.go index 13f7fffe083..9a1affc8267 100644 --- a/tests/server/member/member_test.go +++ b/tests/server/member/member_test.go @@ -21,10 +21,18 @@ import ( "fmt" "io" "net/http" + "strings" "sync" "testing" "time" +<<<<<<< HEAD +======= + "github.com/stretchr/testify/require" + clientv3 "go.etcd.io/etcd/client/v3" + "go.uber.org/goleak" + +>>>>>>> 703f4bb25d (server, member, election: document and test the election client pinning (#11110)) "github.com/pingcap/errors" "github.com/pingcap/failpoint" "github.com/pingcap/kvproto/pkg/pdpb" @@ -439,3 +447,118 @@ func sendRequest(re *require.Assertions, wg *sync.WaitGroup, done <-chan bool, a time.Sleep(10 * time.Millisecond) } } + +func TestElectionClientStaysOnLocalMember(t *testing.T) { + re := require.New(t) + // Speed up endpoint discovery. + re.NoError(failpoint.Enable("github.com/tikv/pd/pkg/utils/etcdutil/fastTick", "return(true)")) + defer func() { + re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/utils/etcdutil/fastTick")) + }() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + cluster, err := tests.NewTestCluster(ctx, 3) + re.NoError(err) + defer cluster.Destroy() + re.NoError(cluster.RunInitialServers()) + re.NotEmpty(cluster.WaitLeader()) + + for name, svr := range cluster.GetServers() { + s := svr.GetServer() + localURLs := strings.Split(s.GetConfig().AdvertiseClientUrls, ",") + // Confirm discovery is active before checking the election client. + testutil.Eventually(re, func() bool { + return len(s.GetClient().Endpoints()) > len(localURLs) + }) + re.Equal(localURLs, s.GetMember().Client().Endpoints(), name) + } +} + +func TestPDLeaderStepsDownWhenLeaseIsLostWithStaleEtcdLeaderView(t *testing.T) { + re := require.New(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + cluster, err := tests.NewTestCluster(ctx, 3) + re.NoError(err) + defer cluster.Destroy() + re.NoError(cluster.RunInitialServers()) + + oldLeaderName := cluster.WaitLeader() + re.NotEmpty(oldLeaderName) + oldLeaderServer := cluster.GetServer(oldLeaderName) + oldLeader := oldLeaderServer.GetServer() + var newEtcdLeaderServer *tests.TestServer + for name, svr := range cluster.GetServers() { + if name != oldLeaderName { + newEtcdLeaderServer = svr + break + } + } + re.NotNil(newEtcdLeaderServer) + + // Keep the colocation check stale so lease loss is the only step-down signal. + re.NoError(failpoint.Enable("github.com/tikv/pd/pkg/member/staleEtcdLeaderView", + fmt.Sprintf("return(\"%s\")", oldLeaderName))) + defer func() { + re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/member/staleEtcdLeaderView")) + }() + re.Equal(oldLeader.GetMember().ID(), oldLeader.GetMember().GetEtcdLeader()) + + // Move etcd leadership while the old member still observes itself as leader. + testutil.Eventually(re, func() bool { + return oldLeaderServer.MoveEtcdLeader(oldLeaderServer.GetServerID(), newEtcdLeaderServer.GetServerID()) == nil + }) + + // Negative control: a stale view alone must not end a term with a valid lease. + time.Sleep(2 * time.Second) + re.True(oldLeader.IsServing()) + re.Equal(oldLeaderName, cluster.GetLeader()) + + lease := oldLeader.GetMember().GetLeadership().GetLease() + re.NotNil(lease) + leaseID := lease.GetID() + re.NotEqual(clientv3.NoLease, leaseID) + _, err = newEtcdLeaderServer.GetServer().GetClient().Revoke(ctx, leaseID) + re.NoError(err) + + testutil.Eventually(re, func() bool { + return !oldLeader.IsServing() && oldLeader.GetRaftCluster() == nil + }, testutil.WithWaitFor(30*time.Second)) + re.NotEqual(oldLeaderName, waitLeaderChange(re, cluster, oldLeaderName)) +} + +func TestPDLeaderStepsDownWhenRenewalsFail(t *testing.T) { + re := require.New(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + cluster, err := tests.NewTestCluster(ctx, 3) + re.NoError(err) + defer cluster.Destroy() + re.NoError(cluster.RunInitialServers()) + + oldLeaderName := cluster.WaitLeader() + re.NotEmpty(oldLeaderName) + oldLeaderServer := cluster.GetServer(oldLeaderName) + oldLeader := oldLeaderServer.GetServer() + + // Disable the colocation signal. + re.NoError(failpoint.Enable("github.com/tikv/pd/pkg/member/staleEtcdLeaderView", + fmt.Sprintf("return(\"%s\")", oldLeaderName))) + defer func() { + re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/member/staleEtcdLeaderView")) + }() + re.True(oldLeader.IsServing()) + + // Keep the etcd lease alive while only the old leader observes renewal failures. + re.NoError(failpoint.Enable("github.com/tikv/pd/pkg/election/keepAliveFailed", + fmt.Sprintf("return(\"leader election@%s\")", oldLeaderName))) + defer func() { + re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/election/keepAliveFailed")) + }() + + testutil.Eventually(re, func() bool { + return !oldLeader.IsServing() + }, testutil.WithWaitFor(30*time.Second)) + re.NotEqual(oldLeaderName, waitLeaderChange(re, cluster, oldLeaderName)) +}