db: track memtable flush, manifest update durations in TableIngestInfo

This patch tracks the duration of memtable flushes (for flushable
ingests) and MANIFEST updates in our TableIngestInfo. This gives
us more observability into slow ingests.

Informs: #4970

Author: annrpom

event.go

@@ -535,6 +535,12 @@ type TableIngestInfo struct {
 	// flushable.
 	flushable bool
 	Err       error
+	// WaitFlushDuration is the time spent waiting for memtable flushes to
+	// complete, given that an overlap between ingesting sstables and memtables
+	// exists.
+	WaitFlushDuration time.Duration
+	// ManifestUpdateDuration is the time spent updating the manifest.
+	ManifestUpdateDuration time.Duration
 }
 
 func (i TableIngestInfo) String() string {
@@ -549,7 +555,8 @@ func (i TableIngestInfo) SafeFormat(w redact.SafePrinter, _ rune) {
 	}
 
 	if i.flushable {
-		w.Printf("[JOB %d] ingested as flushable", redact.Safe(i.JobID))
+		w.Printf("[JOB %d] ingested as flushable, memtable flushes took %.1fs:", redact.Safe(i.JobID),
+			redact.Safe(i.WaitFlushDuration.Seconds()))
 	} else {
 		w.Printf("[JOB %d] ingested", redact.Safe(i.JobID))
 	}
@@ -566,6 +573,7 @@ func (i TableIngestInfo) SafeFormat(w redact.SafePrinter, _ rune) {
 		w.Printf(" %s%s (%s)", redact.Safe(levelStr), t.FileNum,
 			redact.Safe(humanize.Bytes.Uint64(t.Size)))
 	}
+	w.Printf("; manifest update took %.1fs", redact.Safe(i.ManifestUpdateDuration.Seconds()))
 }
 
 // TableStatsInfo contains the info for a table stats loaded event.

event_listener_test.go

@@ -55,6 +55,13 @@ func TestEventListener(t *testing.T) {
 				info.InputBytes = 100
 				flushEnd(info)
 			}
+			tableIngested := lel.TableIngested
+			lel.TableIngested = func(info TableIngestInfo) {
+				// Make deterministic.
+				info.WaitFlushDuration = 200 * time.Millisecond
+				info.ManifestUpdateDuration = 100 * time.Millisecond
+				tableIngested(info)
+			}
 			opts := &Options{
 				// The table stats collector runs asynchronously and its
 				// timing is less predictable. It increments nextJobID, which

ingest.go

@@ -11,6 +11,7 @@ import (
 	"sort"
 	"time"
 
+	"github.com/cockroachdb/crlib/crtime"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/cache"
@@ -1499,6 +1500,7 @@ func (d *DB) ingest(ctx context.Context, args ingestArgs) (IngestOperationStats,
 	var mut *memTable
 	// asFlushable indicates whether the sstable was ingested as a flushable.
 	var asFlushable bool
+	var waitFlushStart crtime.Mono
 	prepare := func(seqNum base.SeqNum) {
 		// Note that d.commit.mu is held by commitPipeline when calling prepare.
 
@@ -1607,7 +1609,6 @@ func (d *DB) ingest(ctx context.Context, args ingestArgs) (IngestOperationStats,
 				d.mu.log.manager.ElevateWriteStallThresholdForFailover()) &&
 			!d.opts.Experimental.DisableIngestAsFlushable() && !hasRemoteFiles &&
 			(!args.ExciseSpan.Valid() || d.FormatMajorVersion() >= FormatFlushableIngestExcises)
-
 		if !canIngestFlushable {
 			// We're not able to ingest as a flushable,
 			// so we must synchronously flush.
@@ -1628,6 +1629,7 @@ func (d *DB) ingest(ctx context.Context, args ingestArgs) (IngestOperationStats,
 			mut = d.mu.mem.mutable
 			mut.writerRef()
 			mem.flushForced = true
+			waitFlushStart = crtime.NowMono()
 			d.maybeScheduleFlush()
 			return
 		}
@@ -1642,6 +1644,8 @@ func (d *DB) ingest(ctx context.Context, args ingestArgs) (IngestOperationStats,
 	}
 
 	var ve *manifest.VersionEdit
+	var waitFlushDuration time.Duration
+	var manifestUpdateDuration time.Duration
 	apply := func(seqNum base.SeqNum) {
 		if err != nil || asFlushable {
 			// An error occurred during prepare.
@@ -1683,11 +1687,12 @@ func (d *DB) ingest(ctx context.Context, args ingestArgs) (IngestOperationStats,
 		// finish.
 		if mem != nil {
 			<-mem.flushed
+			waitFlushDuration = waitFlushStart.Elapsed()
 		}
 
 		// Assign the sstables to the correct level in the LSM and apply the
 		// version edit.
-		ve, err = d.ingestApply(ctx, jobID, loadResult, mut, args.ExciseSpan, args.ExciseBoundsPolicy, seqNum)
+		ve, manifestUpdateDuration, err = d.ingestApply(ctx, jobID, loadResult, mut, args.ExciseSpan, args.ExciseBoundsPolicy, seqNum)
 	}
 
 	// Only one ingest can occur at a time because if not, one would block waiting
@@ -1724,9 +1729,11 @@ func (d *DB) ingest(ctx context.Context, args ingestArgs) (IngestOperationStats,
 	var stats IngestOperationStats
 	if loadResult.fileCount() > 0 {
 		info := TableIngestInfo{
-			JobID:     int(jobID),
-			Err:       err,
-			flushable: asFlushable,
+			JobID:                  int(jobID),
+			Err:                    err,
+			flushable:              asFlushable,
+			WaitFlushDuration:      waitFlushDuration,
+			ManifestUpdateDuration: manifestUpdateDuration,
 		}
 		if len(loadResult.local) > 0 {
 			info.GlobalSeqNum = loadResult.local[0].SmallestSeqNum
@@ -1905,7 +1912,7 @@ func (d *DB) ingestApply(
 	exciseSpan KeyRange,
 	exciseBoundsPolicy exciseBoundsPolicy,
 	exciseSeqNum base.SeqNum,
-) (*manifest.VersionEdit, error) {
+) (*manifest.VersionEdit, time.Duration, error) {
 	d.mu.Lock()
 	defer d.mu.Unlock()
 
@@ -1917,6 +1924,7 @@ func (d *DB) ingestApply(
 	}
 	var metrics levelMetricsDelta
 
+	manifestUpdateStart := crtime.NowMono()
 	// Determine the target level inside UpdateVersionLocked. This prevents two
 	// concurrent ingestion jobs from using the same version to determine the
 	// target level, and also provides serialization with concurrent compaction
@@ -2164,8 +2172,9 @@ func (d *DB) ingestApply(
 		}, nil
 	})
 	if err != nil {
-		return nil, err
+		return nil, 0, err
 	}
+	manifestUpdateDuration := manifestUpdateStart.Elapsed()
 
 	// Check for any EventuallyFileOnlySnapshots that could be watching for
 	// an excise on this span. There should be none as the
@@ -2211,7 +2220,7 @@ func (d *DB) ingestApply(
 		}
 	}
 	d.maybeValidateSSTablesLocked(toValidate)
-	return ve, nil
+	return ve, manifestUpdateDuration, nil
 }
 
 // maybeValidateSSTablesLocked adds the slice of newTableEntrys to the pending

internal/strparse/strparse.go

@@ -175,7 +175,7 @@ func (p *Parser) Uint32() uint32 {
 	return uint32(x)
 }
 
-// Uint64 parses the next token as a sequence number.
+// SeqNum parses the next token as a sequence number.
 func (p *Parser) SeqNum() base.SeqNum {
 	return base.ParseSeqNum(p.Next())
 }

replay/testdata/collect/start_stop

@@ -14,7 +14,7 @@ ingest
 created src/000002.sst
 created src/000003.sst
 created src/000004.sst
-[JOB 0] ingested L0:000002 (10KB), L0:000003 (10KB), L0:000004 (10KB)
+[JOB 0] ingested L0:000002 (10KB), L0:000003 (10KB), L0:000004 (10KB); manifest update took 0.1s
 
 
 wait

replay/workload_capture_test.go

@@ -128,6 +128,10 @@ func TestWorkloadCollector(t *testing.T) {
 					_, err = currentManifest.Write(randData(25))
 					require.NoError(t, err)
 				}
+				// Override the default duration values for TableIngestInfo to
+				// ensure deterministic output.
+				ingestInfo.WaitFlushDuration = 200 * time.Millisecond
+				ingestInfo.ManifestUpdateDuration = 100 * time.Millisecond
 				fmt.Fprint(&buf, ingestInfo.String())
 				c.onTableIngest(ingestInfo)
 				return buf.String()

testdata/event_listener

@@ -238,7 +238,7 @@ sync: db
 remove: db/MANIFEST-000011
 [JOB 10] MANIFEST deleted 000011
 remove: ext/0
-[JOB 10] ingested L0:000015 (757B)
+[JOB 10] ingested L0:000015 (755B); manifest update took 0.1s
 
 metrics
 ----
@@ -350,7 +350,7 @@ sync: wal
 [JOB 13] WAL created 000020
 remove: ext/a
 remove: ext/b
-[JOB 11] ingested as flushable 000017 (757B), 000018 (757B)
+[JOB 11] ingested as flushable, memtable flushes took 0.2s: 000017 (755B), 000018 (755B); manifest update took 0.1s
 sync-data: wal/000020.log
 close: wal/000020.log
 create: wal/000021.log