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,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

@@ -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
+	// Embedded runtime metrics from record.LogWriterMetrics
+	record.LogWriterMetrics
+}
+
 var (
 	// FsyncLatencyBuckets are prometheus histogram buckets suitable for a histogram
 	// that records latencies for fsyncs.
@@ -552,6 +571,12 @@ var (
 		prometheus.ExponentialBucketsRange(float64(time.Millisecond*5), float64(10*time.Second), 50)...,
 	)
 
+	// WALFileOpLatencyBuckets are histogram buckets for all WAL file operations.
+	WALFileOpLatencyBuckets = append(
+		prometheus.LinearBuckets(0.0, 0.01, 50),                 // 0 to 0.5ms in 10μs increments
+		prometheus.ExponentialBucketsRange(1.0, 10000.0, 50)..., // 1ms to 10s
+	)
+
 	// SecondaryCacheIOBuckets exported to enable exporting from package pebble to
 	// enable exporting metrics with below buckets in CRDB.
 	SecondaryCacheIOBuckets = sharedcache.IOBuckets

open.go

@@ -289,10 +289,19 @@ 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 WAL file operation histogram for filesystem operation tracking
+	walFileOpHistogram := prometheus.NewHistogram(prometheus.HistogramOpts{
+		Name:    "pebble_wal_file_op_duration_nanoseconds",
+		Help:    "Histogram of WAL file operation latencies in nanoseconds",
+		Buckets: WALFileOpLatencyBuckets,
 	})
 
+	// Initialize WAL metrics structure
+	d.mu.log.metrics = WALMetrics{
+		PrimaryFileOpLatency: walFileOpHistogram,
+		// SecondaryFileOpLatency could be populated for failover scenarios
+	}
+
 	walOpts := wal.Options{
 		Primary:              dirs.WALPrimary,
 		Secondary:            dirs.WALSecondary,
@@ -302,7 +311,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,
+		WALFileOpHistogram:   walFileOpHistogram,
 		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
+
+		// WAL file operation latency histogram
+		walFileOpHistogram WALFileOpHistogram
+		// 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,46 @@ 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"
+	}
+}
+
+// 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
+
+	// Directory type for this LogWriter (Primary, Secondary, or Unknown)
+	DirectoryType DirectoryType
+
+	// 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 +621,8 @@ func NewLogWriter(
 
 	f := &r.flusher
 	f.minSyncInterval = logWriterConfig.WALMinSyncInterval
-	f.fsyncLatency = logWriterConfig.WALFsyncLatency
+	f.walFileOpHistogram = logWriterConfig.WALFileOpHistogram
+	f.directoryType = logWriterConfig.DirectoryType
 
 	go func() {
 		pprof.Do(context.Background(), walSyncLabels, r.flushLoop)
@@ -725,9 +766,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 +825,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 +857,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 +937,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 +949,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 },
 		})