Repository: matrixorigin/matrixone
Pull request #4030 (2022-07-15 15:47:58)

logservice,hakeeper: added TestHAKeeperClusterCanBootstrapAndRepairLogShard
**What type of PR is this?**

- [ ] API-change
- [x] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [x] Test and CI
- [ ] Code Refactoring

**Which issue(s) this PR fixes:**

issue #

**What this PR does / why we need it:**

Added a test TestHAKeeperClusterCanBootstrapAndRepairLogShard. Fixed a few minor issues that prevented the new test from completion. 

**Special notes for your reviewer:**

Not Available

**Additional documentation (e.g. design docs, usage docs, etc.):**

Not Available



Patch:
diff --git a/pkg/hakeeper/rsm.go b/pkg/hakeeper/rsm.go
index eb7c3dbfd..09658ada7 100644
--- a/pkg/hakeeper/rsm.go
+++ b/pkg/hakeeper/rsm.go
@@ -198,6 +198,7 @@ func (s *stateMachine) handleUpdateCommandsCmd(cmd []byte) sm.Result {
 	for _, c := range b.Commands {
 		if c.Bootstrapping {
 			if s.state.State != pb.HAKeeperBootstrapping {
+				plog.Errorf("ignored bootstrapping cmd: %s", c.LogString())
 				return sm.Result{}
 			}
 		}
@@ -215,6 +216,7 @@ func (s *stateMachine) handleUpdateCommandsCmd(cmd []byte) sm.Result {
 				Commands: make([]pb.ScheduleCommand, 0),
 			}
 		}
+		plog.Infof("adding schedule command to hakeeper rsm: %s", c.LogString())
 		l.Commands = append(l.Commands, c)
 		s.state.ScheduleCommands[c.UUID] = l
 	}
diff --git a/pkg/logservice/commands.go b/pkg/logservice/commands.go
index f74e87a5d..d52ea5cea 100644
--- a/pkg/logservice/commands.go
+++ b/pkg/logservice/commands.go
@@ -21,6 +21,7 @@ import (
 
 func (s *Service) handleCommands(cmds []pb.ScheduleCommand) {
 	for _, cmd := range cmds {
+		plog.Infof("%s applying cmd: %s", s.ID(), cmd.LogString())
 		if cmd.GetConfigChange() != nil {
 			plog.Infof("applying schedule command: %s", cmd.LogString())
 			switch cmd.ConfigChange.ChangeType {
diff --git a/pkg/logservice/config.go b/pkg/logservice/config.go
index 9b02cb339..7caccfb42 100644
--- a/pkg/logservice/config.go
+++ b/pkg/logservice/config.go
@@ -48,6 +48,9 @@ type Config struct {
 	GossipAddress        string
 	GossipListenAddress  string
 	GossipSeedAddresses  []string
+	// DisableHAKeeperTicker disables the HAKeeper ticker in test. Never set this
+	// field to true in production
+	DisableHAKeeperTicker bool
 }
 
 // Validate validates the configuration.
diff --git a/pkg/logservice/hakeeper_checker.go b/pkg/logservice/hakeeper_checker.go
index 8be089a48..2fbecff48 100644
--- a/pkg/logservice/hakeeper_checker.go
+++ b/pkg/logservice/hakeeper_checker.go
@@ -178,6 +178,9 @@ func (l *store) bootstrap(term uint64, state *pb.CheckerState) {
 		return
 	}
 	if len(cmds) > 0 {
+		for _, c := range cmds {
+			plog.Infof("bootstrap cmd: %s", c.LogString())
+		}
 		ctx, cancel := context.WithTimeout(context.Background(), hakeeperDefaultTimeout)
 		defer cancel()
 		if err := l.addScheduleCommands(ctx, term, cmds); err != nil {
diff --git a/pkg/logservice/hakeeper_client.go b/pkg/logservice/hakeeper_client.go
index df06fdaef..991fc74fb 100644
--- a/pkg/logservice/hakeeper_client.go
+++ b/pkg/logservice/hakeeper_client.go
@@ -100,6 +100,7 @@ func newHAKeeperClient(ctx context.Context,
 		c.addr = addr
 		c.client = cc
 		isHAKeeper, err := c.checkRemote(ctx)
+		plog.Infof("isHAKeeper: %t, err: %v", isHAKeeper, err)
 		if err == nil && isHAKeeper {
 			return c, nil
 		} else if err != nil {
diff --git a/pkg/logservice/service.go b/pkg/logservice/service.go
index 5f2b83317..f62fd161c 100644
--- a/pkg/logservice/service.go
+++ b/pkg/logservice/service.go
@@ -20,13 +20,13 @@ package logservice
 import (
 	"context"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/fagongzi/goetty/v2"
 	"github.com/lni/dragonboat/v4/logger"
 
 	"github.com/matrixorigin/matrixone/pkg/common/morpc"
-	"github.com/matrixorigin/matrixone/pkg/hakeeper"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 )
 
@@ -120,7 +120,9 @@ func NewService(cfg Config) (*Service, error) {
 
 func (s *Service) Close() (err error) {
 	err = firstError(err, s.server.Close())
-	err = firstError(err, s.store.close())
+	if s.store != nil {
+		err = firstError(err, s.store.close())
+	}
 	return err
 }
 
@@ -340,13 +342,8 @@ func (s *Service) handleDNHeartbeat(req pb.Request) pb.Response {
 
 func (s *Service) handleCheckHAKeeper(req pb.Request) pb.Response {
 	resp := getResponse(req)
-	hb := s.store.getHeartbeatMessage()
-	for _, replicaInfo := range hb.Replicas {
-		si := replicaInfo.LogShardInfo
-		if si.ShardID == hakeeper.DefaultHAKeeperShardID {
-			resp.IsHAKeeper = true
-			return resp
-		}
+	if atomic.LoadUint64(&s.store.haKeeperReplicaID) != 0 {
+		resp.IsHAKeeper = true
 	}
 	return resp
 }
diff --git a/pkg/logservice/store.go b/pkg/logservice/store.go
index 6ada11d06..ac51a722f 100644
--- a/pkg/logservice/store.go
+++ b/pkg/logservice/store.go
@@ -17,6 +17,7 @@ package logservice
 import (
 	"context"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/cockroachdb/errors"
@@ -30,6 +31,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/common/stopper"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper/bootstrap"
+	"github.com/matrixorigin/matrixone/pkg/hakeeper/checkers"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 )
 
@@ -127,6 +129,7 @@ func newLogStore(cfg Config) (*store, error) {
 	ls := &store{
 		cfg:     cfg,
 		nh:      nh,
+		checker: checkers.NewCoordinator(),
 		alloc:   newIDAllocator(),
 		stopper: stopper.NewStopper("log-store"),
 	}
@@ -154,16 +157,19 @@ func (l *store) id() string {
 
 func (l *store) startHAKeeperReplica(replicaID uint64,
 	initialReplicas map[uint64]dragonboat.Target, join bool) error {
-	l.haKeeperReplicaID = replicaID
 	raftConfig := getRaftConfig(hakeeper.DefaultHAKeeperShardID, replicaID)
 	if err := l.nh.StartReplica(initialReplicas,
 		join, hakeeper.NewStateMachine, raftConfig); err != nil {
 		return err
 	}
-	if err := l.stopper.RunTask(func(ctx context.Context) {
-		l.ticker(ctx)
-	}); err != nil {
-		return err
+	atomic.StoreUint64(&l.haKeeperReplicaID, replicaID)
+	if !l.cfg.DisableHAKeeperTicker {
+		if err := l.stopper.RunTask(func(ctx context.Context) {
+			l.ticker(ctx)
+		}); err != nil {
+			return err
+		}
+		plog.Infof("HAKeeper ticker restarted")
 	}
 	return nil
 }
@@ -659,6 +665,14 @@ func (l *store) getHeartbeatMessage() pb.LogStoreHeartbeat {
 	}
 	nhi := l.nh.GetNodeHostInfo(opts)
 	for _, ci := range nhi.ShardInfoList {
+		if ci.Pending {
+			plog.Infof("shard %d is pending, not included into the heartbeat",
+				ci.ShardID)
+			continue
+		}
+		if ci.ConfigChangeIndex == 0 {
+			panic("ci.ConfigChangeIndex is 0")
+		}
 		replicaInfo := pb.LogReplicaInfo{
 			LogShardInfo: pb.LogShardInfo{
 				ShardID:  ci.ShardID,


Test patch:
diff --git a/pkg/logservice/hakeeper_checker_test.go b/pkg/logservice/hakeeper_checker_test.go
index 92a0e0e59..6e9c82a12 100644
--- a/pkg/logservice/hakeeper_checker_test.go
+++ b/pkg/logservice/hakeeper_checker_test.go
@@ -16,6 +16,7 @@ package logservice
 
 import (
 	"context"
+	"os"
 	"testing"
 	"time"
 
@@ -131,6 +132,218 @@ func runHAKeeperStoreTest(t *testing.T, startLogReplica bool, fn func(*testing.T
 	fn(t, store)
 }
 
+func runHAKeeperClusterTest(t *testing.T, fn func(*testing.T, []*Service)) {
+	defer leaktest.AfterTest(t)()
+	cfg1 := Config{
+		FS:                    vfs.NewStrictMem(),
+		DeploymentID:          1,
+		RTTMillisecond:        5,
+		DataDir:               "data-1",
+		ServiceAddress:        "127.0.0.1:9002",
+		RaftAddress:           "127.0.0.1:9000",
+		GossipAddress:         "127.0.0.1:9001",
+		GossipSeedAddresses:   []string{"127.0.0.1:9011", "127.0.0.1:9021", "127.0.0.1:9031"},
+		DisableHAKeeperTicker: true,
+	}
+	cfg2 := Config{
+		FS:                    vfs.NewStrictMem(),
+		DeploymentID:          1,
+		RTTMillisecond:        5,
+		DataDir:               "data-2",
+		ServiceAddress:        "127.0.0.1:9012",
+		RaftAddress:           "127.0.0.1:9010",
+		GossipAddress:         "127.0.0.1:9011",
+		GossipSeedAddresses:   []string{"127.0.0.1:9001", "127.0.0.1:9021", "127.0.0.1:9031"},
+		DisableHAKeeperTicker: true,
+	}
+	cfg3 := Config{
+		FS:                    vfs.NewStrictMem(),
+		DeploymentID:          1,
+		RTTMillisecond:        5,
+		DataDir:               "data-3",
+		ServiceAddress:        "127.0.0.1:9022",
+		RaftAddress:           "127.0.0.1:9020",
+		GossipAddress:         "127.0.0.1:9021",
+		GossipSeedAddresses:   []string{"127.0.0.1:9001", "127.0.0.1:9011", "127.0.0.1:9031"},
+		DisableHAKeeperTicker: true,
+	}
+	cfg4 := Config{
+		FS:                    vfs.NewStrictMem(),
+		DeploymentID:          1,
+		RTTMillisecond:        5,
+		DataDir:               "data-4",
+		ServiceAddress:        "127.0.0.1:9032",
+		RaftAddress:           "127.0.0.1:9030",
+		GossipAddress:         "127.0.0.1:9031",
+		GossipSeedAddresses:   []string{"127.0.0.1:9001", "127.0.0.1:9011", "127.0.0.1:9021"},
+		DisableHAKeeperTicker: true,
+	}
+	service1, err := NewService(cfg1)
+	require.NoError(t, err)
+	defer func() {
+		//assert.NoError(t, service1.Close())
+	}()
+	service2, err := NewService(cfg2)
+	require.NoError(t, err)
+	defer func() {
+		assert.NoError(t, service2.Close())
+	}()
+	service3, err := NewService(cfg3)
+	require.NoError(t, err)
+	defer func() {
+		assert.NoError(t, service3.Close())
+	}()
+	service4, err := NewService(cfg4)
+	require.NoError(t, err)
+	defer func() {
+		assert.NoError(t, service4.Close())
+	}()
+
+	peers := make(map[uint64]dragonboat.Target)
+	peers[1] = service1.ID()
+	peers[2] = service2.ID()
+	peers[3] = service3.ID()
+	assert.NoError(t, service1.store.startHAKeeperReplica(1, peers, false))
+	assert.NoError(t, service2.store.startHAKeeperReplica(2, peers, false))
+	assert.NoError(t, service3.store.startHAKeeperReplica(3, peers, false))
+	fn(t, []*Service{service1, service2, service3, service4})
+}
+
+func TestHAKeeperClusterCanBootstrapAndRepairLogShard(t *testing.T) {
+	if os.Getenv("LONG_TEST") == "" {
+		// this test will fail on go1.18 when -race is enabled, as it always
+		// timeout. go1.19 has a much faster -race implementation and it works fine
+		t.Skip("Skipping long test")
+	}
+
+	fn := func(t *testing.T, services []*Service) {
+		// bootstrap the cluster, 1 DN 1 Log shard, Log and HAKeeper have
+		// 3 replicas
+		store1 := services[0].store
+		state, err := store1.getCheckerState()
+		require.NoError(t, err)
+		assert.Equal(t, pb.HAKeeperCreated, state.State)
+		require.NoError(t, store1.setInitialClusterInfo(1, 1, 3))
+		state, err = store1.getCheckerState()
+		require.NoError(t, err)
+		assert.Equal(t, pb.HAKeeperBootstrapping, state.State)
+
+		sendHeartbeat := func(ss []*Service) {
+			for _, s := range ss {
+				m := s.store.getHeartbeatMessage()
+				ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+				defer cancel()
+				assert.NoError(t, s.store.addLogStoreHeartbeat(ctx, m))
+			}
+		}
+		sendHeartbeat(services[:3])
+
+		// find out the leader HAKeeper store as we need the term value
+		var term uint64
+		var leaderStore *store
+		for _, s := range services[:3] {
+			isLeader, curTerm, err := s.store.isLeaderHAKeeper()
+			require.NoError(t, err)
+			if isLeader {
+				term = curTerm
+				leaderStore = s.store
+				break
+			}
+		}
+		require.NotNil(t, leaderStore)
+		require.True(t, term > 0)
+
+		// bootstrap the cluster
+		state, err = leaderStore.getCheckerState()
+		require.NoError(t, err)
+		leaderStore.bootstrap(term, state)
+
+		state, err = leaderStore.getCheckerState()
+		require.NoError(t, err)
+		assert.Equal(t, pb.HAKeeperBootstrapCommandsReceived, state.State)
+		assert.Equal(t, uint64(checkBootstrapInterval), leaderStore.bootstrapCheckInterval)
+		require.NotNil(t, leaderStore.bootstrapMgr)
+		assert.False(t, leaderStore.bootstrapMgr.CheckBootstrap(state.LogState))
+
+		// get and apply all bootstrap schedule commands
+		ctx, cancel := context.WithTimeout(context.Background(), 100*time.Second)
+		defer cancel()
+		for _, s := range services[:3] {
+			cb, err := s.store.getCommandBatch(ctx, s.store.id())
+			require.NoError(t, err)
+			if len(cb.Commands) > 0 {
+				s.handleStartReplica(cb.Commands[0])
+			}
+		}
+
+		// check bootstrap can be completed
+		for i := 0; i < 100; i++ {
+			sendHeartbeat(services[:3])
+			state, err = leaderStore.getCheckerState()
+			require.NoError(t, err)
+			leaderStore.checkBootstrap(state)
+
+			state, err = leaderStore.getCheckerState()
+			require.NoError(t, err)
+			if state.State != pb.HAKeeperRunning {
+				// FIXME: why wait here?
+				time.Sleep(50 * time.Millisecond)
+			} else {
+				break
+			}
+			if i == 99 {
+				t.Fatalf("failed to complete bootstrap")
+			}
+		}
+
+		// stop store 1
+		require.NoError(t, services[0].Close())
+		// no service.Close can be repeatedly called
+		services[0].store = nil
+		services = services[1:]
+
+		// wait for HAKeeper to repair the cluster
+		for i := 0; i < 5000; i++ {
+			m := services[0].store.getHeartbeatMessage()
+			assert.NoError(t, services[0].store.addLogStoreHeartbeat(ctx, m))
+			m = services[1].store.getHeartbeatMessage()
+			assert.NoError(t, services[1].store.addLogStoreHeartbeat(ctx, m))
+			m = services[2].store.getHeartbeatMessage()
+			assert.NoError(t, services[0].store.addLogStoreHeartbeat(ctx, m))
+
+			for _, s := range services {
+				if hasShard(s.store, 0) {
+					s.store.hakeeperTick()
+					s.store.hakeeperCheck()
+				}
+
+				cb, err := services[0].store.getCommandBatch(ctx, s.store.id())
+				for _, cmd := range cb.Commands {
+					plog.Infof("store returned schedule command: %s", cmd.LogString())
+				}
+				require.NoError(t, err)
+				s.handleCommands(cb.Commands)
+			}
+
+			notReady := false
+			for _, s := range services {
+				if !hasShard(s.store, 0) || !hasShard(s.store, 1) {
+					notReady = true
+					break
+				}
+			}
+			if notReady {
+				time.Sleep(time.Millisecond)
+			} else {
+				plog.Infof("repair completed, i: %d", i)
+				return
+			}
+		}
+		t.Fatalf("failed to repair shards")
+	}
+	runHAKeeperClusterTest(t, fn)
+}
+
 func TestGetCheckerState(t *testing.T) {
 	fn := func(t *testing.T, store *store) {
 		state, err := store.getCheckerState()
@@ -153,8 +366,6 @@ func TestSetInitialClusterInfo(t *testing.T) {
 	runHAKeeperStoreTest(t, false, fn)
 }
 
-// FIXME: re-enable this test
-
 func TestFailedBootstrap(t *testing.T) {
 	testBootstrap(t, true)
 }
@@ -213,7 +424,6 @@ func testBootstrap(t *testing.T, fail bool) {
 			require.Equal(t, 1, len(cb.Commands))
 			service := &Service{store: store}
 			service.handleStartReplica(cb.Commands[0])
-			//service.handleStartReplica(cb.Commands[1])
 
 			for i := 0; i < 100; i++ {
 				ctx, cancel := context.WithTimeout(context.Background(), time.Second)
diff --git a/pkg/logservice/service_test.go b/pkg/logservice/service_test.go
index 8ebe5bbb2..6a631dcd6 100644
--- a/pkg/logservice/service_test.go
+++ b/pkg/logservice/service_test.go
@@ -32,7 +32,7 @@ import (
 )
 
 const (
-	testServiceAddress = "localhost:9000"
+	testServiceAddress = "127.0.0.1:9000"
 )
 
 func getServiceTestConfig() Config {
diff --git a/pkg/logservice/store_test.go b/pkg/logservice/store_test.go
index 17bd3e70f..07f125c89 100644
--- a/pkg/logservice/store_test.go
+++ b/pkg/logservice/store_test.go
@@ -339,9 +339,15 @@ func TestGetHeartbeatMessage(t *testing.T) {
 		assert.NoError(t, store.startReplica(10, 1, peers, false))
 		assert.NoError(t, store.startHAKeeperReplica(1, peers, false))
 
-		m := store.getHeartbeatMessage()
-		// hakeeper shard is included
-		assert.Equal(t, 3, len(m.Replicas))
+		for i := 0; i < 5000; i++ {
+			m := store.getHeartbeatMessage()
+			if len(m.Replicas) != 3 {
+				time.Sleep(time.Millisecond)
+			} else {
+				return
+			}
+		}
+		t.Fatalf("failed to get all replicas details from heartbeat message")
 	}
 	runStoreTest(t, fn)
 }
@@ -489,6 +495,16 @@ func TestRemoveReplica(t *testing.T) {
 	}
 }
 
+func hasShard(s *store, shardID uint64) bool {
+	hb := s.getHeartbeatMessage()
+	for _, info := range hb.Replicas {
+		if info.ShardID == shardID {
+			return true
+		}
+	}
+	return false
+}
+
 func hasReplica(s *store, shardID uint64, replicaID uint64) bool {
 	hb := s.getHeartbeatMessage()
 	for _, info := range hb.Replicas {