Repository: matrixorigin/matrixcube
Pull request #572 (2021-12-17 12:44:45)

Implement log and snapshot compaction as event worker actionns
**What type of PR is this?**

- [ ] API-change
- [ ] BUG
- [x] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

**Which issue(s) this PR fixes:**

Fixes #

**What this PR does / why we need it:**

Not Available

**Special notes for your reviewer:**

Not Available

**Additional documentation (e.g. design docs, usage docs, etc.):**

Not Available



Patch:
diff --git a/raftstore/replica_apply_result.go b/raftstore/replica_apply_result.go
index 4821089..dbe3099 100644
--- a/raftstore/replica_apply_result.go
+++ b/raftstore/replica_apply_result.go
@@ -17,14 +17,11 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/fagongzi/util/protoc"
-	"go.etcd.io/etcd/raft/v3"
 	"go.etcd.io/etcd/raft/v3/raftpb"
 	"go.uber.org/zap"
 
 	"github.com/matrixorigin/matrixcube/components/log"
 	"github.com/matrixorigin/matrixcube/components/prophet/pb/metapb"
-	"github.com/matrixorigin/matrixcube/pb/meta"
 	"github.com/matrixorigin/matrixcube/pb/rpc"
 )
 
@@ -134,65 +131,12 @@ func (pr *replica) applyUpdateLabels(result updateLabelsResult) {
 }
 
 func (pr *replica) applyCompactionResult(r compactionResult) {
-	if r.index == 0 {
-		return
-	}
-	pr.logger.Info("log compaction called",
-		log.IndexField(r.index))
-	// generate a dummy snapshot so we can run the log compaction.
-	// this dummy snapshot will be used to establish the marker position of
-	// the LogReader on startup.
-	// such dummy snapshot will never be loaded, as its Index value is not
-	// greater than data storage's persistentLogIndex value.
-	term, err := pr.lr.Term(r.index)
-	if err != nil {
-		pr.logger.Error("failed to get term value",
-			zap.Error(err),
-			log.IndexField(r.index))
-		if err == raft.ErrCompacted || err == raft.ErrUnavailable {
-			// skip this compaction operation as we can't establish the marker
-			// position.
-			pr.logger.Info("skipped a compaction request",
-				log.IndexField(r.index))
-			return
-		}
-		panic(err)
-	}
-	// this is a dummy snapshot meaning there is no on disk snapshot image.
-	// we are not supposed to apply such dummy snapshot. the dummy flag is
-	// used for debugging purposes.
-	si := meta.SnapshotInfo{
-		Dummy: true,
-	}
-	rd := raft.Ready{
-		Snapshot: raftpb.Snapshot{
-			Metadata: raftpb.SnapshotMetadata{
-				Index: r.index,
-				Term:  term,
-			},
-			Data: protoc.MustMarshal(&si),
-		},
-	}
-	wc := pr.logdb.NewWorkerContext()
-	defer wc.Close()
-	if err := pr.logdb.SaveRaftState(pr.shardID, pr.replicaID, rd, wc); err != nil {
-		panic(err)
-	}
-	pr.logger.Info("dummy snapshot saved",
-		log.IndexField(r.index))
-	// update LogReader's range info to make the compacted entries invisible to
-	// raft.
-	if err := pr.lr.Compact(r.index); err != nil {
-		if err != raft.ErrCompacted {
-			// TODO: check whether any error should be tolerated.
-			panic(err)
-		}
-	}
-	if err := pr.logdb.RemoveEntriesTo(pr.shardID, pr.replicaID, r.index); err != nil {
-		panic(err)
+	if r.index > 0 {
+		pr.addAction(action{
+			actionType:  logCompactionAction,
+			targetIndex: r.index,
+		})
 	}
-	pr.logger.Info("compaction completed",
-		log.IndexField(r.index))
 }
 
 func (pr *replica) applyConfChange(cp configChangeResult) {
diff --git a/raftstore/replica_event_loop.go b/raftstore/replica_event_loop.go
index 1d1b0a8..144726e 100644
--- a/raftstore/replica_event_loop.go
+++ b/raftstore/replica_event_loop.go
@@ -18,6 +18,8 @@ import (
 	"time"
 
 	"go.etcd.io/etcd/raft/v3"
+	"go.etcd.io/etcd/raft/v3/raftpb"
+	trackerPkg "go.etcd.io/etcd/raft/v3/tracker"
 	"go.uber.org/zap"
 
 	"github.com/fagongzi/util/protoc"
@@ -29,7 +31,6 @@ import (
 	"github.com/matrixorigin/matrixcube/pb/rpc"
 	"github.com/matrixorigin/matrixcube/util"
 	"github.com/matrixorigin/matrixcube/util/uuid"
-	trackerPkg "go.etcd.io/etcd/raft/v3/tracker"
 )
 
 const (
@@ -37,12 +38,13 @@ const (
 )
 
 type action struct {
-	actionType     actionType
-	splitCheckData splitCheckData
-	targetIndex    uint64
-	readMetrics    readMetrics
-	epoch          Epoch
-	actionCallback func(interface{})
+	actionType         actionType
+	snapshotCompaction snapshotCompactionDetails
+	splitCheckData     splitCheckData
+	targetIndex        uint64
+	readMetrics        readMetrics
+	epoch              Epoch
+	actionCallback     func(interface{})
 }
 
 type readMetrics struct {
@@ -58,6 +60,11 @@ type splitCheckData struct {
 	ctx       []byte
 }
 
+type snapshotCompactionDetails struct {
+	snapshot           raftpb.Snapshot
+	persistentLogIndex uint64
+}
+
 type actionType int
 
 const (
@@ -69,6 +76,8 @@ const (
 	updateReadMetrics
 	checkLogCommittedAction
 	checkLogAppliedAction
+	logCompactionAction
+	snapshotCompactionAction
 )
 
 func (pr *replica) addAdminRequest(adminType rpc.AdminCmdType, request protoc.PB) {
@@ -312,6 +321,11 @@ func (pr *replica) handleAction(items []interface{}) bool {
 			pr.doCheckLogApplied(act)
 		case checkCompactLogAction:
 			pr.doCheckLogCompact(pr.rn.Status().Progress, pr.rn.LastIndex())
+		case logCompactionAction:
+			pr.doLogCompaction(act.targetIndex)
+		case snapshotCompactionAction:
+			pr.doSnapshotCompaction(act.snapshotCompaction.snapshot,
+				act.snapshotCompaction.persistentLogIndex)
 		}
 	}
 
@@ -483,19 +497,37 @@ func (pr *replica) doCheckLogCompact(progresses map[uint64]trackerPkg.Progress,
 	firstIndex := pr.getFirstIndex()
 	if minReplicatedIndex < firstIndex ||
 		minReplicatedIndex-firstIndex <= pr.store.cfg.Raft.RaftLog.CompactThreshold {
-		return
-	}
-	if appliedIndex > firstIndex &&
+		pr.logger.Debug("maybe skip requesting log compaction",
+			zap.Uint64("min-replicated-index", minReplicatedIndex),
+			zap.Uint64("applied-index", minReplicatedIndex),
+			zap.Uint64("last-index", lastIndex),
+			zap.Uint64("first-index", firstIndex),
+			zap.Uint64("threshold", pr.store.cfg.Raft.RaftLog.CompactThreshold))
+		compactIndex = 0
+	}
+
+	// check wether to force compaction or not
+	if compactIndex == 0 &&
+		appliedIndex > firstIndex &&
 		appliedIndex-firstIndex >= pr.store.cfg.Raft.RaftLog.ForceCompactCount {
 		compactIndex = appliedIndex
-	} else if pr.stats.raftLogSizeHint >= pr.store.cfg.Raft.RaftLog.ForceCompactBytes {
+	} else if compactIndex == 0 &&
+		pr.stats.raftLogSizeHint >= pr.store.cfg.Raft.RaftLog.ForceCompactBytes {
 		compactIndex = appliedIndex
 	}
+
 	if compactIndex == 0 {
+		pr.logger.Debug("requesting log compaction skipped",
+			zap.Uint64("min-replicated-index", minReplicatedIndex),
+			zap.Uint64("applied-index", minReplicatedIndex),
+			zap.Uint64("last-index", lastIndex),
+			zap.Uint64("first-index", firstIndex),
+			zap.Uint64("threshold", pr.store.cfg.Raft.RaftLog.CompactThreshold))
 		return
 	}
+
 	if compactIndex > minReplicatedIndex {
-		pr.logger.Info("some replica lag is too large, maybe sent a snapshot later",
+		pr.logger.Debug("some replica lag is too large, maybe sent a snapshot later",
 			zap.Uint64("lag", compactIndex-minReplicatedIndex))
 	}
 	compactIndex--
@@ -508,3 +540,71 @@ func (pr *replica) doCheckLogCompact(progresses map[uint64]trackerPkg.Progress,
 		CompactIndex: compactIndex,
 	})
 }
+
+func (pr *replica) doLogCompaction(index uint64) {
+	if index == 0 {
+		return
+	}
+	pr.logger.Info("log compaction action handled",
+		log.IndexField(index))
+	// generate a dummy snapshot so we can run the log compaction.
+	// this dummy snapshot will be used to establish the marker position of
+	// the LogReader on startup.
+	// such dummy snapshot will never be loaded, as its Index value is not
+	// greater than data storage's persistentLogIndex value.
+	term, err := pr.lr.Term(index)
+	if err != nil {
+		pr.logger.Error("failed to get term value",
+			zap.Error(err),
+			log.IndexField(index))
+		if err == raft.ErrCompacted || err == raft.ErrUnavailable {
+			// skip this compaction operation as we can't establish the marker
+			// position.
+			pr.logger.Info("skipped a compaction action",
+				log.IndexField(index))
+			return
+		}
+		panic(err)
+	}
+	// this is a dummy snapshot meaning there is no on disk snapshot image.
+	// we are not supposed to apply such dummy snapshot. the dummy flag is
+	// used for debugging purposes.
+	si := meta.SnapshotInfo{
+		Dummy: true,
+	}
+	rd := raft.Ready{
+		Snapshot: raftpb.Snapshot{
+			Metadata: raftpb.SnapshotMetadata{
+				Index: index,
+				Term:  term,
+			},
+			Data: protoc.MustMarshal(&si),
+		},
+	}
+	wc := pr.logdb.NewWorkerContext()
+	defer wc.Close()
+	if err := pr.logdb.SaveRaftState(pr.shardID, pr.replicaID, rd, wc); err != nil {
+		panic(err)
+	}
+	pr.logger.Info("dummy snapshot saved",
+		log.IndexField(index))
+	// update LogReader's range info to make the compacted entries invisible to
+	// raft.
+	if err := pr.lr.Compact(index); err != nil {
+		if err != raft.ErrCompacted {
+			// TODO: check whether any error should be tolerated.
+			panic(err)
+		}
+	}
+	if err := pr.logdb.RemoveEntriesTo(pr.shardID, pr.replicaID, index); err != nil {
+		panic(err)
+	}
+	pr.logger.Info("compaction completed",
+		log.IndexField(index))
+}
+
+func (pr *replica) doSnapshotCompaction(ss raftpb.Snapshot, index uint64) {
+	if err := pr.snapshotCompaction(ss, index); err != nil {
+		panic(err)
+	}
+}
diff --git a/raftstore/replica_snapshot.go b/raftstore/replica_snapshot.go
index 8cc9cf1..7e30a4d 100644
--- a/raftstore/replica_snapshot.go
+++ b/raftstore/replica_snapshot.go
@@ -128,13 +128,13 @@ func (r *replica) applySnapshot(ss raftpb.Snapshot) error {
 	if err != nil {
 		return err
 	}
-	// FIXME: change this to an event worker action
-	if err := r.snapshotCompaction(ss, persistentLogIndex); err != nil {
-		logger.Error("snapshot compaction failed",
-			zap.Error(err))
-		return err
-	}
-
+	r.addAction(action{
+		actionType: snapshotCompactionAction,
+		snapshotCompaction: snapshotCompactionDetails{
+			snapshot:           ss,
+			persistentLogIndex: persistentLogIndex,
+		},
+	})
 	if r.aware != nil {
 		r.aware.Updated(md.Metadata.Shard)
 	}


Test patch:
diff --git a/raftstore/replica_event_loop_test.go b/raftstore/replica_event_loop_test.go
index 730e6b5..e996251 100644
--- a/raftstore/replica_event_loop_test.go
+++ b/raftstore/replica_event_loop_test.go
@@ -133,6 +133,7 @@ func TestApplyInitialSnapshot(t *testing.T) {
 		assert.Equal(t, ss.Metadata.Term, r.sm.metadataMu.term)
 		assert.Equal(t, shard, r.sm.metadataMu.shard)
 
+		r.handleAction(make([]interface{}, readyBatchSize))
 		sms, err := r.sm.dataStorage.GetInitialStates()
 		assert.NoError(t, err)
 		assert.Equal(t, 1, len(sms))
@@ -245,6 +246,20 @@ func TestDoCheckCompactLog(t *testing.T) {
 	}, 102)
 	assert.Equal(t, int64(0), pr.requests.Len())
 
+	// force count, if minReplicated - first == CompactThreshold
+	pr.store.cfg.Raft.RaftLog.ForceCompactCount = 1
+	pr.store.cfg.Raft.RaftLog.CompactThreshold = 1
+	pr.stats.raftLogSizeHint = 0
+	pr.sm.setFirstIndex(100)
+	pr.appliedIndex = 102
+	pr.doCheckLogCompact(map[uint64]trackerPkg.Progress{
+		1: {Match: 101},
+	}, 101)
+	v, _ := pr.requests.Peek()
+	req := &rpc.CompactLogRequest{}
+	protoc.MustUnmarshal(req, v.(reqCtx).req.Cmd)
+	assert.Equal(t, uint64(101), req.CompactIndex)
+
 	// force count
 	pr.store.cfg.Raft.RaftLog.ForceCompactCount = 1
 	pr.store.cfg.Raft.RaftLog.ForceCompactBytes = 1000
@@ -255,10 +270,10 @@ func TestDoCheckCompactLog(t *testing.T) {
 	pr.doCheckLogCompact(map[uint64]trackerPkg.Progress{
 		1: {Match: 101},
 	}, 101)
-	v, _ := pr.requests.Peek()
-	req := &rpc.CompactLogRequest{}
+	v, _ = pr.requests.Peek()
+	req = &rpc.CompactLogRequest{}
 	protoc.MustUnmarshal(req, v.(reqCtx).req.Cmd)
-	assert.Equal(t, uint64(100), req.CompactIndex)
+	assert.Equal(t, uint64(101), req.CompactIndex)
 
 	// force bytes
 	pr.store.cfg.Raft.RaftLog.ForceCompactCount = 1000
diff --git a/raftstore/replica_event_raft_ready_test.go b/raftstore/replica_event_raft_ready_test.go
index d30c0ca..9099a43 100644
--- a/raftstore/replica_event_raft_ready_test.go
+++ b/raftstore/replica_event_raft_ready_test.go
@@ -253,6 +253,7 @@ func TestApplyReceivedSnapshot(t *testing.T) {
 		assert.Equal(t, ss.Metadata.Term, r.sm.metadataMu.term)
 		assert.Equal(t, shard, r.sm.metadataMu.shard)
 
+		r.handleAction(make([]interface{}, readyBatchSize))
 		sms, err := r.sm.dataStorage.GetInitialStates()
 		assert.NoError(t, err)
 		assert.Equal(t, 1, len(sms))
diff --git a/raftstore/replica_snapshot_test.go b/raftstore/replica_snapshot_test.go
index 64cd0a0..0281550 100644
--- a/raftstore/replica_snapshot_test.go
+++ b/raftstore/replica_snapshot_test.go
@@ -29,6 +29,7 @@ import (
 	"github.com/matrixorigin/matrixcube/storage/kv/mem"
 	"github.com/matrixorigin/matrixcube/util/fileutil"
 	"github.com/matrixorigin/matrixcube/util/leaktest"
+	"github.com/matrixorigin/matrixcube/util/task"
 	"github.com/matrixorigin/matrixcube/vfs"
 )
 
@@ -66,8 +67,12 @@ func runReplicaSnapshotTest(t *testing.T,
 	sm := newStateMachine(logger, ds, ldb, shard, replicaRec, nil, nil)
 	sm.updateAppliedIndexTerm(100, 1)
 	r := &replica{
+		startedC: make(chan struct{}),
+		store: &store{
+			workerPool: newWorkerPool(logger, ldb, nil, 96),
+		},
+		actions:     task.New(32),
 		storeID:     100,
-		store:       &store{},
 		logger:      logger,
 		logdb:       ldb,
 		sm:          sm,
@@ -76,6 +81,7 @@ func runReplicaSnapshotTest(t *testing.T,
 		replica:     replicaRec,
 		lr:          lr,
 	}
+	r.setStarted()
 	fn(t, r, fs)
 }
 
@@ -147,6 +153,8 @@ func TestReplicaSnapshotCanBeApplied(t *testing.T) {
 
 		r.replica = Replica{}
 		assert.NoError(t, r.applySnapshot(ss))
+		r.handleAction(make([]interface{}, readyBatchSize))
+
 		// applySnapshot will have the persistentLogIndex value updated
 		persistentLogIndex, err = r.getPersistentLogIndex()
 		assert.NoError(t, err)
diff --git a/raftstore/replica_state_machine_exec_test.go b/raftstore/replica_state_machine_exec_test.go
index f44f8b7..2f3c34e 100644
--- a/raftstore/replica_state_machine_exec_test.go
+++ b/raftstore/replica_state_machine_exec_test.go
@@ -347,6 +347,8 @@ func TestDoExecCompactLog(t *testing.T) {
 	assert.NotNil(t, result.adminResult)
 	pr.lr.SetRange(0, 100)
 	pr.handleApplyResult(result)
+	// actual compaction is done as an action by the event worker
+	pr.handleAction(make([]interface{}, readyBatchSize))
 	state, err := pr.sm.logdb.ReadRaftState(pr.shardID, pr.replicaID, 0)
 	assert.NoError(t, err)
 	assert.Equal(t, uint64(3), state.FirstIndex)