Repository: petermattis/pebble
Pull request #123 (2019-05-06 00:57:00)

log EventListener events by default
For events available to `EventListener` for which the user has not
provided any custom handler, log them to the `db.Options.Logger`.

petermattis: 

<!-- Reviewable:start -->
This change is [<img src="https://reviewable.io/review_button.png" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/petermattis/pebble/123)
<!-- Reviewable:end -->

petermattis: @ajkr I see I've caused I few conflicts here. Do you want me to take this PR over?
petermattis: I've updated this PR so that instead of providing a default `EventListener` which logs, there is now a `db.NewLoggingEventListener` function. Minor other adjustments. 

Patch:
diff --git a/compaction.go b/compaction.go
index d10be56..0e1702f 100644
--- a/compaction.go
+++ b/compaction.go
@@ -744,7 +744,9 @@ func (d *DB) compact() {
 	defer d.mu.Unlock()
 	if err := d.compact1(); err != nil {
 		// TODO(peter): count consecutive compaction errors and backoff.
-		_ = err
+		if d.opts.EventListener != nil && d.opts.EventListener.BackgroundError != nil {
+			d.opts.EventListener.BackgroundError(err)
+		}
 	}
 	d.mu.compact.compacting = false
 	// The previous compaction may have produced too many files in a
@@ -775,9 +777,17 @@ func (d *DB) compact1() (err error) {
 
 	jobID := d.mu.nextJobID
 	d.mu.nextJobID++
+	info := db.CompactionInfo{
+		JobID: jobID,
+	}
 	if d.opts.EventListener != nil && d.opts.EventListener.CompactionBegin != nil {
-		info := db.CompactionInfo{
-			JobID: jobID,
+		info.Input.Level = c.startLevel
+		info.Output.Level = c.outputLevel
+		for i := range c.inputs {
+			for j := range c.inputs[i] {
+				m := &c.inputs[i][j]
+				info.Input.Tables[i] = append(info.Input.Tables[i], m.tableInfo(d.dirname))
+			}
 		}
 		d.opts.EventListener.CompactionBegin(info)
 	}
@@ -785,19 +795,8 @@ func (d *DB) compact1() (err error) {
 	ve, pendingOutputs, err := d.compactDiskTables(c)
 
 	if d.opts.EventListener != nil && d.opts.EventListener.CompactionEnd != nil {
-		info := db.CompactionInfo{
-			JobID: jobID,
-			Err:   err,
-		}
-		if err != nil {
-			info.Input.Level = c.startLevel
-			info.Output.Level = c.outputLevel
-			for i := range c.inputs {
-				for j := range c.inputs[i] {
-					m := &c.inputs[i][j]
-					info.Input.Tables[i] = append(info.Input.Tables[i], m.tableInfo(d.dirname))
-				}
-			}
+		info.Err = err
+		if err == nil {
 			for i := range ve.newFiles {
 				e := &ve.newFiles[i]
 				info.Output.Tables = append(info.Output.Tables, e.meta.tableInfo(d.dirname))
diff --git a/db/event.go b/db/event.go
index a323ff8..510d6df 100644
--- a/db/event.go
+++ b/db/event.go
@@ -140,3 +140,94 @@ type EventListener struct {
 	// WALDeleted is invoked after a WAL has been delete.
 	WALDeleted func(WALDeleteInfo)
 }
+
+// NewLoggingEventListener creates an EventListener that logs all events to the
+// specified logger.
+func NewLoggingEventListener(logger Logger) *EventListener {
+	return &EventListener{
+		BackgroundError: func(err error) {
+			logger.Infof("background error: %s", err)
+		},
+		CompactionBegin: func(info CompactionInfo) {
+			if len(info.Input.Tables[1]) > 0 {
+				logger.Infof("[JOB %d] compacting %d:L%d and %d:L%d to L%d",
+					info.JobID, len(info.Input.Tables[0]), info.Input.Level,
+					len(info.Input.Tables[1]), info.Output.Level, info.Output.Level)
+			} else {
+				logger.Infof("[JOB %d] compacting %d:L%d to L%d",
+					info.JobID, len(info.Input.Tables[0]), info.Input.Level,
+					info.Output.Level)
+			}
+		},
+		CompactionEnd: func(info CompactionInfo) {
+			if info.Err != nil {
+				logger.Infof("[JOB %d] compaction to L%d error: %s",
+					info.JobID, info.Output.Level, info.Err)
+				return
+			}
+
+			if len(info.Input.Tables[1]) > 0 {
+				logger.Infof("[JOB %d] compacted %d:L%d and %d:L%d to L%d",
+					info.JobID, len(info.Input.Tables[0]), info.Input.Level,
+					len(info.Input.Tables[1]), info.Output.Level, info.Output.Level)
+			} else {
+				logger.Infof("[JOB %d] compacted %d:L%d to L%d",
+					info.JobID, len(info.Input.Tables[0]), info.Input.Level,
+					info.Output.Level)
+			}
+		},
+		FlushBegin: func(info FlushInfo) {
+			logger.Infof("[JOB %d] flushing", info.JobID)
+		},
+		FlushEnd: func(info FlushInfo) {
+			if info.Err != nil {
+				logger.Infof("[JOB %d] flush error: %s", info.JobID, info.Err)
+				return
+			}
+
+			logger.Infof("[JOB %d] flushed sstable %06d", info.JobID, info.Output.FileNum)
+		},
+		TableDeleted: func(info TableDeleteInfo) {
+			if info.Err != nil {
+				logger.Infof("[JOB %d] sstable delete error %06d: %s", info.JobID,
+					info.FileNum, info.Err)
+				return
+			}
+
+			logger.Infof("[JOB %d] sstable deleted %06d", info.JobID, info.FileNum)
+		},
+		TableIngested: func(info TableIngestInfo) {
+			if info.Err != nil {
+				logger.Infof("[JOB %d] ingest error: %s", info.Err)
+				return
+			}
+
+			var plural string
+			if len(info.Tables) != 1 {
+				plural = "s"
+			}
+			logger.Infof("[JOB %d] ingested %d sstable%s", info.JobID, len(info.Tables), plural)
+		},
+		WALCreated: func(info WALCreateInfo) {
+			if info.Err != nil {
+				logger.Infof("[JOB %d] WAL create error: %s", info.Err)
+				return
+			}
+
+			if info.RecycledFileNum == 0 {
+				logger.Infof("[JOB %d] WAL created %06d", info.JobID, info.FileNum)
+			} else {
+				logger.Infof("[JOB %d] WAL created %06d (recycled %06d)", info.JobID,
+					info.FileNum, info.RecycledFileNum)
+			}
+		},
+		WALDeleted: func(info WALDeleteInfo) {
+			if info.Err != nil {
+				logger.Infof("[JOB %d] WAL delete error: %s", info.JobID, info.Err)
+				return
+			}
+
+			logger.Infof("[JOB %d] WAL deleted %06d", info.JobID, info.FileNum)
+		},
+	}
+}


Test patch:
diff --git a/event_listener_test.go b/event_listener_test.go
index ad89dbe..406bacc 100644
--- a/event_listener_test.go
+++ b/event_listener_test.go
@@ -34,6 +34,20 @@ func (b *syncedBuffer) Write(p []byte) (n int, err error) {
 	return b.buf.Write(p)
 }
 
+func (b *syncedBuffer) Infof(format string, args ...interface{}) {
+	s := fmt.Sprintf(format, args...)
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.buf.Write([]byte(s))
+	if n := len(s); n == 0 || s[n-1] != '\n' {
+		b.buf.Write([]byte("\n"))
+	}
+}
+
+func (b *syncedBuffer) Fatalf(format string, args ...interface{}) {
+	panic(fmt.Sprintf(format, args...))
+}
+
 func (b *syncedBuffer) String() string {
 	b.mu.Lock()
 	defer b.mu.Unlock()
@@ -84,38 +98,6 @@ func (f loggingFile) Sync() error {
 	return f.File.Sync()
 }
 
-func newLoggingEventListener(w io.Writer) *db.EventListener {
-	return &db.EventListener{
-		CompactionBegin: func(info db.CompactionInfo) {
-			fmt.Fprintf(w, "#%d: compaction begin: L%d -> L%d\n", info.JobID,
-				info.Input.Level, info.Input.Level+1)
-		},
-		CompactionEnd: func(info db.CompactionInfo) {
-			fmt.Fprintf(w, "#%d: compaction end: L%d -> L%d\n", info.JobID,
-				info.Input.Level, info.Input.Level+1)
-		},
-		FlushBegin: func(info db.FlushInfo) {
-			fmt.Fprintf(w, "#%d: flush begin\n", info.JobID)
-		},
-		FlushEnd: func(info db.FlushInfo) {
-			fmt.Fprintf(w, "#%d: flush end: %d\n", info.JobID, info.Output.FileNum)
-		},
-		TableDeleted: func(info db.TableDeleteInfo) {
-			fmt.Fprintf(w, "#%d: table deleted: %d\n", info.JobID, info.FileNum)
-		},
-		TableIngested: func(info db.TableIngestInfo) {
-			fmt.Fprintf(w, "#%d: table ingested\n", info.JobID)
-		},
-		WALCreated: func(info db.WALCreateInfo) {
-			fmt.Fprintf(w, "#%d: WAL created: %d recycled=%d\n",
-				info.JobID, info.FileNum, info.RecycledFileNum)
-		},
-		WALDeleted: func(info db.WALDeleteInfo) {
-			fmt.Fprintf(w, "#%d: WAL deleted: %d\n", info.JobID, info.FileNum)
-		},
-	}
-}
-
 // Verify event listener actions, as well as expected filesystem operations.
 func TestEventListener(t *testing.T) {
 	var d *DB
@@ -133,7 +115,7 @@ func TestEventListener(t *testing.T) {
 			var err error
 			d, err = Open("db", &db.Options{
 				FS:                  loggingFS{mem, &buf},
-				EventListener:       newLoggingEventListener(&buf),
+				EventListener:       db.NewLoggingEventListener(&buf),
 				MaxManifestFileSize: 1,
 				WALDir:              "wal",
 			})
diff --git a/testdata/event_listener b/testdata/event_listener
index ca07702..2f3fc12 100644
--- a/testdata/event_listener
+++ b/testdata/event_listener
@@ -24,12 +24,12 @@ sync: wal/000002.log
 create: wal/000005.log
 sync: wal
 sync: wal/000002.log
-#2: WAL created: 5 recycled=0
-#3: flush begin
+[JOB 2] WAL created 000005
+[JOB 3] flushing
 create: db/000006.sst
 sync: db/000006.sst
 sync: db
-#3: flush end: 6
+[JOB 3] flushed sstable 000006
 create: db/MANIFEST-000007
 sync: db/MANIFEST-000007
 create: db/CURRENT.000007.dbtmp
@@ -44,31 +44,31 @@ rename: wal/000002.log -> wal/000008.log
 create: wal/000008.log
 sync: wal
 sync: wal/000005.log
-#4: WAL created: 8 recycled=2
-#5: flush begin
+[JOB 4] WAL created 000008 (recycled 000002)
+[JOB 5] flushing
 create: db/000009.sst
 sync: db/000009.sst
 sync: db
-#5: flush end: 9
+[JOB 5] flushed sstable 000009
 create: db/MANIFEST-000010
 sync: db/MANIFEST-000010
 create: db/CURRENT.000010.dbtmp
 sync: db/CURRENT.000010.dbtmp
 rename: db/CURRENT.000010.dbtmp -> db/CURRENT
 sync: db
-#6: compaction begin: L0 -> L1
+[JOB 6] compacting 2:L0 to L6
 create: db/000011.sst
 sync: db/000011.sst
 sync: db
-#6: compaction end: L0 -> L1
+[JOB 6] compacted 2:L0 to L6
 create: db/MANIFEST-000012
 sync: db/MANIFEST-000012
 create: db/CURRENT.000012.dbtmp
 sync: db/CURRENT.000012.dbtmp
 rename: db/CURRENT.000012.dbtmp -> db/CURRENT
 sync: db
-#6: table deleted: 6
-#6: table deleted: 9
+[JOB 6] sstable deleted 000006
+[JOB 6] sstable deleted 000009
 
 ingest
 ----
@@ -80,7 +80,7 @@ create: db/CURRENT.000014.dbtmp
 sync: db/CURRENT.000014.dbtmp
 rename: db/CURRENT.000014.dbtmp -> db/CURRENT
 sync: db
-#7: table ingested
+[JOB 7] ingested 1 sstable
 
 metrics
 ----