From 347c019c23a4444b6039ce0a6313a0a9e5c0a313 Mon Sep 17 00:00:00 2001 From: Leon Fattakhov Date: Mon, 10 Oct 2022 16:05:18 -0400 Subject: [PATCH] metrics: expose pebble fsync latency as prometheus metric MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Leverage the metric callback exposed in pebble to update the store’s histogram. However since the callbacks must be setup in `pebble.Open` but the metric to update is part of the `kvserver/metric.go`. We store the callbacks on the `pebble.onMetricEvent pebble.MetricEventListener` and craft a struct that wraps the callbacks with a `nil` check to ensure that they have been set inside of `kvserver/store.go` store start method:`Store.Start()`. This allows for deferring the setting of the callback handlers until we have access to the metrics to update. Diagram of the above description: ``` type Pebble struct { onMetricEvent struct { +---> SomeCallback func(duration time.Duration) <--------------+ | } | | } | | | Checks | func NewPebble(...) { | | ... | | cfg.Opts.MetricEventListener = pebble.MetricEventListener{ | | SomeCallback: func(duration time.Duration) { | S | if p.onMetricEvent.SomeCallback != nil { +-------------^ e | p.onMetricEvent.SomeCallback(duration) +-------------v t | } | s | }, | | } | | | | db, err := pebble.Open(cfg.StorageConfig.Dir, cfg.Opts) | | ... | Calls | } | | | | func (s *Store) Start(...) { | | ... | ^--- s.engine.RegisterMetricEventListener( | pebble.MetricEventListener{ | SomeCallback: func(duration time.Duration) { | s.metrics.FsyncLatency.RecordValue(duration)<--------v } }) ... } ``` Release note: None --- monitoring/rules/aggregation.rules.yml | 12 ++++++++++++ pkg/kv/kvserver/metrics.go | 9 +++++++++ pkg/kv/kvserver/store.go | 4 ++++ pkg/storage/engine.go | 5 +++++ pkg/storage/pebble.go | 16 ++++++++-------- 5 files changed, 38 insertions(+), 8 deletions(-) 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.