db: improve metrics and logs to measure all filesystem ops

Enhance the log writer to instrument all filesystem operations done
for writing the log file in the latency histogram (create, write, sync, etc.).

Author: xxmplus

commit_test.go

@@ -20,7 +20,6 @@ import (
 	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/record"
 	"github.com/cockroachdb/pebble/vfs"
-	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/require"
 )
 
@@ -253,7 +252,6 @@ func TestCommitPipelineWALClose(t *testing.T) {
 	}
 	p := newCommitPipeline(testEnv)
 	wal = record.NewLogWriter(sf, 0 /* logNum */, record.LogWriterConfig{
-		WALFsyncLatency:     prometheus.NewHistogram(prometheus.HistogramOpts{}),
 		QueueSemChan:        p.logSyncQSem,
 		WriteWALSyncOffsets: func() bool { return false },
 	})
@@ -395,7 +393,6 @@ func BenchmarkCommitPipeline(b *testing.B) {
 					p := newCommitPipeline(nullCommitEnv)
 					wal = record.NewLogWriter(io.Discard, 0, /* logNum */
 						record.LogWriterConfig{
-							WALFsyncLatency:     prometheus.NewHistogram(prometheus.HistogramOpts{}),
 							QueueSemChan:        p.logSyncQSem,
 							WriteWALSyncOffsets: func() bool { return false },
 						})

db.go

@@ -31,15 +31,13 @@ import (
 	"github.com/cockroachdb/pebble/objstorage"
 	"github.com/cockroachdb/pebble/objstorage/remote"
 	"github.com/cockroachdb/pebble/rangekey"
-	"github.com/cockroachdb/pebble/record"
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/sstable/blob"
 	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/cockroachdb/pebble/vfs/atomicfs"
 	"github.com/cockroachdb/pebble/wal"
 	"github.com/cockroachdb/tokenbucket"
-	"github.com/prometheus/client_golang/prometheus"
 )
 
 const (
@@ -376,13 +374,7 @@ type DB struct {
 			// commitPipeline.mu and DB.mu to be held when rotating the WAL/memtable
 			// (i.e. makeRoomForWrite). Can be nil.
 			writer  wal.Writer
-			metrics struct {
-				// fsyncLatency has its own internal synchronization, and is not
-				// protected by mu.
-				fsyncLatency prometheus.Histogram
-				// Updated whenever a wal.Writer is closed.
-				record.LogWriterMetrics
-			}
+			metrics WALMetrics
 		}
 
 		mem struct {
@@ -1962,8 +1954,7 @@ func (d *DB) Metrics() *Metrics {
 	metrics.BlobFiles.ReferencedValueSize = blobStats.ReferencedValueSize
 	metrics.BlobFiles.ReferencedBackingValueSize = blobStats.ReferencedBackingValueSize
 
-	metrics.LogWriter.FsyncLatency = d.mu.log.metrics.fsyncLatency
-	if err := metrics.LogWriter.Merge(&d.mu.log.metrics.LogWriterMetrics); err != nil {
+	if err := metrics.WALMetrics.Merge(&d.mu.log.metrics.LogWriterMetrics); err != nil {
 		d.opts.Logger.Errorf("metrics error: %s", err)
 	}
 	metrics.Flush.WriteThroughput = d.mu.compact.flushWriteThroughput

metrics.go

@@ -481,10 +481,7 @@ type Metrics struct {
 		Failover wal.FailoverStats
 	}
 
-	LogWriter struct {
-		FsyncLatency prometheus.Histogram
-		record.LogWriterMetrics
-	}
+	WALMetrics WALMetrics
 
 	CategoryStats []block.CategoryStatsAggregate
 
@@ -539,6 +536,41 @@ func (cm *CompressionMetrics) MergeWith(o *CompressionMetrics) {
 	cm.Zstd.Add(o.Zstd)
 }
 
+// DirectoryContext identifies which WAL directory an operation targets
+type DirectoryContext int
+
+const (
+	// DirectoryPrimary indicates operation targets the primary WAL directory
+	DirectoryPrimary DirectoryContext = iota
+	// DirectorySecondary indicates operation targets the secondary WAL directory
+	DirectorySecondary
+	// DirectoryUnknown indicates directory context is unknown or not applicable
+	DirectoryUnknown
+)
+
+// WALMetrics contains directory-specific latency histograms for all WAL operations
+type WALMetrics struct {
+	// Directory-specific latency histograms for all filesystem operations
+	Primary struct {
+		CreateLatency  prometheus.Histogram // File creation operations
+		WriteLatency   prometheus.Histogram // Write operations
+		FsyncLatency   prometheus.Histogram // Fsync operations
+		CloseLatency   prometheus.Histogram // File close operations
+		StatLatency    prometheus.Histogram // File stat operations
+		OpenDirLatency prometheus.Histogram // Directory open operations
+	}
+	Secondary struct {
+		CreateLatency  prometheus.Histogram // File creation operations
+		WriteLatency   prometheus.Histogram // Write operations
+		FsyncLatency   prometheus.Histogram // Fsync operations
+		CloseLatency   prometheus.Histogram // File close operations
+		StatLatency    prometheus.Histogram // File stat operations
+		OpenDirLatency prometheus.Histogram // Directory open operations
+	}
+	// Embedded runtime metrics from record.LogWriterMetrics
+	record.LogWriterMetrics
+}
+
 var (
 	// FsyncLatencyBuckets are prometheus histogram buckets suitable for a histogram
 	// that records latencies for fsyncs.
@@ -547,6 +579,30 @@ var (
 		prometheus.ExponentialBucketsRange(float64(time.Millisecond*5), float64(10*time.Second), 50)...,
 	)
 
+	// WALCreateLatencyBuckets are histogram buckets for file creation operations.
+	WALCreateLatencyBuckets = append(
+		prometheus.LinearBuckets(0.0, float64(time.Microsecond*50), 40),
+		prometheus.ExponentialBucketsRange(float64(time.Millisecond*2), float64(5*time.Second), 40)...,
+	)
+
+	// WALWriteLatencyBuckets are histogram buckets for write operations.
+	WALWriteLatencyBuckets = append(
+		prometheus.LinearBuckets(0.0, float64(time.Microsecond*10), 50),
+		prometheus.ExponentialBucketsRange(float64(time.Millisecond*1), float64(2*time.Second), 40)...,
+	)
+
+	// WALStatLatencyBuckets are histogram buckets for stat operations.
+	WALStatLatencyBuckets = append(
+		prometheus.LinearBuckets(0.0, float64(time.Microsecond*5), 20),
+		prometheus.ExponentialBucketsRange(float64(time.Microsecond*100), float64(200*time.Millisecond), 30)...,
+	)
+
+	// WALOpenDirLatencyBuckets are histogram buckets for directory open operations.
+	WALOpenDirLatencyBuckets = append(
+		prometheus.LinearBuckets(0.0, float64(time.Microsecond*20), 50),
+		prometheus.ExponentialBucketsRange(float64(time.Millisecond*1), float64(3*time.Second), 40)...,
+	)
+
 	// SecondaryCacheIOBuckets exported to enable exporting from package pebble to
 	// enable exporting metrics with below buckets in CRDB.
 	SecondaryCacheIOBuckets = sharedcache.IOBuckets

open.go

@@ -324,9 +324,56 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 		d.mu.mem.queue = append(d.mu.mem.queue, entry)
 	}
 
-	d.mu.log.metrics.fsyncLatency = prometheus.NewHistogram(prometheus.HistogramOpts{
-		Buckets: FsyncLatencyBuckets,
-	})
+	// Create enhanced WAL metrics for comprehensive filesystem operation tracking
+	walFileMetrics := record.WALFileMetrics{
+		CreateLatency: prometheus.NewHistogram(prometheus.HistogramOpts{
+			Buckets: WALCreateLatencyBuckets,
+		}),
+		WriteLatency: prometheus.NewHistogram(prometheus.HistogramOpts{
+			Buckets: WALWriteLatencyBuckets,
+		}),
+		FsyncLatency: prometheus.NewHistogram(prometheus.HistogramOpts{
+			Buckets: FsyncLatencyBuckets, // Reuse existing sync buckets
+		}),
+		CloseLatency: prometheus.NewHistogram(prometheus.HistogramOpts{
+			Buckets: FsyncLatencyBuckets, // Close is similar to sync in timing
+		}),
+		StatLatency: prometheus.NewHistogram(prometheus.HistogramOpts{
+			Buckets: WALStatLatencyBuckets,
+		}),
+		OpenDirLatency: prometheus.NewHistogram(prometheus.HistogramOpts{
+			Buckets: WALOpenDirLatencyBuckets,
+		}),
+	}
+
+	// Initialize enhanced LogWriter metrics structure
+	d.mu.log.metrics = WALMetrics{
+		Primary: struct {
+			CreateLatency  prometheus.Histogram
+			WriteLatency   prometheus.Histogram
+			FsyncLatency   prometheus.Histogram
+			CloseLatency   prometheus.Histogram
+			StatLatency    prometheus.Histogram
+			OpenDirLatency prometheus.Histogram
+		}{
+			CreateLatency:  walFileMetrics.CreateLatency,
+			WriteLatency:   walFileMetrics.WriteLatency,
+			FsyncLatency:   walFileMetrics.FsyncLatency,
+			CloseLatency:   walFileMetrics.CloseLatency,
+			StatLatency:    walFileMetrics.StatLatency,
+			OpenDirLatency: walFileMetrics.OpenDirLatency,
+		},
+		Secondary: struct {
+			CreateLatency  prometheus.Histogram
+			WriteLatency   prometheus.Histogram
+			FsyncLatency   prometheus.Histogram
+			CloseLatency   prometheus.Histogram
+			StatLatency    prometheus.Histogram
+			OpenDirLatency prometheus.Histogram
+		}{
+			// Empty for now, could be populated for failover scenarios
+		},
+	}
 
 	walOpts := wal.Options{
 		Primary:              wal.Dir{FS: opts.FS, Dirname: walDirname},
@@ -337,7 +384,7 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 		BytesPerSync:         opts.WALBytesPerSync,
 		PreallocateSize:      d.walPreallocateSize,
 		MinSyncInterval:      opts.WALMinSyncInterval,
-		FsyncLatency:         d.mu.log.metrics.fsyncLatency,
+		WALFileMetrics:       walFileMetrics,
 		QueueSemChan:         d.commit.logSyncQSem,
 		Logger:               opts.Logger,
 		EventListener:        walEventListenerAdaptor{l: opts.EventListener},

record/log_writer.go

@@ -455,8 +455,12 @@ type LogWriter struct {
 		err error
 		// minSyncInterval is the minimum duration between syncs.
 		minSyncInterval durationFunc
-		fsyncLatency    prometheus.Histogram
-		pending         []*block
+
+		// Enhanced WAL metrics for all filesystem operations
+		walFileMetrics WALFileMetrics
+		// Directory type for metrics recording
+		directoryType DirectoryType
+		pending       []*block
 		// Pushing and popping from pendingSyncs does not require flusher mutex to
 		// be held.
 		pendingSyncs pendingSyncs
@@ -486,10 +490,52 @@ type LogWriter struct {
 	emitFragment func(n int, p []byte) (remainingP []byte)
 }
 
+// DirectoryType identifies which WAL directory an operation targets
+type DirectoryType int
+
+const (
+	// DirectoryPrimary indicates operation targets the primary WAL directory
+	DirectoryPrimary DirectoryType = iota
+	// DirectorySecondary indicates operation targets the secondary WAL directory
+	DirectorySecondary
+	// DirectoryUnknown indicates directory context is unknown or not applicable
+	DirectoryUnknown
+)
+
+// String returns the string representation of DirectoryType.
+func (d DirectoryType) String() string {
+	switch d {
+	case DirectoryPrimary:
+		return "Primary"
+	case DirectorySecondary:
+		return "Secondary"
+	case DirectoryUnknown:
+		return "Unknown"
+	default:
+		return "Invalid"
+	}
+}
+
+// WALFileMetrics contains all latency histograms for individual WAL file operations
+type WALFileMetrics struct {
+	CreateLatency  prometheus.Histogram // File creation operations
+	WriteLatency   prometheus.Histogram // Write operations
+	FsyncLatency   prometheus.Histogram // Fsync operations
+	CloseLatency   prometheus.Histogram // File close operations
+	StatLatency    prometheus.Histogram // File stat operations
+	OpenDirLatency prometheus.Histogram // Directory open operations
+}
+
 // LogWriterConfig is a struct used for configuring new LogWriters
 type LogWriterConfig struct {
 	WALMinSyncInterval durationFunc
-	WALFsyncLatency    prometheus.Histogram
+
+	// Directory type for this LogWriter (Primary, Secondary, or Unknown)
+	DirectoryType DirectoryType
+
+	// Enhanced WAL metrics for this directory
+	WALFileMetrics WALFileMetrics
+
 	// QueueSemChan is an optional channel to pop from when popping from
 	// LogWriter.flusher.syncQueue. It functions as a semaphore that prevents
 	// the syncQueue from overflowing (which will cause a panic). All production
@@ -581,7 +627,8 @@ func NewLogWriter(
 
 	f := &r.flusher
 	f.minSyncInterval = logWriterConfig.WALMinSyncInterval
-	f.fsyncLatency = logWriterConfig.WALFsyncLatency
+	f.walFileMetrics = logWriterConfig.WALFileMetrics
+	f.directoryType = logWriterConfig.DirectoryType
 
 	go func() {
 		pprof.Do(context.Background(), walSyncLabels, r.flushLoop)
@@ -725,9 +772,9 @@ func (w *LogWriter) flushLoop(context.Context) {
 		writtenOffset += uint64(len(data))
 		synced, syncLatency, bytesWritten, err := w.flushPending(data, pending, snap)
 		f.Lock()
-		if synced && f.fsyncLatency != nil {
+		if synced && f.walFileMetrics.FsyncLatency != nil {
 			w.syncedOffset.Store(writtenOffset)
-			f.fsyncLatency.Observe(float64(syncLatency))
+			f.walFileMetrics.FsyncLatency.Observe(float64(syncLatency))
 		}
 		f.err = err
 		if f.err != nil {
@@ -784,7 +831,12 @@ func (w *LogWriter) flushPending(
 	}
 	if n := len(data); err == nil && n > 0 {
 		bytesWritten += int64(n)
+		// Measure write latency
+		writeStart := crtime.NowMono()
 		_, err = w.w.Write(data)
+		if writeLatency := writeStart.Elapsed(); w.flusher.walFileMetrics.WriteLatency != nil {
+			w.flusher.walFileMetrics.WriteLatency.Observe(float64(writeLatency))
+		}
 	}
 
 	synced = !snap.empty()
@@ -811,7 +863,13 @@ func (w *LogWriter) syncWithLatency() (time.Duration, error) {
 }
 
 func (w *LogWriter) flushBlock(b *block) error {
-	if _, err := w.w.Write(b.buf[b.flushed:]); err != nil {
+	// Measure write latency for block flush
+	writeStart := crtime.NowMono()
+	_, err := w.w.Write(b.buf[b.flushed:])
+	if writeLatency := writeStart.Elapsed(); w.flusher.walFileMetrics.WriteLatency != nil {
+		w.flusher.walFileMetrics.WriteLatency.Observe(float64(writeLatency))
+	}
+	if err != nil {
 		return err
 	}
 	b.written.Store(0)
@@ -885,8 +943,8 @@ func (w *LogWriter) closeInternal(lastQueuedRecord PendingSyncIndex) error {
 		syncLatency, err = w.syncWithLatency()
 	}
 	f.Lock()
-	if err == nil && f.fsyncLatency != nil {
-		f.fsyncLatency.Observe(float64(syncLatency))
+	if err == nil && f.walFileMetrics.FsyncLatency != nil {
+		f.walFileMetrics.FsyncLatency.Observe(float64(syncLatency))
 	}
 	free := w.free.blocks
 	f.Unlock()
@@ -897,7 +955,12 @@ func (w *LogWriter) closeInternal(lastQueuedRecord PendingSyncIndex) error {
 		w.pendingSyncsBackingIndex.externalSyncQueueCallback(lastQueuedRecord, err)
 	}
 	if w.c != nil {
+		// Measure close latency
+		closeStart := crtime.NowMono()
 		cerr := w.c.Close()
+		if closeLatency := closeStart.Elapsed(); w.flusher.walFileMetrics.CloseLatency != nil {
+			w.flusher.walFileMetrics.CloseLatency.Observe(float64(closeLatency))
+		}
 		w.c = nil
 		err = firstError(err, cerr)
 	}