Merge #157932

157932: batchevel: make cmd_truncate_log use the log engine r=iskettaneh a=iskettaneh

This is the last known place where touch the unreplicated rangeID local keyspace. This commit provides the LogEngine through EvalCtx so that cmd_truncate_log can use it to compute the stats.

Fixes: #157895

Release note: None

Co-authored-by: iskettaneh <[email protected]>

Author: craig[bot]

pkg/kv/kvserver/batcheval/cmd_truncate_log.go

@@ -110,6 +110,16 @@ func TruncateLog(
 	start := keys.RaftLogKey(rangeID, firstIndex)
 	end := keys.RaftLogKey(rangeID, args.Index)
 
+	// TODO(pav-kv): GetCompactedIndex, GetTerm, and NewReader calls can disagree
+	// on the state of the log since we don't hold any Replica locks here. Move
+	// the computation inside Replica where locking can be controlled precisely.
+	//
+	// Use the log engine to compute stats for the raft log.
+	// TODO(#136358): After we precisely maintain the Raft Log size, we could stop
+	// needing the Log Engine to compute the stats.
+	logReader := cArgs.EvalCtx.LogEngine().NewReader(storage.StandardDurability)
+	defer logReader.Close()
+
 	// Compute the stats delta that were to occur should the log entries be
 	// purged. We do this as a side effect of seeing a new TruncatedState,
 	// downstream of Raft.
@@ -118,7 +128,7 @@ func TruncateLog(
 	// are not tracked in the raft log delta. The delta will be adjusted below
 	// raft.
 	// We can pass zero as nowNanos because we're only interested in SysBytes.
-	ms, err := storage.ComputeStats(ctx, readWriter, start, end, 0 /* nowNanos */)
+	ms, err := storage.ComputeStats(ctx, logReader, start, end, 0 /* nowNanos */)
 	if err != nil {
 		return result.Result{}, errors.Wrap(err, "while computing stats of Raft log freed by truncation")
 	}

pkg/kv/kvserver/batcheval/cmd_truncate_log_test.go

@@ -64,22 +64,35 @@ func TestTruncateLog(t *testing.T) {
 	)
 
 	st := cluster.MakeTestingClusterSettings()
+
+	// The truncate command takes the stateEng as any other command. However, it
+	// also accepts the log engine via EvalCtx for the stats computation.
+	stateEng := storage.NewDefaultInMemForTesting()
+	logEng := storage.NewDefaultInMemForTesting()
+	defer stateEng.Close()
+	defer logEng.Close()
+
 	evalCtx := &MockEvalCtx{
 		ClusterSettings: st,
 		Desc:            &roachpb.RangeDescriptor{RangeID: rangeID},
 		Term:            term,
 		CompactedIndex:  compactedIndex,
+		LogEngine:       logEng,
 	}
 
-	eng := storage.NewDefaultInMemForTesting()
-	defer eng.Close()
-
 	truncState := kvserverpb.RaftTruncatedState{
 		Index: compactedIndex + 2,
 		Term:  term,
 	}
 
-	putTruncatedState(t, eng, rangeID, truncState)
+	putTruncatedState(t, stateEng, rangeID, truncState)
+
+	// Write some raft log entries to the log engine to create non-zero stats.
+	// These entries will be "truncated" by our request.
+	for i := compactedIndex + 1; i < compactedIndex+8; i++ {
+		key := keys.RaftLogKey(rangeID, kvpb.RaftIndex(i))
+		require.NoError(t, logEng.PutUnversioned(key, []byte("some-data")))
+	}
 
 	// Send a truncation request.
 	req := kvpb.TruncateLogRequest{
@@ -91,10 +104,8 @@ func TestTruncateLog(t *testing.T) {
 		Args:    &req,
 	}
 	resp := &kvpb.TruncateLogResponse{}
-	res, err := TruncateLog(ctx, eng, cArgs, resp)
-	if err != nil {
-		t.Fatal(err)
-	}
+	res, err := TruncateLog(ctx, stateEng, cArgs, resp)
+	require.NoError(t, err)
 
 	expTruncState := kvserverpb.RaftTruncatedState{
 		Index: req.Index - 1,
@@ -103,9 +114,22 @@ func TestTruncateLog(t *testing.T) {
 
 	// The unreplicated key that we see should be the initial truncated
 	// state (it's only updated below Raft).
-	gotTruncatedState := readTruncStates(t, eng, rangeID)
+	gotTruncatedState := readTruncStates(t, stateEng, rangeID)
 	assert.Equal(t, truncState, gotTruncatedState)
 
 	assert.NotNil(t, res.Replicated.RaftTruncatedState)
 	assert.Equal(t, expTruncState, *res.Replicated.RaftTruncatedState)
+
+	// Verify the stats match what's actually in the log engine.
+	start := keys.RaftLogKey(rangeID, compactedIndex+1)
+	end := keys.RaftLogKey(rangeID, compactedIndex+8)
+	expectedStats, err := storage.ComputeStats(ctx, logEng, start, end, 0)
+	require.NoError(t, err)
+	assert.Equal(t, -expectedStats.SysBytes, res.Replicated.RaftLogDelta,
+		"RaftLogDelta should match stats computed from log engine")
+
+	// The state machine engine's stats should be zero.
+	zeroStats, err := storage.ComputeStats(ctx, stateEng, start, end, 0)
+	require.NoError(t, err)
+	require.Zero(t, zeroStats.SysBytes)
 }

pkg/kv/kvserver/batcheval/eval_context.go

@@ -19,6 +19,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/readsummary/rspb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
+	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/storage/enginepb"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/limit"
@@ -62,6 +63,8 @@ type EvalContext interface {
 	GetCompactedIndex() kvpb.RaftIndex
 	GetTerm(index kvpb.RaftIndex) (kvpb.RaftTerm, error)
 	GetLeaseAppliedIndex() kvpb.LeaseAppliedIndex
+	// LogEngine returns the engine that stores the raft log.
+	LogEngine() storage.Engine
 
 	Desc() *roachpb.RangeDescriptor
 	ContainsKey(key roachpb.Key) bool
@@ -175,6 +178,7 @@ type MockEvalCtx struct {
 	GCThreshold            hlc.Timestamp
 	Term                   kvpb.RaftTerm
 	CompactedIndex         kvpb.RaftIndex
+	LogEngine              storage.Engine
 	CanCreateTxnRecordFn   func() (bool, kvpb.TransactionAbortedReason)
 	MinTxnCommitTSFn       func() hlc.Timestamp
 	LastReplicaGCTimestamp hlc.Timestamp
@@ -223,6 +227,14 @@ func (m *mockEvalCtxImpl) GetConcurrencyManager() concurrency.Manager {
 		panic("ConcurrencyManager not configured")
 	}
 }
+
+func (m *mockEvalCtxImpl) LogEngine() storage.Engine {
+	if m.MockEvalCtx.LogEngine != nil {
+		return m.MockEvalCtx.LogEngine
+	}
+	panic("LogEngine not configured")
+}
+
 func (m *mockEvalCtxImpl) NodeID() roachpb.NodeID {
 	return m.MockEvalCtx.NodeID
 }

pkg/kv/kvserver/replica.go

@@ -1409,6 +1409,11 @@ func (r *Replica) StoreID() roachpb.StoreID {
 	return r.store.StoreID()
 }
 
+// LogEngine returns the log engine.
+func (r *Replica) LogEngine() storage.Engine {
+	return r.store.LogEngine()
+}
+
 // EvalKnobs returns the EvalContext's Knobs.
 func (r *Replica) EvalKnobs() kvserverbase.BatchEvalTestingKnobs {
 	return r.store.cfg.TestingKnobs.EvalKnobs

pkg/kv/kvserver/replica_eval_context_span.go

@@ -19,6 +19,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/spanset"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
+	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/storage/enginepb"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/mon"
@@ -95,6 +96,11 @@ func (rec *SpanSetReplicaEvalContext) GetLeaseAppliedIndex() kvpb.LeaseAppliedIn
 	return rec.i.GetLeaseAppliedIndex()
 }
 
+// LogEngine returns the log engine.
+func (rec *SpanSetReplicaEvalContext) LogEngine() storage.Engine {
+	return rec.i.LogEngine()
+}
+
 // IsFirstRange returns true iff the replica belongs to the first range.
 func (rec *SpanSetReplicaEvalContext) IsFirstRange() bool {
 	return rec.i.IsFirstRange()