From 3cc5f2e059d5890112bbe8ac38fa604765cb841c Mon Sep 17 00:00:00 2001 From: alrevuelta Date: Thu, 27 Aug 2026 12:50:05 +0200 Subject: [PATCH 1/2] Fix pebble db metrics --- sei-db/db_engine/pebbledb/mvcc/db.go | 14 +++--- sei-db/db_engine/pebbledb/mvcc/metrics.go | 54 +++++++++++------------ 2 files changed, 34 insertions(+), 34 deletions(-) diff --git a/sei-db/db_engine/pebbledb/mvcc/db.go b/sei-db/db_engine/pebbledb/mvcc/db.go index 421fe4dcf8..039c591c78 100644 --- a/sei-db/db_engine/pebbledb/mvcc/db.go +++ b/sei-db/db_engine/pebbledb/mvcc/db.go @@ -1406,13 +1406,13 @@ func (db *Database) collectAndRecordMetrics(ctx context.Context) { m := db.storage.Metrics() // Compaction metrics - report raw counts - otelMetrics.compactionCount.Add(ctx, m.Compact.Count) + otelMetrics.compactionCount.Record(ctx, m.Compact.Count) otelMetrics.compactionDuration.Record(ctx, m.Compact.Duration.Seconds()) // Flush metrics - report raw counts - otelMetrics.flushCount.Add(ctx, m.Flush.Count) + otelMetrics.flushCount.Record(ctx, m.Flush.Count) otelMetrics.flushDuration.Record(ctx, m.Flush.WriteThroughput.WorkDuration.Seconds()) - otelMetrics.flushBytesWritten.Add(ctx, m.Flush.WriteThroughput.Bytes) + otelMetrics.flushBytesWritten.Record(ctx, m.Flush.WriteThroughput.Bytes) // Storage metrics per level with level as attribute for level := 0; level < len(m.Levels); level++ { @@ -1421,8 +1421,8 @@ func (db *Database) collectAndRecordMetrics(ctx context.Context) { otelMetrics.sstableCount.Record(ctx, levelMetrics.TablesCount, metric.WithAttributes(levelAttr)) otelMetrics.sstableTotalSize.Record(ctx, levelMetrics.TablesSize, metric.WithAttributes(levelAttr)) - otelMetrics.compactionBytesRead.Add(ctx, int64(levelMetrics.TableBytesIn), metric.WithAttributes(levelAttr)) //nolint:gosec - otelMetrics.compactionBytesWritten.Add(ctx, int64(levelMetrics.TableBytesCompacted), metric.WithAttributes(levelAttr)) //nolint:gosec + otelMetrics.compactionBytesRead.Record(ctx, int64(levelMetrics.TableBytesIn), metric.WithAttributes(levelAttr)) + otelMetrics.compactionBytesWritten.Record(ctx, int64(levelMetrics.TableBytesCompacted), metric.WithAttributes(levelAttr)) } // Memtable metrics @@ -1433,7 +1433,7 @@ func (db *Database) collectAndRecordMetrics(ctx context.Context) { otelMetrics.walSize.Record(ctx, int64(m.WAL.Size)) //nolint:gosec // Cache metrics - report raw counts - otelMetrics.cacheHits.Add(ctx, m.BlockCache.Hits) - otelMetrics.cacheMisses.Add(ctx, m.BlockCache.Misses) + otelMetrics.cacheHits.Record(ctx, m.BlockCache.Hits) + otelMetrics.cacheMisses.Record(ctx, m.BlockCache.Misses) otelMetrics.cacheSize.Record(ctx, m.BlockCache.Size) } diff --git a/sei-db/db_engine/pebbledb/mvcc/metrics.go b/sei-db/db_engine/pebbledb/mvcc/metrics.go index 7a7a6dddad..334d6f14eb 100644 --- a/sei-db/db_engine/pebbledb/mvcc/metrics.go +++ b/sei-db/db_engine/pebbledb/mvcc/metrics.go @@ -17,13 +17,13 @@ var ( importLatency metric.Float64Histogram batchWriteLatency metric.Float64Histogram - compactionCount metric.Int64Counter - compactionDuration metric.Float64Histogram - compactionBytesRead metric.Int64Counter - compactionBytesWritten metric.Int64Counter - flushCount metric.Int64Counter - flushDuration metric.Float64Histogram - flushBytesWritten metric.Int64Counter + compactionCount metric.Int64Gauge + compactionDuration metric.Float64Gauge + compactionBytesRead metric.Int64Gauge + compactionBytesWritten metric.Int64Gauge + flushCount metric.Int64Gauge + flushDuration metric.Float64Gauge + flushBytesWritten metric.Int64Gauge sstableCount metric.Int64Gauge sstableTotalSize metric.Int64Gauge @@ -31,8 +31,8 @@ var ( memtableTotalSize metric.Int64Gauge walSize metric.Int64Gauge - cacheHits metric.Int64Counter - cacheMisses metric.Int64Counter + cacheHits metric.Int64Gauge + cacheMisses metric.Int64Gauge cacheSize metric.Int64Gauge batchSize metric.Int64Histogram @@ -79,40 +79,40 @@ var ( metric.WithUnit("s"), )), - compactionCount: must(meter.Int64Counter( + compactionCount: must(meter.Int64Gauge( "pebble_compaction_count", - metric.WithDescription("Total number of compactions"), + metric.WithDescription("Total number of compactions since the DB opened"), metric.WithUnit("{count}"), )), - compactionDuration: must(meter.Float64Histogram( + compactionDuration: must(meter.Float64Gauge( "pebble_compaction_duration", - metric.WithDescription("Duration of compaction operations"), + metric.WithDescription("Cumulative time spent compacting since the DB opened"), metric.WithUnit("s"), )), - compactionBytesRead: must(meter.Int64Counter( + compactionBytesRead: must(meter.Int64Gauge( "pebble_compaction_bytes_read", - metric.WithDescription("Total bytes read during compaction"), + metric.WithDescription("Total bytes read during compaction since the DB opened"), metric.WithUnit("By"), )), - compactionBytesWritten: must(meter.Int64Counter( + compactionBytesWritten: must(meter.Int64Gauge( "pebble_compaction_bytes_written", - metric.WithDescription("Total bytes written during compaction"), + metric.WithDescription("Total bytes written during compaction since the DB opened"), metric.WithUnit("By"), )), - flushCount: must(meter.Int64Counter( + flushCount: must(meter.Int64Gauge( "pebble_flush_count", - metric.WithDescription("Total number of memtable flushes"), + metric.WithDescription("Total number of memtable flushes since the DB opened"), metric.WithUnit("{count}"), )), - flushDuration: must(meter.Float64Histogram( + flushDuration: must(meter.Float64Gauge( "pebble_flush_duration", - metric.WithDescription("Duration of memtable flush operations"), + metric.WithDescription("Cumulative time spent flushing memtables since the DB opened"), metric.WithUnit("s"), )), - flushBytesWritten: must(meter.Int64Counter( + flushBytesWritten: must(meter.Int64Gauge( "pebble_flush_bytes_written", - metric.WithDescription("Total bytes written during memtable flushes"), + metric.WithDescription("Total bytes written during memtable flushes since the DB opened"), metric.WithUnit("By"), )), @@ -142,14 +142,14 @@ var ( metric.WithUnit("By"), )), - cacheHits: must(meter.Int64Counter( + cacheHits: must(meter.Int64Gauge( "pebble_cache_hits", - metric.WithDescription("Total number of cache hits"), + metric.WithDescription("Total number of cache hits since the DB opened"), metric.WithUnit("{count}"), )), - cacheMisses: must(meter.Int64Counter( + cacheMisses: must(meter.Int64Gauge( "pebble_cache_misses", - metric.WithDescription("Total number of cache misses"), + metric.WithDescription("Total number of cache misses since the DB opened"), metric.WithUnit("{count}"), )), cacheSize: must(meter.Int64Gauge( From 2b8b377ee750b3adb954525f87be02cc1472a3d6 Mon Sep 17 00:00:00 2001 From: alrevuelta Date: Fri, 28 Aug 2026 10:21:36 +0200 Subject: [PATCH 2/2] Remove duplicated metrics --- sei-db/db_engine/pebbledb/mvcc/batch.go | 21 ++-- sei-db/db_engine/pebbledb/mvcc/db.go | 118 ++++++------------ .../db_engine/pebbledb/mvcc/db_ascending.go | 6 +- sei-db/db_engine/pebbledb/mvcc/iterator.go | 6 +- .../pebbledb/mvcc/iterator_ascending.go | 6 +- sei-db/db_engine/pebbledb/mvcc/metrics.go | 102 +-------------- 6 files changed, 72 insertions(+), 187 deletions(-) diff --git a/sei-db/db_engine/pebbledb/mvcc/batch.go b/sei-db/db_engine/pebbledb/mvcc/batch.go index 7071930e89..9842825d7b 100644 --- a/sei-db/db_engine/pebbledb/mvcc/batch.go +++ b/sei-db/db_engine/pebbledb/mvcc/batch.go @@ -20,6 +20,7 @@ type Batch struct { ops []batchOp descending bool operationMetrics *pebbledbmetrics.OperationMetrics + dbName string } type batchOp struct { @@ -29,7 +30,7 @@ type batchOp struct { } // NewBatch creates a new Batch using the supplied MVCC encoding mode. -func NewBatch(storage *pebble.DB, version int64, descending bool, operationMetrics ...*pebbledbmetrics.OperationMetrics) (*Batch, error) { +func NewBatch(storage *pebble.DB, version int64, descending bool, dbName string, operationMetrics ...*pebbledbmetrics.OperationMetrics) (*Batch, error) { if version < 0 { return nil, fmt.Errorf("version must be non-negative") } @@ -45,6 +46,7 @@ func NewBatch(storage *pebble.DB, version int64, descending bool, operationMetri ops: make([]batchOp, 0, 16), descending: descending, operationMetrics: metrics, + dbName: dbName, }, nil } @@ -77,7 +79,7 @@ func (b *Batch) Delete(storeKey string, key []byte) error { func (b *Batch) Write() error { writeCount := int64(len(b.ops) + 1) // includes latest-version metadata. - err := writeBatchOps(b.storage, b.ops, func(batch *pebble.Batch) error { + err := writeBatchOps(b.storage, b.ops, b.dbName, func(batch *pebble.Batch) error { var versionBz [VersionSize]byte binary.LittleEndian.PutUint64(versionBz[:], uint64(b.version)) //nolint:gosec // block heights are non-negative and fit in int64 if err := batch.Set([]byte(latestVersionKey), versionBz[:], nil); err != nil { @@ -97,10 +99,11 @@ type RawBatch struct { ops []batchOp descending bool operationMetrics *pebbledbmetrics.OperationMetrics + dbName string } // NewRawBatch creates a new RawBatch using the supplied MVCC encoding mode. -func NewRawBatch(storage *pebble.DB, descending bool, operationMetrics ...*pebbledbmetrics.OperationMetrics) (*RawBatch, error) { +func NewRawBatch(storage *pebble.DB, descending bool, dbName string, operationMetrics ...*pebbledbmetrics.OperationMetrics) (*RawBatch, error) { var metrics *pebbledbmetrics.OperationMetrics if len(operationMetrics) > 0 { metrics = operationMetrics[0] @@ -111,6 +114,7 @@ func NewRawBatch(storage *pebble.DB, descending bool, operationMetrics ...*pebbl ops: make([]batchOp, 0, 16), descending: descending, operationMetrics: metrics, + dbName: dbName, }, nil } @@ -154,7 +158,7 @@ func (b *Batch) HardDelete(storeKey string, key []byte) error { func (b *RawBatch) Write() error { writeCount := int64(len(b.ops)) - err := writeBatchOps(b.storage, b.ops, nil) + err := writeBatchOps(b.storage, b.ops, b.dbName, nil) if err == nil && b.operationMetrics != nil { b.operationMetrics.AddWrite(writeCount) } @@ -165,7 +169,7 @@ func (b *RawBatch) Write() error { // otel metrics, and commits. The optional beforeCommit hook runs on the // pebble batch right before commit (used by Batch.Write to stamp the // latest-version metadata key). -func writeBatchOps(storage *pebble.DB, ops []batchOp, beforeCommit func(*pebble.Batch) error) (err error) { +func writeBatchOps(storage *pebble.DB, ops []batchOp, dbName string, beforeCommit func(*pebble.Batch) error) (err error) { startTime := time.Now() batchSize := int64(len(ops)) defer func() { @@ -173,9 +177,12 @@ func writeBatchOps(storage *pebble.DB, ops []batchOp, beforeCommit func(*pebble. otelMetrics.batchWriteLatency.Record( ctx, time.Since(startTime).Seconds(), - metric.WithAttributes(attribute.Bool("success", err == nil)), + metric.WithAttributes( + attribute.Bool("success", err == nil), + attribute.String("db", dbName), + ), ) - otelMetrics.batchSize.Record(ctx, batchSize) + otelMetrics.batchSize.Record(ctx, batchSize, metric.WithAttributes(attribute.String("db", dbName))) }() batch := storage.NewBatch() diff --git a/sei-db/db_engine/pebbledb/mvcc/db.go b/sei-db/db_engine/pebbledb/mvcc/db.go index 039c591c78..ad5fdb247a 100644 --- a/sei-db/db_engine/pebbledb/mvcc/db.go +++ b/sei-db/db_engine/pebbledb/mvcc/db.go @@ -106,6 +106,11 @@ type Database struct { // Cancel function for background metrics collection metricsCancel context.CancelFunc + // dbName identifies this instance as the "db" attribute on every otel + // metric it records, so multiple Database instances in one process (e.g. + // SeparateEVMSubDBs) don't share unattributed series. + dbName string + operationMetrics *pebbledbmetrics.OperationMetrics } @@ -207,18 +212,17 @@ func OpenDB(dataDir string, config config.StateStoreConfig) (types.StateStore, e return nil, fmt.Errorf("failed to retrieve latest version: %w", err) } + dbName := filepath.Base(dataDir) database := &Database{ - storage: db, - asyncWriteWG: sync.WaitGroup{}, - config: config, - earliestVersion: atomic.Int64{}, - latestVersion: atomic.Int64{}, - descending: descending, - pendingChanges: make(chan VersionedChangesets, config.AsyncWriteBuffer), - operationMetrics: pebbledbmetrics.NewOperationMetrics( - config.EnableReadWriteMetrics, - filepath.Base(dataDir), - ), + storage: db, + asyncWriteWG: sync.WaitGroup{}, + config: config, + earliestVersion: atomic.Int64{}, + latestVersion: atomic.Int64{}, + descending: descending, + pendingChanges: make(chan VersionedChangesets, config.AsyncWriteBuffer), + dbName: dbName, + operationMetrics: pebbledbmetrics.NewOperationMetrics(config.EnableReadWriteMetrics, dbName), } database.latestVersion.Store(latestVersion) database.earliestVersion.Store(earliestVersion) @@ -247,10 +251,11 @@ func OpenDB(dataDir string, config config.StateStoreConfig) (types.StateStore, e database.asyncWriteWG.Add(1) go database.writeAsyncInBackground() - // Start background metrics collection + // Start background metrics collection for Pebble-internal stats + // (compaction, flush, sstable, memtable, WAL, cache). metricsCtx, metricsCancel := context.WithCancel(context.Background()) database.metricsCancel = metricsCancel - go database.collectMetricsInBackground(metricsCtx) + pebbledbmetrics.NewPebbleMetrics(metricsCtx, db, dbName, 10*time.Second) return database, nil } @@ -594,7 +599,8 @@ func (db *Database) recordPruneOutcome(err error) { } else { failures = db.pruneFailures.Add(1) } - otelMetrics.pruneConsecutiveFailures.Record(context.Background(), failures) + otelMetrics.pruneConsecutiveFailures.Record(context.Background(), failures, + metric.WithAttributes(attribute.String("db", db.dbName))) } // Retrieves earliest version from db, if not found, return 0 @@ -651,7 +657,10 @@ func (db *Database) ApplyChangesetSync(version int64, changeset []*proto.NamedCh otelMetrics.applyChangesetLatency.Record( context.Background(), time.Since(startTime).Seconds(), - metric.WithAttributes(attribute.Bool("success", _err == nil)), + metric.WithAttributes( + attribute.Bool("success", _err == nil), + attribute.String("db", db.dbName), + ), ) }() // Check if version is 0 and change it to 1 @@ -662,7 +671,7 @@ func (db *Database) ApplyChangesetSync(version int64, changeset []*proto.NamedCh } // Create batch and persist latest version in the batch - b, err := NewBatch(db.storage, version, db.descending, db.operationMetrics) + b, err := NewBatch(db.storage, version, db.descending, db.dbName, db.operationMetrics) if err != nil { return err } @@ -697,12 +706,16 @@ func (db *Database) ApplyChangesetAsync(version int64, changesets []*proto.Named otelMetrics.applyChangesetAsyncLatency.Record( context.Background(), time.Since(startTime).Seconds(), - metric.WithAttributes(attribute.Bool("success", _err == nil)), + metric.WithAttributes( + attribute.Bool("success", _err == nil), + attribute.String("db", db.dbName), + ), ) // Record pending queue depth otelMetrics.pendingChangesQueueDepth.Record( context.Background(), int64(len(db.pendingChanges)), + metric.WithAttributes(attribute.String("db", db.dbName)), ) }() // Write to WAL @@ -846,6 +859,7 @@ func (db *Database) getDescending(storeKey string, targetVersion int64, key []by metric.WithAttributes( attribute.Bool("success", _err == nil), attribute.String("store", storeKey), + attribute.String("db", db.dbName), ), ) }() @@ -892,6 +906,7 @@ func (db *Database) pruneDescending(version int64) (_err error) { time.Since(startTime).Seconds(), metric.WithAttributes( attribute.Bool("success", _err == nil), + attribute.String("db", db.dbName), ), ) }() @@ -1045,7 +1060,7 @@ func (db *Database) iteratorDescending(ctx context.Context, storeKey string, ver return nil, fmt.Errorf("failed to create PebbleDB iterator: %w", err) } - return finishMVCCIterator(newPebbleDBIterator(ctx, itr, storePrefix(storeKey), start, end, version, db.GetEarliestVersion(), false, db.config.UseDefaultComparer, storeKey, db.operationMetrics)) + return finishMVCCIterator(newPebbleDBIterator(ctx, itr, storePrefix(storeKey), start, end, version, db.GetEarliestVersion(), false, db.config.UseDefaultComparer, storeKey, db.operationMetrics, db.dbName)) } func (db *Database) reverseIteratorDescending(ctx context.Context, storeKey string, version int64, start, end []byte) (dbm.Iterator, error) { @@ -1071,7 +1086,7 @@ func (db *Database) reverseIteratorDescending(ctx context.Context, storeKey stri return nil, fmt.Errorf("failed to create PebbleDB iterator: %w", err) } - return finishMVCCIterator(newPebbleDBIterator(ctx, itr, storePrefix(storeKey), start, end, version, db.GetEarliestVersion(), true, db.config.UseDefaultComparer, storeKey, db.operationMetrics)) + return finishMVCCIterator(newPebbleDBIterator(ctx, itr, storePrefix(storeKey), start, end, version, db.GetEarliestVersion(), true, db.config.UseDefaultComparer, storeKey, db.operationMetrics, db.dbName)) } func getMVCCSliceDescending(db *pebble.DB, storeKey string, key []byte, version int64) (_ []byte, err error) { @@ -1172,6 +1187,7 @@ func (db *Database) Import(version int64, ch <-chan types.SnapshotNode) (_err er time.Since(startTime).Seconds(), metric.WithAttributes( attribute.Bool("success", _err == nil), + attribute.String("db", db.dbName), ), ) }() @@ -1180,7 +1196,7 @@ func (db *Database) Import(version int64, ch <-chan types.SnapshotNode) (_err er worker := func() { defer wg.Done() - batch, err := NewBatch(db.storage, version, db.descending, db.operationMetrics) + batch, err := NewBatch(db.storage, version, db.descending, db.dbName, db.operationMetrics) if err != nil { panic(err) } @@ -1201,7 +1217,7 @@ func (db *Database) Import(version int64, ch <-chan types.SnapshotNode) (_err er panic(err) } - batch, err = NewBatch(db.storage, version, db.descending, db.operationMetrics) + batch, err = NewBatch(db.storage, version, db.descending, db.dbName, db.operationMetrics) if err != nil { panic(err) } @@ -1286,7 +1302,7 @@ func (db *Database) RawIterate(storeKey string, fn func(key []byte, value []byte func (db *Database) DeleteKeysAtVersion(module string, version int64) error { - batch, err := NewBatch(db.storage, version, db.descending, db.operationMetrics) + batch, err := NewBatch(db.storage, version, db.descending, db.dbName, db.operationMetrics) if err != nil { return fmt.Errorf("failed to create deletion batch for module %q: %w", module, err) } @@ -1306,7 +1322,7 @@ func (db *Database) DeleteKeysAtVersion(module string, version int64) error { return true } deleteCounter = 0 - batch, err = NewBatch(db.storage, version, db.descending, db.operationMetrics) + batch, err = NewBatch(db.storage, version, db.descending, db.dbName, db.operationMetrics) if err != nil { fmt.Printf("Error creating a new deletion batch for module %q: %v\n", module, err) return true @@ -1381,59 +1397,3 @@ func valTombstoned(value []byte) bool { return true } - -// collectMetricsInBackground periodically collects PebbleDB internal metrics -func (db *Database) collectMetricsInBackground(ctx context.Context) { - ticker := time.NewTicker(10 * time.Second) // Collect metrics every 10 seconds - defer ticker.Stop() - - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - db.collectAndRecordMetrics(ctx) - } - } -} - -// collectAndRecordMetrics collects PebbleDB internal metrics and records them -func (db *Database) collectAndRecordMetrics(ctx context.Context) { - if db.storage == nil { - return - } - - m := db.storage.Metrics() - - // Compaction metrics - report raw counts - otelMetrics.compactionCount.Record(ctx, m.Compact.Count) - otelMetrics.compactionDuration.Record(ctx, m.Compact.Duration.Seconds()) - - // Flush metrics - report raw counts - otelMetrics.flushCount.Record(ctx, m.Flush.Count) - otelMetrics.flushDuration.Record(ctx, m.Flush.WriteThroughput.WorkDuration.Seconds()) - otelMetrics.flushBytesWritten.Record(ctx, m.Flush.WriteThroughput.Bytes) - - // Storage metrics per level with level as attribute - for level := 0; level < len(m.Levels); level++ { - levelMetrics := m.Levels[level] - levelAttr := attribute.Int("level", level) - - otelMetrics.sstableCount.Record(ctx, levelMetrics.TablesCount, metric.WithAttributes(levelAttr)) - otelMetrics.sstableTotalSize.Record(ctx, levelMetrics.TablesSize, metric.WithAttributes(levelAttr)) - otelMetrics.compactionBytesRead.Record(ctx, int64(levelMetrics.TableBytesIn), metric.WithAttributes(levelAttr)) - otelMetrics.compactionBytesWritten.Record(ctx, int64(levelMetrics.TableBytesCompacted), metric.WithAttributes(levelAttr)) - } - - // Memtable metrics - otelMetrics.memtableCount.Record(ctx, m.MemTable.Count) - otelMetrics.memtableTotalSize.Record(ctx, int64(m.MemTable.Size)) //nolint:gosec - - // WAL metrics - otelMetrics.walSize.Record(ctx, int64(m.WAL.Size)) //nolint:gosec - - // Cache metrics - report raw counts - otelMetrics.cacheHits.Record(ctx, m.BlockCache.Hits) - otelMetrics.cacheMisses.Record(ctx, m.BlockCache.Misses) - otelMetrics.cacheSize.Record(ctx, m.BlockCache.Size) -} diff --git a/sei-db/db_engine/pebbledb/mvcc/db_ascending.go b/sei-db/db_engine/pebbledb/mvcc/db_ascending.go index 83dc1d625c..d6c5ed966a 100644 --- a/sei-db/db_engine/pebbledb/mvcc/db_ascending.go +++ b/sei-db/db_engine/pebbledb/mvcc/db_ascending.go @@ -47,6 +47,7 @@ func (db *Database) getAscending(storeKey string, targetVersion int64, key []byt metric.WithAttributes( attribute.Bool("success", _err == nil), attribute.String("store", storeKey), + attribute.String("db", db.dbName), ), ) }() @@ -104,6 +105,7 @@ func (db *Database) pruneAscending(version int64) (_err error) { time.Since(startTime).Seconds(), metric.WithAttributes( attribute.Bool("success", _err == nil), + attribute.String("db", db.dbName), ), ) }() @@ -253,7 +255,7 @@ func (db *Database) iteratorAscending(ctx context.Context, storeKey string, vers return nil, fmt.Errorf("failed to create PebbleDB iterator: %w", err) } - return finishMVCCIterator(newAscendingIterator(ctx, itr, storePrefix(storeKey), start, end, version, db.GetEarliestVersion(), false, storeKey, db.operationMetrics)) + return finishMVCCIterator(newAscendingIterator(ctx, itr, storePrefix(storeKey), start, end, version, db.GetEarliestVersion(), false, storeKey, db.operationMetrics, db.dbName)) } func (db *Database) reverseIteratorAscending(ctx context.Context, storeKey string, version int64, start, end []byte) (dbm.Iterator, error) { @@ -279,7 +281,7 @@ func (db *Database) reverseIteratorAscending(ctx context.Context, storeKey strin return nil, fmt.Errorf("failed to create PebbleDB iterator: %w", err) } - return finishMVCCIterator(newAscendingIterator(ctx, itr, storePrefix(storeKey), start, end, version, db.GetEarliestVersion(), true, storeKey, db.operationMetrics)) + return finishMVCCIterator(newAscendingIterator(ctx, itr, storePrefix(storeKey), start, end, version, db.GetEarliestVersion(), true, storeKey, db.operationMetrics, db.dbName)) } func getMVCCSliceAscending(db *pebble.DB, storeKey string, key []byte, version int64) ([]byte, error) { diff --git a/sei-db/db_engine/pebbledb/mvcc/iterator.go b/sei-db/db_engine/pebbledb/mvcc/iterator.go index 76ebc0d5e5..d7b4f821fb 100644 --- a/sei-db/db_engine/pebbledb/mvcc/iterator.go +++ b/sei-db/db_engine/pebbledb/mvcc/iterator.go @@ -37,6 +37,7 @@ type iterator struct { readCount int64 storeKey string operationMetrics *pebbledbmetrics.OperationMetrics + dbName string ctx context.Context err error @@ -61,7 +62,7 @@ func finishMVCCIterator(itr dbm.Iterator) (dbm.Iterator, error) { return itr, nil } -func newPebbleDBIterator(ctx context.Context, src *pebble.Iterator, prefix, mvccStart, mvccEnd []byte, version int64, earliestVersion int64, reverse bool, useDefaultComparer bool, storeKey string, operationMetrics *pebbledbmetrics.OperationMetrics) *iterator { +func newPebbleDBIterator(ctx context.Context, src *pebble.Iterator, prefix, mvccStart, mvccEnd []byte, version int64, earliestVersion int64, reverse bool, useDefaultComparer bool, storeKey string, operationMetrics *pebbledbmetrics.OperationMetrics, dbName string) *iterator { // Return invalid iterator if requested iterator height is lower than earliest version after pruning if version < earliestVersion { return &iterator{ @@ -75,6 +76,7 @@ func newPebbleDBIterator(ctx context.Context, src *pebble.Iterator, prefix, mvcc useDefaultComparer: useDefaultComparer, storeKey: storeKey, operationMetrics: operationMetrics, + dbName: dbName, ctx: ctx, } } @@ -98,6 +100,7 @@ func newPebbleDBIterator(ctx context.Context, src *pebble.Iterator, prefix, mvcc useDefaultComparer: useDefaultComparer, storeKey: storeKey, operationMetrics: operationMetrics, + dbName: dbName, ctx: ctx, } @@ -381,6 +384,7 @@ func (itr *iterator) Close() error { metric.WithAttributes( attribute.Bool("reverse", itr.reverse), attribute.String("store", itr.storeKey), + attribute.String("db", itr.dbName), ), ) if itr.operationMetrics != nil { diff --git a/sei-db/db_engine/pebbledb/mvcc/iterator_ascending.go b/sei-db/db_engine/pebbledb/mvcc/iterator_ascending.go index a841cbfdc7..0b1eb46129 100644 --- a/sei-db/db_engine/pebbledb/mvcc/iterator_ascending.go +++ b/sei-db/db_engine/pebbledb/mvcc/iterator_ascending.go @@ -41,13 +41,14 @@ type ascendingIterator struct { readCount int64 storeKey string operationMetrics *pebbledbmetrics.OperationMetrics + dbName string ctx context.Context err error closeSync sync.Once } -func newAscendingIterator(ctx context.Context, src *pebble.Iterator, prefix, mvccStart, mvccEnd []byte, version int64, earliestVersion int64, reverse bool, storeKey string, operationMetrics *pebbledbmetrics.OperationMetrics) *ascendingIterator { +func newAscendingIterator(ctx context.Context, src *pebble.Iterator, prefix, mvccStart, mvccEnd []byte, version int64, earliestVersion int64, reverse bool, storeKey string, operationMetrics *pebbledbmetrics.OperationMetrics, dbName string) *ascendingIterator { // Return invalid iterator if requested iterator height is lower than earliest version after pruning if version < earliestVersion { return &ascendingIterator{ @@ -60,6 +61,7 @@ func newAscendingIterator(ctx context.Context, src *pebble.Iterator, prefix, mvc reverse: reverse, storeKey: storeKey, operationMetrics: operationMetrics, + dbName: dbName, ctx: ctx, } } @@ -82,6 +84,7 @@ func newAscendingIterator(ctx context.Context, src *pebble.Iterator, prefix, mvc reverse: reverse, storeKey: storeKey, operationMetrics: operationMetrics, + dbName: dbName, ctx: ctx, } @@ -380,6 +383,7 @@ func (itr *ascendingIterator) Close() error { metric.WithAttributes( attribute.Bool("reverse", itr.reverse), attribute.String("store", itr.storeKey), + attribute.String("db", itr.dbName), ), ) if itr.operationMetrics != nil { diff --git a/sei-db/db_engine/pebbledb/mvcc/metrics.go b/sei-db/db_engine/pebbledb/mvcc/metrics.go index 334d6f14eb..e5c7195538 100644 --- a/sei-db/db_engine/pebbledb/mvcc/metrics.go +++ b/sei-db/db_engine/pebbledb/mvcc/metrics.go @@ -8,6 +8,11 @@ import ( var ( meter = otel.Meter("seidb_pebble") + // otelMetrics holds MVCC operation-level instruments, shared across every + // Database instance in the process; each Record call attaches a "db" + // attribute for the specific instance. Pebble-internal stats (compaction, + // flush, sstable, memtable, WAL, cache) are reported separately, once per + // instance, by pebbledb.PebbleMetrics. otelMetrics = struct { getLatency metric.Float64Histogram applyChangesetLatency metric.Float64Histogram @@ -17,24 +22,6 @@ var ( importLatency metric.Float64Histogram batchWriteLatency metric.Float64Histogram - compactionCount metric.Int64Gauge - compactionDuration metric.Float64Gauge - compactionBytesRead metric.Int64Gauge - compactionBytesWritten metric.Int64Gauge - flushCount metric.Int64Gauge - flushDuration metric.Float64Gauge - flushBytesWritten metric.Int64Gauge - - sstableCount metric.Int64Gauge - sstableTotalSize metric.Int64Gauge - memtableCount metric.Int64Gauge - memtableTotalSize metric.Int64Gauge - walSize metric.Int64Gauge - - cacheHits metric.Int64Gauge - cacheMisses metric.Int64Gauge - cacheSize metric.Int64Gauge - batchSize metric.Int64Histogram pendingChangesQueueDepth metric.Int64Gauge iteratorIterations metric.Float64Histogram @@ -79,85 +66,6 @@ var ( metric.WithUnit("s"), )), - compactionCount: must(meter.Int64Gauge( - "pebble_compaction_count", - metric.WithDescription("Total number of compactions since the DB opened"), - metric.WithUnit("{count}"), - )), - compactionDuration: must(meter.Float64Gauge( - "pebble_compaction_duration", - metric.WithDescription("Cumulative time spent compacting since the DB opened"), - metric.WithUnit("s"), - )), - compactionBytesRead: must(meter.Int64Gauge( - "pebble_compaction_bytes_read", - metric.WithDescription("Total bytes read during compaction since the DB opened"), - metric.WithUnit("By"), - )), - compactionBytesWritten: must(meter.Int64Gauge( - "pebble_compaction_bytes_written", - metric.WithDescription("Total bytes written during compaction since the DB opened"), - metric.WithUnit("By"), - )), - - flushCount: must(meter.Int64Gauge( - "pebble_flush_count", - metric.WithDescription("Total number of memtable flushes since the DB opened"), - metric.WithUnit("{count}"), - )), - flushDuration: must(meter.Float64Gauge( - "pebble_flush_duration", - metric.WithDescription("Cumulative time spent flushing memtables since the DB opened"), - metric.WithUnit("s"), - )), - flushBytesWritten: must(meter.Int64Gauge( - "pebble_flush_bytes_written", - metric.WithDescription("Total bytes written during memtable flushes since the DB opened"), - metric.WithUnit("By"), - )), - - sstableCount: must(meter.Int64Gauge( - "pebble_sstable_count", - metric.WithDescription("Current number of SSTables at each level"), - metric.WithUnit("{count}"), - )), - sstableTotalSize: must(meter.Int64Gauge( - "pebble_sstable_total_size", - metric.WithDescription("Total size of SSTables at each level"), - metric.WithUnit("By"), - )), - memtableCount: must(meter.Int64Gauge( - "pebble_memtable_count", - metric.WithDescription("Current number of memtables"), - metric.WithUnit("{count}"), - )), - memtableTotalSize: must(meter.Int64Gauge( - "pebble_memtable_total_size", - metric.WithDescription("Total size of all memtables"), - metric.WithUnit("By"), - )), - walSize: must(meter.Int64Gauge( - "pebble_wal_size", - metric.WithDescription("Current size of Write-Ahead Log"), - metric.WithUnit("By"), - )), - - cacheHits: must(meter.Int64Gauge( - "pebble_cache_hits", - metric.WithDescription("Total number of cache hits since the DB opened"), - metric.WithUnit("{count}"), - )), - cacheMisses: must(meter.Int64Gauge( - "pebble_cache_misses", - metric.WithDescription("Total number of cache misses since the DB opened"), - metric.WithUnit("{count}"), - )), - cacheSize: must(meter.Int64Gauge( - "pebble_cache_size", - metric.WithDescription("Current cache size"), - metric.WithUnit("By"), - )), - batchSize: must(meter.Int64Histogram( "pebble_batch_size", metric.WithDescription("Size of batches written to PebbleDB"),