vfs: extend disk-health checking to filesystem metadata operations

Previously, disk stalls while syncing a filesystem directory were undetected,
because the health-checking filesystem's OpenDir method did not wrap the
directory handle.

Additionally, disk stalls during all filesystem metadata operations (eg, file
creation, removal, rename, etc) were undetected. These operations are performed
directly on a vfs.FS (not a vfs.File), and require an adjusted implementation
to accommodate concurrent, write-oriented operations.

This commit implements a scheme in which filesystem metadata operations lock a
vfs.FS-wide mutex to acquire a 'slot' into which they record their start time.
When an operation completes, the operation atomically zeroes its slot without
reacquiring the mutex. A long-running goroutine listens on a ticker,
periodically acquiring the vfs.FS mutex and scanning the slots for start times
older than a threshold.

The long-running goroutine is launched lazily by the first filesystem operation
to acquire the mutex. A new `io.Closer` return value of `WithDiskHealthChecks`
may be used to stop the long-running goroutine.

Fix cockroachdb#1643.

db.go

@@ -1241,6 +1241,11 @@ func (d *DB) Close() error {
 	d.deleters.Wait()
 	d.compactionSchedulers.Wait()
 	d.mu.Lock()
+
+	// If the options include a closer to 'close' the filesystem, close it.
+	if d.opts.private.fsCloser != nil {
+		d.opts.private.fsCloser.Close()
+	}
 	return err
 }
 

options.go

@@ -7,6 +7,7 @@ package pebble
 import (
 	"bytes"
 	"fmt"
+	"io"
 	"runtime"
 	"strconv"
 	"strings"
@@ -703,6 +704,16 @@ type Options struct {
 		// default is 1 MB/s. Currently disabled as this option has no effect while
 		// private.enablePacing is false.
 		minFlushRate int
+
+		// fsCloser holds a closer that should be invoked after a DB using these
+		// Options is closed. This is used to automatically stop the
+		// long-running goroutine associated with the disk-health-checking FS.
+		// See the initialization of FS in EnsureDefaults. Note that care has
+		// been taken to ensure that it is still safe to continue using the FS
+		// after this closer has been invoked. However, if write operations
+		// against the FS are made after the DB is closed, the FS may leak a
+		// goroutine indefinitely.
+		fsCloser io.Closer
 	}
 }
 
@@ -826,7 +837,7 @@ func (o *Options) EnsureDefaults() *Options {
 	}
 
 	if o.FS == nil {
-		o.FS = vfs.WithDiskHealthChecks(vfs.Default, 5*time.Second,
+		o.FS, o.private.fsCloser = vfs.WithDiskHealthChecks(vfs.Default, 5*time.Second,
 			func(name string, duration time.Duration) {
 				o.EventListener.DiskSlow(DiskSlowInfo{
 					Path:     name,