db: add support for blob files to versionSet

Adapt the versionSet to write extant blob files to new manifests during a
manifest rollover and support tracking zombie and obsolete blob files.

Informs #112.

Author: jbowens

metrics.go

@@ -312,6 +312,36 @@ type Metrics struct {
 		}
 	}
 
+	BlobFiles struct {
+		// The count of all live blob files.
+		LiveCount uint64
+		// The size of all live blob files.
+		LiveSize uint64
+		// The count of all obsolete blob files.
+		ObsoleteCount uint64
+		// The size of all obsolete blob files.
+		ObsoleteSize uint64
+		// The count of all zombie blob files.
+		ZombieCount uint64
+		// The size of all zombie blob files.
+		ZombieSize uint64
+		// Local file sizes.
+		Local struct {
+			// LiveSize is the number of bytes in live blob files.
+			LiveSize uint64
+			// LiveCount is the number of live blob files.
+			LiveCount uint64
+			// ObsoleteSize is the number of bytes in obsolete blob files.
+			ObsoleteSize uint64
+			// ObsoleteCount is the number of obsolete blob files.
+			ObsoleteCount uint64
+			// ZombieSize is the number of bytes in zombie blob files.
+			ZombieSize uint64
+			// ZombieCount is the number of zombie blob files.
+			ZombieCount uint64
+		}
+	}
+
 	FileCache CacheMetrics
 
 	// Count of the number of open sstable iterators.

objstorage/objstorage.go

@@ -395,6 +395,13 @@ func Copy(ctx context.Context, r ReadHandle, out Writable, offset, length uint64
 	return nil
 }
 
+// IsLocalBlobFile returns true if a blob file with the given fileNum exists and is
+// local.
+func IsLocalBlobFile(provider Provider, fileNum base.DiskFileNum) bool {
+	meta, err := provider.Lookup(base.FileTypeBlob, fileNum)
+	return err == nil && !meta.IsRemote()
+}
+
 // IsLocalTable returns true if a table with the given fileNum exists and is
 // local.
 func IsLocalTable(provider Provider, fileNum base.DiskFileNum) bool {

obsolete_files.go

@@ -439,7 +439,7 @@ func (d *DB) scanObsoleteFiles(list []string, flushableIngests []*ingestedFlusha
 
 	d.mu.versions.obsoleteTables = mergeObjectInfos(d.mu.versions.obsoleteTables, obsoleteTables)
 	d.mu.versions.obsoleteBlobs = mergeObjectInfos(d.mu.versions.obsoleteBlobs, obsoleteBlobs)
-	d.mu.versions.updateObsoleteTableMetricsLocked()
+	d.mu.versions.updateObsoleteObjectMetricsLocked()
 	d.mu.versions.obsoleteManifests = mergeObsoleteFiles(d.mu.versions.obsoleteManifests, obsoleteManifests)
 	d.mu.versions.obsoleteOptions = mergeObsoleteFiles(d.mu.versions.obsoleteOptions, obsoleteOptions)
 }