Repository: petermattis/pebble
Pull request #188 (2019-07-18 21:23:39)

Fix metrics WAL file tracking
The tracking of the number of WAL files was buggy, causing incorrect and
impossible values (very large positive values due to subtracting 1 from
a count of 0). Also noticed that we weren't exposing the number of
obsolete (dead) log files. That count is now exposed via
WAL.ObsoleteFiles.


Patch:
diff --git a/compaction.go b/compaction.go
index f0886ac..2cd3172 100644
--- a/compaction.go
+++ b/compaction.go
@@ -997,7 +997,7 @@ func (d *DB) runCompaction(c *compaction) (
 	var prevBytesIterated uint64
 	var iterCount int
 	totalBytes := d.memTableTotalBytes()
-	refreshDirtyBytesThreshold := uint64(d.opts.MemTableSize * 5/100)
+	refreshDirtyBytesThreshold := uint64(d.opts.MemTableSize * 5 / 100)
 
 	for key, val := iter.First(); key != nil; key, val = iter.Next() {
 		// Slow down memtable flushing to match fill rate.
@@ -1007,7 +1007,7 @@ func (d *DB) runCompaction(c *compaction) (
 			// byte count requires grabbing DB.mu which is expensive.
 			if iterCount >= 1000 || c.bytesIterated > refreshDirtyBytesThreshold {
 				totalBytes = d.memTableTotalBytes()
-				refreshDirtyBytesThreshold = c.bytesIterated + uint64(d.opts.MemTableSize * 5/100)
+				refreshDirtyBytesThreshold = c.bytesIterated + uint64(d.opts.MemTableSize*5/100)
 				iterCount = 0
 			}
 			iterCount++
@@ -1024,7 +1024,7 @@ func (d *DB) runCompaction(c *compaction) (
 			// occur if memtable flushing can keep up with the pace of incoming
 			// writes. If writes come in faster than how fast the memtable can flush,
 			// flushing proceeds at maximum (unthrottled) speed.
-			if dirtyBytes <= uint64(d.opts.MemTableSize * 105/100) {
+			if dirtyBytes <= uint64(d.opts.MemTableSize*105/100) {
 				burst := d.flushLimiter.Burst()
 				for flushAmount > uint64(burst) {
 					err := d.flushLimiter.WaitN(context.Background(), burst)
@@ -1151,6 +1151,7 @@ func (d *DB) scanObsoleteFiles(list []string) {
 	}
 
 	d.mu.log.queue = merge(d.mu.log.queue, obsoleteLogs)
+	d.mu.versions.metrics.WAL.Files += int64(len(obsoleteLogs))
 	d.mu.versions.obsoleteTables = merge(d.mu.versions.obsoleteTables, obsoleteTables)
 	d.mu.versions.obsoleteManifests = merge(d.mu.versions.obsoleteManifests, obsoleteManifests)
 	d.mu.versions.obsoleteOptions = merge(d.mu.versions.obsoleteOptions, obsoleteOptions)
@@ -1178,7 +1179,7 @@ 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))
+			d.mu.versions.metrics.WAL.Files -= int64(len(obsoleteLogs))
 			break
 		}
 	}
diff --git a/db.go b/db.go
index 2d59fdc..05159f9 100644
--- a/db.go
+++ b/db.go
@@ -760,8 +760,10 @@ func (d *DB) AsyncFlush() error {
 // Metrics returns metrics about the database.
 func (d *DB) Metrics() *VersionMetrics {
 	metrics := &VersionMetrics{}
+	recycledLogs := d.logRecycler.count()
 	d.mu.Lock()
 	*metrics = d.mu.versions.metrics
+	metrics.WAL.ObsoleteFiles = int64(recycledLogs)
 	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++ {
diff --git a/log_recycler.go b/log_recycler.go
index 5eb74a2..69fd7bb 100644
--- a/log_recycler.go
+++ b/log_recycler.go
@@ -55,6 +55,12 @@ func (r *logRecycler) peek() uint64 {
 	return r.mu.logNums[0]
 }
 
+func (r *logRecycler) count() int {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return len(r.mu.logNums)
+}
+
 // pop removes the log number at the head of the recycling queue, enforcing
 // that it matches the specifed logNum. An error is returned of the recycling
 // queue is empty or the head log number does not match the specified one.
diff --git a/metrics.go b/metrics.go
index 000af14..2b27992 100644
--- a/metrics.go
+++ b/metrics.go
@@ -15,7 +15,7 @@ import (
 // size of the files, and compaction related metrics.
 type LevelMetrics struct {
 	// The total number of files in the level.
-	NumFiles uint64
+	NumFiles int64
 	// The total size in bytes of the files in the level.
 	Size uint64
 	// The level's compaction score.
@@ -71,7 +71,9 @@ func (m *LevelMetrics) format(buf *bytes.Buffer) {
 type VersionMetrics struct {
 	WAL struct {
 		// Number of live WAL files.
-		Files uint64
+		Files int64
+		// Number of obsolete WAL files.
+		ObsoleteFiles int64
 		// 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
diff --git a/open.go b/open.go
index 6b11e72..3dbb78d 100644
--- a/open.go
+++ b/open.go
@@ -213,6 +213,7 @@ func Open(dirname string, opts *Options) (*DB, error) {
 		PreallocateSize: d.walPreallocateSize(),
 	})
 	d.mu.log.LogWriter = record.NewLogWriter(logFile, ve.logNumber)
+	d.mu.versions.metrics.WAL.Files++
 
 	// Write a new manifest to disk.
 	if err := d.mu.versions.logAndApply(0, &ve, d.dataDir); err != nil {
diff --git a/version_set.go b/version_set.go
index f2e5b92..335dcfe 100644
--- a/version_set.go
+++ b/version_set.go
@@ -289,7 +289,7 @@ func (vs *versionSet) logAndApply(jobID int, ve *versionEdit, dir vfs.File) erro
 	}
 	for i := range vs.metrics.Levels {
 		l := &vs.metrics.Levels[i]
-		l.NumFiles = uint64(len(newVersion.files[i]))
+		l.NumFiles = int64(len(newVersion.files[i]))
 		l.Size = uint64(totalSize(newVersion.files[i]))
 	}
 	return nil


Test patch:
diff --git a/log_recycler_test.go b/log_recycler_test.go
index 9590bb2..2c7933a 100644
--- a/log_recycler_test.go
+++ b/log_recycler_test.go
@@ -74,8 +74,9 @@ func TestLogRecycler(t *testing.T) {
 }
 
 func TestRecycleLogs(t *testing.T) {
+	mem := vfs.NewMem()
 	d, err := Open("", &Options{
-		FS: vfs.NewMem(),
+		FS: mem,
 	})
 	if err != nil {
 		t.Fatal(err)
@@ -86,6 +87,11 @@ func TestRecycleLogs(t *testing.T) {
 		defer d.mu.Unlock()
 		return d.mu.log.queue[len(d.mu.log.queue)-1]
 	}
+	logCount := func() int {
+		d.mu.Lock()
+		defer d.mu.Unlock()
+		return len(d.mu.log.queue)
+	}
 
 	// Flush the memtable a few times, forcing rotation of the WAL. We should see
 	// the recycled logs change as expected.
@@ -108,4 +114,21 @@ func TestRecycleLogs(t *testing.T) {
 	if err := d.Close(); err != nil {
 		t.Fatal(err)
 	}
+
+	d, err = Open("", &Options{
+		FS: mem,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	metrics := d.Metrics()
+	if n := logCount(); n != int(metrics.WAL.Files) {
+		t.Fatalf("expected %d WAL files, but found %d", n, metrics.WAL.Files)
+	}
+	if n := d.logRecycler.count(); n != int(metrics.WAL.ObsoleteFiles) {
+		t.Fatalf("expected %d obsolete WAL files, but found %d", n, metrics.WAL.ObsoleteFiles)
+	}
+	if err := d.Close(); err != nil {
+		t.Fatal(err)
+	}
 }
diff --git a/testdata/event_listener b/testdata/event_listener
index 28f3577..d21ea90 100644
--- a/testdata/event_listener
+++ b/testdata/event_listener
@@ -94,7 +94,7 @@ sync: db
 metrics
 ----
 level__files____size___score______in__ingest____move____read___write___w-amp
-  WAL      0    27 B       -    32 B       -       -       -    81 B     2.5
+  WAL      1    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