diff --git a/pkg/embed/operator.go b/pkg/embed/operator.go index d3aaaa05ff94d..5b9fb0cb9554f 100644 --- a/pkg/embed/operator.go +++ b/pkg/embed/operator.go @@ -78,6 +78,7 @@ type fileServiceCloser interface { const ( defaultHAKeeperRunningTimeout = 2 * time.Minute testingHAKeeperRunningTimeout = 5 * time.Minute + clusterConditionCheckInterval = 100 * time.Millisecond ) func newService( @@ -456,6 +457,7 @@ func (op *operator) waitHAKeeperRunning( client logservice.CNHAKeeperClient, ) error { // wait HAKeeper running + lastLogTime := time.Now().Add(-time.Second) for { state, err := client.GetClusterState(ctx) if errors.Is(err, context.DeadlineExceeded) { @@ -464,8 +466,11 @@ func (op *operator) waitHAKeeperRunning( if moerr.IsMoErrCode(err, moerr.ErrNoHAKeeper) || state.State != logpb.HAKeeperRunning { // not ready - op.reset.logger.Info("hakeeper not ready, retry") - if err := waitStartupRetry(ctx, op.cfg.HAKeeperRunningRetryInterval.Duration); err != nil { + if time.Since(lastLogTime) >= time.Second { + op.reset.logger.Info("hakeeper not ready, retry") + lastLogTime = time.Now() + } + if err := waitStartupRetry(ctx, op.clusterConditionCheckInterval()); err != nil { return err } continue @@ -481,6 +486,10 @@ func (op *operator) hakeeperRunningTimeout() time.Duration { return defaultHAKeeperRunningTimeout } +func (op *operator) clusterConditionCheckInterval() time.Duration { + return clusterConditionCheckInterval +} + func (op *operator) waitAnyShardReadyLocked(client logservice.CNHAKeeperClient) error { ctx, cancel := context.WithTimeoutCause(context.TODO(), time.Second*30, moerr.CauseWaitAnyShardReadyLocked) defer cancel() @@ -489,6 +498,7 @@ func (op *operator) waitAnyShardReadyLocked(client logservice.CNHAKeeperClient) func (op *operator) waitAnyShardReady(ctx context.Context, client logservice.CNHAKeeperClient) error { // wait shard ready + lastLogTime := time.Now().Add(-time.Second) for { if ok, err := func() (bool, error) { details, err := client.GetClusterDetails(ctx) @@ -512,7 +522,10 @@ func (op *operator) waitAnyShardReady(ctx context.Context, client logservice.CNH return true, nil } } - op.reset.logger.Info("shard not ready") + if time.Since(lastLogTime) >= time.Second { + op.reset.logger.Info("shard not ready") + lastLogTime = time.Now() + } return false, nil }(); err != nil { return err @@ -520,7 +533,7 @@ func (op *operator) waitAnyShardReady(ctx context.Context, client logservice.CNH op.reset.logger.Info("shard ready") return nil } - if err := waitStartupRetry(ctx, op.cfg.TNShardReadyRetryInterval.Duration); err != nil { + if err := waitStartupRetry(ctx, op.clusterConditionCheckInterval()); err != nil { return err } } diff --git a/pkg/embed/operator_test.go b/pkg/embed/operator_test.go index cb940b57655fe..7d8ec24bb2830 100644 --- a/pkg/embed/operator_test.go +++ b/pkg/embed/operator_test.go @@ -169,6 +169,12 @@ func TestHAKeeperRunningTimeout(t *testing.T) { assert.Equal(t, 5*time.Minute, (&operator{testing: true}).hakeeperRunningTimeout()) } +func TestClusterConditionCheckInterval(t *testing.T) { + interval := (&operator{}).clusterConditionCheckInterval() + assert.Greater(t, interval, time.Duration(0)) + assert.Less(t, interval, time.Second) +} + func TestWaitClusterConditionClosesHAKeeperClient(t *testing.T) { waitErr := errors.New("wait failed") closeErr := errors.New("close failed") diff --git a/pkg/logservice/service.go b/pkg/logservice/service.go index ace3dc2b9a223..30d40522f0c0a 100644 --- a/pkg/logservice/service.go +++ b/pkg/logservice/service.go @@ -74,6 +74,7 @@ type Service struct { stopper *stopper.Stopper haClient LogHAKeeperClient fileService fileservice.FileService + heartbeatC chan struct{} shutdownC chan struct{} options struct { @@ -116,6 +117,7 @@ func NewService( cfg: cfg, stopper: stopper.NewStopper("log-service"), fileService: fileService, + heartbeatC: make(chan struct{}, 1), shutdownC: shutdownC, } for _, opt := range opts { @@ -136,6 +138,7 @@ func NewService( service.runtime.Logger().Error("failed to create log store", zap.Error(err)) return nil, err } + store.bootstrapCommandsAdded = service.requestHeartbeat if err := store.loadMetadata(); err != nil { _ = store.close() return nil, err diff --git a/pkg/logservice/service_bootstrap.go b/pkg/logservice/service_bootstrap.go index 5ee6399d63cfa..50a1f6a2d7227 100644 --- a/pkg/logservice/service_bootstrap.go +++ b/pkg/logservice/service_bootstrap.go @@ -153,6 +153,16 @@ func (s *Service) BootstrapHAKeeper(ctx context.Context, cfg Config) error { return nil default: } + ready, err := s.store.waitHAKeeperLeaderReady(ctx, hakeeperDefaultTimeout) + if err != nil { + if restoreConfigured { + return err + } + return nil + } + if !ready { + continue + } s.runtime.SubLogger(runtime.SystemInit).Info("before initial cluster info") applied, err := s.store.setInitialClusterInfoWithRecoveryResult( numOfLogShards, @@ -184,6 +194,7 @@ func (s *Service) BootstrapHAKeeper(ctx context.Context, cfg Config) error { initialClusterProposed = true s.runtime.SubLogger(runtime.SystemInit).Info("initial cluster info set", zap.Bool("applied", applied)) + s.requestHeartbeat() break } if backup != nil { diff --git a/pkg/logservice/service_commands.go b/pkg/logservice/service_commands.go index 43972ddbf7604..2fd52cc66941c 100644 --- a/pkg/logservice/service_commands.go +++ b/pkg/logservice/service_commands.go @@ -182,6 +182,8 @@ func (s *Service) heartbeatWorker(ctx context.Context) { select { case <-ctx.Done(): return + case <-s.heartbeatC: + s.heartbeat(ctx) case <-ticker.C: s.heartbeat(ctx) // I'd call this an ugly hack to just workaround select's @@ -195,6 +197,16 @@ func (s *Service) heartbeatWorker(ctx context.Context) { } } +func (s *Service) requestHeartbeat() { + if s.heartbeatC == nil { + return + } + select { + case s.heartbeatC <- struct{}{}: + default: + } +} + func (s *Service) checkReplicaHealth(ctx context.Context) { details, err := s.store.getClusterDetails(ctx) if err != nil { diff --git a/pkg/logservice/store.go b/pkg/logservice/store.go index a603fb48c22c9..a4dcfbbf423b1 100644 --- a/pkg/logservice/store.go +++ b/pkg/logservice/store.go @@ -139,8 +139,10 @@ type store struct { tickerStopper *stopper.Stopper runtime runtime.Runtime - bootstrapCheckCycles uint64 - bootstrapMgr *bootstrap.Manager + bootstrapCheckCycles uint64 + bootstrapMgr *bootstrap.Manager + lastBootstrapLogTime time.Time + bootstrapCommandsAdded func() taskScheduler hakeeper.TaskScheduler @@ -1230,8 +1232,8 @@ func (l *store) ticker(ctx context.Context) { defer func() { l.runtime.Logger().Info("HAKeeper ticker stopped") }() - haTicker := time.NewTicker(l.cfg.HAKeeperCheckInterval.Duration) - defer haTicker.Stop() + haTimer := time.NewTimer(l.initialHAKeeperCheckInterval()) + defer haTimer.Stop() // moving task schedule from the ticker normal routine to a // separate goroutine can avoid the hakeeper's health check and tick update @@ -1245,8 +1247,9 @@ func (l *store) ticker(ctx context.Context) { select { case <-ticker.C: l.hakeeperTick() - case <-haTicker.C: - l.hakeeperCheck() + case <-haTimer.C: + state := l.hakeeperCheck() + haTimer.Reset(l.nextHAKeeperCheckInterval(state)) case <-ctx.Done(): return } @@ -1268,6 +1271,33 @@ func (l *store) isLeaderHAKeeper() (bool, uint64, error) { return ok && replicaID != 0 && leaderID == replicaID, term, nil } +func (l *store) waitHAKeeperLeaderReady(ctx context.Context, maxWait time.Duration) (bool, error) { + if leaderID, _, ok, err := l.nh.GetLeaderID(hakeeper.DefaultHAKeeperShardID); err == nil && ok && leaderID != 0 { + return true, nil + } + if maxWait <= 0 { + return false, nil + } + + ticker := time.NewTicker(time.Millisecond * 20) + defer ticker.Stop() + timer := time.NewTimer(maxWait) + defer timer.Stop() + for { + leaderID, _, ok, err := l.nh.GetLeaderID(hakeeper.DefaultHAKeeperShardID) + if err == nil && ok && leaderID != 0 { + return true, nil + } + select { + case <-ctx.Done(): + return false, moerr.AttachCause(ctx, ctx.Err()) + case <-timer.C: + return false, nil + case <-ticker.C: + } + } +} + // TODO: add test for this func (l *store) hakeeperTick() { isLeader, _, err := l.isLeaderHAKeeper() diff --git a/pkg/logservice/store_hakeeper_check.go b/pkg/logservice/store_hakeeper_check.go index 4c9cd59d6faab..7a12ad0c0e39a 100644 --- a/pkg/logservice/store_hakeeper_check.go +++ b/pkg/logservice/store_hakeeper_check.go @@ -18,6 +18,7 @@ import ( "context" "fmt" "os" + "strings" "sync/atomic" "time" @@ -31,9 +32,10 @@ import ( ) const ( - minIDAllocCapacity uint64 = 1024 - defaultIDBatchSize uint64 = 1024 * 10 - checkBootstrapCycles = 100 + minIDAllocCapacity uint64 = 1024 + defaultIDBatchSize uint64 = 1024 * 10 + checkBootstrapCycles = 100 + bootstrapHAKeeperCheckInterval = 100 * time.Millisecond ) var ( @@ -252,16 +254,37 @@ func (l *store) getCheckerStateFromLeader() (*pb.CheckerState, uint64) { var debugPrintHAKeeperState atomic.Bool -func (l *store) hakeeperCheck() { +func (l *store) initialHAKeeperCheckInterval() time.Duration { + interval := l.cfg.HAKeeperCheckInterval.Duration + if interval > bootstrapHAKeeperCheckInterval { + return bootstrapHAKeeperCheckInterval + } + return interval +} + +func (l *store) nextHAKeeperCheckInterval(state *pb.CheckerState) time.Duration { + interval := l.cfg.HAKeeperCheckInterval.Duration + if state != nil && state.State != pb.HAKeeperRunning { + if interval > bootstrapHAKeeperCheckInterval { + return bootstrapHAKeeperCheckInterval + } + } + return interval +} + +func (l *store) hakeeperCheck() *pb.CheckerState { state, term := l.getCheckerStateFromLeader() if state == nil { - return + return nil } switch state.State { case pb.HAKeeperCreated: - l.runtime.Logger().Warn("waiting for initial cluster info to be set, check skipped") - return + if time.Since(l.lastBootstrapLogTime) >= time.Second { + l.runtime.Logger().Warn("waiting for initial cluster info to be set, check skipped") + l.lastBootstrapLogTime = time.Now() + } + return state case pb.HAKeeperBootstrapping: l.bootstrap(term, state) case pb.HAKeeperBootstrapCommandsReceived: @@ -277,6 +300,7 @@ func (l *store) hakeeperCheck() { default: panic("unknown HAKeeper state") } + return state } func (l *store) assertHAKeeperState(s pb.HAKeeperState) { @@ -358,6 +382,13 @@ func (l *store) bootstrap(term uint64, state *pb.CheckerState) { } cmds, err := l.getScheduleCommand(false, term, state) if err != nil { + if isBootstrapWaitingForLogStores(err) { + if time.Since(l.lastBootstrapLogTime) >= time.Second { + l.runtime.Logger().Info("waiting for log stores before bootstrap", zap.Error(err)) + l.lastBootstrapLogTime = time.Now() + } + return + } l.runtime.Logger().Error("failed to get bootstrap schedule commands", zap.Error(err)) return } @@ -398,9 +429,17 @@ func (l *store) bootstrap(term uint64, state *pb.CheckerState) { l.bootstrapCheckCycles = checkBootstrapCycles l.bootstrapMgr = bootstrap.NewBootstrapManager(state.ClusterInfo) l.assertHAKeeperState(pb.HAKeeperBootstrapCommandsReceived) + if l.bootstrapCommandsAdded != nil { + l.bootstrapCommandsAdded() + } } } +func isBootstrapWaitingForLogStores(err error) bool { + return moerr.IsMoErrCode(err, moerr.ErrInternal) && + strings.Contains(err.Error(), "not enough log stores") +} + func (l *store) checkBootstrap(state *pb.CheckerState) { l.checkBootstrapWithSetter(state, l.setBootstrapState) } diff --git a/pkg/logservice/store_hakeeper_check_test.go b/pkg/logservice/store_hakeeper_check_test.go index 967754ddfc302..ac0fd3e94fa63 100644 --- a/pkg/logservice/store_hakeeper_check_test.go +++ b/pkg/logservice/store_hakeeper_check_test.go @@ -36,6 +36,7 @@ import ( "github.com/matrixorigin/matrixone/pkg/pb/metadata" "github.com/matrixorigin/matrixone/pkg/pb/task" "github.com/matrixorigin/matrixone/pkg/taskservice" + "github.com/matrixorigin/matrixone/pkg/util/toml" ) func TestIDAllocatorDefaultState(t *testing.T) { @@ -46,6 +47,29 @@ func TestIDAllocatorDefaultState(t *testing.T) { assert.Equal(t, uint64(0), v) } +func TestNextHAKeeperCheckIntervalUsesFastBootstrapInterval(t *testing.T) { + s := &store{ + cfg: Config{ + HAKeeperCheckInterval: toml.Duration{Duration: 3 * time.Second}, + }, + } + + require.Equal(t, bootstrapHAKeeperCheckInterval, s.initialHAKeeperCheckInterval()) + require.Equal(t, 3*time.Second, s.nextHAKeeperCheckInterval(nil)) + require.Equal(t, bootstrapHAKeeperCheckInterval, s.nextHAKeeperCheckInterval(&pb.CheckerState{ + State: pb.HAKeeperCreated, + })) + require.Equal(t, bootstrapHAKeeperCheckInterval, s.nextHAKeeperCheckInterval(&pb.CheckerState{ + State: pb.HAKeeperBootstrapping, + })) + require.Equal(t, bootstrapHAKeeperCheckInterval, s.nextHAKeeperCheckInterval(&pb.CheckerState{ + State: pb.HAKeeperBootstrapCommandsReceived, + })) + require.Equal(t, 3*time.Second, s.nextHAKeeperCheckInterval(&pb.CheckerState{ + State: pb.HAKeeperRunning, + })) +} + func TestIDAllocatorCapacity(t *testing.T) { tests := []struct { next uint64 @@ -940,11 +964,16 @@ func testBootstrap(t *testing.T, fail bool, remoteRecoveryPending bool) { state, err = store.getCheckerState() require.NoError(t, err) + bootstrapCommandsAdded := false + store.bootstrapCommandsAdded = func() { + bootstrapCommandsAdded = true + } store.bootstrap(term, state) state, err = store.getCheckerState() require.NoError(t, err) assert.Equal(t, pb.HAKeeperBootstrapCommandsReceived, state.State) + assert.True(t, bootstrapCommandsAdded) assert.Equal(t, uint64(checkBootstrapCycles), store.bootstrapCheckCycles) require.NotNil(t, store.bootstrapMgr) assert.False(t, store.bootstrapMgr.CheckBootstrap(state.LogState)) diff --git a/pkg/tests/service/logservice.go b/pkg/tests/service/logservice.go index c7d0a7bb5b7a8..2cb6516686697 100644 --- a/pkg/tests/service/logservice.go +++ b/pkg/tests/service/logservice.go @@ -15,6 +15,7 @@ package service import ( + "context" "path/filepath" "sync" @@ -226,13 +227,11 @@ func (c *testCluster) startHAKeeperReplica() error { } // setInitialClusterInfo initializes cluster information. -func (c *testCluster) setInitialClusterInfo() error { - errChan := make(chan error, 1) - +func (c *testCluster) setInitialClusterInfo(ctx context.Context) error { initialize := func() { var err error defer func() { - errChan <- err + c.log.initialInfoErr = err }() selected := c.selectHAkeeperServices() @@ -240,7 +239,8 @@ func (c *testCluster) setInitialClusterInfo() error { c.logger.Info("initialize cluster information") - err = selected[0].SetInitialClusterInfo( + leader := c.WaitHAKeeperLeader(ctx) + err = leader.SetInitialClusterInfo( c.opt.initial.logShardNum, c.opt.initial.tnShardNum, c.opt.initial.logReplicaNum, @@ -255,7 +255,7 @@ func (c *testCluster) setInitialClusterInfo() error { // initialize cluster only once c.log.once.Do(initialize) - return <-errChan + return c.log.initialInfoErr } // listHAKeeperService lists all log services that start hakeeper. diff --git a/pkg/tests/service/service.go b/pkg/tests/service/service.go index d41d6b92e1af3..efd40b4ba5a72 100644 --- a/pkg/tests/service/service.go +++ b/pkg/tests/service/service.go @@ -257,7 +257,8 @@ type testCluster struct { } log struct { - once sync.Once + once sync.Once + initialInfoErr error sync.Mutex cfgs []logservice.Config @@ -1507,7 +1508,7 @@ func (c *testCluster) startLogServices(ctx context.Context) error { } // initialize cluster information - if err := c.setInitialClusterInfo(); err != nil { + if err := c.setInitialClusterInfo(ctx); err != nil { return err } diff --git a/pkg/tests/service/service_test.go b/pkg/tests/service/service_test.go index 9eb2742e75a24..64918373f8299 100644 --- a/pkg/tests/service/service_test.go +++ b/pkg/tests/service/service_test.go @@ -17,7 +17,9 @@ package service import ( "context" "testing" + "time" + "github.com/lni/dragonboat/v4" "github.com/lni/goutils/leaktest" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -26,6 +28,7 @@ import ( "github.com/matrixorigin/matrixone/pkg/common/stopper" "github.com/matrixorigin/matrixone/pkg/logservice" logpb "github.com/matrixorigin/matrixone/pkg/pb/logservice" + "github.com/matrixorigin/matrixone/pkg/taskservice" ) const ( @@ -46,6 +49,79 @@ func TestClusterAdmissionCoversServiceClusterLifecycle(t *testing.T) { require.Nil(t, c.mu.admission) } +func TestSetInitialClusterInfoUsesHAKeeperLeader(t *testing.T) { + follower := &initialClusterInfoLogService{id: "follower"} + leader := &initialClusterInfoLogService{id: "leader", leader: true} + c := &testCluster{ + t: t, + logger: zap.NewNop(), + } + c.opt.initial.logServiceNum = 2 + c.opt.initial.logShardNum = 3 + c.opt.initial.tnShardNum = 4 + c.opt.initial.logReplicaNum = 5 + c.log.svcs = []LogService{follower, leader} + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + require.NoError(t, c.setInitialClusterInfo(ctx)) + + require.Equal(t, 0, follower.initialClusterInfoCalls) + require.Equal(t, 1, leader.initialClusterInfoCalls) + require.Equal(t, [3]uint64{3, 4, 5}, leader.initialClusterInfoArgs) +} + +func TestSetInitialClusterInfoCanBeCalledAgain(t *testing.T) { + leader := &initialClusterInfoLogService{id: "leader", leader: true} + c := &testCluster{ + t: t, + logger: zap.NewNop(), + } + c.opt.initial.logServiceNum = 1 + c.log.svcs = []LogService{leader} + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + require.NoError(t, c.setInitialClusterInfo(ctx)) + require.NoError(t, c.setInitialClusterInfo(ctx)) + require.Equal(t, 1, leader.initialClusterInfoCalls) +} + +type initialClusterInfoLogService struct { + id string + leader bool + initialClusterInfoCalls int + initialClusterInfoArgs [3]uint64 +} + +func (s *initialClusterInfoLogService) Start() error { return nil } +func (s *initialClusterInfoLogService) Close() error { return nil } +func (s *initialClusterInfoLogService) Status() ServiceStatus { + return ServiceStarted +} +func (s *initialClusterInfoLogService) ID() string { return s.id } +func (s *initialClusterInfoLogService) IsLeaderHakeeper() (bool, error) { + return s.leader, nil +} +func (s *initialClusterInfoLogService) GetClusterState() (*logpb.CheckerState, error) { + return nil, nil +} +func (s *initialClusterInfoLogService) SetInitialClusterInfo( + numOfLogShards, numOfTNShards, numOfLogReplicas uint64, +) error { + s.initialClusterInfoCalls++ + s.initialClusterInfoArgs = [3]uint64{numOfLogShards, numOfTNShards, numOfLogReplicas} + return nil +} +func (s *initialClusterInfoLogService) StartHAKeeperReplica( + replicaID uint64, initialReplicas map[uint64]dragonboat.Target, join bool, +) error { + return nil +} +func (s *initialClusterInfoLogService) GetTaskService() (taskservice.TaskService, bool) { + return nil, false +} + func TestClusterStart(t *testing.T) { defer leaktest.AfterTest(t)() if testing.Short() {