db: add BlobFileCreated and BlobFileDeleted events

Add two new events to the EventListener: one invoked when a new blob file is
created and when a blob file is deleted. For now these are largely untested
while blob files are still being integrated.

Informs #112.

Author: jbowens

compaction.go

@@ -174,6 +174,15 @@ func (k compactionKind) String() string {
 	return "?"
 }
 
+// compactingOrFlushing returns "flushing" if the compaction kind is a flush,
+// otherwise it returns "compacting".
+func (k compactionKind) compactingOrFlushing() string {
+	if k == compactionKindFlush {
+		return "flushing"
+	}
+	return "compacting"
+}
+
 // compaction is a table compaction from one level to the next, starting from a
 // given version.
 type compaction struct {
@@ -3143,7 +3152,7 @@ func (d *DB) compactAndWrite(
 		}
 		// Create a new table.
 		writerOpts := d.opts.MakeWriterOptions(c.outputLevel.level, tableFormat)
-		objMeta, tw, err := d.newCompactionOutput(jobID, c, writerOpts)
+		objMeta, tw, err := d.newCompactionOutputTable(jobID, c, writerOpts)
 		if err != nil {
 			return runner.Finish().WithError(err)
 		}
@@ -3271,43 +3280,56 @@ func (c *compaction) makeVersionEdit(result compact.Result) (*versionEdit, error
 	return ve, nil
 }
 
-// newCompactionOutput creates an object for a new table produced by a
+// newCompactionOutputTable creates an object for a new table produced by a
 // compaction or flush.
-func (d *DB) newCompactionOutput(
+func (d *DB) newCompactionOutputTable(
 	jobID JobID, c *compaction, writerOpts sstable.WriterOptions,
 ) (objstorage.ObjectMetadata, sstable.RawWriter, error) {
-	writable, objMeta, err := d.newCompactionOutputObj(jobID, c, base.FileTypeTable)
+	writable, objMeta, err := d.newCompactionOutputObj(c, base.FileTypeTable)
 	if err != nil {
 		return objstorage.ObjectMetadata{}, nil, err
 	}
-
-	var reason string
-	if c.kind == compactionKindFlush {
-		reason = "flushing"
-	} else {
-		reason = "compacting"
-	}
 	d.opts.EventListener.TableCreated(TableCreateInfo{
 		JobID:   int(jobID),
-		Reason:  reason,
+		Reason:  c.kind.compactingOrFlushing(),
 		Path:    d.objProvider.Path(objMeta),
 		FileNum: objMeta.DiskFileNum,
 	})
-
 	writerOpts.SetInternal(sstableinternal.WriterOptions{
 		CacheOpts: sstableinternal.CacheOptions{
 			CacheHandle: d.cacheHandle,
 			FileNum:     objMeta.DiskFileNum,
 		},
 	})
-
 	tw := sstable.NewRawWriterWithCPUMeasurer(writable, writerOpts, c.grantHandle)
 	return objMeta, tw, nil
 }
 
+// Allow the newCompactionOutputBlob method to be unused for now.
+// TODO(jackson): Hook this up.
+var _ = (*DB).newCompactionOutputBlob
+
+// newCompactionOutputBlob creates an object for a new blob produced by a
+// compaction or flush.
+func (d *DB) newCompactionOutputBlob(
+	jobID JobID, c *compaction,
+) (objstorage.Writable, objstorage.ObjectMetadata, error) {
+	writable, objMeta, err := d.newCompactionOutputObj(c, base.FileTypeBlob)
+	if err != nil {
+		return nil, objstorage.ObjectMetadata{}, err
+	}
+	d.opts.EventListener.BlobFileCreated(BlobFileCreateInfo{
+		JobID:   int(jobID),
+		Reason:  c.kind.compactingOrFlushing(),
+		Path:    d.objProvider.Path(objMeta),
+		FileNum: objMeta.DiskFileNum,
+	})
+	return writable, objMeta, nil
+}
+
 // newCompactionOutputObj creates an object produced by a compaction or flush.
 func (d *DB) newCompactionOutputObj(
-	jobID JobID, c *compaction, typ base.FileType,
+	c *compaction, typ base.FileType,
 ) (objstorage.Writable, objstorage.ObjectMetadata, error) {
 	diskFileNum := d.mu.versions.getNextDiskFileNum()
 

event.go

@@ -97,6 +97,48 @@ func (i LevelInfo) SafeFormat(w redact.SafePrinter, _ rune) {
 		redact.Safe(i.Score))
 }
 
+// BlobFileCreateInfo contains the info for a blob file creation event.
+type BlobFileCreateInfo struct {
+	JobID int
+	// Reason is the reason for the table creation: "compacting", "flushing", or
+	// "ingesting".
+	Reason  string
+	Path    string
+	FileNum base.DiskFileNum
+}
+
+func (i BlobFileCreateInfo) String() string {
+	return redact.StringWithoutMarkers(i)
+}
+
+// SafeFormat implements redact.SafeFormatter.
+func (i BlobFileCreateInfo) SafeFormat(w redact.SafePrinter, _ rune) {
+	w.Printf("[JOB %d] %s: blob file created %s",
+		redact.Safe(i.JobID), redact.Safe(i.Reason), i.FileNum)
+}
+
+// BlobFileDeleteInfo contains the info for a blob file deletion event.
+type BlobFileDeleteInfo struct {
+	JobID   int
+	Path    string
+	FileNum base.DiskFileNum
+	Err     error
+}
+
+func (i BlobFileDeleteInfo) String() string {
+	return redact.StringWithoutMarkers(i)
+}
+
+// SafeFormat implements redact.SafeFormatter.
+func (i BlobFileDeleteInfo) SafeFormat(w redact.SafePrinter, _ rune) {
+	if i.Err != nil {
+		w.Printf("[JOB %d] blob file delete error %s: %s",
+			redact.Safe(i.JobID), i.FileNum, i.Err)
+		return
+	}
+	w.Printf("[JOB %d] blob file deleted %s", redact.Safe(i.JobID), i.FileNum)
+}
+
 // CompactionInfo contains the info for a compaction event.
 type CompactionInfo struct {
 	// JobID is the ID of the compaction job.
@@ -728,6 +770,12 @@ type EventListener struct {
 	// operation such as flush or compaction.
 	BackgroundError func(error)
 
+	// BlobFileCreated is invoked after a blob file has been created.
+	BlobFileCreated func(BlobFileCreateInfo)
+
+	// BlobFileDeleted is invoked after a blob file has been deleted.
+	BlobFileDeleted func(BlobFileDeleteInfo)
+
 	// DataCorruption is invoked when an on-disk corruption is detected. It should
 	// not block, as it is called synchronously in read paths.
 	DataCorruption func(DataCorruptionInfo)
@@ -827,6 +875,12 @@ func (l *EventListener) EnsureDefaults(logger Logger) {
 			l.BackgroundError = func(error) {}
 		}
 	}
+	if l.BlobFileCreated == nil {
+		l.BlobFileCreated = func(info BlobFileCreateInfo) {}
+	}
+	if l.BlobFileDeleted == nil {
+		l.BlobFileDeleted = func(info BlobFileDeleteInfo) {}
+	}
 	if l.DataCorruption == nil {
 		if logger != nil {
 			l.DataCorruption = func(info DataCorruptionInfo) {
@@ -912,6 +966,12 @@ func MakeLoggingEventListener(logger Logger) EventListener {
 		BackgroundError: func(err error) {
 			logger.Errorf("background error: %s", err)
 		},
+		BlobFileCreated: func(info BlobFileCreateInfo) {
+			logger.Infof("%s", info)
+		},
+		BlobFileDeleted: func(info BlobFileDeleteInfo) {
+			logger.Infof("%s", info)
+		},
 		DataCorruption: func(info DataCorruptionInfo) {
 			logger.Errorf("%s", info)
 		},
@@ -990,6 +1050,14 @@ func TeeEventListener(a, b EventListener) EventListener {
 			a.BackgroundError(err)
 			b.BackgroundError(err)
 		},
+		BlobFileCreated: func(info BlobFileCreateInfo) {
+			a.BlobFileCreated(info)
+			b.BlobFileCreated(info)
+		},
+		BlobFileDeleted: func(info BlobFileDeleteInfo) {
+			a.BlobFileDeleted(info)
+			b.BlobFileDeleted(info)
+		},
 		DataCorruption: func(info DataCorruptionInfo) {
 			a.DataCorruption(info)
 			b.DataCorruption(info)

obsolete_files.go

@@ -286,7 +286,13 @@ func (cm *cleanupManager) deleteObsoleteObject(
 			FileNum: fileNum,
 			Err:     err,
 		})
-		// TODO(jackson): Add BlobFileDeleted event.
+	case base.FileTypeBlob:
+		cm.opts.EventListener.BlobFileDeleted(BlobFileDeleteInfo{
+			JobID:   int(jobID),
+			Path:    path,
+			FileNum: fileNum,
+			Err:     err,
+		})
 	}
 }