diff --git a/monitoring/rules/aggregation.rules.yml b/monitoring/rules/aggregation.rules.yml
index bf2320872521..a01d0c88b27d 100644
--- a/monitoring/rules/aggregation.rules.yml
+++ b/monitoring/rules/aggregation.rules.yml
@@ -95,3 +95,15 @@ groups:
     expr: histogram_quantile(0.95, raft_process_commandcommit_latency_bucket:rate1m)
   - record: raft_process_commandcommit_latency:rate1m:quantile_99
     expr: histogram_quantile(0.99, raft_process_commandcommit_latency_bucket:rate1m)
+  - record: pebble_fsync_latency_bucket:rate1m
+    expr: rate(pebble_fsync_latency_bucket{job="cockroachdb"}[1m])
+  - record: pebble_fsync_latency:rate1m:quantile_50
+    expr: histogram_quantile(0.5, pebble_fsync_latency_bucket:rate1m)
+  - record: pebble_fsync_latency:rate1m:quantile_75
+    expr: histogram_quantile(0.75, pebble_fsync_latency_bucket:rate1m)
+  - record: pebble_fsync_latency:rate1m:quantile_90
+    expr: histogram_quantile(0.9, pebble_fsync_latency_bucket:rate1m)
+  - record: pebble_fsync_latency:rate1m:quantile_95
+    expr: histogram_quantile(0.95, pebble_fsync_latency_bucket:rate1m)
+  - record: pebble_fsync_latency:rate1m:quantile_99
+    expr: histogram_quantile(0.99, pebble_fsync_latency_bucket:rate1m)
diff --git a/pkg/kv/kvserver/metrics.go b/pkg/kv/kvserver/metrics.go
index d436b89883ee..86cf5171f441 100644
--- a/pkg/kv/kvserver/metrics.go
+++ b/pkg/kv/kvserver/metrics.go
@@ -1669,6 +1669,13 @@ Note that the measurement does not include the duration for replicating the eval
 		Measurement: "Flush Utilization",
 		Unit:        metric.Unit_PERCENT,
 	}
+
+	metaPebbleFsyncLatency = metric.Metadata{
+		Name:        "pebble.fsync.latency",
+		Help:        "The pebble write ahead log writer fsync latency",
+		Measurement: "Fsync Latency",
+		Unit:        metric.Unit_NANOSECONDS,
+	}
 )
 
 // StoreMetrics is the set of metrics for a given store.
@@ -1965,6 +1972,7 @@ type StoreMetrics struct {
 	ReplicaWriteBatchEvaluationLatency *metric.Histogram
 
 	FlushUtilization *metric.GaugeFloat64
+	FsyncLatency     *metric.Histogram
 }
 
 type tenantMetricsRef struct {
@@ -2503,6 +2511,7 @@ func newStoreMetrics(histogramWindow time.Duration) *StoreMetrics {
 			metaReplicaWriteBatchEvaluationLatency, histogramWindow, metric.IOLatencyBuckets,
 		),
 		FlushUtilization: metric.NewGaugeFloat64(metaPebbleFlushUtilization),
+		FsyncLatency:     metric.NewHistogram(metaPebbleFsyncLatency, histogramWindow, metric.IOLatencyBuckets),
 	}
 
 	{
diff --git a/pkg/kv/kvserver/store.go b/pkg/kv/kvserver/store.go
index 18f1d4e81879..873bb80b5744 100644
--- a/pkg/kv/kvserver/store.go
+++ b/pkg/kv/kvserver/store.go
@@ -1880,6 +1880,10 @@ func (s *Store) Start(ctx context.Context, stopper *stop.Stopper) error {
 	})
 	s.metrics.registry.AddMetricStruct(s.intentResolver.Metrics)
 
+	s.engine.RegisterMetricEventListener(pebble.MetricEventListener{WALFsyncLatency: func(duration time.Duration) {
+		s.metrics.FsyncLatency.RecordValue(duration.Microseconds())
+	}})
+
 	// Create the raft log truncator and register the callback.
 	s.raftTruncator = makeRaftLogTruncator(s.cfg.AmbientCtx, (*storeForTruncatorImpl)(s), stopper)
 	{
diff --git a/pkg/storage/engine.go b/pkg/storage/engine.go
index b3302e2d57c1..916dd8a11745 100644
--- a/pkg/storage/engine.go
+++ b/pkg/storage/engine.go
@@ -929,6 +929,11 @@ type Engine interface {
 	// of the callback since it could cause a deadlock (since the callback may
 	// be invoked while holding mutexes).
 	RegisterFlushCompletedCallback(cb func())
+	// RegisterMetricEventListener registers the provided listener. The listener
+	// has a collection of callback functions. Call this will cause the listener
+	// to replaces the existing listener struct. Each callback in the
+	// pebble.MetricEventListener is invoked separately.
+	RegisterMetricEventListener(listener pebble.MetricEventListener)
 	// Filesystem functionality.
 	fs.FS
 	// CreateCheckpoint creates a checkpoint of the engine in the given directory,
diff --git a/pkg/storage/pebble.go b/pkg/storage/pebble.go
index 457f08c4b209..79e84b7d1428 100644
--- a/pkg/storage/pebble.go
+++ b/pkg/storage/pebble.go
@@ -682,7 +682,7 @@ type Pebble struct {
 	diskSlowCount        int64
 	diskStallCount       int64
 
-	onMetricCallback pebble.MetricCallbacks
+	onMetricEvent pebble.MetricEventListener
 
 	// Relevant options copied over from pebble.Options.
 	fs            vfs.FS
@@ -933,10 +933,10 @@ func NewPebble(ctx context.Context, cfg PebbleConfig) (p *Pebble, err error) {
 		return nil, err
 	}
 
-	cfg.Opts.OnMetrics = pebble.MetricCallbacks{
-		LogWriterFsyncLatency: func(duration int64) {
-			if p.onMetricCallback.LogWriterFsyncLatency != nil {
-				p.onMetricCallback.LogWriterFsyncLatency(duration)
+	cfg.Opts.MetricEventListener = pebble.MetricEventListener{
+		WALFsyncLatency: func(duration time.Duration) {
+			if p.onMetricEvent.WALFsyncLatency != nil {
+				p.onMetricEvent.WALFsyncLatency(duration)
 			}
 		},
 	}
@@ -1703,9 +1703,9 @@ func (p *Pebble) RegisterFlushCompletedCallback(cb func()) {
 	p.mu.Unlock()
 }
 
-// RegisterMetricCallbacks implements the Engine interface
-func (p *Pebble) RegisterMetricCallbacks(callbacks pebble.MetricCallbacks) {
-	p.onMetricCallback = callbacks
+// RegisterMetricEventListener implements the Engine interface
+func (p *Pebble) RegisterMetricEventListener(listener pebble.MetricEventListener) {
+	p.onMetricEvent = listener
 }
 
 // Remove implements the FS interface.