db: fix obsolete file metric underflows

Previously a race existed in the calculation of obsolete file metrics resulting
in underflow. The values within versionSet.metrics were reset and recalculated
to reflect the set of files in versionSet.obsolete{Tables,Blobs} whenever new
files were added to obsoleteFiles. Additionally, the cleanup manager invoked a
callback to decrease versionSet.metrics whenever a table was deleted.

The recalculation of versionSet.metrics could reset the metrics to less than
the sum of outstanding pending deletes. When the cleanup manager eventually
deleted the pending tables, these metrics would underflow.

This commit fixes the bug by maintaining separate stats for all files that have
been enqueued for the cleanup manager and all files that have been successfully
deleted. The volume of outstanding, pending deletions is the difference between
the two.

For now, there's an additional wart that the set of files that are sitting in
versionSet.obsolete{Table,Blobs} are still separately tracked.

Fix #4811.

Author: jbowens

db.go

@@ -453,10 +453,16 @@ type DB struct {
 			noOngoingFlushStartTime crtime.Mono
 		}
 
-		// Non-zero when file cleaning is disabled. The disabled count acts as a
-		// reference count to prohibit file cleaning. See
-		// DB.{disable,Enable}FileDeletions().
-		disableFileDeletions int
+		fileDeletions struct {
+			// Non-zero when file cleaning is disableCount. The disableCount
+			// count acts as a reference count to prohibit file cleaning. See
+			// DB.{disable,enable}FileDeletions().
+			disableCount int
+			// queuedStats holds cumulative stats for files that have been
+			// queued for deletion by the cleanup manager. These stats are
+			// monotonically increasing for the *DB's lifetime.
+			queuedStats obsoleteObjectStats
+		}
 
 		snapshots struct {
 			// The list of active snapshots.
@@ -2055,6 +2061,7 @@ func (d *DB) AsyncFlush() (<-chan struct{}, error) {
 func (d *DB) Metrics() *Metrics {
 	metrics := &Metrics{}
 	walStats := d.mu.log.manager.Stats()
+	completedObsoleteFileStats := d.cleanupManager.CompletedStats()
 
 	d.mu.Lock()
 	vers := d.mu.versions.currentVersion()
@@ -2114,6 +2121,30 @@ func (d *DB) Metrics() *Metrics {
 	metrics.Table.ZombieCount = int64(d.mu.versions.zombieTables.Count())
 	metrics.Table.ZombieSize = d.mu.versions.zombieTables.TotalSize()
 	metrics.Table.Local.ZombieCount, metrics.Table.Local.ZombieSize = d.mu.versions.zombieTables.LocalStats()
+
+	// The obsolete blob/table metrics have a subtle calculation:
+	//
+	// (A) The vs.metrics.{Table,BlobFiles}.[Local.]{ObsoleteCount,ObsoleteSize}
+	// fields reflect the set of files currently sitting in
+	// vs.obsolete{Tables,Blobs} but not yet enqueued to the cleanup manager.
+	//
+	// (B) The d.mu.fileDeletions.queuedStats field holds the set of files that have
+	// been queued for deletion by the cleanup manager.
+	//
+	// (C) The cleanup manager also maintains cumulative stats for the set of
+	// files that have been deleted.
+	//
+	// The value of currently pending obsolete files is (A) + (B) - (C).
+	pendingObsoleteFileStats := d.mu.fileDeletions.queuedStats
+	pendingObsoleteFileStats.Sub(completedObsoleteFileStats)
+	metrics.Table.Local.ObsoleteCount += pendingObsoleteFileStats.tablesLocal.count
+	metrics.Table.Local.ObsoleteSize += pendingObsoleteFileStats.tablesLocal.size
+	metrics.Table.ObsoleteCount += int64(pendingObsoleteFileStats.tablesAll.count)
+	metrics.Table.ObsoleteSize += pendingObsoleteFileStats.tablesAll.size
+	metrics.BlobFiles.Local.ObsoleteCount += pendingObsoleteFileStats.blobFilesLocal.count
+	metrics.BlobFiles.Local.ObsoleteSize += pendingObsoleteFileStats.blobFilesLocal.size
+	metrics.BlobFiles.ObsoleteCount += pendingObsoleteFileStats.blobFilesAll.count
+	metrics.BlobFiles.ObsoleteSize += pendingObsoleteFileStats.blobFilesAll.size
 	metrics.private.optionsFileSize = d.optionsFileSize
 
 	// TODO(jackson): Consider making these metrics optional.

obsolete_files.go

@@ -33,10 +33,9 @@ type DeleteCleaner = base.DeleteCleaner
 type ArchiveCleaner = base.ArchiveCleaner
 
 type cleanupManager struct {
-	opts            *Options
-	objProvider     objstorage.Provider
-	onTableDeleteFn func(fileSize uint64, isLocal bool)
-	deletePacer     *deletionPacer
+	opts        *Options
+	objProvider objstorage.Provider
+	deletePacer *deletionPacer
 
 	// jobsCh is used as the cleanup job queue.
 	jobsCh chan *cleanupJob
@@ -47,12 +46,21 @@ type cleanupManager struct {
 		sync.Mutex
 		// totalJobs is the total number of enqueued jobs (completed or in progress).
 		totalJobs              int
+		completedStats         obsoleteObjectStats
 		completedJobs          int
 		completedJobsCond      sync.Cond
 		jobsQueueWarningIssued bool
 	}
 }
 
+// CompletedStats returns the stats summarizing objects deleted. The returned
+// stats increase monotonically over the lifetime of the DB.
+func (m *cleanupManager) CompletedStats() obsoleteObjectStats {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.mu.completedStats
+}
+
 // We can queue this many jobs before we have to block EnqueueJob.
 const jobsQueueDepth = 1000
 
@@ -74,20 +82,17 @@ func (of *obsoleteFile) needsPacing() bool {
 type cleanupJob struct {
 	jobID         JobID
 	obsoleteFiles []obsoleteFile
+	stats         obsoleteObjectStats
 }
 
 // openCleanupManager creates a cleanupManager and starts its background goroutine.
 // The cleanupManager must be Close()d.
 func openCleanupManager(
-	opts *Options,
-	objProvider objstorage.Provider,
-	onTableDeleteFn func(fileSize uint64, isLocal bool),
-	getDeletePacerInfo func() deletionPacerInfo,
+	opts *Options, objProvider objstorage.Provider, getDeletePacerInfo func() deletionPacerInfo,
 ) *cleanupManager {
 	cm := &cleanupManager{
-		opts:            opts,
-		objProvider:     objProvider,
-		onTableDeleteFn: onTableDeleteFn,
+		opts:        opts,
+		objProvider: objProvider,
 		deletePacer: newDeletionPacer(
 			crtime.NowMono(),
 			opts.FreeSpaceThresholdBytes,
@@ -119,10 +124,13 @@ func (cm *cleanupManager) Close() {
 }
 
 // EnqueueJob adds a cleanup job to the manager's queue.
-func (cm *cleanupManager) EnqueueJob(jobID JobID, obsoleteFiles []obsoleteFile) {
+func (cm *cleanupManager) EnqueueJob(
+	jobID JobID, obsoleteFiles []obsoleteFile, stats obsoleteObjectStats,
+) {
 	job := &cleanupJob{
 		jobID:         jobID,
 		obsoleteFiles: obsoleteFiles,
+		stats:         stats,
 	}
 
 	// Report deleted bytes to the pacer, which can use this data to potentially
@@ -174,7 +182,6 @@ func (cm *cleanupManager) mainLoop() {
 			switch of.fileType {
 			case base.FileTypeTable:
 				cm.maybePace(&tb, &of)
-				cm.onTableDeleteFn(of.fileSize, of.isLocal)
 				cm.deleteObsoleteObject(of.fileType, job.jobID, of.fileNum)
 			case base.FileTypeBlob:
 				cm.maybePace(&tb, &of)
@@ -185,6 +192,7 @@ func (cm *cleanupManager) mainLoop() {
 		}
 		cm.mu.Lock()
 		cm.mu.completedJobs++
+		cm.mu.completedStats.Add(job.stats)
 		cm.mu.completedJobsCond.Broadcast()
 		cm.maybeLogLocked()
 		cm.mu.Unlock()
@@ -320,18 +328,6 @@ func (d *DB) getDeletionPacerInfo() deletionPacerInfo {
 	return pacerInfo
 }
 
-// onObsoleteTableDelete is called to update metrics when an sstable is deleted.
-func (d *DB) onObsoleteTableDelete(fileSize uint64, isLocal bool) {
-	d.mu.Lock()
-	d.mu.versions.metrics.Table.ObsoleteCount--
-	d.mu.versions.metrics.Table.ObsoleteSize -= fileSize
-	if isLocal {
-		d.mu.versions.metrics.Table.Local.ObsoleteCount--
-		d.mu.versions.metrics.Table.Local.ObsoleteSize -= fileSize
-	}
-	d.mu.Unlock()
-}
-
 // scanObsoleteFiles scans the filesystem for files that are no longer needed
 // and adds those to the internal lists of obsolete files. Note that the files
 // are not actually deleted by this method. A subsequent call to
@@ -448,7 +444,7 @@ func (d *DB) scanObsoleteFiles(list []string, flushableIngests []*ingestedFlusha
 //
 // d.mu must be held when calling this method.
 func (d *DB) disableFileDeletions() {
-	d.mu.disableFileDeletions++
+	d.mu.fileDeletions.disableCount++
 	d.mu.Unlock()
 	defer d.mu.Lock()
 	d.cleanupManager.Wait()
@@ -459,11 +455,11 @@ func (d *DB) disableFileDeletions() {
 //
 // d.mu must be held when calling this method.
 func (d *DB) enableFileDeletions() {
-	if d.mu.disableFileDeletions <= 0 {
+	if d.mu.fileDeletions.disableCount <= 0 {
 		panic("pebble: file deletion disablement invariant violated")
 	}
-	d.mu.disableFileDeletions--
-	if d.mu.disableFileDeletions > 0 {
+	d.mu.fileDeletions.disableCount--
+	if d.mu.fileDeletions.disableCount > 0 {
 		return
 	}
 	d.deleteObsoleteFiles(d.newJobIDLocked())
@@ -478,7 +474,7 @@ type fileInfo = base.FileInfo
 // Does nothing if file deletions are disabled (see disableFileDeletions). A
 // cleanup job will be scheduled when file deletions are re-enabled.
 func (d *DB) deleteObsoleteFiles(jobID JobID) {
-	if d.mu.disableFileDeletions > 0 {
+	if d.mu.fileDeletions.disableCount > 0 {
 		return
 	}
 	_, noRecycle := d.opts.Cleaner.(base.NeedsFileContents)
@@ -524,6 +520,16 @@ func (d *DB) deleteObsoleteFiles(jobID JobID) {
 	obsoleteOptions := d.mu.versions.obsoleteOptions
 	d.mu.versions.obsoleteOptions = nil
 
+	// Compute the stats for the files being queued for deletion and add them to
+	// the running total. These stats will be used during DB.Metrics() to
+	// calculate the count and size of pending obsolete files by diffing these
+	// stats and the stats reported by the cleanup manager.
+	var objectStats obsoleteObjectStats
+	objectStats.tablesAll, objectStats.tablesLocal = calculateObsoleteObjectStats(obsoleteTables)
+	objectStats.blobFilesAll, objectStats.blobFilesLocal = calculateObsoleteObjectStats(obsoleteBlobs)
+	d.mu.fileDeletions.queuedStats.Add(objectStats)
+	d.mu.versions.updateObsoleteObjectMetricsLocked()
+
 	// Release d.mu while preparing the cleanup job and possibly waiting.
 	// Note the unusual order: Unlock and then Lock.
 	d.mu.Unlock()
@@ -552,7 +558,7 @@ func (d *DB) deleteObsoleteFiles(jobID JobID) {
 		d.fileCache.Evict(f.fileNum, base.FileTypeBlob)
 	}
 	if len(filesToDelete) > 0 {
-		d.cleanupManager.EnqueueJob(jobID, filesToDelete)
+		d.cleanupManager.EnqueueJob(jobID, filesToDelete, objectStats)
 	}
 	if d.opts.private.testingAlwaysWaitForCleanup {
 		d.cleanupManager.Wait()
@@ -601,6 +607,54 @@ func (o objectInfo) asObsoleteFile(fs vfs.FS, fileType base.FileType, dirname st
 	}
 }
 
+func calculateObsoleteObjectStats(files []obsoleteFile) (total, local countAndSize) {
+	for _, of := range files {
+		if of.isLocal {
+			local.count++
+			local.size += of.fileSize
+		}
+		total.count++
+		total.size += of.fileSize
+	}
+	return total, local
+}
+
+type obsoleteObjectStats struct {
+	tablesLocal    countAndSize
+	tablesAll      countAndSize
+	blobFilesLocal countAndSize
+	blobFilesAll   countAndSize
+}
+
+func (s *obsoleteObjectStats) Add(other obsoleteObjectStats) {
+	s.tablesLocal.Add(other.tablesLocal)
+	s.tablesAll.Add(other.tablesAll)
+	s.blobFilesLocal.Add(other.blobFilesLocal)
+	s.blobFilesAll.Add(other.blobFilesAll)
+}
+
+func (s *obsoleteObjectStats) Sub(other obsoleteObjectStats) {
+	s.tablesLocal.Sub(other.tablesLocal)
+	s.tablesAll.Sub(other.tablesAll)
+	s.blobFilesLocal.Sub(other.blobFilesLocal)
+	s.blobFilesAll.Sub(other.blobFilesAll)
+}
+
+type countAndSize struct {
+	count uint64
+	size  uint64
+}
+
+func (c *countAndSize) Add(other countAndSize) {
+	c.count += other.count
+	c.size += other.size
+}
+
+func (c *countAndSize) Sub(other countAndSize) {
+	c.count = invariants.SafeSub(c.count, other.count)
+	c.size = invariants.SafeSub(c.size, other.size)
+}
+
 func makeZombieObjects() zombieObjects {
 	return zombieObjects{
 		objs: make(map[base.DiskFileNum]objectInfo),

open.go

@@ -391,7 +391,7 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 
 	d.mu.log.manager = walManager
 
-	d.cleanupManager = openCleanupManager(opts, d.objProvider, d.onObsoleteTableDelete, d.getDeletionPacerInfo)
+	d.cleanupManager = openCleanupManager(opts, d.objProvider, d.getDeletionPacerInfo)
 
 	if manifestExists && !opts.DisableConsistencyCheck {
 		curVersion := d.mu.versions.currentVersion()

version_set.go

@@ -1197,6 +1197,11 @@ func (vs *versionSet) addObsolete(obsolete manifest.ObsoleteFiles) {
 }
 
 func (vs *versionSet) updateObsoleteObjectMetricsLocked() {
+	// TODO(jackson): Ideally we would update vs.fileDeletions.queuedStats to
+	// include the files on vs.obsolete{Tables,Blobs}, but there's subtlety in
+	// deduplicating the files before computing the stats. It might also be
+	// possible to refactor to remove the vs.obsolete{Tables,Blobs} intermediary
+	// step. Revisit this.
 	vs.metrics.Table.ObsoleteCount = int64(len(vs.obsoleteTables))
 	vs.metrics.Table.ObsoleteSize = 0
 	vs.metrics.Table.Local.ObsoleteSize = 0