diff --git a/server/etcd_start_test.go b/server/etcd_start_test.go new file mode 100644 index 0000000000..bbace96f37 --- /dev/null +++ b/server/etcd_start_test.go @@ -0,0 +1,132 @@ +// 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" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/pingcap/errors" + + "github.com/tikv/pd/pkg/errs" +) + +const ( + testCheckInterval = 5 * time.Millisecond + testNoProgressTimeout = 50 * time.Millisecond +) + +// constApplied returns an appliedIndex getter that never advances, simulating a +// hung etcd that makes no apply progress. +func constApplied(v uint64) func() uint64 { + return func() uint64 { return v } +} + +// advancingApplied returns an appliedIndex getter that advances on every call, +// simulating a healthy etcd that keeps applying its raft log. +func advancingApplied() func() uint64 { + var idx atomic.Uint64 + return func() uint64 { return idx.Add(1) } +} + +func TestWaitEtcdReadyProgress(t *testing.T) { + t.Run("ready immediately", func(t *testing.T) { + re := require.New(t) + ready := make(chan struct{}) + close(ready) + err := waitEtcdReadyProgress(context.Background(), ready, nil, nil, constApplied(0), + testCheckInterval, testNoProgressTimeout) + re.NoError(err) + }) + + t.Run("fails fast on async etcd error", func(t *testing.T) { + re := require.New(t) + // etcd reports a fatal startup error before ever becoming ready. + errCh := make(chan error, 1) + errCh <- errors.New("boom") + start := time.Now() + err := waitEtcdReadyProgress(context.Background(), make(chan struct{}), nil, errCh, constApplied(0), + testCheckInterval, testNoProgressTimeout) + re.Error(err) + // The real cause is wrapped in ErrStartEtcd, not swallowed by a timeout. + re.ErrorContains(err, "PD:etcd:ErrStartEtcd") + re.ErrorContains(err, "boom") + // It must not wait out the whole no-progress window. + re.Less(time.Since(start), testNoProgressTimeout) + }) + + t.Run("fails fast when etcd server stops before ready", func(t *testing.T) { + re := require.New(t) + // The server terminates before becoming ready without publishing an error + // on errCh (e.g. an internal EtcdServer.run failure). + stopped := make(chan struct{}) + close(stopped) + start := time.Now() + err := waitEtcdReadyProgress(context.Background(), make(chan struct{}), stopped, nil, constApplied(0), + testCheckInterval, testNoProgressTimeout) + re.Error(err) + re.ErrorContains(err, "PD:etcd:ErrStartEtcd") + re.Less(time.Since(start), testNoProgressTimeout) + }) + + t.Run("keeps waiting while applied index advances", func(t *testing.T) { + re := require.New(t) + // ready only fires well after the no-progress timeout would have elapsed; + // a fixed deadline would have killed this, but steady apply progress must + // keep the wait alive. + ready := make(chan struct{}) + go func() { + time.Sleep(4 * testNoProgressTimeout) + close(ready) + }() + start := time.Now() + err := waitEtcdReadyProgress(context.Background(), ready, nil, nil, advancingApplied(), + testCheckInterval, testNoProgressTimeout) + re.NoError(err) + // It must have outlived the no-progress window instead of bailing at it. + re.GreaterOrEqual(time.Since(start), 3*testNoProgressTimeout) + }) + + t.Run("gives up when no apply progress", func(t *testing.T) { + re := require.New(t) + // ready never fires and applied index is stuck: a genuine hang. + start := time.Now() + err := waitEtcdReadyProgress(context.Background(), make(chan struct{}), nil, nil, constApplied(7), + testCheckInterval, testNoProgressTimeout) + re.Error(err) + re.True(errors.ErrorEqual(err, errs.ErrCancelStartEtcd)) + // It should wait roughly the whole no-progress window before giving up. + re.GreaterOrEqual(time.Since(start), testNoProgressTimeout-testCheckInterval) + }) + + t.Run("honors ctx cancellation", func(t *testing.T) { + re := require.New(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { + time.Sleep(2 * testCheckInterval) + cancel() + }() + // applied index advances, so only ctx cancellation can end the wait. + err := waitEtcdReadyProgress(ctx, make(chan struct{}), nil, nil, advancingApplied(), + testCheckInterval, testNoProgressTimeout) + re.Error(err) + re.True(errors.ErrorEqual(err, errs.ErrCancelStartEtcd)) + }) +} diff --git a/server/server.go b/server/server.go index 58e3caa31e..7850927827 100644 --- a/server/server.go +++ b/server/server.go @@ -114,9 +114,21 @@ const ( lostPDLeaderReElectionFactor = 10 ) -// EtcdStartTimeout the timeout of the startup etcd. +// EtcdStartTimeout is the timeout for starting the embedded etcd. It is used as +// a no-progress timeout rather than a fixed overall deadline: startEtcd only +// gives up when etcd makes no apply progress for this duration, so a +// slow-but-healthy startup that keeps catching up its raft log is not killed. +// See waitEtcdReady. var EtcdStartTimeout = time.Minute * 5 +// etcdStartProgressCheckInterval is how often startEtcd polls the embedded +// etcd applied index to tell a slow-but-progressing startup apart from a hang. +const etcdStartProgressCheckInterval = time.Second + +// etcdStartProgressLogInterval rate-limits the "still making progress" log, so a +// recovery that runs for many minutes does not emit one line per poll. +const etcdStartProgressLogInterval = 30 * time.Second + var ( // WithLabelValues is a heavy operation, define variable to avoid call it every time. etcdTermGauge = etcdStateGauge.WithLabelValues("term") @@ -345,9 +357,6 @@ func CreateServer(ctx context.Context, cfg *config.Config, services []string, le } func (s *Server) startEtcd(ctx context.Context) (retErr error) { - newCtx, cancel := context.WithTimeout(ctx, EtcdStartTimeout) - defer cancel() - etcd, err := embed.StartEtcd(s.etcdCfg) if err != nil { return errs.ErrStartEtcd.Wrap(err).GenWithStackByCause() @@ -395,11 +404,11 @@ func (s *Server) startEtcd(ctx context.Context) (retErr error) { return err } - select { - // Wait etcd until it is ready to use - case <-etcd.Server.ReadyNotify(): - case <-newCtx.Done(): - return errs.ErrCancelStartEtcd.FastGenByArgs() + // Wait until etcd is ready to use. Instead of a fixed overall deadline, only + // give up when etcd makes no apply progress for EtcdStartTimeout, so a slow + // but healthy startup that is still catching up its raft log is not killed. + if err = waitEtcdReady(ctx, etcd); err != nil { + return err } // Start the etcd and HTTP clients, then init the member. @@ -407,7 +416,9 @@ func (s *Server) startEtcd(ctx context.Context) (retErr error) { if err != nil { return err } - err = s.initMember(newCtx, etcd) + initCtx, cancel := context.WithTimeout(ctx, EtcdStartTimeout) + defer cancel() + err = s.initMember(initCtx, etcd) if err != nil { return err } @@ -416,6 +427,91 @@ func (s *Server) startEtcd(ctx context.Context) (retErr error) { return nil } +// waitEtcdReady blocks until the embedded etcd is ready to serve requests. +// +// Rather than bounding the wait by a fixed deadline, it treats the etcd applied +// index as a liveness signal: as long as the applied index keeps advancing, the +// startup is still making progress (e.g. applying its raft log while catching +// up) and we keep waiting. Only when there is no apply progress for +// EtcdStartTimeout do we give up. This distinguishes a slow-but-healthy startup +// from a genuine hang, such as a removed member that can never rejoin the +// quorum. It also fails fast when etcd reports a startup error or stops before +// becoming ready, and honors ctx cancellation for normal shutdown. +func waitEtcdReady(ctx context.Context, etcd *embed.Etcd) error { + return waitEtcdReadyProgress(ctx, etcd.Server.ReadyNotify(), etcd.Server.StopNotify(), + etcd.Err(), etcd.Server.AppliedIndex, etcdStartProgressCheckInterval, EtcdStartTimeout) +} + +// waitEtcdReadyProgress is the testable core of waitEtcdReady, decoupled from +// *embed.Etcd so it can be driven without a real server. It keeps waiting while +// appliedIndex advances and reports a no-progress failure only when appliedIndex +// does not advance for noProgressTimeout. It fails fast when: +// - etcd reports an asynchronous startup error on errCh (e.g. listener or +// serving failures); +// - the etcd server stops before becoming ready (stopped), e.g. an internal +// EtcdServer.run failure that may not surface on errCh — the underlying +// error is preserved when etcd published one; +// - ctx is canceled (normal shutdown). +func waitEtcdReadyProgress( + ctx context.Context, + ready, stopped <-chan struct{}, + errCh <-chan error, + appliedIndex func() uint64, + checkInterval, noProgressTimeout time.Duration, +) error { + ticker := time.NewTicker(checkInterval) + defer ticker.Stop() + + lastApplied := appliedIndex() + lastProgress := time.Now() + lastLog := lastProgress + for { + select { + case <-ready: + return nil + case err := <-errCh: + return errs.ErrStartEtcd.Wrap(err).GenWithStackByCause() + case <-stopped: + // The server terminated before becoming ready; surface the real + // cause if etcd published one, otherwise report the stop itself. + select { + case err := <-errCh: + return errs.ErrStartEtcd.Wrap(err).GenWithStackByCause() + default: + return errs.ErrStartEtcd.GenWithStackByArgs() + } + case <-ctx.Done(): + return errs.ErrCancelStartEtcd.FastGenByArgs() + case <-ticker.C: + applied := appliedIndex() + // Sample the wall clock here rather than reusing the ticker's + // scheduled time: a tick buffered during a scheduling or GC pause + // carries a stale timestamp that could otherwise trip a false + // no-progress timeout on the following tick. + now := time.Now() + if applied > lastApplied { + lastApplied = applied + lastProgress = now + // Rate-limit: this path can run for many minutes during a slow + // recovery, so avoid logging on every poll. + if now.Sub(lastLog) >= etcdStartProgressLogInterval { + log.Info("etcd is not ready yet but still making apply progress, keep waiting", + zap.Uint64("applied-index", applied)) + lastLog = now + } + continue + } + if stalled := now.Sub(lastProgress); stalled >= noProgressTimeout { + log.Warn("etcd is not ready and made no apply progress, stop waiting", + zap.Uint64("applied-index", applied), + zap.Duration("no-progress-duration", stalled), + zap.Duration("timeout", noProgressTimeout)) + return errs.ErrCancelStartEtcd.FastGenByArgs() + } + } + } +} + func (s *Server) initGRPCServiceLabels() { for name, serviceInfo := range s.grpcServer.GetServiceInfo() { if name == gRPCServiceName {