Skip to content

Fix pebble db metrics #4031

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
alrevuelta wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Fix pebble db metrics #4031

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3cc5f2e
Fix pebble db metrics
alrevuelta Aug 27, 2026
2b8b377
Remove duplicated metrics
alrevuelta Aug 28, 2026
08e13a0
Merge branch 'main' into fix-pebble-metrics
alrevuelta Aug 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 14 additions & 7 deletions sei-db/db_engine/pebbledb/mvcc/batch.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ type Batch struct {
ops []batchOp
descending bool
operationMetrics *pebbledbmetrics.OperationMetrics
dbName string
}

type batchOp struct {
Expand All @@ -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")
}
Expand All @@ -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
}

Expand Down Expand Up @@ -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 {
Expand All @@ -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]
Expand All @@ -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
}

Expand Down Expand Up @@ -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)
}
Expand All @@ -165,17 +169,20 @@ 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() {
ctx := context.Background()
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()
Expand Down
118 changes: 39 additions & 79 deletions sei-db/db_engine/pebbledb/mvcc/db.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
}

Expand Down Expand Up @@ -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)

@cursor cursor Bot Aug 28, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

DB metric names collide across stores

Medium Severity

dbName is filepath.Base(dataDir), so default store paths (state_store/cosmos/pebbledb, state_store/evm/pebbledb, ledger/receipt/pebbledb) all label series db="pebbledb". Gauges then overwrite each other and counters mix increments, so the new attribute does not separate SeparateEVMSubDBs or other in-process Pebble instances.

Additional Locations (1)
Fix in Cursor Fix in Web

Reviewed by Cursor Bugbot for commit 08e13a0. Configure here.

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)
Expand Down Expand Up @@ -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
}
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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
}
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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),
),
)
}()
Expand Down Expand Up @@ -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),
),
)
}()
Expand Down Expand Up @@ -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) {
Expand All @@ -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) {
Expand Down Expand Up @@ -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),
),
)
}()
Expand All @@ -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)
}
Expand All @@ -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)
}
Expand Down Expand Up @@ -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)
}
Expand All @@ -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
Expand Down Expand Up @@ -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.Add(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.flushDuration.Record(ctx, m.Flush.WriteThroughput.WorkDuration.Seconds())
otelMetrics.flushBytesWritten.Add(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.Add(ctx, int64(levelMetrics.TableBytesIn), metric.WithAttributes(levelAttr)) //nolint:gosec
otelMetrics.compactionBytesWritten.Add(ctx, int64(levelMetrics.TableBytesCompacted), metric.WithAttributes(levelAttr)) //nolint:gosec
}

// 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.Add(ctx, m.BlockCache.Hits)
otelMetrics.cacheMisses.Add(ctx, m.BlockCache.Misses)
otelMetrics.cacheSize.Record(ctx, m.BlockCache.Size)
}
6 changes: 4 additions & 2 deletions sei-db/db_engine/pebbledb/mvcc/db_ascending.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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),
),
)
}()
Expand Down Expand Up @@ -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),
),
)
}()
Expand Down Expand Up @@ -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) {
Expand All @@ -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) {
Expand Down
Loading
Loading