diff --git a/sei-db/db_engine/pebbledb/mvcc/batch.go b/sei-db/db_engine/pebbledb/mvcc/batch.go index 7071930e89..d06620d034 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,13 @@ 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 +52,7 @@ func NewBatch(storage *pebble.DB, version int64, descending bool, operationMetri ops: make([]batchOp, 0, 16), descending: descending, operationMetrics: metrics, + dbName: dbName, }, nil } @@ -77,9 +85,12 @@ 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 + 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 { return fmt.Errorf("failed to set latest version in batch: %w", err) } @@ -97,10 +108,16 @@ 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 +128,7 @@ func NewRawBatch(storage *pebble.DB, descending bool, operationMetrics ...*pebbl ops: make([]batchOp, 0, 16), descending: descending, operationMetrics: metrics, + dbName: dbName, }, nil } @@ -154,7 +172,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 +183,12 @@ 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 +196,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 421fe4dcf8..b645300944 100644 --- a/sei-db/db_engine/pebbledb/mvcc/db.go +++ b/sei-db/db_engine/pebbledb/mvcc/db.go @@ -106,6 +106,10 @@ 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 don't share unattributed series. + dbName string + operationMetrics *pebbledbmetrics.OperationMetrics } @@ -207,18 +211,20 @@ func OpenDB(dataDir string, config config.StateStoreConfig) (types.StateStore, e return nil, fmt.Errorf("failed to retrieve latest version: %w", err) } + dbName := dataDir + if abs, absErr := filepath.Abs(dataDir); absErr == nil { + dbName = abs + } 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 +253,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 +601,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 +659,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 +673,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 +708,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 @@ -798,7 +813,12 @@ func (db *Database) Iterator(storeKey string, version int64, start, end []byte) return db.IteratorWithContext(context.Background(), storeKey, version, start, end) } -func (db *Database) IteratorWithContext(ctx context.Context, storeKey string, version int64, start, end []byte) (dbm.Iterator, error) { +func (db *Database) IteratorWithContext( + ctx context.Context, + storeKey string, + version int64, + start, end []byte, +) (dbm.Iterator, error) { if db.descending { return db.iteratorDescending(ctx, storeKey, version, start, end) } @@ -809,7 +829,12 @@ func (db *Database) ReverseIterator(storeKey string, version int64, start, end [ return db.ReverseIteratorWithContext(context.Background(), storeKey, version, start, end) } -func (db *Database) ReverseIteratorWithContext(ctx context.Context, storeKey string, version int64, start, end []byte) (dbm.Iterator, error) { +func (db *Database) ReverseIteratorWithContext( + ctx context.Context, + storeKey string, + version int64, + start, end []byte, +) (dbm.Iterator, error) { if db.descending { return db.reverseIteratorDescending(ctx, storeKey, version, start, end) } @@ -846,6 +871,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 +918,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), ), ) }() @@ -1022,7 +1049,12 @@ func (db *Database) pruneDescending(version int64) (_err error) { return db.compactPrunedRange(firstDeletedKey, lastDeletedKey) } -func (db *Database) iteratorDescending(ctx context.Context, storeKey string, version int64, start, end []byte) (dbm.Iterator, error) { +func (db *Database) iteratorDescending( + ctx context.Context, + storeKey string, + version int64, + start, end []byte, +) (dbm.Iterator, error) { if (start != nil && len(start) == 0) || (end != nil && len(end) == 0) { return nil, errorutils.ErrKeyEmpty } @@ -1045,10 +1077,30 @@ 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) { +func (db *Database) reverseIteratorDescending( + ctx context.Context, + storeKey string, + version int64, + start, end []byte, +) (dbm.Iterator, error) { if (start != nil && len(start) == 0) || (end != nil && len(end) == 0) { return nil, errorutils.ErrKeyEmpty } @@ -1071,7 +1123,22 @@ 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 +1239,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 +1248,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 +1269,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 +1354,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 +1374,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 +1449,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) -} diff --git a/sei-db/db_engine/pebbledb/mvcc/db_ascending.go b/sei-db/db_engine/pebbledb/mvcc/db_ascending.go index 83dc1d625c..cf016a999f 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), ), ) }() @@ -181,7 +183,8 @@ func (db *Database) pruneAscending(version int64) (_err error) { // Delete a key if another entry for that key exists at a larger version than original but leq to the prune height // Also delete a key if it has been tombstoned and its version is leq to the prune height // Also delete a key if KeepLastVersion is false and version is leq to the prune height - if prevVersionDecoded <= version && (bytes.Equal(prevKey, currKey) || valTombstoned(prevValEncoded) || !db.config.KeepLastVersion) { + if prevVersionDecoded <= version && + (bytes.Equal(prevKey, currKey) || valTombstoned(prevValEncoded) || !db.config.KeepLastVersion) { err = batch.Delete(prevKeyEncoded, nil) if err != nil { return err @@ -232,7 +235,12 @@ func (db *Database) pruneAscending(version int64) (_err error) { return db.compactPrunedRange(firstDeletedKey, lastDeletedKey) } -func (db *Database) iteratorAscending(ctx context.Context, storeKey string, version int64, start, end []byte) (dbm.Iterator, error) { +func (db *Database) iteratorAscending( + ctx context.Context, + storeKey string, + version int64, + start, end []byte, +) (dbm.Iterator, error) { if (start != nil && len(start) == 0) || (end != nil && len(end) == 0) { return nil, errorutils.ErrKeyEmpty } @@ -253,10 +261,29 @@ 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) { +func (db *Database) reverseIteratorAscending( + ctx context.Context, + storeKey string, + version int64, + start, end []byte, +) (dbm.Iterator, error) { if (start != nil && len(start) == 0) || (end != nil && len(end) == 0) { return nil, errorutils.ErrKeyEmpty } @@ -279,7 +306,21 @@ 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..cf868dbd7c 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,18 @@ 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 +87,7 @@ func newPebbleDBIterator(ctx context.Context, src *pebble.Iterator, prefix, mvcc useDefaultComparer: useDefaultComparer, storeKey: storeKey, operationMetrics: operationMetrics, + dbName: dbName, ctx: ctx, } } @@ -98,6 +111,7 @@ func newPebbleDBIterator(ctx context.Context, src *pebble.Iterator, prefix, mvcc useDefaultComparer: useDefaultComparer, storeKey: storeKey, operationMetrics: operationMetrics, + dbName: dbName, ctx: ctx, } @@ -381,6 +395,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..d4014cb3ba 100644 --- a/sei-db/db_engine/pebbledb/mvcc/iterator_ascending.go +++ b/sei-db/db_engine/pebbledb/mvcc/iterator_ascending.go @@ -41,13 +41,24 @@ 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 +71,7 @@ func newAscendingIterator(ctx context.Context, src *pebble.Iterator, prefix, mvc reverse: reverse, storeKey: storeKey, operationMetrics: operationMetrics, + dbName: dbName, ctx: ctx, } } @@ -82,6 +94,7 @@ func newAscendingIterator(ctx context.Context, src *pebble.Iterator, prefix, mvc reverse: reverse, storeKey: storeKey, operationMetrics: operationMetrics, + dbName: dbName, ctx: ctx, } @@ -380,6 +393,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 7a7a6dddad..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.Int64Counter - compactionDuration metric.Float64Histogram - compactionBytesRead metric.Int64Counter - compactionBytesWritten metric.Int64Counter - flushCount metric.Int64Counter - flushDuration metric.Float64Histogram - flushBytesWritten metric.Int64Counter - - sstableCount metric.Int64Gauge - sstableTotalSize metric.Int64Gauge - memtableCount metric.Int64Gauge - memtableTotalSize metric.Int64Gauge - walSize metric.Int64Gauge - - cacheHits metric.Int64Counter - cacheMisses metric.Int64Counter - cacheSize metric.Int64Gauge - batchSize metric.Int64Histogram pendingChangesQueueDepth metric.Int64Gauge iteratorIterations metric.Float64Histogram @@ -79,85 +66,6 @@ var ( metric.WithUnit("s"), )), - compactionCount: must(meter.Int64Counter( - "pebble_compaction_count", - metric.WithDescription("Total number of compactions"), - metric.WithUnit("{count}"), - )), - compactionDuration: must(meter.Float64Histogram( - "pebble_compaction_duration", - metric.WithDescription("Duration of compaction operations"), - metric.WithUnit("s"), - )), - compactionBytesRead: must(meter.Int64Counter( - "pebble_compaction_bytes_read", - metric.WithDescription("Total bytes read during compaction"), - metric.WithUnit("By"), - )), - compactionBytesWritten: must(meter.Int64Counter( - "pebble_compaction_bytes_written", - metric.WithDescription("Total bytes written during compaction"), - metric.WithUnit("By"), - )), - - flushCount: must(meter.Int64Counter( - "pebble_flush_count", - metric.WithDescription("Total number of memtable flushes"), - metric.WithUnit("{count}"), - )), - flushDuration: must(meter.Float64Histogram( - "pebble_flush_duration", - metric.WithDescription("Duration of memtable flush operations"), - metric.WithUnit("s"), - )), - flushBytesWritten: must(meter.Int64Counter( - "pebble_flush_bytes_written", - metric.WithDescription("Total bytes written during memtable flushes"), - 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.Int64Counter( - "pebble_cache_hits", - metric.WithDescription("Total number of cache hits"), - metric.WithUnit("{count}"), - )), - cacheMisses: must(meter.Int64Counter( - "pebble_cache_misses", - metric.WithDescription("Total number of cache misses"), - 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"),