metrics: clean up setting of obsolete metrics

There is no reason to actively maintain the stats for the
`obsoleteTables`/`obsoleteBlobs` lists. We can calculate them
when we get the metrics, making things much easier to follow.

The "dual" meaning of these fields - inside `versionSet.Metrics`
they do not include enqueued files but in the result of `Metrics()`
they do include them - was the root cause of the delete pacer
issue #5424.

Author: RaduBerinde

db.go

@@ -1944,13 +1944,17 @@ func (d *DB) Metrics() *Metrics {
 	metrics.Table.Zombie.All.Bytes = d.mu.versions.zombieTables.TotalSize()
 	metrics.Table.Zombie.Local.Count, metrics.Table.Zombie.Local.Bytes = d.mu.versions.zombieTables.LocalStats()
 
-	// The obsolete blob/table metrics have a subtle calculation: the initial
-	// metrics we copied from vs.metrics reflect (in {Table,BlobFiles}.Obsolete
-	// the set of files currently sitting in vs.obsolete{Tables,Blobs} but not yet
-	// enqueued to the delete pacer. To complete the metrics, we add the currently
-	// enqueued files.
-	metrics.Table.Obsolete.Accumulate(deletePacerMetrics.InQueue.Tables)
-	metrics.BlobFiles.Obsolete.Accumulate(deletePacerMetrics.InQueue.BlobFiles)
+	// Populate obsolete blob/table metrics from both the not-yet-enqueued lists
+	// in the versionSet, and what is already in the delete pacer queue.
+	metrics.Table.Obsolete = deletePacerMetrics.InQueue.Tables
+	for _, fi := range d.mu.versions.obsoleteTables {
+		metrics.Table.Obsolete.Inc(fi.FileSize, fi.IsLocal)
+	}
+	metrics.BlobFiles.Obsolete = deletePacerMetrics.InQueue.BlobFiles
+	for _, fi := range d.mu.versions.obsoleteBlobs {
+		metrics.BlobFiles.Obsolete.Inc(fi.FileSize, fi.IsLocal)
+	}
+
 	metrics.private.optionsFileSize = d.optionsFileSize
 
 	d.mu.versions.logLock()

obsolete_files.go

@@ -201,7 +201,6 @@ func (d *DB) scanObsoleteFiles(list []string, flushableIngests []*ingestedFlusha
 	d.mu.versions.obsoleteBlobs = mergeObsoleteFiles(d.mu.versions.obsoleteBlobs, obsoleteBlobs)
 	d.mu.versions.obsoleteManifests = mergeObsoleteFiles(d.mu.versions.obsoleteManifests, obsoleteManifests)
 	d.mu.versions.obsoleteOptions = mergeObsoleteFiles(d.mu.versions.obsoleteOptions, obsoleteOptions)
-	d.mu.versions.updateObsoleteObjectMetricsLocked()
 }
 
 // disableFileDeletions disables file deletions and then waits for any
@@ -289,10 +288,6 @@ func (d *DB) deleteObsoleteFiles(jobID JobID) {
 	obsoleteOptions := d.mu.versions.obsoleteOptions
 	d.mu.versions.obsoleteOptions = nil
 
-	// Update the obsolete object metrics in d.mu.versions.metrics. These metrics
-	// will be combined with the metrics from the delete pacer in DB.Metrics().
-	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()

version_set.go

@@ -76,6 +76,8 @@ type versionSet struct {
 	// on the creation of every version.
 	obsoleteFn func(manifest.ObsoleteFiles)
 	// obsolete{Tables,Blobs,Manifests,Options} are sorted by file number ascending.
+	// TODO(jackson, radu): Investigate if we still need this intermediary step or
+	// we can directly enqueue all deletions.
 	obsoleteTables    []deletepacer.ObsoleteFile
 	obsoleteBlobs     []deletepacer.ObsoleteFile
 	obsoleteManifests []deletepacer.ObsoleteFile
@@ -1038,7 +1040,6 @@ func (vs *versionSet) addObsoleteLocked(obsolete manifest.ObsoleteFiles) {
 			asObsoleteFile(vs.fs, base.FileTypeBlob, vs.dirname)
 	}
 	vs.obsoleteBlobs = mergeObsoleteFiles(vs.obsoleteBlobs, newlyObsoleteBlobFiles)
-	vs.updateObsoleteObjectMetricsLocked()
 }
 
 // addObsolete will acquire DB.mu, so DB.mu must not be held when this is
@@ -1049,20 +1050,6 @@ func (vs *versionSet) addObsolete(obsolete manifest.ObsoleteFiles) {
 	vs.addObsoleteLocked(obsolete)
 }
 
-func (vs *versionSet) updateObsoleteObjectMetricsLocked() {
-	// TODO(jackson, radu): Investigate if we still need the
-	// vs.obsolete{Tables,Blobs} intermediary step or we can directly enqueue all
-	// deletions.
-	vs.metrics.Table.Obsolete = metrics.TableCountsAndSizes{}
-	for _, fi := range vs.obsoleteTables {
-		vs.metrics.Table.Obsolete.Inc(fi.FileSize, fi.IsLocal)
-	}
-	vs.metrics.BlobFiles.Obsolete = metrics.BlobFileCountsAndSizes{}
-	for _, fi := range vs.obsoleteBlobs {
-		vs.metrics.BlobFiles.Obsolete.Inc(fi.FileSize, fi.IsLocal)
-	}
-}
-
 // This method sets the following fields of m.Levels[*]:
 //   - [Virtual]Tables{Count,Size}
 //   - Sublevels