fix: speed up hakeeper bootstrap

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,15 +522,18 @@ 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
 		} else if ok {
 			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
 		}
 	}

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")

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

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 {

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 {

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()

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)
 }

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))