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

@@ -33,14 +33,12 @@ 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/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 {
@@ -1972,8 +1964,9 @@ 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 {
+	metrics.WALMetrics.PrimaryFileOpLatency = d.mu.log.metrics.PrimaryFileOpLatency
+	metrics.WALMetrics.SecondaryFileOpLatency = d.mu.log.metrics.SecondaryFileOpLatency
+	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

@@ -486,10 +486,7 @@ type Metrics struct {
 	// deletion. These can be relevant if free disk space is unexplainably low.
 	DeletePacer deletepacer.Metrics
 
-	LogWriter struct {
-		FsyncLatency prometheus.Histogram
-		record.LogWriterMetrics
-	}
+	WALMetrics WALMetrics
 
 	CategoryStats []block.CategoryStatsAggregate
 
@@ -544,6 +541,28 @@ 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 {
+	// PrimaryFileOpLatency tracks all file operations for the primary directory
+	PrimaryFileOpLatency prometheus.Histogram
+	// SecondaryFileOpLatency tracks all file operations for the secondary directory
+	SecondaryFileOpLatency prometheus.Histogram
+	// Updated whenever a wal.Writer is closed
+	record.LogWriterMetrics
+}
+
 var (
 	// FsyncLatencyBuckets are prometheus histogram buckets suitable for a histogram
 	// that records latencies for fsyncs.

open.go

@@ -289,9 +289,8 @@ 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,
-	})
+	// Initialize WAL metrics structure (histograms populated below)
+	d.mu.log.metrics = WALMetrics{}
 
 	walOpts := wal.Options{
 		Primary:              dirs.WALPrimary,
@@ -302,17 +301,33 @@ 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,
 		QueueSemChan:         d.commit.logSyncQSem,
 		Logger:               opts.Logger,
 		EventListener:        walEventListenerAdaptor{l: opts.EventListener},
 		WriteWALSyncOffsets:  func() bool { return d.FormatMajorVersion() >= FormatWALSyncChunks },
 	}
+
+	// Create and assign WAL file operation histograms
+	walPrimaryFileOpHistogram := prometheus.NewHistogram(prometheus.HistogramOpts{
+		Buckets: FsyncLatencyBuckets,
+	})
+	d.mu.log.metrics.PrimaryFileOpLatency = walPrimaryFileOpHistogram
+	walOpts.PrimaryFileOpHistogram = walPrimaryFileOpHistogram
+
+	// Configure failover-specific histograms and options
 	if !opts.ReadOnly && opts.WALFailover != nil {
-		walOpts.FailoverOptions = opts.WALFailover.FailoverOptions
-		walOpts.FailoverWriteAndSyncLatency = prometheus.NewHistogram(prometheus.HistogramOpts{
+		walSecondaryFileOpHistogram := prometheus.NewHistogram(prometheus.HistogramOpts{
+			Buckets: FsyncLatencyBuckets,
+		})
+		d.mu.log.metrics.SecondaryFileOpLatency = walSecondaryFileOpHistogram
+		walOpts.SecondaryFileOpHistogram = walSecondaryFileOpHistogram
+
+		walFailoverWriteAndSyncHistogram := prometheus.NewHistogram(prometheus.HistogramOpts{
 			Buckets: FsyncLatencyBuckets,
 		})
+		walOpts.FailoverWriteAndSyncLatency = walFailoverWriteAndSyncHistogram
+
+		walOpts.FailoverOptions = opts.WALFailover.FailoverOptions
 	}
 	walDirs := walOpts.Dirs()
 	var recoveryDirLocks base.DirLockSet

record/log_writer.go

@@ -455,8 +455,10 @@ type LogWriter struct {
 		err error
 		// minSyncInterval is the minimum duration between syncs.
 		minSyncInterval durationFunc
-		fsyncLatency    prometheus.Histogram
-		pending         []*block
+
+		// WAL file operation latency histogram
+		walFileOpHistogram WALFileOpHistogram
+		pending            []*block
 		// Pushing and popping from pendingSyncs does not require flusher mutex to
 		// be held.
 		pendingSyncs pendingSyncs
@@ -486,10 +488,17 @@ type LogWriter struct {
 	emitFragment func(n int, p []byte) (remainingP []byte)
 }
 
+// WALFileOpHistogram is a prometheus histogram for tracking WAL file operation latencies
+// (create, write, fsync, close, stat, opendir).
+type WALFileOpHistogram = prometheus.Histogram
+
 // LogWriterConfig is a struct used for configuring new LogWriters
 type LogWriterConfig struct {
 	WALMinSyncInterval durationFunc
-	WALFsyncLatency    prometheus.Histogram
+
+	// WAL file operation latency histogram for this directory
+	WALFileOpHistogram WALFileOpHistogram
+
 	// 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 +590,7 @@ func NewLogWriter(
 
 	f := &r.flusher
 	f.minSyncInterval = logWriterConfig.WALMinSyncInterval
-	f.fsyncLatency = logWriterConfig.WALFsyncLatency
+	f.walFileOpHistogram = logWriterConfig.WALFileOpHistogram
 
 	go func() {
 		pprof.Do(context.Background(), walSyncLabels, r.flushLoop)
@@ -725,9 +734,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.walFileOpHistogram != nil {
 			w.syncedOffset.Store(writtenOffset)
-			f.fsyncLatency.Observe(float64(syncLatency))
+			f.walFileOpHistogram.Observe(float64(syncLatency))
 		}
 		f.err = err
 		if f.err != nil {
@@ -784,7 +793,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.walFileOpHistogram != nil {
+			w.flusher.walFileOpHistogram.Observe(float64(writeLatency))
+		}
 	}
 
 	synced = !snap.empty()
@@ -811,7 +825,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.walFileOpHistogram != nil {
+		w.flusher.walFileOpHistogram.Observe(float64(writeLatency))
+	}
+	if err != nil {
 		return err
 	}
 	b.written.Store(0)
@@ -885,8 +905,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.walFileOpHistogram != nil {
+		f.walFileOpHistogram.Observe(float64(syncLatency))
 	}
 	free := w.free.blocks
 	f.Unlock()
@@ -897,7 +917,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.walFileOpHistogram != nil {
+			w.flusher.walFileOpHistogram.Observe(float64(closeLatency))
+		}
 		w.c = nil
 		err = firstError(err, cerr)
 	}

record/log_writer_test.go

@@ -150,7 +150,6 @@ func TestSyncError(t *testing.T) {
 
 	injectedErr := errors.New("injected error")
 	w := NewLogWriter(syncErrorFile{f, injectedErr}, 0, LogWriterConfig{
-		WALFsyncLatency:     prometheus.NewHistogram(prometheus.HistogramOpts{}),
 		WriteWALSyncOffsets: func() bool { return false },
 	})
 
@@ -193,7 +192,6 @@ func (f *syncFile) Sync() error {
 func TestSyncRecord(t *testing.T) {
 	f := &syncFile{}
 	w := NewLogWriter(f, 0, LogWriterConfig{
-		WALFsyncLatency:     prometheus.NewHistogram(prometheus.HistogramOpts{}),
 		WriteWALSyncOffsets: func() bool { return false },
 	})
 
@@ -221,7 +219,6 @@ func TestSyncRecordWithSignalChan(t *testing.T) {
 		semChan <- struct{}{}
 	}
 	w := NewLogWriter(f, 0, LogWriterConfig{
-		WALFsyncLatency:     prometheus.NewHistogram(prometheus.HistogramOpts{}),
 		QueueSemChan:        semChan,
 		WriteWALSyncOffsets: func() bool { return false },
 	})
@@ -273,7 +270,6 @@ func TestMinSyncInterval(t *testing.T) {
 		WALMinSyncInterval: func() time.Duration {
 			return minSyncInterval
 		},
-		WALFsyncLatency:     prometheus.NewHistogram(prometheus.HistogramOpts{}),
 		WriteWALSyncOffsets: func() bool { return false },
 	})
 
@@ -346,7 +342,6 @@ func TestMinSyncIntervalClose(t *testing.T) {
 		WALMinSyncInterval: func() time.Duration {
 			return minSyncInterval
 		},
-		WALFsyncLatency:     prometheus.NewHistogram(prometheus.HistogramOpts{}),
 		WriteWALSyncOffsets: func() bool { return false },
 	})
 
@@ -398,7 +393,6 @@ func TestMetricsWithoutSync(t *testing.T) {
 	f := &syncFileWithWait{}
 	f.writeWG.Add(1)
 	w := NewLogWriter(f, 0, LogWriterConfig{
-		WALFsyncLatency:     prometheus.NewHistogram(prometheus.HistogramOpts{}),
 		WriteWALSyncOffsets: func() bool { return false },
 	})
 	offset, err := w.SyncRecord([]byte("hello"), nil, nil)
@@ -444,10 +438,9 @@ func TestMetricsWithSync(t *testing.T) {
 	})
 
 	w := NewLogWriter(f, 0, LogWriterConfig{
-		WALFsyncLatency:     syncLatencyMicros,
 		WriteWALSyncOffsets: func() bool { return false },
-	},
-	)
+		WALFileOpHistogram:  syncLatencyMicros,
+	})
 	var wg sync.WaitGroup
 	wg.Add(100)
 	for i := 0; i < 100; i++ {
@@ -551,7 +544,6 @@ func TestQueueWALBlocks(t *testing.T) {
 		return nil
 	}))
 	w := NewLogWriter(f, 0, LogWriterConfig{
-		WALFsyncLatency:     prometheus.NewHistogram(prometheus.HistogramOpts{}),
 		WriteWALSyncOffsets: func() bool { return false },
 	})
 	const numBlocks = 1024
@@ -662,7 +654,6 @@ func TestSyncRecordGeneralized(t *testing.T) {
 	lastSync := int64(-1)
 	cbChan := make(chan struct{}, 2)
 	w := NewLogWriter(f, 0, LogWriterConfig{
-		WALFsyncLatency: prometheus.NewHistogram(prometheus.HistogramOpts{}),
 		ExternalSyncQueueCallback: func(doneSync PendingSyncIndex, err error) {
 			require.NoError(t, err)
 			require.Equal(t, lastSync+1, doneSync.Index)
@@ -714,7 +705,6 @@ func TestSyncRecordGeneralizedWithCloseError(t *testing.T) {
 	lastSync := int64(-1)
 	cbChan := make(chan struct{}, 2)
 	w := NewLogWriter(f, 0, LogWriterConfig{
-		WALFsyncLatency: prometheus.NewHistogram(prometheus.HistogramOpts{}),
 		ExternalSyncQueueCallback: func(doneSync PendingSyncIndex, err error) {
 			if doneSync.Index == 1 {
 				require.Error(t, err)
@@ -755,7 +745,6 @@ func TestSyncRecordGeneralizedWithCloseError(t *testing.T) {
 func writeWALSyncRecords(t *testing.T, numRecords int, recordSizes []int) *syncFile {
 	f := &syncFile{}
 	w := NewLogWriter(f, 1, LogWriterConfig{
-		WALFsyncLatency:     prometheus.NewHistogram(prometheus.HistogramOpts{}),
 		WriteWALSyncOffsets: func() bool { return true },
 	})
 	var syncErr error
@@ -864,7 +853,6 @@ func BenchmarkQueueWALBlocks(b *testing.B) {
 					return nil
 				}))
 				w := NewLogWriter(f, 0, LogWriterConfig{
-					WALFsyncLatency:     prometheus.NewHistogram(prometheus.HistogramOpts{}),
 					WriteWALSyncOffsets: func() bool { return false },
 				})
 
@@ -897,7 +885,6 @@ func BenchmarkWriteWALBlocksAllocs(b *testing.B) {
 		b.StopTimer()
 		f := vfstest.DiscardFile
 		w := NewLogWriter(f, 0, LogWriterConfig{
-			WALFsyncLatency:           prometheus.NewHistogram(prometheus.HistogramOpts{}),
 			ExternalSyncQueueCallback: func(doneSync PendingSyncIndex, err error) {},
 			WriteWALSyncOffsets:       func() bool { return false },
 		})