Repository: matrixorigin/matrixcube
Pull request #540 (2021-12-15 04:10:47)

Allow apply result metrics to be ignored
**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 b59a397..8752540 100644
--- a/raftstore/replica_apply_result.go
+++ b/raftstore/replica_apply_result.go
@@ -29,23 +29,24 @@ import (
 )
 
 type applyResult struct {
-	shardID     uint64
-	index       uint64
-	adminResult *adminResult
-	metrics     applyMetrics
+	shardID       uint64
+	index         uint64
+	adminResult   *adminResult
+	ignoreMetrics bool
+	metrics       applyMetrics
 }
 
 func (res *applyResult) hasSplitResult() bool {
-	return nil != res.adminResult && res.adminResult.splitResult != nil
+	return nil != res.adminResult &&
+		res.adminResult.adminType == rpc.AdminCmdType_BatchSplit
 }
 
-// TODO: should probably use values not pointers
 type adminResult struct {
 	adminType            rpc.AdminCmdType
-	configChangeResult   *configChangeResult
-	splitResult          *splitResult
-	compactionResult     *compactionResult
-	updateMetadataResult *updateMetadataResult
+	configChangeResult   configChangeResult
+	splitResult          splitResult
+	compactionResult     compactionResult
+	updateMetadataResult updateMetadataResult
 }
 
 type updateMetadataResult struct {
@@ -74,7 +75,9 @@ func (pr *replica) notifyPendingProposal(id []byte,
 
 func (pr *replica) handleApplyResult(result applyResult) {
 	pr.updateAppliedIndex(result)
-	pr.updateMetricsHints(result)
+	if !result.ignoreMetrics {
+		pr.updateMetricsHints(result)
+	}
 	if result.adminResult != nil {
 		pr.handleAdminResult(result)
 	}
@@ -95,9 +98,7 @@ func (pr *replica) updateMetricsHints(result applyResult) {
 		pr.stats.approximateSize = result.metrics.approximateDiffHint
 	} else {
 		pr.stats.deleteKeysHint += result.metrics.deleteKeysHint
-		if result.metrics.approximateDiffHint > 0 {
-			pr.stats.approximateSize = result.metrics.approximateDiffHint
-		}
+		pr.stats.approximateSize = result.metrics.approximateDiffHint
 	}
 }
 
@@ -114,13 +115,13 @@ func (pr *replica) handleAdminResult(result applyResult) {
 	}
 }
 
-func (pr *replica) applyUpdateMetadataResult(cp *updateMetadataResult) {
+func (pr *replica) applyUpdateMetadataResult(cp updateMetadataResult) {
 	for _, cc := range cp.changes {
 		pr.rn.ApplyConfChange(cc)
 	}
 }
 
-func (pr *replica) applyCompactionResult(r *compactionResult) {
+func (pr *replica) applyCompactionResult(r compactionResult) {
 	pr.logger.Info("log compaction called",
 		zap.Uint64("index", r.index))
 	// generate a dummy snapshot so we can run the log compaction.
@@ -175,7 +176,7 @@ func (pr *replica) applyCompactionResult(r *compactionResult) {
 	}
 }
 
-func (pr *replica) applyConfChange(cp *configChangeResult) {
+func (pr *replica) applyConfChange(cp configChangeResult) {
 	if cp.index == 0 {
 		// TODO: when the entry was treated as a NoOP, configChangeResult should be
 		// nil and applyConfChange() should be called in the first place.
@@ -248,7 +249,7 @@ func (pr *replica) applyConfChange(cp *configChangeResult) {
 		log.ShardField("shard", pr.getShard()))
 }
 
-func (pr *replica) applySplit(result *splitResult) {
+func (pr *replica) applySplit(result splitResult) {
 	pr.logger.Info("shard split applied, current shard will destory",
 		zap.Int("new-shards-count", len(result.newShards)))
 
diff --git a/raftstore/replica_state_machine.go b/raftstore/replica_state_machine.go
index 77cd079..183fd24 100644
--- a/raftstore/replica_state_machine.go
+++ b/raftstore/replica_state_machine.go
@@ -180,22 +180,26 @@ func (d *stateMachine) applyCommittedEntries(entries []raftpb.Entry) {
 			// noop entry with empty payload proposed by the leader at the beginning
 			// of its term
 			d.updateAppliedIndexTerm(entry.Index, entry.Term)
-			d.resultHandler.handleApplyResult(applyResult{index: entry.Index})
+			d.resultHandler.handleApplyResult(applyResult{
+				index:         entry.Index,
+				ignoreMetrics: true,
+			})
 			continue
 		}
 
-		d.applyRequestBatch(d.applyCtx)
+		ignoreMetrics := d.applyRequestBatch(d.applyCtx)
 		result := applyResult{
-			shardID:     d.shardID,
-			adminResult: d.applyCtx.adminResult,
-			index:       entry.Index,
-			metrics:     d.applyCtx.metrics,
+			shardID:       d.shardID,
+			adminResult:   d.applyCtx.adminResult,
+			index:         entry.Index,
+			ignoreMetrics: ignoreMetrics,
+			metrics:       d.applyCtx.metrics,
 		}
 		if isConfigChangeEntry(entry) {
 			if result.adminResult == nil {
 				result.adminResult = &adminResult{
 					adminType:          rpc.AdminCmdType_ConfigChange,
-					configChangeResult: &configChangeResult{},
+					configChangeResult: configChangeResult{},
 				}
 			} else {
 				result.adminResult.configChangeResult.confChange = d.applyCtx.v2cc
@@ -232,7 +236,9 @@ func (d *stateMachine) notifyShardRemoved(ctx *applyContext) {
 		resp, isConfigChangeRequestBatch(ctx.req))
 }
 
-func (d *stateMachine) applyRequestBatch(ctx *applyContext) {
+// applyRequestBatch returns a boolean value indicating whether to skip
+// updating the metrics.
+func (d *stateMachine) applyRequestBatch(ctx *applyContext) bool {
 	// FIXME: update impacted tests
 	// if sc, ok := d.store.cfg.Test.Shards[d.shardID]; ok && sc.SkipApply {
 	//	return
@@ -240,9 +246,9 @@ func (d *stateMachine) applyRequestBatch(ctx *applyContext) {
 	if d.isRemoved() {
 		d.logger.Fatal("applying entries on removed replica")
 	}
-
 	var err error
 	var resp rpc.ResponseBatch
+	ignoreMetrics := true
 	if !d.checkEpoch(ctx.req) {
 		if ce := d.logger.Check(zap.DebugLevel, "apply committed log skipped"); ce != nil {
 			ce.Write(zap.Uint64("index", ctx.index),
@@ -270,6 +276,7 @@ func (d *stateMachine) applyRequestBatch(ctx *applyContext) {
 			if ce := d.logger.Check(zap.DebugLevel, "apply write requests"); ce != nil {
 				ce.Write(zap.Uint64("index", ctx.index))
 			}
+			ignoreMetrics = false
 			resp = d.execWriteRequest(ctx)
 		}
 
@@ -283,6 +290,7 @@ func (d *stateMachine) applyRequestBatch(ctx *applyContext) {
 	// executeContext
 	d.resultHandler.notifyPendingProposal(ctx.req.Header.ID,
 		resp, isConfigChangeRequestBatch(ctx.req))
+	return ignoreMetrics
 }
 
 func (d *stateMachine) close() {
diff --git a/raftstore/replica_state_machine_exec.go b/raftstore/replica_state_machine_exec.go
index 2eaeb75..819699e 100644
--- a/raftstore/replica_state_machine_exec.go
+++ b/raftstore/replica_state_machine_exec.go
@@ -71,7 +71,7 @@ func (d *stateMachine) doExecCompactLog(ctx *applyContext) (rpc.ResponseBatch, e
 	resp := newAdminResponseBatch(rpc.AdminCmdType_CompactLog, &rpc.CompactLogResponse{})
 	ctx.adminResult = &adminResult{
 		adminType: rpc.AdminCmdType_CompactLog,
-		compactionResult: &compactionResult{
+		compactionResult: compactionResult{
 			index: compactIndex,
 		},
 	}
@@ -188,7 +188,7 @@ func (d *stateMachine) doExecConfigChange(ctx *applyContext) (rpc.ResponseBatch,
 	})
 	ctx.adminResult = &adminResult{
 		adminType: rpc.AdminCmdType_ConfigChange,
-		configChangeResult: &configChangeResult{
+		configChangeResult: configChangeResult{
 			index:   ctx.index,
 			changes: []rpc.ConfigChangeRequest{req},
 			shard:   res,
@@ -288,7 +288,7 @@ func (d *stateMachine) doExecSplit(ctx *applyContext) (rpc.ResponseBatch, error)
 	})
 	ctx.adminResult = &adminResult{
 		adminType: rpc.AdminCmdType_BatchSplit,
-		splitResult: &splitResult{
+		splitResult: splitResult{
 			newShards: newShards,
 		},
 	}
@@ -416,7 +416,7 @@ func (d *stateMachine) doUpdateMetadata(ctx *applyContext) (rpc.ResponseBatch, e
 	resp := newAdminResponseBatch(rpc.AdminCmdType_UpdateMetadata, &rpc.UpdateMetadataResponse{})
 	ctx.adminResult = &adminResult{
 		adminType: rpc.AdminCmdType_UpdateMetadata,
-		updateMetadataResult: &updateMetadataResult{
+		updateMetadataResult: updateMetadataResult{
 			changes: cc,
 		},
 	}


Test patch:
diff --git a/raftstore/replica_apply_result_test.go b/raftstore/replica_apply_result_test.go
index 3990ebc..719a165 100644
--- a/raftstore/replica_apply_result_test.go
+++ b/raftstore/replica_apply_result_test.go
@@ -17,6 +17,7 @@ import (
 	"testing"
 
 	"github.com/matrixorigin/matrixcube/pb/meta"
+	"github.com/matrixorigin/matrixcube/pb/rpc"
 	"github.com/matrixorigin/matrixcube/storage"
 	skv "github.com/matrixorigin/matrixcube/storage/kv"
 	"github.com/matrixorigin/matrixcube/util/leaktest"
@@ -35,7 +36,7 @@ func TestApplySplit(t *testing.T) {
 	kv := pr.sm.dataStorage.(storage.KVStorageWrapper).GetKVStorage()
 	assert.NoError(t, kv.Set(skv.EncodeDataKey([]byte{1}, nil), []byte("v"), false))
 
-	result := &splitResult{
+	result := splitResult{
 		newShards: []Shard{
 			{ID: 2, Start: []byte{1}, End: []byte{5}, Replicas: []Replica{{ID: 200}}},
 			{ID: 3, Start: []byte{5}, End: []byte{10}, Replicas: []Replica{{ID: 300}}},
@@ -93,27 +94,72 @@ func TestUpdateMetricsHints(t *testing.T) {
 	s := NewSingleTestClusterStore(t).GetStore(0).(*store)
 	pr := newTestReplica(Shard{ID: 1}, Replica{ID: 1}, s)
 
-	pr.updateMetricsHints(applyResult{metrics: applyMetrics{approximateDiffHint: 1, deleteKeysHint: 1, writtenBytes: 1, writtenKeys: 1}})
+	pr.updateMetricsHints(applyResult{
+		metrics: applyMetrics{
+			approximateDiffHint: 1,
+			deleteKeysHint:      1,
+			writtenBytes:        1,
+			writtenKeys:         1,
+		},
+	})
 	assert.Equal(t, uint64(1), pr.stats.approximateSize)
 	assert.Equal(t, uint64(1), pr.stats.deleteKeysHint)
 	assert.Equal(t, uint64(1), pr.stats.writtenBytes)
 	assert.Equal(t, uint64(1), pr.stats.writtenKeys)
 
-	pr.updateMetricsHints(applyResult{metrics: applyMetrics{approximateDiffHint: 3, deleteKeysHint: 1, writtenBytes: 1, writtenKeys: 1}})
+	pr.updateMetricsHints(applyResult{
+		metrics: applyMetrics{
+			approximateDiffHint: 3,
+			deleteKeysHint:      1,
+			writtenBytes:        1,
+			writtenKeys:         1,
+		},
+	})
 	assert.Equal(t, uint64(3), pr.stats.approximateSize)
 	assert.Equal(t, uint64(2), pr.stats.deleteKeysHint)
 	assert.Equal(t, uint64(2), pr.stats.writtenBytes)
 	assert.Equal(t, uint64(2), pr.stats.writtenKeys)
 
 	pr.updateMetricsHints(applyResult{})
-	assert.Equal(t, uint64(3), pr.stats.approximateSize)
+	assert.Equal(t, uint64(0), pr.stats.approximateSize)
 	assert.Equal(t, uint64(2), pr.stats.deleteKeysHint)
 	assert.Equal(t, uint64(2), pr.stats.writtenBytes)
 	assert.Equal(t, uint64(2), pr.stats.writtenKeys)
 
-	pr.updateMetricsHints(applyResult{adminResult: &adminResult{splitResult: &splitResult{}}})
+	pr.updateMetricsHints(applyResult{
+		adminResult: &adminResult{
+			adminType:   rpc.AdminCmdType_BatchSplit,
+			splitResult: splitResult{},
+		},
+	})
 	assert.Equal(t, uint64(0), pr.stats.approximateSize)
 	assert.Equal(t, uint64(0), pr.stats.deleteKeysHint)
 	assert.Equal(t, uint64(2), pr.stats.writtenBytes)
 	assert.Equal(t, uint64(2), pr.stats.writtenKeys)
 }
+
+func TestUpdateMetricsHintsCanBeIgnored(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	s := NewSingleTestClusterStore(t).GetStore(0).(*store)
+	pr := newTestReplica(Shard{ID: 1}, Replica{ID: 1}, s)
+	pr.updateMetricsHints(applyResult{
+		metrics: applyMetrics{
+			approximateDiffHint: 1,
+			deleteKeysHint:      1,
+			writtenBytes:        1,
+			writtenKeys:         1,
+		},
+	})
+	assert.Equal(t, uint64(1), pr.stats.approximateSize)
+	assert.Equal(t, uint64(1), pr.stats.deleteKeysHint)
+	assert.Equal(t, uint64(1), pr.stats.writtenBytes)
+	assert.Equal(t, uint64(1), pr.stats.writtenKeys)
+
+	pr.handleApplyResult(applyResult{
+		ignoreMetrics: true,
+	})
+	assert.Equal(t, uint64(1), pr.stats.approximateSize)
+	pr.handleApplyResult(applyResult{})
+	assert.Equal(t, uint64(0), pr.stats.approximateSize)
+}