db: track and report long-standing iterators

We wire up iterators to an `inflight.Tracker`. Every 5 minutes we log
a report on iterators that have been open for more than 1 minute.

Author: RaduBerinde

db.go

@@ -21,6 +21,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/arenaskl"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/cache"
+	"github.com/cockroachdb/pebble/internal/inflight"
 	"github.com/cockroachdb/pebble/internal/invalidating"
 	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/keyspan"
@@ -545,6 +546,8 @@ type DB struct {
 	// the time at database Open; may be used to compute metrics like effective
 	// compaction concurrency
 	openedAt time.Time
+
+	iterTracker *inflight.Tracker
 }
 
 var _ Reader = (*DB)(nil)
@@ -1111,6 +1114,10 @@ func (d *DB) newIter(
 		dbi.batch.batch = batch
 		dbi.batch.batchSeqNum = batch.nextSeqNum()
 	}
+	if !dbi.batchOnlyIter {
+		dbi.tracker = d.iterTracker
+		dbi.trackerHandle = d.iterTracker.Start()
+	}
 	return finishInitializingIter(ctx, buf)
 }
 
@@ -1121,6 +1128,7 @@ func (d *DB) newIter(
 func finishInitializingIter(ctx context.Context, buf *iterAlloc) *Iterator {
 	// Short-hand.
 	dbi := &buf.dbi
+
 	var memtables flushableList
 	if dbi.readState != nil {
 		memtables = dbi.readState.memtables
@@ -1616,6 +1624,9 @@ func (d *DB) Close() error {
 		err = firstError(err, errors.Errorf("leaked snapshots: %d open snapshots on DB %p", v, d))
 	}
 
+	d.iterTracker.Close()
+	d.iterTracker = nil
+
 	return err
 }
 

iterator.go

@@ -16,6 +16,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/bytealloc"
 	"github.com/cockroachdb/pebble/internal/humanize"
+	"github.com/cockroachdb/pebble/internal/inflight"
 	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/keyspan"
 	"github.com/cockroachdb/pebble/internal/keyspan/keyspanimpl"
@@ -209,12 +210,14 @@ type Iterator struct {
 	// short-lived (since they pin memtables and sstables), (b) plumbing a
 	// context into every method is very painful, (c) they do not (yet) respect
 	// context cancellation and are only used for tracing.
-	ctx       context.Context
-	opts      IterOptions
-	merge     Merge
-	comparer  *base.Comparer
-	iter      internalIterator
-	pointIter topLevelIterator
+	ctx           context.Context
+	tracker       *inflight.Tracker
+	trackerHandle inflight.Handle
+	opts          IterOptions
+	merge         Merge
+	comparer      *base.Comparer
+	iter          internalIterator
+	pointIter     topLevelIterator
 	// Either readState or version is set, but not both.
 	readState *readState
 	version   *manifest.Version
@@ -2362,6 +2365,9 @@ const maxKeyBufCacheSize = 4 << 10 // 4 KB
 // It is not valid to call any method, including Close, after the iterator
 // has been closed.
 func (i *Iterator) Close() error {
+	if i.tracker != nil {
+		i.tracker.Stop(i.trackerHandle)
+	}
 	// Close the child iterator before releasing the readState because when the
 	// readState is released sstables referenced by the readState may be deleted
 	// which will fail on Windows if the sstables are still open by the child
@@ -2895,6 +2901,10 @@ func (i *Iterator) CloneWithContext(ctx context.Context, opts CloneOptions) (*It
 		newIterRangeKey:     i.newIterRangeKey,
 		seqNum:              i.seqNum,
 	}
+	if i.tracker != nil {
+		dbi.tracker = i.tracker
+		dbi.trackerHandle = i.tracker.Start()
+	}
 	if i.batch != nil {
 		dbi.batch = &buf.batchState
 		dbi.batch.batch = i.batch.batch

open.go

@@ -23,6 +23,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/arenaskl"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/cache"
+	"github.com/cockroachdb/pebble/internal/inflight"
 	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/keyspan"
 	"github.com/cockroachdb/pebble/internal/manifest"
@@ -204,6 +205,11 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 	} else {
 		d.compactionScheduler = newConcurrencyLimitScheduler(defaultTimeSource{})
 	}
+	const iterTrackerPollInterval = 5 * time.Minute
+	const iterTrackerMaxAge = 1 * time.Minute
+	d.iterTracker = inflight.NewPollingTracker(iterTrackerPollInterval, iterTrackerMaxAge, func(report string) {
+		d.opts.Logger.Infof("Long-lived iterators detected:\n%s", report)
+	})
 
 	defer func() {
 		// If an error or panic occurs during open, attempt to release the manually
@@ -236,6 +242,7 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 			if d.mu.versions.manifestFile != nil {
 				_ = d.mu.versions.manifestFile.Close()
 			}
+			d.iterTracker.Close()
 			if r != nil {
 				panic(r)
 			}

scan_internal.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/inflight"
 	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/keyspan"
 	"github.com/cockroachdb/pebble/internal/keyspan/keyspanimpl"
@@ -187,6 +188,7 @@ func (d *DB) newInternalIter(
 	*dbi = scanInternalIterator{
 		ctx:             ctx,
 		db:              d,
+		trackerHandle:   d.iterTracker.Start(),
 		comparer:        d.opts.Comparer,
 		merge:           d.opts.Merger.Merge,
 		readState:       readState,
@@ -578,6 +580,7 @@ type IteratorLevel struct {
 type scanInternalIterator struct {
 	ctx              context.Context
 	db               *DB
+	trackerHandle    inflight.Handle
 	opts             ScanInternalOptions
 	comparer         *base.Comparer
 	merge            Merge
@@ -1283,6 +1286,7 @@ func (i *scanInternalIterator) error() error {
 
 // Close closes this iterator, and releases any pooled objects.
 func (i *scanInternalIterator) Close() error {
+	i.db.iterTracker.Stop(i.trackerHandle)
 	err := i.iter.Close()
 	err = errors.CombineErrors(err, i.blobValueFetcher.Close())
 	if i.readState != nil {