Repository: matrixorigin/matrixone
Pull request #4262 (2022-07-28 13:24:24)

hakeeper: remove schedule commands after heartbeat
**What type of PR is this?**

- [ ] API-change
- [ ] BUG
- [x] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

**Which issue(s) this PR fixes:**

issue #

**What this PR does / why we need it:**

remove schedule commands after heartbeat so log is not flooded with useless info. 

**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 2af05f6bb..75b61885f 100644
--- a/pkg/hakeeper/rsm.go
+++ b/pkg/hakeeper/rsm.go
@@ -59,8 +59,6 @@ const (
 )
 
 type StateQuery struct{}
-type logShardIDQuery struct{ name string }
-type logShardIDQueryResult struct{ id uint64 }
 type ScheduleCommandQuery struct{ UUID string }
 type ClusterDetailsQuery struct{ Cfg Config }
 
@@ -225,6 +223,18 @@ func (s *stateMachine) handleUpdateCommandsCmd(cmd []byte) sm.Result {
 	return sm.Result{}
 }
 
+func (s *stateMachine) getCommandBatch(uuid string) sm.Result {
+	if batch, ok := s.state.ScheduleCommands[uuid]; ok {
+		delete(s.state.ScheduleCommands, uuid)
+		data, err := batch.Marshal()
+		if err != nil {
+			panic(err)
+		}
+		return sm.Result{Data: data}
+	}
+	return sm.Result{}
+}
+
 func (s *stateMachine) handleCNHeartbeat(cmd []byte) sm.Result {
 	data := parseHeartbeatCmd(cmd)
 	var hb pb.CNStoreHeartbeat
@@ -232,7 +242,7 @@ func (s *stateMachine) handleCNHeartbeat(cmd []byte) sm.Result {
 		panic(err)
 	}
 	s.state.CNState.Update(hb, s.state.Tick)
-	return sm.Result{}
+	return s.getCommandBatch(hb.UUID)
 }
 
 func (s *stateMachine) handleDNHeartbeat(cmd []byte) sm.Result {
@@ -242,7 +252,7 @@ func (s *stateMachine) handleDNHeartbeat(cmd []byte) sm.Result {
 		panic(err)
 	}
 	s.state.DNState.Update(hb, s.state.Tick)
-	return sm.Result{}
+	return s.getCommandBatch(hb.UUID)
 }
 
 func (s *stateMachine) handleLogHeartbeat(cmd []byte) sm.Result {
@@ -252,7 +262,7 @@ func (s *stateMachine) handleLogHeartbeat(cmd []byte) sm.Result {
 		panic(err)
 	}
 	s.state.LogState.Update(hb, s.state.Tick)
-	return sm.Result{}
+	return s.getCommandBatch(hb.UUID)
 }
 
 func (s *stateMachine) handleTick(cmd []byte) sm.Result {
@@ -404,14 +414,6 @@ func (s *stateMachine) handleStateQuery() interface{} {
 	return result
 }
 
-func (s *stateMachine) handleShardIDQuery(name string) *logShardIDQueryResult {
-	id, ok := s.state.LogShards[name]
-	if ok {
-		return &logShardIDQueryResult{id: id}
-	}
-	return &logShardIDQueryResult{}
-}
-
 func (s *stateMachine) handleScheduleCommandQuery(uuid string) *pb.CommandBatch {
 	if batch, ok := s.state.ScheduleCommands[uuid]; ok {
 		return &batch
@@ -465,9 +467,7 @@ func (s *stateMachine) handleClusterDetailsQuery(cfg Config) *pb.ClusterDetails
 }
 
 func (s *stateMachine) Lookup(query interface{}) (interface{}, error) {
-	if q, ok := query.(*logShardIDQuery); ok {
-		return s.handleShardIDQuery(q.name), nil
-	} else if _, ok := query.(*StateQuery); ok {
+	if _, ok := query.(*StateQuery); ok {
 		return s.handleStateQuery(), nil
 	} else if q, ok := query.(*ScheduleCommandQuery); ok {
 		return s.handleScheduleCommandQuery(q.UUID), nil
diff --git a/pkg/logservice/service.go b/pkg/logservice/service.go
index ba8bdfe4c..2d2c63c4b 100644
--- a/pkg/logservice/service.go
+++ b/pkg/logservice/service.go
@@ -314,11 +314,7 @@ func (s *Service) handleLogHeartbeat(req pb.Request) pb.Response {
 	defer cancel()
 	hb := req.LogHeartbeat
 	resp := getResponse(req)
-	if err := s.store.addLogStoreHeartbeat(ctx, hb); err != nil {
-		resp.ErrorCode, resp.ErrorMessage = toErrorCode(err)
-		return resp
-	}
-	if cb, err := s.store.getCommandBatch(ctx, hb.UUID); err != nil {
+	if cb, err := s.store.addLogStoreHeartbeat(ctx, hb); err != nil {
 		resp.ErrorCode, resp.ErrorMessage = toErrorCode(err)
 		return resp
 	} else {
@@ -346,11 +342,7 @@ func (s *Service) handleDNHeartbeat(req pb.Request) pb.Response {
 	defer cancel()
 	hb := req.DNHeartbeat
 	resp := getResponse(req)
-	if err := s.store.addDNStoreHeartbeat(ctx, hb); err != nil {
-		resp.ErrorCode, resp.ErrorMessage = toErrorCode(err)
-		return resp
-	}
-	if cb, err := s.store.getCommandBatch(ctx, hb.UUID); err != nil {
+	if cb, err := s.store.addDNStoreHeartbeat(ctx, hb); err != nil {
 		resp.ErrorCode, resp.ErrorMessage = toErrorCode(err)
 		return resp
 	} else {
diff --git a/pkg/logservice/store.go b/pkg/logservice/store.go
index c98aea5df..8686e1ac0 100644
--- a/pkg/logservice/store.go
+++ b/pkg/logservice/store.go
@@ -389,15 +389,18 @@ func handleNotHAKeeperError(err error) error {
 }
 
 func (l *store) addLogStoreHeartbeat(ctx context.Context,
-	hb pb.LogStoreHeartbeat) error {
+	hb pb.LogStoreHeartbeat) (pb.CommandBatch, error) {
 	data := MustMarshal(&hb)
 	cmd := hakeeper.GetLogStoreHeartbeatCmd(data)
 	session := l.nh.GetNoOPSession(hakeeper.DefaultHAKeeperShardID)
-	if _, err := l.propose(ctx, session, cmd); err != nil {
+	if result, err := l.propose(ctx, session, cmd); err != nil {
 		plog.Errorf("propose failed, %v", err)
-		return handleNotHAKeeperError(err)
+		return pb.CommandBatch{}, handleNotHAKeeperError(err)
+	} else {
+		var cb pb.CommandBatch
+		MustUnmarshal(&cb, result.Data)
+		return cb, nil
 	}
-	return nil
 }
 
 func (l *store) addCNStoreHeartbeat(ctx context.Context,
@@ -413,15 +416,18 @@ func (l *store) addCNStoreHeartbeat(ctx context.Context,
 }
 
 func (l *store) addDNStoreHeartbeat(ctx context.Context,
-	hb pb.DNStoreHeartbeat) error {
+	hb pb.DNStoreHeartbeat) (pb.CommandBatch, error) {
 	data := MustMarshal(&hb)
 	cmd := hakeeper.GetDNStoreHeartbeatCmd(data)
 	session := l.nh.GetNoOPSession(hakeeper.DefaultHAKeeperShardID)
-	if _, err := l.propose(ctx, session, cmd); err != nil {
+	if result, err := l.propose(ctx, session, cmd); err != nil {
 		plog.Errorf("propose failed, %v", err)
-		return handleNotHAKeeperError(err)
+		return pb.CommandBatch{}, handleNotHAKeeperError(err)
+	} else {
+		var cb pb.CommandBatch
+		MustUnmarshal(&cb, result.Data)
+		return cb, nil
 	}
-	return nil
 }
 
 func (l *store) getCommandBatch(ctx context.Context,


Test patch:
diff --git a/pkg/hakeeper/rsm_test.go b/pkg/hakeeper/rsm_test.go
index cad10d3bc..db847e3a7 100644
--- a/pkg/hakeeper/rsm_test.go
+++ b/pkg/hakeeper/rsm_test.go
@@ -557,3 +557,17 @@ func TestHandleInitialClusterRequestCmd(t *testing.T) {
 	assert.Equal(t, pb.HAKeeperBootstrapping, rsm.state.State)
 	assert.Equal(t, K8SIDRangeEnd, rsm.state.NextID)
 }
+
+func TestGetCommandBatch(t *testing.T) {
+	rsm := NewStateMachine(0, 1).(*stateMachine)
+	cb := pb.CommandBatch{
+		Term: 12345,
+	}
+	rsm.state.ScheduleCommands["uuid1"] = cb
+	result := rsm.getCommandBatch("uuid1")
+	var ncb pb.CommandBatch
+	require.NoError(t, ncb.Unmarshal(result.Data))
+	assert.Equal(t, cb, ncb)
+	_, ok := rsm.state.ScheduleCommands["uuid1"]
+	assert.False(t, ok)
+}
diff --git a/pkg/logservice/hakeeper_checker_test.go b/pkg/logservice/hakeeper_checker_test.go
index 7344d79de..bde5ef093 100644
--- a/pkg/logservice/hakeeper_checker_test.go
+++ b/pkg/logservice/hakeeper_checker_test.go
@@ -246,7 +246,8 @@ func TestHAKeeperCanBootstrapAndRepairShards(t *testing.T) {
 				m := s.store.getHeartbeatMessage()
 				ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 				defer cancel()
-				assert.NoError(t, s.store.addLogStoreHeartbeat(ctx, m))
+				_, err := s.store.addLogStoreHeartbeat(ctx, m)
+				assert.NoError(t, err)
 			}
 		}
 		sendHeartbeat(services[:3])
@@ -258,7 +259,8 @@ func TestHAKeeperCanBootstrapAndRepairShards(t *testing.T) {
 		}
 		ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
 		defer cancel()
-		require.NoError(t, services[0].store.addDNStoreHeartbeat(ctx, dnMsg))
+		_, err = services[0].store.addDNStoreHeartbeat(ctx, dnMsg)
+		require.NoError(t, err)
 
 		// find out the leader HAKeeper store as we need the term value
 		var term uint64
@@ -329,13 +331,15 @@ func TestHAKeeperCanBootstrapAndRepairShards(t *testing.T) {
 		}
 		dnMsg.Shards = append(dnMsg.Shards, dnShardInfo)
 		// as if DN is running
-		require.NoError(t, services[0].store.addDNStoreHeartbeat(ctx, dnMsg))
+		_, err = services[0].store.addDNStoreHeartbeat(ctx, dnMsg)
+		require.NoError(t, err)
 		// 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))
+		_, err = services[0].store.addDNStoreHeartbeat(ctx, dnMsg2)
+		require.NoError(t, err)
 
 		// stop store 1
 		require.NoError(t, services[0].Close())
@@ -346,22 +350,29 @@ func TestHAKeeperCanBootstrapAndRepairShards(t *testing.T) {
 		// wait for HAKeeper to repair the Log & HAKeeper shards
 		dnRepaired := false
 		for i := 0; i < 5000; i++ {
+			plog.Infof("iteration %d", i)
 			tn := func() (bool, error) {
 				ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 				defer cancel()
 				m := services[0].store.getHeartbeatMessage()
-				if err := services[0].store.addLogStoreHeartbeat(ctx, m); err != nil {
+				if cb, err := services[0].store.addLogStoreHeartbeat(ctx, m); err != nil {
 					return false, err
+				} else {
+					services[0].handleCommands(cb.Commands)
 				}
 				m = services[1].store.getHeartbeatMessage()
-				if err := services[1].store.addLogStoreHeartbeat(ctx, m); err != nil {
+				if cb, err := services[1].store.addLogStoreHeartbeat(ctx, m); err != nil {
 					return false, err
+				} else {
+					services[1].handleCommands(cb.Commands)
 				}
 				m = services[2].store.getHeartbeatMessage()
-				if err := services[0].store.addLogStoreHeartbeat(ctx, m); err != nil {
+				if cb, err := services[0].store.addLogStoreHeartbeat(ctx, m); err != nil {
 					return false, err
+				} else {
+					services[2].handleCommands(cb.Commands)
 				}
-				if err := services[0].store.addDNStoreHeartbeat(ctx, dnMsg2); err != nil {
+				if _, err := services[0].store.addDNStoreHeartbeat(ctx, dnMsg2); err != nil {
 					return false, err
 				}
 
@@ -384,15 +395,6 @@ func TestHAKeeperCanBootstrapAndRepairShards(t *testing.T) {
 							}
 						}
 					}
-
-					cb, err = services[0].store.getCommandBatch(ctx, s.store.id())
-					if err != nil {
-						return false, err
-					}
-					for _, cmd := range cb.Commands {
-						plog.Infof("store returned schedule command: %s", cmd.LogString())
-					}
-					s.handleCommands(cb.Commands)
 				}
 
 				logRepaired := true
@@ -467,13 +469,15 @@ func testBootstrap(t *testing.T, fail bool) {
 		m := store.getHeartbeatMessage()
 		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 		defer cancel()
-		assert.NoError(t, store.addLogStoreHeartbeat(ctx, m))
+		_, err = store.addLogStoreHeartbeat(ctx, m)
+		assert.NoError(t, err)
 
 		dnMsg := pb.DNStoreHeartbeat{
 			UUID:   uuid.New().String(),
 			Shards: make([]pb.DNShardInfo, 0),
 		}
-		assert.NoError(t, store.addDNStoreHeartbeat(ctx, dnMsg))
+		_, err = store.addDNStoreHeartbeat(ctx, dnMsg)
+		assert.NoError(t, err)
 
 		_, term, err := store.isLeaderHAKeeper()
 		require.NoError(t, err)
@@ -517,7 +521,8 @@ func testBootstrap(t *testing.T, fail bool) {
 				ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 				defer cancel()
 				m := store.getHeartbeatMessage()
-				assert.NoError(t, store.addLogStoreHeartbeat(ctx, m))
+				_, err = store.addLogStoreHeartbeat(ctx, m)
+				assert.NoError(t, err)
 
 				state, err = store.getCheckerState()
 				require.NoError(t, err)
diff --git a/pkg/logservice/store_test.go b/pkg/logservice/store_test.go
index b4384c3e8..f8a7aefad 100644
--- a/pkg/logservice/store_test.go
+++ b/pkg/logservice/store_test.go
@@ -362,7 +362,8 @@ func TestAddHeartbeat(t *testing.T) {
 		m := store.getHeartbeatMessage()
 		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 		defer cancel()
-		assert.NoError(t, store.addLogStoreHeartbeat(ctx, m))
+		_, err := store.addLogStoreHeartbeat(ctx, m)
+		assert.NoError(t, err)
 
 		cnMsg := pb.CNStoreHeartbeat{
 			UUID: store.id(),
@@ -374,7 +375,8 @@ func TestAddHeartbeat(t *testing.T) {
 			Shards: make([]pb.DNShardInfo, 0),
 		}
 		dnMsg.Shards = append(dnMsg.Shards, pb.DNShardInfo{ShardID: 2, ReplicaID: 3})
-		assert.NoError(t, store.addDNStoreHeartbeat(ctx, dnMsg))
+		_, err = store.addDNStoreHeartbeat(ctx, dnMsg)
+		assert.NoError(t, err)
 	}
 	runStoreTest(t, fn)
 }