Merge #144537

144537: logstore: make the caller responsible for metrics r=tbg a=pav-kv

Epic: CRDB-46488

Co-authored-by: Pavel Kalinnikov <[email protected]>

Author: craig[bot]

pkg/kv/kvserver/client_manual_proposal_test.go

@@ -12,7 +12,6 @@ import (
 	"path/filepath"
 	"sync"
 	"testing"
-	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/keys"
@@ -35,7 +34,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/testutils/testcluster"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
-	"github.com/cockroachdb/cockroach/pkg/util/metric"
 	"github.com/cockroachdb/cockroach/pkg/util/stop"
 	"github.com/stretchr/testify/require"
 )
@@ -215,9 +213,6 @@ LIMIT
 			Responses: []raftpb.Message{{}}, // need >0 responses so StoreEntries will sync
 		}
 
-		fakeMeta := metric.Metadata{
-			Name: "fake.meta",
-		}
 		swl := logstore.NewSyncWaiterLoop()
 		stopper := stop.NewStopper()
 		defer stopper.Stop(ctx)
@@ -230,14 +225,6 @@ LIMIT
 			SyncWaiter:  swl,
 			EntryCache:  raftentry.NewCache(1024),
 			Settings:    st,
-			Metrics: logstore.Metrics{
-				RaftLogCommitLatency: metric.NewHistogram(metric.HistogramOptions{
-					Mode:         metric.HistogramModePrometheus,
-					Metadata:     fakeMeta,
-					Duration:     time.Millisecond,
-					BucketConfig: metric.IOLatencyBuckets,
-				}),
-			},
 		}
 
 		wg := &sync.WaitGroup{}
@@ -257,8 +244,6 @@ LIMIT
 
 type wgSyncCallback sync.WaitGroup
 
-func (w *wgSyncCallback) OnLogSync(
-	context.Context, raft.StorageAppendAck, storage.BatchCommitStats,
-) {
+func (w *wgSyncCallback) OnLogSync(context.Context, raft.StorageAppendAck, logstore.WriteStats) {
 	(*sync.WaitGroup)(w).Done()
 }

pkg/kv/kvserver/logstore/logstore.go

@@ -12,6 +12,7 @@ import (
 	"math/rand"
 	"slices"
 	"sync"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
@@ -31,7 +32,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/iterutil"
 	"github.com/cockroachdb/cockroach/pkg/util/metamorphic"
-	"github.com/cockroachdb/cockroach/pkg/util/metric"
 	"github.com/cockroachdb/crlib/crtime"
 	"github.com/cockroachdb/errors"
 )
@@ -119,9 +119,10 @@ type AppendStats struct {
 	NonBlocking bool
 }
 
-// Metrics contains metrics specific to the log storage.
-type Metrics struct {
-	RaftLogCommitLatency metric.IHistogram
+// WriteStats contains stats about a write to raft storage.
+type WriteStats struct {
+	CommitDur time.Duration
+	storage.BatchCommitStats
 }
 
 // LogStore is a stub of a separated Raft log storage.
@@ -133,7 +134,6 @@ type LogStore struct {
 	SyncWaiter  *SyncWaiterLoop
 	EntryCache  *raftentry.Cache
 	Settings    *cluster.Settings
-	Metrics     Metrics
 
 	DisableSyncLogWriteToss bool // for testing only
 }
@@ -146,7 +146,7 @@ type LogStore struct {
 //
 // commitStats is populated iff this was a non-blocking sync.
 type SyncCallback interface {
-	OnLogSync(context.Context, raft.StorageAppendAck, storage.BatchCommitStats)
+	OnLogSync(context.Context, raft.StorageAppendAck, WriteStats)
 }
 
 func newStoreEntriesBatch(eng storage.Engine) storage.Batch {
@@ -273,7 +273,6 @@ func (s *LogStore) storeEntriesAndCommitBatch(
 			cb:             cb,
 			onDone:         m.Ack(),
 			batch:          batch,
-			metrics:        s.Metrics,
 			logCommitBegin: stats.PebbleBegin,
 		}
 		s.SyncWaiter.enqueue(ctx, batch, waiterCallback)
@@ -287,9 +286,8 @@ func (s *LogStore) storeEntriesAndCommitBatch(
 		stats.PebbleEnd = crtime.NowMono()
 		stats.PebbleCommitStats = batch.CommitStats()
 		if wantsSync {
-			logCommitEnd := stats.PebbleEnd
-			s.Metrics.RaftLogCommitLatency.RecordValue(logCommitEnd.Sub(stats.PebbleBegin).Nanoseconds())
-			cb.OnLogSync(ctx, m.Ack(), storage.BatchCommitStats{})
+			commitDur := stats.PebbleEnd.Sub(stats.PebbleBegin)
+			cb.OnLogSync(ctx, m.Ack(), WriteStats{CommitDur: commitDur})
 		}
 	}
 	stats.Sync = wantsSync
@@ -344,17 +342,16 @@ type nonBlockingSyncWaiterCallback struct {
 	onDone raft.StorageAppendAck
 	// Used to extract stats. This is the batch that has been synced.
 	batch storage.WriteBatch
-	// Used to record Metrics.
-	metrics        Metrics
+	// Used to measure raft storage write/sync latency.
 	logCommitBegin crtime.Mono
 }
 
 // run is the callback's logic. It is executed on the SyncWaiterLoop goroutine.
 func (cb *nonBlockingSyncWaiterCallback) run() {
-	dur := cb.logCommitBegin.Elapsed().Nanoseconds()
-	cb.metrics.RaftLogCommitLatency.RecordValue(dur)
-	commitStats := cb.batch.CommitStats()
-	cb.cb.OnLogSync(cb.ctx, cb.onDone, commitStats)
+	cb.cb.OnLogSync(cb.ctx, cb.onDone, WriteStats{
+		CommitDur:        cb.logCommitBegin.Elapsed(),
+		BatchCommitStats: cb.batch.CommitStats(),
+	})
 	cb.release()
 }
 

pkg/kv/kvserver/logstore/logstore_bench_test.go

@@ -10,7 +10,6 @@ import (
 	"fmt"
 	"math/rand"
 	"testing"
-	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/raftentry"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/raftlog"
@@ -20,7 +19,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/util/humanizeutil"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
-	"github.com/cockroachdb/cockroach/pkg/util/metric"
 	"github.com/stretchr/testify/require"
 )
 
@@ -34,9 +32,7 @@ func (b *discardBatch) Commit(bool) error {
 
 type noopSyncCallback struct{}
 
-func (noopSyncCallback) OnLogSync(
-	context.Context, raft.StorageAppendAck, storage.BatchCommitStats,
-) {
+func (noopSyncCallback) OnLogSync(context.Context, raft.StorageAppendAck, WriteStats) {
 }
 
 func BenchmarkLogStore_StoreEntries(b *testing.B) {
@@ -65,14 +61,6 @@ func runBenchmarkLogStore_StoreEntries(b *testing.B, bytes int64) {
 		StateLoader: NewStateLoader(rangeID),
 		EntryCache:  ec,
 		Settings:    st,
-		Metrics: Metrics{
-			RaftLogCommitLatency: metric.NewHistogram(metric.HistogramOptions{
-				Mode:         metric.HistogramModePrometheus,
-				Metadata:     metric.Metadata{},
-				Duration:     10 * time.Second,
-				BucketConfig: metric.IOLatencyBuckets,
-			}),
-		},
 	}
 
 	rs := RaftState{

pkg/kv/kvserver/replica_init.go

@@ -224,9 +224,6 @@ func newUninitializedReplicaWithoutRaftGroup(
 		SyncWaiter: store.syncWaiters[int(rangeID)%len(store.syncWaiters)],
 		EntryCache: store.raftEntryCache,
 		Settings:   store.cfg.Settings,
-		Metrics: logstore.Metrics{
-			RaftLogCommitLatency: store.metrics.RaftLogCommitLatency,
-		},
 		DisableSyncLogWriteToss: buildutil.CrdbTestBuild &&
 			store.TestingKnobs().DisableSyncLogWriteToss,
 	}

pkg/kv/kvserver/replica_raft.go

@@ -1800,7 +1800,7 @@ func (r *Replica) maybeCoalesceHeartbeat(
 type replicaSyncCallback Replica
 
 func (r *replicaSyncCallback) OnLogSync(
-	ctx context.Context, ack raft.StorageAppendAck, commitStats storage.BatchCommitStats,
+	ctx context.Context, ack raft.StorageAppendAck, stats logstore.WriteStats,
 ) {
 	repl := (*Replica)(r)
 	// The log mark is non-empty only if this was a non-empty log append that
@@ -1814,8 +1814,10 @@ func (r *replicaSyncCallback) OnLogSync(
 	}
 	// Send MsgStorageAppend's responses.
 	repl.sendStorageAck(ctx, ack, false /* willDeliver */)
-	if commitStats.TotalDuration > defaultReplicaRaftMuWarnThreshold {
-		log.Infof(repl.raftCtx, "slow non-blocking raft commit: %s", commitStats)
+
+	r.store.metrics.RaftLogCommitLatency.RecordValue(stats.CommitDur.Nanoseconds())
+	if stats.TotalDuration > defaultReplicaRaftMuWarnThreshold {
+		log.Infof(repl.raftCtx, "slow non-blocking raft commit: %s", stats.BatchCommitStats)
 	}
 }