diff --git a/errors.toml b/errors.toml index 33acad6fbfb..d0cf5435c5b 100644 --- a/errors.toml +++ b/errors.toml @@ -746,6 +746,11 @@ error = ''' etcd leader not found ''' +["PD:member:ErrEtcdLeaderTransferTargetCheck"] +error = ''' +etcd leader transfer target check failed: %s +''' + ["PD:member:ErrMarshalMember"] error = ''' marshal member failed diff --git a/pkg/errs/errno.go b/pkg/errs/errno.go index a887d539f69..57457ad7c37 100644 --- a/pkg/errs/errno.go +++ b/pkg/errs/errno.go @@ -120,10 +120,11 @@ var ( // member errors var ( - ErrEtcdLeaderNotFound = errors.Normalize("etcd leader not found", errors.RFCCodeText("PD:member:ErrEtcdLeaderNotFound")) - ErrMarshalMember = errors.Normalize("marshal member failed", errors.RFCCodeText("PD:member:ErrMarshalMember")) - ErrMarshalParticipant = errors.Normalize("marshal participant failed", errors.RFCCodeText("PD:member:ErrMarshalParticipant")) - ErrCheckCampaign = errors.Normalize("check campaign failed", errors.RFCCodeText("PD:member:ErrCheckCampaign")) + ErrEtcdLeaderNotFound = errors.Normalize("etcd leader not found", errors.RFCCodeText("PD:member:ErrEtcdLeaderNotFound")) + ErrMarshalMember = errors.Normalize("marshal member failed", errors.RFCCodeText("PD:member:ErrMarshalMember")) + ErrMarshalParticipant = errors.Normalize("marshal participant failed", errors.RFCCodeText("PD:member:ErrMarshalParticipant")) + ErrCheckCampaign = errors.Normalize("check campaign failed", errors.RFCCodeText("PD:member:ErrCheckCampaign")) + ErrEtcdLeaderTransferTargetCheck = errors.Normalize("etcd leader transfer target check failed: %s", errors.RFCCodeText("PD:member:ErrEtcdLeaderTransferTargetCheck")) ) // core errors diff --git a/pkg/member/member.go b/pkg/member/member.go index e8eaa52d296..907fea1e29e 100644 --- a/pkg/member/member.go +++ b/pkg/member/member.go @@ -66,6 +66,28 @@ type Member struct { lastLeaderUpdatedTime atomic.Value } +// MoveEtcdLeaderOption customizes embedded etcd leader transfer. +type MoveEtcdLeaderOption func(*moveEtcdLeaderOptions) + +type moveEtcdLeaderOptions struct { + targetChecker func(context.Context, uint64) error +} + +// WithTargetChecker checks whether the target member can become leader before transfer. +func WithTargetChecker(checker func(context.Context, uint64) error) MoveEtcdLeaderOption { + return func(opts *moveEtcdLeaderOptions) { + opts.targetChecker = checker + } +} + +func newMoveEtcdLeaderOptions(opts ...MoveEtcdLeaderOption) *moveEtcdLeaderOptions { + options := &moveEtcdLeaderOptions{} + for _, opt := range opts { + opt(options) + } + return options +} + // NewMember create a new Member. func NewMember(etcd *embed.Etcd, client *clientv3.Client, id uint64) *Member { return &Member{ @@ -280,7 +302,7 @@ func (m *Member) Resign() { } // CheckPriority checks whether the etcd leader should be moved according to the priority. -func (m *Member) CheckPriority(ctx context.Context) { +func (m *Member) CheckPriority(ctx context.Context, opts ...MoveEtcdLeaderOption) { myPriority, err := m.GetMemberLeaderPriority(m.ID()) if err != nil { log.Error("failed to load leader priority", errs.ZapError(err)) @@ -299,9 +321,19 @@ func (m *Member) CheckPriority(ctx context.Context) { return } if myPriority > leaderPriority { - err := m.MoveEtcdLeader(ctx, etcdLeader, m.ID()) + err := m.MoveEtcdLeader(ctx, etcdLeader, m.ID(), opts...) if err != nil { - log.Error("failed to transfer etcd leader", errs.ZapError(err)) + if errors.ErrorEqual(err, errs.ErrEtcdLeaderTransferTargetCheck) { + log.Warn("skip transferring etcd leader by priority because target check failed", + zap.Uint64("from", etcdLeader), + zap.Uint64("to", m.ID()), + errs.ZapError(err)) + } else { + log.Error("failed to transfer etcd leader by priority", + zap.Uint64("from", etcdLeader), + zap.Uint64("to", m.ID()), + errs.ZapError(err)) + } } else { log.Info("transfer etcd leader", zap.Uint64("from", etcdLeader), @@ -311,9 +343,15 @@ func (m *Member) CheckPriority(ctx context.Context) { } // MoveEtcdLeader tries to transfer etcd leader. -func (m *Member) MoveEtcdLeader(ctx context.Context, old, new uint64) error { +func (m *Member) MoveEtcdLeader(ctx context.Context, old, new uint64, opts ...MoveEtcdLeaderOption) error { moveCtx, cancel := context.WithTimeout(ctx, moveLeaderTimeout) defer cancel() + options := newMoveEtcdLeaderOptions(opts...) + if options.targetChecker != nil { + if err := options.targetChecker(moveCtx, new); err != nil { + return errs.ErrEtcdLeaderTransferTargetCheck.GenWithStackByArgs(err.Error()) + } + } err := m.etcd.Server.MoveLeader(moveCtx, old, new) if err != nil { return errs.ErrEtcdMoveLeader.Wrap(err).GenWithStackByCause() @@ -353,7 +391,7 @@ func (m *Member) InitMemberInfo(advertiseClientUrls, advertisePeerUrls, name str // ResignEtcdLeader resigns current PD's etcd leadership. If nextLeader is empty, all // other pd-servers can campaign. -func (m *Member) ResignEtcdLeader(ctx context.Context, from string, nextEtcdLeader string) error { +func (m *Member) ResignEtcdLeader(ctx context.Context, from string, nextEtcdLeader string, opts ...MoveEtcdLeaderOption) error { log.Info("try to resign etcd leader to next pd-server", zap.String("from", from), zap.String("to", nextEtcdLeader)) // Determine next etcd leader candidates. var etcdLeaderIDs []uint64 @@ -375,6 +413,30 @@ func (m *Member) ResignEtcdLeader(ctx context.Context, from string, nextEtcdLead if len(etcdLeaderIDs) == 0 { return errors.New("no valid pd to transfer etcd leader") } + options := newMoveEtcdLeaderOptions(opts...) + if options.targetChecker != nil { + readyEtcdLeaderIDs := make([]uint64, 0, len(etcdLeaderIDs)) + var lastErr error + for _, id := range etcdLeaderIDs { + checkCtx, cancel := context.WithTimeout(ctx, moveLeaderTimeout) + err := options.targetChecker(checkCtx, id) + cancel() + if err != nil { + lastErr = err + log.Warn("skip unready pd when resigning etcd leader", zap.Uint64("target-member-id", id), errs.ZapError(err)) + continue + } + readyEtcdLeaderIDs = append(readyEtcdLeaderIDs, id) + } + if len(readyEtcdLeaderIDs) == 0 { + errMsg := "no ready pd to transfer etcd leader" + if lastErr != nil { + errMsg += ": " + lastErr.Error() + } + return errs.ErrEtcdLeaderTransferTargetCheck.GenWithStackByArgs(errMsg) + } + etcdLeaderIDs = readyEtcdLeaderIDs + } nextEtcdLeaderID := etcdLeaderIDs[rand.IntN(len(etcdLeaderIDs))] return m.MoveEtcdLeader(ctx, m.ID(), nextEtcdLeaderID) } diff --git a/pkg/utils/apiutil/apiutil.go b/pkg/utils/apiutil/apiutil.go index da6369ae59c..e257e7b8167 100644 --- a/pkg/utils/apiutil/apiutil.go +++ b/pkg/utils/apiutil/apiutil.go @@ -438,6 +438,8 @@ func ReadJSONRespondError(rd *render.Render, w http.ResponseWriter, body io.Read const ( // CorePath the core group, is at REST path `/pd/api/v1`. CorePath = "/pd/api/v1" + // CoreV2Path is the core group REST path for API v2. + CoreV2Path = "/pd/api/v2" // ExtensionsPath the named groups are REST at `/pd/apis/{GROUP_NAME}/{Version}`. ExtensionsPath = "/pd/apis" ) diff --git a/pkg/versioninfo/feature.go b/pkg/versioninfo/feature.go index 24aed22dc9d..a68e0e64258 100644 --- a/pkg/versioninfo/feature.go +++ b/pkg/versioninfo/feature.go @@ -49,6 +49,8 @@ const ( HotScheduleWithQuery // SwitchWithess supports switch between witness and non-witness. SwitchWitness + // ReadyAPI supports checking whether PD has finished initial region loading. + ReadyAPI ) var featuresDict = map[Feature]string{ @@ -62,6 +64,7 @@ var featuresDict = map[Feature]string{ ConfChangeV2: "5.0.0", HotScheduleWithQuery: "5.2.0", SwitchWitness: "6.6.0", + ReadyAPI: "8.5.2", } // MinSupportedVersion returns the minimum support version for the specified feature. @@ -73,3 +76,13 @@ func MinSupportedVersion(v Feature) *semver.Version { version := MustParseVersion(target) return version } + +// IsReadyAPISupported returns whether PD supports the ready API for initial region loading. +func IsReadyAPISupported(pdVersion *semver.Version) bool { + if pdVersion == nil { + return false + } + // ReadyAPI was introduced in a patch release, so use strict semver comparison + // instead of IsFeatureSupported's same-minor compatibility semantics. + return !pdVersion.LessThan(*MinSupportedVersion(ReadyAPI)) +} diff --git a/pkg/versioninfo/versioninfo_test.go b/pkg/versioninfo/versioninfo_test.go index 6c40882f1f1..bb369e4140c 100644 --- a/pkg/versioninfo/versioninfo_test.go +++ b/pkg/versioninfo/versioninfo_test.go @@ -46,3 +46,23 @@ func TestIsHotScheduleWithCPUSupported(t *testing.T) { re.Equal(test.expect, IsHotScheduleWithCPUSupported(MustParseVersion(test.version)), test.version) } } + +func TestIsReadyAPISupported(t *testing.T) { + re := require.New(t) + re.False(IsReadyAPISupported(nil)) + + tests := []struct { + version string + expect bool + }{ + {"8.5.1", false}, + {"v8.5.1", false}, + {"8.5.2", true}, + {"v8.5.2", true}, + {"8.5.3", true}, + {"9.0.0", true}, + } + for _, test := range tests { + re.Equal(test.expect, IsReadyAPISupported(MustParseVersion(test.version)), test.version) + } +} diff --git a/server/api/member.go b/server/api/member.go index b4b3a0a6103..b5e27662e8c 100644 --- a/server/api/member.go +++ b/server/api/member.go @@ -29,6 +29,7 @@ import ( "github.com/pingcap/log" "github.com/tikv/pd/pkg/errs" + "github.com/tikv/pd/pkg/member" "github.com/tikv/pd/pkg/utils/apiutil" "github.com/tikv/pd/pkg/utils/etcdutil" "github.com/tikv/pd/pkg/utils/keypath" @@ -281,7 +282,8 @@ func (h *leaderHandler) GetLeader(w http.ResponseWriter, _ *http.Request) { // @Failure 500 {string} string "PD server failed to proceed the request." // @Router /leader/resign [post] func (h *leaderHandler) ResignLeader(w http.ResponseWriter, _ *http.Request) { - err := h.svr.GetMember().ResignEtcdLeader(h.svr.Context(), h.svr.Name(), "") + err := h.svr.GetMember().ResignEtcdLeader(h.svr.Context(), h.svr.Name(), "", + member.WithTargetChecker(h.svr.CheckMemberReadyForLeaderTransfer)) if err != nil { h.rd.JSON(w, http.StatusInternalServerError, err.Error()) return @@ -300,7 +302,8 @@ func (h *leaderHandler) ResignLeader(w http.ResponseWriter, _ *http.Request) { // @Failure 500 {string} string "PD server failed to proceed the request." // @Router /leader/transfer/{nextLeader} [post] func (h *leaderHandler) TransferLeader(w http.ResponseWriter, r *http.Request) { - err := h.svr.GetMember().ResignEtcdLeader(h.svr.Context(), h.svr.Name(), mux.Vars(r)["next_leader"]) + err := h.svr.GetMember().ResignEtcdLeader(h.svr.Context(), h.svr.Name(), mux.Vars(r)["next_leader"], + member.WithTargetChecker(h.svr.CheckMemberReadyForLeaderTransfer)) if err != nil { h.rd.JSON(w, http.StatusInternalServerError, err.Error()) return diff --git a/server/apiv2/middlewares/microservice_redirector.go b/server/apiv2/middlewares/microservice_redirector.go index ca5adeeefd5..0cd6d66ea5b 100644 --- a/server/apiv2/middlewares/microservice_redirector.go +++ b/server/apiv2/middlewares/microservice_redirector.go @@ -75,7 +75,7 @@ func MicroserviceRedirector(rules ...serverapi.RedirectRule) gin.HandlerFunc { // AffinityMicroserviceRedirector only forwards affinity GET requests to the scheduling service. func AffinityMicroserviceRedirector() gin.HandlerFunc { - pdAffinityPath := "/pd/api/v2/affinity-groups" + pdAffinityPath := apiutil.CoreV2Path + "/affinity-groups" targetAffinityPath := scheapi.APIPathPrefix + "/affinity-groups" return MicroserviceRedirector( serverapi.RedirectRule{ diff --git a/server/apiv2/router.go b/server/apiv2/router.go index 272ce9bd20b..717347ba73b 100644 --- a/server/apiv2/router.go +++ b/server/apiv2/router.go @@ -33,7 +33,7 @@ var group = apiutil.APIServiceGroup{ PathPrefix: apiV2Prefix, } -const apiV2Prefix = "/pd/api/v2/" +const apiV2Prefix = apiutil.CoreV2Path + "/" // NewV2Handler creates a HTTP handler for API. // diff --git a/server/member_ready.go b/server/member_ready.go new file mode 100644 index 00000000000..a7d316d00d9 --- /dev/null +++ b/server/member_ready.go @@ -0,0 +1,190 @@ +// Copyright 2026 TiKV Project Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package server + +import ( + "context" + "encoding/json" + "io" + "net/http" + "strings" + "time" + + "github.com/pingcap/errors" + "github.com/pingcap/kvproto/pkg/pdpb" + + "github.com/tikv/pd/pkg/utils/apiutil" + "github.com/tikv/pd/pkg/versioninfo" +) + +const memberReadyCheckTimeout = 5 * time.Second + +type targetPDVersion struct { + Version string `json:"version"` +} + +// CheckMemberReadyForLeaderTransfer checks whether the target PD member can be promoted to leader. +func (s *Server) CheckMemberReadyForLeaderTransfer(ctx context.Context, memberID uint64) error { + clientURLs, err := s.getMemberClientURLs(memberID) + if err != nil { + return err + } + + triedURLs, ready, lastErr := s.checkMemberReadyURLs(ctx, memberID, clientURLs) + if ready { + return nil + } + + reloadedClientURLs, err := s.reloadMemberClientURLs(memberID) + if err != nil { + lastErr = errors.Annotatef(err, "failed to reload target pd member %d client urls", memberID) + } else if !sameClientURLs(clientURLs, reloadedClientURLs) { + var reloadedTriedURLs []string + reloadedTriedURLs, ready, lastErr = s.checkMemberReadyURLs(ctx, memberID, reloadedClientURLs) + triedURLs = append(triedURLs, reloadedTriedURLs...) + if ready { + return nil + } + } + return errors.Errorf("target pd member %d is not ready for leader transfer, tried urls %v, last error: %v", memberID, triedURLs, lastErr) +} + +func (s *Server) checkMemberReadyURLs(ctx context.Context, memberID uint64, clientURLs []string) ([]string, bool, error) { + triedURLs := make([]string, 0, len(clientURLs)) + var lastErr error + for _, clientURL := range clientURLs { + triedURLs = append(triedURLs, clientURL) + if err := s.checkMemberReadyURL(ctx, memberID, clientURL); err != nil { + lastErr = err + continue + } + return triedURLs, true, nil + } + return triedURLs, false, lastErr +} + +func sameClientURLs(a, b []string) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +} + +func (s *Server) getMemberClientURLs(memberID uint64) ([]string, error) { + members, err := s.GetMembers() + if err != nil { + return nil, err + } + if clientURLs, found := findMemberClientURLs(members, memberID); found { + if len(clientURLs) > 0 { + return clientURLs, nil + } + return s.reloadMemberClientURLs(memberID) + } + return s.reloadMemberClientURLs(memberID) +} + +func (s *Server) reloadMemberClientURLs(memberID uint64) ([]string, error) { + members, err := s.ReloadMembers() + if err != nil { + return nil, err + } + clientURLs, found := findMemberClientURLs(members, memberID) + if !found { + return nil, errors.Errorf("target pd member %d not found", memberID) + } + if len(clientURLs) == 0 { + return nil, errors.Errorf("target pd member %d has no client url", memberID) + } + return clientURLs, nil +} + +func findMemberClientURLs(members []*pdpb.Member, memberID uint64) ([]string, bool) { + for _, member := range members { + if member.GetMemberId() == memberID { + return member.GetClientUrls(), true + } + } + return nil, false +} + +func (s *Server) checkMemberReadyURL(ctx context.Context, memberID uint64, clientURL string) error { + checkCtx, cancel := context.WithTimeout(ctx, memberReadyCheckTimeout) + defer cancel() + + version, err := s.getTargetPDVersion(checkCtx, clientURL) + if err != nil { + return errors.Annotatef(err, "failed to get target pd member %d version from %s", memberID, clientURL) + } + pdVersion, err := versioninfo.ParseVersion(version) + // A successful /version response with an unparsable version, such as "None" + // from local or dev builds, is treated as a dev/current build, so /ready + // decides readiness. Only a parseable release version older than ReadyAPI + // skips /ready for compatibility. + if err == nil && !versioninfo.IsReadyAPISupported(pdVersion) { + return nil + } + if err := s.checkTargetPDReady(checkCtx, clientURL); err != nil { + return errors.Annotatef(err, "target pd member %d is not ready at %s", memberID, clientURL) + } + return nil +} + +func (s *Server) getTargetPDVersion(ctx context.Context, clientURL string) (string, error) { + body, err := s.getTargetPD(ctx, clientURL, apiutil.CorePath+"/version") + if err != nil { + return "", err + } + version := &targetPDVersion{} + if err := json.Unmarshal(body, version); err != nil { + return "", errors.WithStack(err) + } + return version.Version, nil +} + +func (s *Server) checkTargetPDReady(ctx context.Context, clientURL string) error { + _, err := s.getTargetPD(ctx, clientURL, apiutil.CoreV2Path+"/ready") + return err +} + +func (s *Server) getTargetPD(ctx context.Context, clientURL, path string) ([]byte, error) { + httpClient := s.GetHTTPClient() + if httpClient == nil { + return nil, errors.New("pd http client is nil") + } + url := strings.TrimRight(clientURL, "/") + path + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, http.NoBody) + if err != nil { + return nil, errors.WithStack(err) + } + resp, err := httpClient.Do(req) + if err != nil { + return nil, errors.WithStack(err) + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, errors.WithStack(err) + } + if resp.StatusCode != http.StatusOK { + return nil, errors.Errorf("unexpected status %d from %s: %s", resp.StatusCode, url, string(body)) + } + return body, nil +} diff --git a/server/member_ready_test.go b/server/member_ready_test.go new file mode 100644 index 00000000000..49a1a043dc7 --- /dev/null +++ b/server/member_ready_test.go @@ -0,0 +1,136 @@ +// Copyright 2026 TiKV Project Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package server + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "sync" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/tikv/pd/pkg/utils/apiutil" +) + +const ( + targetPDVersionPath = apiutil.CorePath + "/version" + targetPDReadyPath = apiutil.CoreV2Path + "/ready" +) + +func TestCheckMemberReadyURL(t *testing.T) { + tests := []struct { + name string + version string + versionCode int + readyCode int + expectPaths []string + expectError string + }{ + { + name: "old version skips ready", + version: "v8.5.1", + versionCode: http.StatusOK, + readyCode: http.StatusServiceUnavailable, + expectPaths: []string{targetPDVersionPath}, + }, + { + name: "supported version checks ready", + version: "v8.5.2", + versionCode: http.StatusOK, + readyCode: http.StatusOK, + expectPaths: []string{targetPDVersionPath, targetPDReadyPath}, + }, + { + name: "version request failure", + versionCode: http.StatusInternalServerError, + readyCode: http.StatusOK, + expectPaths: []string{targetPDVersionPath}, + expectError: "failed to get target pd member 1 version", + }, + { + name: "invalid version checks ready", + version: "invalid-version", + versionCode: http.StatusOK, + readyCode: http.StatusOK, + expectPaths: []string{targetPDVersionPath, targetPDReadyPath}, + }, + { + name: "none version checks ready", + version: "None", + versionCode: http.StatusOK, + readyCode: http.StatusOK, + expectPaths: []string{targetPDVersionPath, targetPDReadyPath}, + }, + { + name: "invalid version ready request failure", + version: "invalid-version", + versionCode: http.StatusOK, + readyCode: http.StatusServiceUnavailable, + expectPaths: []string{targetPDVersionPath, targetPDReadyPath}, + expectError: "target pd member 1 is not ready", + }, + { + name: "ready request failure", + version: "v8.5.2", + versionCode: http.StatusOK, + readyCode: http.StatusServiceUnavailable, + expectPaths: []string{targetPDVersionPath, targetPDReadyPath}, + expectError: "target pd member 1 is not ready", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + re := require.New(t) + + var mu sync.Mutex + paths := make([]string, 0, len(test.expectPaths)) + testServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + paths = append(paths, r.URL.Path) + mu.Unlock() + + switch r.URL.Path { + case targetPDVersionPath: + w.WriteHeader(test.versionCode) + if test.versionCode == http.StatusOK { + _, _ = fmt.Fprintf(w, `{"version":%q}`, test.version) + } + case targetPDReadyPath: + w.WriteHeader(test.readyCode) + default: + w.WriteHeader(http.StatusNotFound) + } + })) + defer testServer.Close() + + s := &Server{httpClient: testServer.Client()} + err := s.checkMemberReadyURL(context.Background(), 1, testServer.URL) + if test.expectError == "" { + re.NoError(err) + } else { + re.ErrorContains(err, test.expectError) + } + + mu.Lock() + gotPaths := append([]string(nil), paths...) + mu.Unlock() + re.Equal(test.expectPaths, gotPaths) + }) + } +} diff --git a/server/server.go b/server/server.go index fe0a2a37d1c..4104271e1ae 100644 --- a/server/server.go +++ b/server/server.go @@ -2093,7 +2093,7 @@ func (s *Server) etcdLeaderLoop() { for { select { case <-ticker.C: - s.member.CheckPriority(ctx) + s.member.CheckPriority(ctx, member.WithTargetChecker(s.CheckMemberReadyForLeaderTransfer)) // Note: we reset the ticker here to support updating configuration dynamically. ticker.Reset(s.cfg.LeaderPriorityCheckInterval.Duration) case <-ctx.Done(): diff --git a/tests/integrations/client/http_client_test.go b/tests/integrations/client/http_client_test.go index 03c0399883c..81d0446ad67 100644 --- a/tests/integrations/client/http_client_test.go +++ b/tests/integrations/client/http_client_test.go @@ -826,6 +826,12 @@ func (suite *httpClientTestSuite) TestSetStoreLabelsRejectEngineKey() { } func (suite *httpClientTestSuite) transferLeader(ctx context.Context, re *require.Assertions) { + oldVersion := versioninfo.PDReleaseVersion + versioninfo.PDReleaseVersion = "v8.5.2" + defer func() { + versioninfo.PDReleaseVersion = oldVersion + }() + client := suite.client members, err := client.GetMembers(ctx) re.NoError(err) diff --git a/tests/server/api/member_test.go b/tests/server/api/member_test.go index 6567d0ae249..b2218ffa577 100644 --- a/tests/server/api/member_test.go +++ b/tests/server/api/member_test.go @@ -30,6 +30,7 @@ import ( "github.com/pingcap/kvproto/pkg/pdpb" + "github.com/tikv/pd/pkg/versioninfo" "github.com/tikv/pd/server/api" "github.com/tikv/pd/tests" ) @@ -161,6 +162,12 @@ func (suite *memberTestSuite) TestResignMyself() { func (suite *memberTestSuite) checkResignMyself(cluster *tests.TestCluster) { re := suite.Require() + oldVersion := versioninfo.PDReleaseVersion + versioninfo.PDReleaseVersion = "v8.5.2" + defer func() { + versioninfo.PDReleaseVersion = oldVersion + }() + leader := cluster.GetLeaderServer() addr := leader.GetAddr() + api.APIPrefix + "/api/v1/leader/resign" resp, err := tests.TestDialClient.Post(addr, "", nil) diff --git a/tests/server/member/member_test.go b/tests/server/member/member_test.go index d961aec9c43..a9079bf8f4f 100644 --- a/tests/server/member/member_test.go +++ b/tests/server/member/member_test.go @@ -32,8 +32,11 @@ import ( "github.com/pingcap/failpoint" "github.com/pingcap/kvproto/pkg/pdpb" + "github.com/tikv/pd/pkg/errs" + "github.com/tikv/pd/pkg/member" "github.com/tikv/pd/pkg/utils/testutil" "github.com/tikv/pd/pkg/utils/typeutil" + "github.com/tikv/pd/pkg/versioninfo" "github.com/tikv/pd/server/config" "github.com/tikv/pd/tests" ) @@ -175,18 +178,85 @@ func TestLeaderPriority(t *testing.T) { }) } +func TestLeaderPrioritySkipsUnreadyTarget(t *testing.T) { + restorePDReleaseVersion := setPDReleaseVersionForTest() + defer restorePDReleaseVersion() + + re := require.New(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + cluster, err := tests.NewTestCluster(ctx, 3, func(conf *config.Config, _ string) { + conf.LeaderPriorityCheckInterval = typeutil.NewDuration(200 * time.Millisecond) + }) + defer cluster.Destroy() + re.NoError(err) + + re.NoError(cluster.RunInitialServers()) + leaderName := cluster.WaitLeader() + re.NotEmpty(leaderName) + testutil.Eventually(re, func() bool { + etcdLeader, err := cluster.GetServer("pd1").GetEtcdLeader() + if err != nil { + return false + } + return cluster.GetLeader() == etcdLeader + }) + leaderName = cluster.GetLeader() + leaderServer := cluster.GetLeaderServer() + re.NoError(leaderServer.BootstrapCluster()) + followers := getFollowerNames(cluster, leaderName) + re.Len(followers, 2) + targetName := followers[0] + targetServer := cluster.GetServer(targetName) + + etcdLeader, err := leaderServer.GetEtcdLeader() + re.NoError(err) + re.Equal(leaderName, etcdLeader) + + failpointName := "github.com/tikv/pd/server/apiv2/handlers/loadRegionSlow" + re.NoError(failpoint.Enable(failpointName, `return("`+targetServer.GetAddr()+`")`)) + failpointEnabled := true + defer func() { + if failpointEnabled { + re.NoError(failpoint.Disable(failpointName)) + } + }() + + post(t, re, leaderServer.GetConfig().ClientUrls+"/pd/api/v1/members/name/"+targetName, `{"leader-priority": 1}`) + targetPDServer := targetServer.GetServer() + targetPDServer.GetMember().CheckPriority(ctx, member.WithTargetChecker(targetPDServer.CheckMemberReadyForLeaderTransfer)) + etcdLeader, err = leaderServer.GetEtcdLeader() + re.NoError(err) + re.Equal(leaderName, etcdLeader) + + re.NoError(failpoint.Disable(failpointName)) + failpointEnabled = false + testutil.Eventually(re, func() bool { + etcdLeader, err = leaderServer.GetEtcdLeader() + if err != nil { + return false + } + return etcdLeader == targetName + }) +} + func post(t *testing.T, re *require.Assertions, url string, body string) { testutil.Eventually(re, func() bool { - res, err := tests.TestDialClient.Post(url, "", bytes.NewBufferString(body)) // #nosec - re.NoError(err) - b, err := io.ReadAll(res.Body) - res.Body.Close() - re.NoError(err) - t.Logf("post %s, status: %v res: %s", url, res.StatusCode, string(b)) - return res.StatusCode == http.StatusOK + status, _ := postOnce(t, re, url, body) + return status == http.StatusOK }) } +func postOnce(t *testing.T, re *require.Assertions, url string, body string) (int, string) { + res, err := tests.TestDialClient.Post(url, "", bytes.NewBufferString(body)) // #nosec + re.NoError(err) + defer res.Body.Close() + b, err := io.ReadAll(res.Body) + re.NoError(err) + t.Logf("post %s, status: %v res: %s", url, res.StatusCode, string(b)) + return res.StatusCode, string(b) +} + func waitEtcdLeaderChange(re *require.Assertions, server *tests.TestServer, old string) string { var leader string testutil.Eventually(re, func() bool { @@ -201,6 +271,9 @@ func waitEtcdLeaderChange(re *require.Assertions, server *tests.TestServer, old } func TestLeaderResign(t *testing.T) { + restorePDReleaseVersion := setPDReleaseVersionForTest() + defer restorePDReleaseVersion() + re := require.New(t) ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -224,7 +297,134 @@ func TestLeaderResign(t *testing.T) { re.Equal(leader1, leader3) } +func TestLeaderTransferChecksTargetReady(t *testing.T) { + restorePDReleaseVersion := setPDReleaseVersionForTest() + defer restorePDReleaseVersion() + + re := require.New(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + cluster, err := tests.NewTestCluster(ctx, 3) + defer cluster.Destroy() + re.NoError(err) + + re.NoError(cluster.RunInitialServers()) + leaderName := cluster.WaitLeader() + re.NotEmpty(leaderName) + leaderServer := cluster.GetLeaderServer() + re.NoError(leaderServer.BootstrapCluster()) + followerName := cluster.GetFollower() + followerServer := cluster.GetServer(followerName) + + failpointName := "github.com/tikv/pd/server/apiv2/handlers/loadRegionSlow" + re.NoError(failpoint.Enable(failpointName, `return("`+followerServer.GetAddr()+`")`)) + failpointEnabled := true + defer func() { + if failpointEnabled { + re.NoError(failpoint.Disable(failpointName)) + } + }() + + status, body := postOnce(t, re, leaderServer.GetConfig().ClientUrls+"/pd/api/v1/leader/transfer/"+followerName, "") + re.Equal(http.StatusInternalServerError, status) + re.Contains(body, "not ready") + re.Equal(leaderName, cluster.GetLeader()) + + re.NoError(failpoint.Disable(failpointName)) + failpointEnabled = false + post(t, re, leaderServer.GetConfig().ClientUrls+"/pd/api/v1/leader/transfer/"+followerName, "") + leader2 := waitLeaderChange(re, cluster, leaderName) + re.Equal(followerName, leader2) +} + +func TestResignEtcdLeaderReturnsTargetCheckError(t *testing.T) { + re := require.New(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + cluster, err := tests.NewTestCluster(ctx, 2) + defer cluster.Destroy() + re.NoError(err) + + re.NoError(cluster.RunInitialServers()) + leaderName := cluster.WaitLeader() + re.NotEmpty(leaderName) + leaderServer := cluster.GetLeaderServer() + err = leaderServer.GetServer().GetMember().ResignEtcdLeader(ctx, leaderName, "", + member.WithTargetChecker(func(context.Context, uint64) error { + return errors.New("not ready") + })) + re.Error(err) + re.True(errors.ErrorEqual(err, errs.ErrEtcdLeaderTransferTargetCheck)) +} + +func TestLeaderResignSkipsUnreadyCandidate(t *testing.T) { + restorePDReleaseVersion := setPDReleaseVersionForTest() + defer restorePDReleaseVersion() + + re := require.New(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + cluster, err := tests.NewTestCluster(ctx, 3) + defer cluster.Destroy() + re.NoError(err) + + re.NoError(cluster.RunInitialServers()) + leaderName := cluster.WaitLeader() + re.NotEmpty(leaderName) + leaderServer := cluster.GetLeaderServer() + re.NoError(leaderServer.BootstrapCluster()) + followers := getFollowerNames(cluster, leaderName) + re.Len(followers, 2) + unreadyName := followers[0] + readyName := followers[1] + unreadyServer := cluster.GetServer(unreadyName) + + failpointName := "github.com/tikv/pd/server/apiv2/handlers/loadRegionSlow" + re.NoError(failpoint.Enable(failpointName, `return("`+unreadyServer.GetAddr()+`")`)) + defer func() { + re.NoError(failpoint.Disable(failpointName)) + }() + + post(t, re, leaderServer.GetConfig().ClientUrls+"/pd/api/v1/leader/resign", "") + leader2 := waitLeaderChange(re, cluster, leaderName) + re.Equal(readyName, leader2) +} + +func TestLeaderResignFailsWhenAllCandidatesUnready(t *testing.T) { + restorePDReleaseVersion := setPDReleaseVersionForTest() + defer restorePDReleaseVersion() + + re := require.New(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + cluster, err := tests.NewTestCluster(ctx, 2) + defer cluster.Destroy() + re.NoError(err) + + re.NoError(cluster.RunInitialServers()) + leaderName := cluster.WaitLeader() + re.NotEmpty(leaderName) + leaderServer := cluster.GetLeaderServer() + re.NoError(leaderServer.BootstrapCluster()) + followerName := cluster.GetFollower() + followerServer := cluster.GetServer(followerName) + + failpointName := "github.com/tikv/pd/server/apiv2/handlers/loadRegionSlow" + re.NoError(failpoint.Enable(failpointName, `return("`+followerServer.GetAddr()+`")`)) + defer func() { + re.NoError(failpoint.Disable(failpointName)) + }() + + status, body := postOnce(t, re, leaderServer.GetConfig().ClientUrls+"/pd/api/v1/leader/resign", "") + re.Equal(http.StatusInternalServerError, status) + re.Contains(body, "no ready pd") + re.Equal(leaderName, cluster.GetLeader()) +} + func TestLeaderResignWithBlock(t *testing.T) { + restorePDReleaseVersion := setPDReleaseVersionForTest() + defer restorePDReleaseVersion() + re := require.New(t) ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -284,6 +484,24 @@ func waitLeaderChange(re *require.Assertions, cluster *tests.TestCluster, old st return leader } +func getFollowerNames(cluster *tests.TestCluster, leaderName string) []string { + followers := make([]string, 0) + for _, server := range cluster.GetConfig().InitialServers { + if server.Name != leaderName { + followers = append(followers, server.Name) + } + } + return followers +} + +func setPDReleaseVersionForTest() func() { + oldVersion := versioninfo.PDReleaseVersion + versioninfo.PDReleaseVersion = "v8.5.2" + return func() { + versioninfo.PDReleaseVersion = oldVersion + } +} + func TestMoveLeader(t *testing.T) { re := require.New(t) ctx, cancel := context.WithCancel(context.Background()) diff --git a/tools/pd-ctl/tests/member/member_test.go b/tools/pd-ctl/tests/member/member_test.go index 4df9e3e1c9e..ac4daf19d78 100644 --- a/tools/pd-ctl/tests/member/member_test.go +++ b/tools/pd-ctl/tests/member/member_test.go @@ -19,6 +19,7 @@ import ( "encoding/json" "strconv" "testing" + "time" "github.com/stretchr/testify/require" @@ -62,21 +63,29 @@ func TestMember(t *testing.T) { // member leader transfer args = []string{"-u", pdAddr, "member", "leader", "transfer", "pd2"} - _, err = tests.ExecuteCommand(cmd, args...) + testutil.Eventually(re, func() bool { + _, err = tests.ExecuteCommand(cmd, args...) + return err == nil + }, testutil.WithWaitFor(30*time.Second), testutil.WithTickInterval(500*time.Millisecond)) re.NoError(err) testutil.Eventually(re, func() bool { - return svr.GetLeader().GetName() == "pd2" - }) + etcdLeader, err := svr.GetEtcdLeader() + return err == nil && etcdLeader == "pd2" && cluster.GetLeader() == "pd2" + }, testutil.WithWaitFor(90*time.Second), testutil.WithTickInterval(time.Second)) // member leader resign re.NotEmpty(cluster.WaitLeader()) args = []string{"-u", pdAddr, "member", "leader", "resign"} - output, err = tests.ExecuteCommand(cmd, args...) - re.Contains(string(output), "Success") + testutil.Eventually(re, func() bool { + output, err = tests.ExecuteCommand(cmd, args...) + return err == nil + }, testutil.WithWaitFor(30*time.Second), testutil.WithTickInterval(500*time.Millisecond)) re.NoError(err) + re.Contains(string(output), "Success") testutil.Eventually(re, func() bool { - return svr.GetLeader().GetName() != "pd2" - }) + leader := cluster.GetLeader() + return leader != "" && leader != "pd2" + }, testutil.WithWaitFor(90*time.Second), testutil.WithTickInterval(time.Second)) // member leader_priority re.NotEmpty(cluster.WaitLeader())