Repository: petermattis/pebble
Pull request #133 (2019-05-10 20:43:00)

Add basic level / compaction metrics
Track level metrics during flushes, compactions, and ingestion. Provide a
pretty-printer for thes metrics which shows per-level num-files, size,
bytes-in, bytes-ingested, bytes-moved, bytes-read, bytes-written, and
write-amplification. Also show a total across all levels, as well as WAL
bytes in, and written.


Patch:
diff --git a/batch.go b/batch.go
index 29b7bce..7801b63 100644
--- a/batch.go
+++ b/batch.go
@@ -974,8 +974,8 @@ func (b *flushableBatch) readyForFlush() bool {
 	return true
 }
 
-func (b *flushableBatch) logNumber() uint64 {
-	return b.logNum
+func (b *flushableBatch) logInfo() (uint64, uint64) {
+	return b.logNum, 0 /* logSize */
 }
 
 // Note: flushableBatchIter mirrors the implementation of batchIter. Keep the
diff --git a/compaction.go b/compaction.go
index 1e42c2a..d10be56 100644
--- a/compaction.go
+++ b/compaction.go
@@ -525,13 +525,22 @@ func (d *DB) flush1() error {
 
 	// The flush succeeded or it produced an empty sstable. In either case we
 	// want to bump the log number.
+	metrics := &LevelMetrics{}
 	ve := &versionEdit{
-		logNumber: d.mu.mem.queue[n].logNumber(),
+		metrics: map[int]*LevelMetrics{
+			0: metrics,
+		},
+	}
+	ve.logNumber, _ = d.mu.mem.queue[n].logInfo()
+	for i := 0; i < n; i++ {
+		_, size := d.mu.mem.queue[i].logInfo()
+		metrics.BytesIn += size
 	}
 	if err != errEmptyTable {
 		ve.newFiles = []newFileEntry{
 			{level: 0, meta: meta},
 		}
+		metrics.BytesWritten = meta.size
 	}
 
 	err = d.mu.versions.logAndApply(ve, d.dataDir)
@@ -835,6 +844,11 @@ func (d *DB) compactDiskTables(c *compaction) (ve *versionEdit, pendingOutputs [
 			newFiles: []newFileEntry{
 				{level: c.outputLevel, meta: *meta},
 			},
+			metrics: map[int]*LevelMetrics{
+				c.outputLevel: &LevelMetrics{
+					BytesMoved: meta.size,
+				},
+			},
 		}, nil, nil
 	}
 
@@ -880,8 +894,17 @@ func (d *DB) compactDiskTables(c *compaction) (ve *versionEdit, pendingOutputs [
 		}
 	}()
 
+	metrics := &LevelMetrics{
+		BytesIn:   totalSize(c.inputs[0]),
+		BytesRead: totalSize(c.inputs[1]),
+	}
+	metrics.BytesRead += metrics.BytesIn
+
 	ve = &versionEdit{
 		deletedFiles: map[deletedFileEntry]bool{},
+		metrics: map[int]*LevelMetrics{
+			c.outputLevel: metrics,
+		},
 	}
 
 	newOutput := func() error {
@@ -937,6 +960,8 @@ func (d *DB) compactDiskTables(c *compaction) (ve *versionEdit, pendingOutputs [
 		meta.smallestSeqNum = writerMeta.SmallestSeqNum
 		meta.largestSeqNum = writerMeta.LargestSeqNum
 
+		metrics.BytesWritten += meta.size
+
 		// The handling of range boundaries is a bit complicated.
 		if n := len(ve.newFiles); n > 1 {
 			// This is not the first output. Bound the smallest range key by the
@@ -1127,9 +1152,11 @@ func (d *DB) deleteObsoleteFiles(jobID int) {
 		if d.mu.log.queue[i] >= d.mu.versions.logNumber {
 			obsoleteLogs = d.mu.log.queue[:i]
 			d.mu.log.queue = d.mu.log.queue[i:]
+			d.mu.versions.metrics.WAL.Files -= uint64(len(obsoleteLogs))
 			break
 		}
 	}
+
 	obsoleteTables := d.mu.versions.obsoleteTables
 	d.mu.versions.obsoleteTables = nil
 
diff --git a/db.go b/db.go
index cceb9f2..f9b3757 100644
--- a/db.go
+++ b/db.go
@@ -33,7 +33,7 @@ type flushable interface {
 	newRangeDelIter(o *db.IterOptions) internalIterator
 	flushed() chan struct{}
 	readyForFlush() bool
-	logNumber() uint64
+	logInfo() (num, size uint64)
 }
 
 // Reader is a readable key/value store.
@@ -169,7 +169,9 @@ type DB struct {
 		versions versionSet
 
 		log struct {
-			queue []uint64
+			queue   []uint64
+			size    uint64
+			bytesIn uint64
 			*record.LogWriter
 		}
 
@@ -396,6 +398,10 @@ func (d *DB) commitWrite(b *Batch) (*memTable, error) {
 	// Switch out the memtable if there was not enough room to store the batch.
 	err := d.makeRoomForWrite(b)
 
+	if err == nil {
+		d.mu.log.bytesIn += uint64(len(b.storage.data))
+	}
+
 	d.mu.Unlock()
 	if err != nil {
 		return nil, err
@@ -405,10 +411,12 @@ func (d *DB) commitWrite(b *Batch) (*memTable, error) {
 		return d.mu.mem.mutable, nil
 	}
 
-	_, err = d.mu.log.WriteRecord(b.storage.data)
+	size, err := d.mu.log.WriteRecord(b.storage.data)
 	if err != nil {
 		panic(err)
 	}
+
+	atomic.StoreUint64(&d.mu.log.size, uint64(size))
 	return d.mu.mem.mutable, err
 }
 
@@ -719,6 +727,28 @@ func (d *DB) AsyncFlush() error {
 	return err
 }
 
+// Metrics returns metrics about the database.
+func (d *DB) Metrics() *VersionMetrics {
+	metrics := &VersionMetrics{}
+	d.mu.Lock()
+	*metrics = d.mu.versions.metrics
+	metrics.WAL.Size = atomic.LoadUint64(&d.mu.log.size)
+	metrics.WAL.BytesIn = d.mu.log.bytesIn // protected by d.mu
+	for i, n := 0, len(d.mu.mem.queue)-1; i < n; i++ {
+		_, size := d.mu.mem.queue[i].logInfo()
+		metrics.WAL.Size += size
+	}
+	metrics.WAL.BytesWritten = metrics.Levels[0].BytesIn + metrics.WAL.Size
+	metrics.Levels[0].Score = float64(metrics.Levels[0].NumFiles) / float64(d.opts.L0CompactionThreshold)
+	if p := d.mu.versions.picker; p != nil {
+		for level := 1; level < numLevels; level++ {
+			metrics.Levels[level].Score = float64(metrics.Levels[level].Size) / float64(p.levelMaxBytes[level])
+		}
+	}
+	d.mu.Unlock()
+	return metrics
+}
+
 func (d *DB) walPreallocateSize() int {
 	// Set the WAL preallocate size to 110% of the memtable size. Note that there
 	// is a bit of apples and oranges in units here as the memtabls size
@@ -783,6 +813,7 @@ func (d *DB) makeRoomForWrite(b *Batch) error {
 
 		var newLogNumber uint64
 		var newLogFile vfs.File
+		var prevLogSize uint64
 		var err error
 
 		if !d.opts.DisableWAL {
@@ -817,6 +848,7 @@ func (d *DB) makeRoomForWrite(b *Batch) error {
 			}
 
 			if err == nil {
+				prevLogSize = uint64(d.mu.log.Size())
 				err = d.mu.log.Close()
 				if err != nil {
 					newLogFile.Close()
@@ -845,6 +877,8 @@ func (d *DB) makeRoomForWrite(b *Batch) error {
 			d.mu.Lock()
 			d.mu.mem.switching = false
 			d.mu.mem.cond.Broadcast()
+
+			d.mu.versions.metrics.WAL.Files++
 		}
 
 		if err != nil {
@@ -861,7 +895,9 @@ func (d *DB) makeRoomForWrite(b *Batch) error {
 			d.mu.log.LogWriter = record.NewLogWriter(newLogFile, newLogNumber)
 		}
 
-		prevLogNumber := d.mu.mem.mutable.logNum
+		imm := d.mu.mem.mutable
+		imm.logSize = prevLogSize
+		prevLogNumber := imm.logNum
 
 		var scheduleFlush bool
 		if b != nil && b.flushable != nil {
@@ -879,7 +915,6 @@ func (d *DB) makeRoomForWrite(b *Batch) error {
 		// also have to wait for all previous immutable tables to
 		// flush. Additionally, the memtable is tied to particular WAL file and we
 		// want to go through the flush path in order to recycle that WAL file.
-		imm := d.mu.mem.mutable
 		d.mu.mem.mutable = newMemTable(d.opts)
 		// NB: When the immutable memtable is flushed to disk it will apply a
 		// versionEdit to the manifest telling it that log files < newLogNumber
diff --git a/ingest.go b/ingest.go
index f60a32d..5d3f9f6 100644
--- a/ingest.go
+++ b/ingest.go
@@ -380,14 +380,22 @@ func (d *DB) ingestApply(meta []*fileMetadata) (*versionEdit, error) {
 
 	ve := &versionEdit{
 		newFiles: make([]newFileEntry, len(meta)),
+		metrics:  make(map[int]*LevelMetrics),
 	}
 	current := d.mu.versions.currentVersion()
 	for i := range meta {
 		// Determine the lowest level in the LSM for which the sstable doesn't
 		// overlap any existing files in the level.
 		m := meta[i]
-		ve.newFiles[i].level = ingestTargetLevel(d.cmp, current, m)
-		ve.newFiles[i].meta = *m
+		f := &ve.newFiles[i]
+		f.level = ingestTargetLevel(d.cmp, current, m)
+		f.meta = *m
+		metrics := ve.metrics[f.level]
+		if metrics == nil {
+			metrics = &LevelMetrics{}
+			ve.metrics[f.level] = metrics
+		}
+		metrics.BytesIngested += m.size
 	}
 	if err := d.mu.versions.logAndApply(ve, d.dataDir); err != nil {
 		return nil, err
diff --git a/internal/humanize/humanize.go b/internal/humanize/humanize.go
new file mode 100644
index 0000000..e44c861
--- /dev/null
+++ b/internal/humanize/humanize.go
@@ -0,0 +1,34 @@
+// Copyright 2019 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package humanize
+
+import (
+	"fmt"
+	"math"
+)
+
+var sizes = []string{"B", "K", "M", "G", "T", "P", "E"}
+
+func logn(n, b float64) float64 {
+	return math.Log(n) / math.Log(b)
+}
+
+// Uint64 produces a human readable representation of the value.
+func Uint64(s uint64) string {
+	const base = 1024
+
+	if s < 10 {
+		return fmt.Sprintf("%d B", s)
+	}
+	e := math.Floor(logn(float64(s), base))
+	suffix := sizes[int(e)]
+	val := math.Floor(float64(s)/math.Pow(base, e)*10+0.5) / 10
+	f := "%.0f %s"
+	if val < 10 {
+		f = "%.1f %s"
+	}
+
+	return fmt.Sprintf(f, val, suffix)
+}
diff --git a/internal/record/log_writer.go b/internal/record/log_writer.go
index d67911f..b8ba8a5 100644
--- a/internal/record/log_writer.go
+++ b/internal/record/log_writer.go
@@ -291,6 +291,11 @@ func (w *LogWriter) WriteRecord(p []byte) (int64, error) {
 	return offset, w.err
 }
 
+// Size returns the current size of the file.
+func (w *LogWriter) Size() int64 {
+	return w.blockNum*blockSize + int64(w.block.written)
+}
+
 func (w *LogWriter) emitFragment(n int, p []byte) []byte {
 	b := w.block
 	i := b.written
diff --git a/mem_table.go b/mem_table.go
index c381370..e3bd091 100644
--- a/mem_table.go
+++ b/mem_table.go
@@ -54,6 +54,7 @@ type memTable struct {
 	flushedCh   chan struct{}
 	tombstones  rangeTombstoneCache
 	logNum      uint64
+	logSize     uint64
 }
 
 // newMemTable returns a new MemTable.
@@ -95,8 +96,8 @@ func (m *memTable) readyForFlush() bool {
 	return atomic.LoadInt32(&m.refs) == 0
 }
 
-func (m *memTable) logNumber() uint64 {
-	return m.logNum
+func (m *memTable) logInfo() (uint64, uint64) {
+	return m.logNum, m.logSize
 }
 
 // Get gets the value for the given key. It returns ErrNotFound if the DB does
diff --git a/metrics.go b/metrics.go
new file mode 100644
index 0000000..000af14
--- /dev/null
+++ b/metrics.go
@@ -0,0 +1,140 @@
+// Copyright 2019 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package pebble
+
+import (
+	"bytes"
+	"fmt"
+
+	"github.com/petermattis/pebble/internal/humanize"
+)
+
+// LevelMetrics holds per-level metrics such as the number of files and total
+// size of the files, and compaction related metrics.
+type LevelMetrics struct {
+	// The total number of files in the level.
+	NumFiles uint64
+	// The total size in bytes of the files in the level.
+	Size uint64
+	// The level's compaction score.
+	Score float64
+	// The number of incoming bytes from other levels read during
+	// compactions. This excludes bytes moved and bytes ingested. For L0 this is
+	// the bytes written to the WAL.
+	BytesIn uint64
+	// The number of bytes ingested.
+	BytesIngested uint64
+	// The number of bytes moved into the level by a "move" compaction.
+	BytesMoved uint64
+	// The number of bytes read for compactions at the level. This includes bytes
+	// read from other levels (BytesIn), as well as bytes read for the level.
+	BytesRead uint64
+	// The number of bytes written during compactions.
+	BytesWritten uint64
+}
+
+// Add updates the counter metrics for the level.
+func (m *LevelMetrics) Add(u *LevelMetrics) {
+	m.BytesIn += u.BytesIn
+	m.BytesIngested += u.BytesIngested
+	m.BytesMoved += u.BytesMoved
+	m.BytesRead += u.BytesRead
+	m.BytesWritten += u.BytesWritten
+}
+
+// WriteAmp computes the write amplification for compactions at this
+// level. Computed as BytesWritten / BytesIn.
+func (m *LevelMetrics) WriteAmp() float64 {
+	if m.BytesIn == 0 {
+		return 0
+	}
+	return float64(m.BytesWritten) / float64(m.BytesIn)
+}
+
+func (m *LevelMetrics) format(buf *bytes.Buffer) {
+	fmt.Fprintf(buf, "%6d %7s %7.2f %7s %7s %7s %7s %7s %7.1f\n",
+		m.NumFiles,
+		humanize.Uint64(m.Size),
+		m.Score,
+		humanize.Uint64(m.BytesIn),
+		humanize.Uint64(m.BytesIngested),
+		humanize.Uint64(m.BytesMoved),
+		humanize.Uint64(m.BytesRead),
+		humanize.Uint64(m.BytesWritten),
+		m.WriteAmp(),
+	)
+}
+
+// VersionMetrics holds metrics for each level.
+type VersionMetrics struct {
+	WAL struct {
+		// Number of live WAL files.
+		Files uint64
+		// Size of the live data in the WAL files. Note that with WAL file
+		// recycling this is less than the actual on-disk size of the WAL files.
+		Size uint64
+		// Number of logical bytes written to the WAL.
+		BytesIn uint64
+		// Number of bytes written to the WAL.
+		BytesWritten uint64
+	}
+	Levels [numLevels]LevelMetrics
+}
+
+func (m *VersionMetrics) formatWAL(buf *bytes.Buffer) {
+	var writeAmp float64
+	if m.WAL.BytesIn > 0 {
+		writeAmp = float64(m.WAL.BytesWritten) / float64(m.WAL.BytesIn)
+	}
+	fmt.Fprintf(buf, "  WAL %6d %7s       - %7s       -       -       - %7s %7.1f\n",
+		m.WAL.Files,
+		humanize.Uint64(m.WAL.Size),
+		humanize.Uint64(m.WAL.BytesIn),
+		humanize.Uint64(m.WAL.BytesWritten),
+		writeAmp)
+}
+
+// Pretty-print the metrics, showing a line for the WAL, a line per-level, and
+// a total:
+//
+//   level__files____size___score______in__ingest____move____read___write___w-amp
+//     WAL      1    53 M       -   744 M       -       -       -   765 M     1.0
+//       0      6   285 M    3.00   712 M     0 B     0 B     0 B   707 M     1.0
+//       1      0     0 B    0.00     0 B     0 B     0 B     0 B     0 B     0.0
+//       2      0     0 B    0.00     0 B     0 B     0 B     0 B     0 B     0.0
+//       3      0     0 B    0.00     0 B     0 B     0 B     0 B     0 B     0.0
+//       4      0     0 B    0.00     0 B     0 B     0 B     0 B     0 B     0.0
+//       5     80   312 M    1.09   328 M     0 B     0 B   580 M   580 M     1.8
+//       6     23   110 M    0.35   110 M     0 B     0 B   146 M   146 M     1.3
+//   total    109   706 M    0.00   765 M     0 B     0 B   726 M   2.1 G     2.9
+//
+// The WAL "in" metric is the size of the batches written to the WAL. The WAL
+// "write" metric is the size of the physical data written to the WAL which
+// includes record fragment overhead. Write amplification is computed as
+// bytes-written / bytes-in, except for the total row where bytes-in is
+// replaced with WAL-bytes-written + bytes-ingested.
+func (m *VersionMetrics) String() string {
+	var buf bytes.Buffer
+	var total LevelMetrics
+	fmt.Fprintf(&buf, "level__files____size___score______in__ingest____move____read___write___w-amp\n")
+	m.formatWAL(&buf)
+	for level := 0; level < numLevels; level++ {
+		l := &m.Levels[level]
+		fmt.Fprintf(&buf, "%5d ", level)
+		l.format(&buf)
+		total.Add(l)
+		total.NumFiles += l.NumFiles
+		total.Size += l.Size
+	}
+	// Compute total bytes-in as the bytes written to the WAL + bytes ingested
+	total.BytesIn = m.WAL.BytesWritten + total.BytesIngested
+	// Add the total bytes-in to the total bytes-written. This is to account for
+	// the bytes written to the log and bytes written externally and then
+	// ingested.
+	total.BytesWritten += total.BytesIn
+	fmt.Fprintf(&buf, "total ")
+	total.format(&buf)
+	return buf.String()
+}
diff --git a/version_edit.go b/version_edit.go
index 60a13bb..9eb6ffa 100644
--- a/version_edit.go
+++ b/version_edit.go
@@ -74,6 +74,7 @@ type versionEdit struct {
 	lastSequence   uint64
 	deletedFiles   map[deletedFileEntry]bool // A set of deletedFileEntry values.
 	newFiles       []newFileEntry
+	metrics        map[int]*LevelMetrics // level -> metrics update
 }
 
 func (v *versionEdit) decode(r io.Reader) error {
diff --git a/version_set.go b/version_set.go
index d874810..d074832 100644
--- a/version_set.go
+++ b/version_set.go
@@ -37,6 +37,8 @@ type versionSet struct {
 	versions versionList
 	picker   *compactionPicker
 
+	metrics VersionMetrics
+
 	obsoleteTables    []uint64
 	obsoleteManifests []uint64
 	obsoleteOptions   []uint64
@@ -255,6 +257,17 @@ func (vs *versionSet) logAndApply(ve *versionEdit, dir vfs.File) error {
 		vs.manifestFileNumber = newManifestFileNumber
 	}
 	vs.picker = picker
+
+	if ve.metrics != nil {
+		for level, update := range ve.metrics {
+			vs.metrics.Levels[level].Add(update)
+		}
+	}
+	for i := range vs.metrics.Levels {
+		l := &vs.metrics.Levels[i]
+		l.NumFiles = uint64(len(newVersion.files[i]))
+		l.Size = uint64(totalSize(newVersion.files[i]))
+	}
 	return nil
 }
 


Test patch:
diff --git a/cmd/pebble/test.go b/cmd/pebble/test.go
index 3154824..7405e50 100644
--- a/cmd/pebble/test.go
+++ b/cmd/pebble/test.go
@@ -240,6 +240,7 @@ func runTest(dir string, t test) {
 
 		case <-done:
 			t.done(time.Since(start))
+			fmt.Printf("%s", db.Metrics())
 			return
 		}
 	}
diff --git a/event_listener_test.go b/event_listener_test.go
index b9e0991..ad89dbe 100644
--- a/event_listener_test.go
+++ b/event_listener_test.go
@@ -183,6 +183,9 @@ func TestEventListener(t *testing.T) {
 			}
 			return buf.String()
 
+		case "metrics":
+			return d.Metrics().String()
+
 		default:
 			return fmt.Sprintf("unknown command: %s", td.Cmd)
 		}
diff --git a/internal/datadriven/datadriven_test.go b/internal/datadriven/datadriven_test.go
index d2a964e..dcbccec 100644
--- a/internal/datadriven/datadriven_test.go
+++ b/internal/datadriven/datadriven_test.go
@@ -17,13 +17,9 @@ package datadriven
 import (
 	"fmt"
 	"testing"
-
-	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 )
 
 func TestDataDriven(t *testing.T) {
-	defer leaktest.AfterTest(t)()
-
 	input := `
 # NB: we allow duplicate args. It's unclear at this time whether this is useful,
 # either way, ScanArgs simply picks the first occurrence.
diff --git a/testdata/event_listener b/testdata/event_listener
index e0569ca..ca07702 100644
--- a/testdata/event_listener
+++ b/testdata/event_listener
@@ -81,3 +81,16 @@ sync: db/CURRENT.000014.dbtmp
 rename: db/CURRENT.000014.dbtmp -> db/CURRENT
 sync: db
 #7: table ingested
+
+metrics
+----
+level__files____size___score______in__ingest____move____read___write___w-amp
+  WAL      0    27 B       -    32 B       -       -       -    81 B     2.5
+    0      0     0 B    0.00    54 B     0 B     0 B     0 B   1.9 K    35.5
+    1      0     0 B    0.00     0 B     0 B     0 B     0 B     0 B     0.0
+    2      0     0 B    0.00     0 B     0 B     0 B     0 B     0 B     0.0
+    3      0     0 B    0.00     0 B     0 B     0 B     0 B     0 B     0.0
+    4      0     0 B    0.00     0 B     0 B     0 B     0 B     0 B     0.0
+    5      1   959 B    0.00     0 B   959 B     0 B     0 B     0 B     0.0
+    6      1   959 B    0.00   1.9 K     0 B     0 B   1.9 K   959 B     0.5
+total      2   1.9 K    0.00   1.0 K   959 B     0 B   1.9 K   3.8 K     3.8