db: make iterator tracking configurable

We disable iterator tracking by default (since the overhead is
measurable). In addition, we allow disabling tracking for specific
iterators via `IterOptions`.

In CRDB we will enable iterator tracking but exempt those in the
hottest batch eval paths.

Author: RaduBerinde

db.go

@@ -1114,8 +1114,8 @@ func (d *DB) newIter(
 		dbi.batch.batch = batch
 		dbi.batch.batchSeqNum = batch.nextSeqNum()
 	}
-	if !dbi.batchOnlyIter {
-		dbi.tracker = d.iterTracker
+	dbi.tracker = d.iterTracker
+	if !dbi.batchOnlyIter && d.iterTracker != nil && !dbi.opts.ExemptFromTracking {
 		dbi.trackerHandle = d.iterTracker.Start()
 	}
 	return finishInitializingIter(ctx, buf)
@@ -1624,8 +1624,10 @@ 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
+	if d.iterTracker != nil {
+		d.iterTracker.Close()
+		d.iterTracker = nil
+	}
 
 	return err
 }

internal/inflight/in_flight.go

@@ -72,6 +72,10 @@ type Tracker struct {
 // A zero Handle is not valid.
 type Handle uint64
 
+func (h Handle) IsValid() bool {
+	return h != 0
+}
+
 // ReportFn is called (typically periodically) with a formatted report that
 // describes entries older than some threshold. An empty string report is never
 // delivered by the polling tracker.

iterator.go

@@ -210,7 +210,9 @@ 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
+	ctx context.Context
+	// TODO(radu): group fields that are always set from `db` fields (tracker,
+	// merge, comparer, fc) into a single struct we can point to.
 	tracker       *inflight.Tracker
 	trackerHandle inflight.Handle
 	opts          IterOptions
@@ -2365,8 +2367,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 {
+	if i.trackerHandle.IsValid() {
 		i.tracker.Stop(i.trackerHandle)
+		i.trackerHandle = 0
 	}
 	// Close the child iterator before releasing the readState because when the
 	// readState is released sstables referenced by the readState may be deleted
@@ -2900,9 +2903,9 @@ func (i *Iterator) CloneWithContext(ctx context.Context, opts CloneOptions) (*It
 		newIters:            i.newIters,
 		newIterRangeKey:     i.newIterRangeKey,
 		seqNum:              i.seqNum,
+		tracker:             i.tracker,
 	}
-	if i.tracker != nil {
-		dbi.tracker = i.tracker
+	if i.tracker != nil && !dbi.opts.ExemptFromTracking {
 		dbi.trackerHandle = i.tracker.Start()
 	}
 	if i.batch != nil {

metamorphic/ops.go

@@ -1181,9 +1181,6 @@ const debugIterators = false
 func (o *newIterOp) run(t *Test, h historyRecorder) {
 	r := t.getReader(o.readerID)
 	opts := iterOptions(t.testOpts.KeyFormat, o.iterOpts)
-	if debugIterators {
-		opts.DebugRangeKeyStack = true
-	}
 
 	var i *pebble.Iterator
 	for {
@@ -1261,9 +1258,14 @@ type newIterUsingCloneOp struct {
 func (o *newIterUsingCloneOp) run(t *Test, h historyRecorder) {
 	iter := t.getIter(o.existingIterID)
 	cloneOpts := pebble.CloneOptions{
-		IterOptions:      iterOptions(t.testOpts.KeyFormat, o.iterOpts),
 		RefreshBatchView: o.refreshBatch,
 	}
+	// We treat the zero options as an indication that we want to keep the options
+	// of the original iterator. This way, we don't depend on iterOptions() and
+	// whether it returns nil or not.
+	if !o.iterOpts.IsZero() {
+		cloneOpts.IterOptions = iterOptions(t.testOpts.KeyFormat, o.iterOpts)
+	}
 	i, err := iter.iter.Clone(cloneOpts)
 	if err != nil {
 		panic(err)
@@ -1377,7 +1379,10 @@ func (o *iterSetOptionsOp) formattedString(kf KeyFormat) string {
 }
 
 func iterOptions(kf KeyFormat, o iterOpts) *pebble.IterOptions {
-	if o.IsZero() && !debugIterators {
+	// Sometimes exempt from tracking.
+	exemptFromTracking := rand.IntN(4) == 0
+	if o.IsZero() && !debugIterators && !exemptFromTracking && rand.IntN(2) == 0 {
+		// Sometimes return nil if we are using the default options.
 		return nil
 	}
 	var lower, upper []byte
@@ -1395,6 +1400,7 @@ func iterOptions(kf KeyFormat, o iterOpts) *pebble.IterOptions {
 			Suffix: o.maskSuffix,
 		},
 		UseL6Filters:       o.useL6Filters,
+		ExemptFromTracking: exemptFromTracking,
 		DebugRangeKeyStack: debugIterators,
 	}
 	if opts.RangeKeyMasking.Suffix != nil {

metamorphic/options.go

@@ -934,6 +934,11 @@ func RandomOptions(rng *rand.Rand, kf KeyFormat, cfg RandomOptionsCfg) *TestOpti
 		}
 	}
 
+	if rand.IntN(2) == 0 {
+		opts.Experimental.IteratorTracking.PollInterval = 100 * time.Millisecond
+		opts.Experimental.IteratorTracking.MaxAge = 10 * time.Second
+	}
+
 	testOpts.seedEFOS = rng.Uint64()
 	testOpts.ingestSplit = rng.IntN(2) == 0
 	opts.Experimental.IngestSplit = func() bool { return testOpts.ingestSplit }

open.go

@@ -205,11 +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)
-	})
+	if iterTrackOpts := opts.Experimental.IteratorTracking; iterTrackOpts.PollInterval > 0 && iterTrackOpts.MaxAge > 0 {
+		d.iterTracker = inflight.NewPollingTracker(iterTrackOpts.PollInterval, iterTrackOpts.MaxAge, 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
@@ -242,7 +242,10 @@ 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 d.iterTracker != nil {
+				d.iterTracker.Close()
+				d.iterTracker = nil
+			}
 			if r != nil {
 				panic(r)
 			}

options.go

@@ -207,6 +207,13 @@ type IterOptions struct {
 	// changed by calling SetOptions.
 	Category block.Category
 
+	// ExemptFromTracking indicates that we should not track the lifetime of the
+	// iterator (used to log information about long-lived iterators). Useful for
+	// hot paths where we know the iterator will be short-lived.
+	ExemptFromTracking bool
+
+	// DebugRangeKeyStack enables additional logging of the range key stack
+	// iterator, via keyspan.InjectLogging. Only used for debugging.
 	DebugRangeKeyStack bool
 
 	// Internal options.
@@ -792,6 +799,22 @@ type Options struct {
 		// virtual table rewrite compactions entirely. The default value is 0.30
 		// (rewrite when >= 30% of backing data is unreferenced).
 		VirtualTableRewriteUnreferencedFraction func() float64
+
+		// IteratorTracking configures periodic logging of iterators held open for
+		// too long.
+		IteratorTracking struct {
+			// PollInterval is the interval at which to log a report of long-lived
+			// iterators. If zero, disables iterator tracking.
+			//
+			// The default value is 0 (disabled).
+			PollInterval time.Duration
+
+			// MaxAge is the age above which iterators are considered long-lived. If
+			// zero, disables iterator tracking.
+			//
+			// The default value is 0 (disabled).
+			MaxAge time.Duration
+		}
 	}
 
 	// Filters is a map from filter policy name to filter policy. It is used for
@@ -1790,6 +1813,13 @@ func (o *Options) String() string {
 	fmt.Fprintf(&buf, "  secondary_cache_size_bytes=%d\n", o.Experimental.SecondaryCacheSizeBytes)
 	fmt.Fprintf(&buf, "  create_on_shared=%d\n", o.Experimental.CreateOnShared)
 
+	if o.Experimental.IteratorTracking.PollInterval != 0 {
+		fmt.Fprintf(&buf, "  iterator_tracking_poll_interval=%s\n", o.Experimental.IteratorTracking.PollInterval)
+	}
+	if o.Experimental.IteratorTracking.MaxAge != 0 {
+		fmt.Fprintf(&buf, "  iterator_tracking_max_age=%s\n", o.Experimental.IteratorTracking.MaxAge)
+	}
+
 	// Private options.
 	//
 	// These options are only encoded if true, because we do not want them to
@@ -2232,6 +2262,10 @@ func (o *Options) Parse(s string, hooks *ParseHooks) error {
 				var createOnSharedInt int64
 				createOnSharedInt, err = strconv.ParseInt(value, 10, 64)
 				o.Experimental.CreateOnShared = remote.CreateOnSharedStrategy(createOnSharedInt)
+			case "iterator_tracking_poll_interval":
+				o.Experimental.IteratorTracking.PollInterval, err = time.ParseDuration(value)
+			case "iterator_tracking_max_age":
+				o.Experimental.IteratorTracking.MaxAge, err = time.ParseDuration(value)
 			default:
 				if hooks != nil && hooks.SkipUnknown != nil && hooks.SkipUnknown(section+"."+key, value) {
 					return nil

options_test.go

@@ -52,6 +52,10 @@ func (o *Options) randomizeForTesting(t testing.TB) {
 		}
 		o.Experimental.ValueSeparationPolicy = func() ValueSeparationPolicy { return policy }
 	}
+	if rand.IntN(2) == 0 {
+		o.Experimental.IteratorTracking.PollInterval = 100 * time.Millisecond
+		o.Experimental.IteratorTracking.MaxAge = 10 * time.Second
+	}
 	o.EnsureDefaults()
 }
 

scan_internal.go

@@ -188,7 +188,6 @@ 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,
@@ -199,6 +198,9 @@ func (d *DB) newInternalIter(
 		seqNum:          seqNum,
 		mergingIter:     &buf.iterAlloc.merging,
 	}
+	if d.iterTracker != nil {
+		dbi.trackerHandle = d.iterTracker.Start()
+	}
 	dbi.blobValueFetcher.Init(&vers.BlobFiles, d.fileCache, block.ReadEnv{},
 		blob.SuggestedCachedReaders(vers.MaxReadAmp()))
 
@@ -1286,7 +1288,10 @@ 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)
+	if i.trackerHandle != 0 {
+		i.db.iterTracker.Stop(i.trackerHandle)
+		i.trackerHandle = 0
+	}
 	err := i.iter.Close()
 	err = errors.CombineErrors(err, i.blobValueFetcher.Close())
 	if i.readState != nil {