Repository: matrixorigin/matrixone
Pull request #4045 (2022-07-16 16:58:23)

hakeeper: added tests for DN bootstrapping and repair
**What type of PR is this?**

- [ ] API-change
- [ ] 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 tests for DN bootstrapping and repair

**Special notes for your reviewer:**

Not Available

**Additional documentation (e.g. design docs, usage docs, etc.):**

Not Available



Patch:
diff --git a/pkg/hakeeper/bootstrap/bootstrap.go b/pkg/hakeeper/bootstrap/bootstrap.go
index 1cd7bff57..df4ab0489 100644
--- a/pkg/hakeeper/bootstrap/bootstrap.go
+++ b/pkg/hakeeper/bootstrap/bootstrap.go
@@ -135,7 +135,8 @@ func (bm *Manager) bootstrapDN(alloc util.IDAllocator, dn pb.DNState) (commands
 		}
 
 		commands = append(commands, pb.ScheduleCommand{
-			UUID: dnStores[i],
+			UUID:          dnStores[i],
+			Bootstrapping: true,
 			ConfigChange: &pb.ConfigChange{
 				Replica: pb.Replica{
 					UUID:      dnStores[i],


Test patch:
diff --git a/pkg/logservice/hakeeper_checker_test.go b/pkg/logservice/hakeeper_checker_test.go
index 6e9c82a12..6ad0e1ab1 100644
--- a/pkg/logservice/hakeeper_checker_test.go
+++ b/pkg/logservice/hakeeper_checker_test.go
@@ -209,7 +209,7 @@ func runHAKeeperClusterTest(t *testing.T, fn func(*testing.T, []*Service)) {
 	fn(t, []*Service{service1, service2, service3, service4})
 }
 
-func TestHAKeeperClusterCanBootstrapAndRepairLogShard(t *testing.T) {
+func TestHAKeeperCanBootstrapAndRepairShards(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
@@ -238,6 +238,15 @@ func TestHAKeeperClusterCanBootstrapAndRepairLogShard(t *testing.T) {
 		}
 		sendHeartbeat(services[:3])
 
+		// fake a DN store
+		dnMsg := pb.DNStoreHeartbeat{
+			UUID:   uuid.New().String(),
+			Shards: make([]pb.DNShardInfo, 0),
+		}
+		ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+		defer cancel()
+		require.NoError(t, services[0].store.addDNStoreHeartbeat(ctx, dnMsg))
+
 		// find out the leader HAKeeper store as we need the term value
 		var term uint64
 		var leaderStore *store
@@ -266,8 +275,6 @@ func TestHAKeeperClusterCanBootstrapAndRepairLogShard(t *testing.T) {
 		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)
@@ -296,13 +303,35 @@ func TestHAKeeperClusterCanBootstrapAndRepairLogShard(t *testing.T) {
 			}
 		}
 
+		// get the DN bootstrap command, it contains DN shard and replica ID
+		cb, err := leaderStore.getCommandBatch(ctx, dnMsg.UUID)
+		require.NoError(t, err)
+		require.Equal(t, 1, len(cb.Commands))
+		cmd := cb.Commands[0]
+		assert.True(t, cmd.Bootstrapping)
+		assert.Equal(t, pb.DnService, cmd.ServiceType)
+		dnShardInfo := pb.DNShardInfo{
+			ShardID:   cmd.ConfigChange.Replica.ShardID,
+			ReplicaID: cmd.ConfigChange.Replica.ReplicaID,
+		}
+		dnMsg.Shards = append(dnMsg.Shards, dnShardInfo)
+		// as if DN is running
+		require.NoError(t, services[0].store.addDNStoreHeartbeat(ctx, dnMsg))
+		// fake a free DN store
+		dnMsg2 := pb.DNStoreHeartbeat{
+			UUID:   uuid.New().String(),
+			Shards: make([]pb.DNShardInfo, 0),
+		}
+		require.NoError(t, services[0].store.addDNStoreHeartbeat(ctx, dnMsg2))
+
 		// 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
+		// wait for HAKeeper to repair the Log & HAKeeper shards
+		dnRepaired := false
 		for i := 0; i < 5000; i++ {
 			m := services[0].store.getHeartbeatMessage()
 			assert.NoError(t, services[0].store.addLogStoreHeartbeat(ctx, m))
@@ -310,6 +339,7 @@ func TestHAKeeperClusterCanBootstrapAndRepairLogShard(t *testing.T) {
 			assert.NoError(t, services[1].store.addLogStoreHeartbeat(ctx, m))
 			m = services[2].store.getHeartbeatMessage()
 			assert.NoError(t, services[0].store.addLogStoreHeartbeat(ctx, m))
+			require.NoError(t, services[0].store.addDNStoreHeartbeat(ctx, dnMsg2))
 
 			for _, s := range services {
 				if hasShard(s.store, 0) {
@@ -317,6 +347,18 @@ func TestHAKeeperClusterCanBootstrapAndRepairLogShard(t *testing.T) {
 					s.store.hakeeperCheck()
 				}
 
+				cb, err = services[0].store.getCommandBatch(ctx, dnMsg2.UUID)
+				require.NoError(t, err)
+				if len(cb.Commands) > 0 {
+					cmd := cb.Commands[0]
+					if cmd.ServiceType == pb.DnService {
+						if cmd.ConfigChange.Replica.ShardID == dnShardInfo.ShardID &&
+							cmd.ConfigChange.Replica.ReplicaID > dnShardInfo.ReplicaID {
+							dnRepaired = true
+						}
+					}
+				}
+
 				cb, err := services[0].store.getCommandBatch(ctx, s.store.id())
 				for _, cmd := range cb.Commands {
 					plog.Infof("store returned schedule command: %s", cmd.LogString())
@@ -325,14 +367,15 @@ func TestHAKeeperClusterCanBootstrapAndRepairLogShard(t *testing.T) {
 				s.handleCommands(cb.Commands)
 			}
 
-			notReady := false
+			logRepaired := true
 			for _, s := range services {
 				if !hasShard(s.store, 0) || !hasShard(s.store, 1) {
-					notReady = true
+					logRepaired = false
 					break
 				}
 			}
-			if notReady {
+			plog.Infof("dnRepairted %t, logRepaired %t", dnRepaired, logRepaired)
+			if !logRepaired || !dnRepaired {
 				time.Sleep(time.Millisecond)
 			} else {
 				plog.Infof("repair completed, i: %d", i)
@@ -392,7 +435,6 @@ func testBootstrap(t *testing.T, fail bool) {
 			UUID:   uuid.New().String(),
 			Shards: make([]pb.DNShardInfo, 0),
 		}
-		dnMsg.Shards = append(dnMsg.Shards, pb.DNShardInfo{ShardID: 2, ReplicaID: 3})
 		assert.NoError(t, store.addDNStoreHeartbeat(ctx, dnMsg))
 
 		_, term, err := store.isLeaderHAKeeper()
@@ -419,9 +461,17 @@ func testBootstrap(t *testing.T, fail bool) {
 			require.NoError(t, err)
 			assert.Equal(t, pb.HAKeeperBootstrapFailed, state.State)
 		} else {
-			cb, err := store.getCommandBatch(ctx, store.id())
+			cb, err := store.getCommandBatch(ctx, dnMsg.UUID)
+			require.NoError(t, err)
+			require.Equal(t, 1, len(cb.Commands))
+			assert.True(t, cb.Commands[0].Bootstrapping)
+			assert.Equal(t, pb.DnService, cb.Commands[0].ServiceType)
+			assert.True(t, cb.Commands[0].ConfigChange.Replica.ReplicaID > 0)
+
+			cb, err = store.getCommandBatch(ctx, store.id())
 			require.NoError(t, err)
 			require.Equal(t, 1, len(cb.Commands))
+			assert.True(t, cb.Commands[0].Bootstrapping)
 			service := &Service{store: store}
 			service.handleStartReplica(cb.Commands[0])