vfs/errorfs: add facilities for error injection in datadriven tests

We frequently use the errorfs package to inject errors into filesystem
operations to test error code paths. This commit builds off this package to
support encoding error injection conditions within datadriven tests through
parsing a small DSL. Future work will build off this to test handling of I/O
errors during iteration.

Informs cockroachdb#1115.
Informs cockroachdb#2994.

compaction_test.go

@@ -2898,7 +2898,7 @@ func TestCompactionErrorCleanup(t *testing.T) {
 	)
 
 	mem := vfs.NewMem()
-	ii := errorfs.OnIndex(math.MaxInt32) // start disabled
+	ii := errorfs.OnIndex(math.MaxInt32, errorfs.Always()) // start disabled
 	opts := (&Options{
 		FS:     errorfs.Wrap(mem, ii),
 		Levels: make([]LevelOptions, numLevels),

data_test.go

@@ -18,6 +18,7 @@ import (
 
 	"github.com/cockroachdb/datadriven"
 	"github.com/cockroachdb/errors"
+	"github.com/cockroachdb/pebble/bloom"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/humanize"
 	"github.com/cockroachdb/pebble/internal/keyspan"
@@ -29,6 +30,7 @@ import (
 	"github.com/cockroachdb/pebble/objstorage/remote"
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/vfs"
+	"github.com/cockroachdb/pebble/vfs/errorfs"
 	"github.com/stretchr/testify/require"
 )
 
@@ -1447,3 +1449,69 @@ func runLSMCmd(td *datadriven.TestData, d *DB) string {
 	}
 	return d.mu.versions.currentVersion().String()
 }
+
+func parseDBOptionsArgs(opts *Options, args []datadriven.CmdArg) error {
+	for _, cmdArg := range args {
+		switch cmdArg.Key {
+		case "inject-errors":
+			injs := make([]errorfs.Injector, len(cmdArg.Vals))
+			for i := 0; i < len(cmdArg.Vals); i++ {
+				inj, err := errorfs.ParseInjectorFromDSL(cmdArg.Vals[i])
+				if err != nil {
+					return err
+				}
+				injs[i] = inj
+			}
+			opts.FS = errorfs.Wrap(opts.FS, errorfs.Any(injs...))
+		case "format-major-version":
+			v, err := strconv.Atoi(cmdArg.Vals[0])
+			if err != nil {
+				return err
+			}
+			// Override the DB version.
+			opts.FormatMajorVersion = FormatMajorVersion(v)
+		case "block-size":
+			v, err := strconv.Atoi(cmdArg.Vals[0])
+			if err != nil {
+				return err
+			}
+			for i := range opts.Levels {
+				opts.Levels[i].BlockSize = v
+			}
+		case "index-block-size":
+			v, err := strconv.Atoi(cmdArg.Vals[0])
+			if err != nil {
+				return err
+			}
+			for i := range opts.Levels {
+				opts.Levels[i].IndexBlockSize = v
+			}
+		case "target-file-size":
+			v, err := strconv.Atoi(cmdArg.Vals[0])
+			if err != nil {
+				return err
+			}
+			for i := range opts.Levels {
+				opts.Levels[i].TargetFileSize = int64(v)
+			}
+		case "bloom-bits-per-key":
+			v, err := strconv.Atoi(cmdArg.Vals[0])
+			if err != nil {
+				return err
+			}
+			fp := bloom.FilterPolicy(v)
+			opts.Filters = map[string]FilterPolicy{fp.Name(): fp}
+			for i := range opts.Levels {
+				opts.Levels[i].FilterPolicy = fp
+			}
+		case "merger":
+			switch cmdArg.Vals[0] {
+			case "appender":
+				opts.Merger = base.DefaultMerger
+			default:
+				return errors.Newf("unrecognized Merger %q\n", cmdArg.Vals[0])
+			}
+		}
+	}
+	return nil
+}

error_test.go

@@ -136,7 +136,7 @@ func TestErrors(t *testing.T) {
 
 	errorCounts := make(map[string]int)
 	for i := int32(0); ; i++ {
-		fs := errorfs.Wrap(vfs.NewMem(), errorfs.OnIndex(i))
+		fs := errorfs.Wrap(vfs.NewMem(), errorfs.OnIndex(i, errorfs.Always()))
 		err := run(fs)
 		if err == nil {
 			t.Logf("success %d\n", i)
@@ -166,7 +166,7 @@ func TestErrors(t *testing.T) {
 func TestRequireReadError(t *testing.T) {
 	run := func(formatVersion FormatMajorVersion, index int32) (err error) {
 		// Perform setup with error injection disabled as it involves writes/background ops.
-		inj := errorfs.OnIndex(-1)
+		inj := errorfs.OnIndex(-1, errorfs.Always())
 		fs := errorfs.Wrap(vfs.NewMem(), inj)
 		opts := &Options{
 			FS:                 fs,

ingest_test.go

@@ -386,7 +386,7 @@ func TestIngestLinkFallback(t *testing.T) {
 	src, err := mem.Create("source")
 	require.NoError(t, err)
 
-	opts := &Options{FS: errorfs.Wrap(mem, errorfs.OnIndex(1))}
+	opts := &Options{FS: errorfs.Wrap(mem, errorfs.OnIndex(1, errorfs.Always()))}
 	opts.EnsureDefaults().WithFSDefaults()
 	objSettings := objstorageprovider.DefaultSettings(opts.FS, "")
 	// Prevent the provider from listing the dir (where we may get an injected error).
@@ -2182,7 +2182,7 @@ func TestIngestError(t *testing.T) {
 		require.NoError(t, w.Set([]byte("d"), nil))
 		require.NoError(t, w.Close())
 
-		inj := errorfs.OnIndex(-1)
+		inj := errorfs.OnIndex(-1, errorfs.Always())
 		d, err := Open("", &Options{
 			FS:                    errorfs.Wrap(mem, inj),
 			Logger:                panicLogger{},

iterator_histories_test.go

@@ -13,8 +13,6 @@ import (
 
 	"github.com/cockroachdb/datadriven"
 	"github.com/cockroachdb/errors"
-	"github.com/cockroachdb/pebble/bloom"
-	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/sstable"
@@ -41,70 +39,22 @@ func TestIterHistories(t *testing.T) {
 			iters[name] = it
 			return it
 		}
+		var opts *Options
 		parseOpts := func(td *datadriven.TestData) (*Options, error) {
-			opts := &Options{
+			opts = &Options{
 				FS:                 vfs.NewMem(),
 				Comparer:           testkeys.Comparer,
 				FormatMajorVersion: FormatRangeKeys,
 				BlockPropertyCollectors: []func() BlockPropertyCollector{
 					sstable.NewTestKeysBlockPropertyCollector,
 				},
 			}
+
 			opts.DisableAutomaticCompactions = true
 			opts.EnsureDefaults()
 			opts.WithFSDefaults()
-
-			for _, cmdArg := range td.CmdArgs {
-				switch cmdArg.Key {
-				case "format-major-version":
-					v, err := strconv.Atoi(cmdArg.Vals[0])
-					if err != nil {
-						return nil, err
-					}
-					// Override the DB version.
-					opts.FormatMajorVersion = FormatMajorVersion(v)
-				case "block-size":
-					v, err := strconv.Atoi(cmdArg.Vals[0])
-					if err != nil {
-						return nil, err
-					}
-					for i := range opts.Levels {
-						opts.Levels[i].BlockSize = v
-					}
-				case "index-block-size":
-					v, err := strconv.Atoi(cmdArg.Vals[0])
-					if err != nil {
-						return nil, err
-					}
-					for i := range opts.Levels {
-						opts.Levels[i].IndexBlockSize = v
-					}
-				case "target-file-size":
-					v, err := strconv.Atoi(cmdArg.Vals[0])
-					if err != nil {
-						return nil, err
-					}
-					for i := range opts.Levels {
-						opts.Levels[i].TargetFileSize = int64(v)
-					}
-				case "bloom-bits-per-key":
-					v, err := strconv.Atoi(cmdArg.Vals[0])
-					if err != nil {
-						return nil, err
-					}
-					fp := bloom.FilterPolicy(v)
-					opts.Filters = map[string]FilterPolicy{fp.Name(): fp}
-					for i := range opts.Levels {
-						opts.Levels[i].FilterPolicy = fp
-					}
-				case "merger":
-					switch cmdArg.Vals[0] {
-					case "appender":
-						opts.Merger = base.DefaultMerger
-					default:
-						return nil, errors.Newf("unrecognized Merger %q\n", cmdArg.Vals[0])
-					}
-				}
+			if err := parseDBOptionsArgs(opts, td.CmdArgs); err != nil {
+				return nil, err
 			}
 			return opts, nil
 		}
@@ -128,10 +78,11 @@ func TestIterHistories(t *testing.T) {
 		datadriven.RunTest(t, path, func(t *testing.T, td *datadriven.TestData) string {
 			switch td.Cmd {
 			case "define":
+				var err error
 				if err := cleanup(); err != nil {
 					return err.Error()
 				}
-				opts, err := parseOpts(td)
+				opts, err = parseOpts(td)
 				if err != nil {
 					return err.Error()
 				}
@@ -140,12 +91,23 @@ func TestIterHistories(t *testing.T) {
 					return err.Error()
 				}
 				return runLSMCmd(td, d)
-
+			case "reopen":
+				var err error
+				if err := cleanup(); err != nil {
+					return err.Error()
+				}
+				if err := parseDBOptionsArgs(opts, td.CmdArgs); err != nil {
+					return err.Error()
+				}
+				d, err = Open("", opts)
+				require.NoError(t, err)
+				return ""
 			case "reset":
+				var err error
 				if err := cleanup(); err != nil {
 					return err.Error()
 				}
-				opts, err := parseOpts(td)
+				opts, err = parseOpts(td)
 				if err != nil {
 					return err.Error()
 				}

sstable/reader_test.go

@@ -647,7 +647,7 @@ func TestInjectedErrors(t *testing.T) {
 			f, err := vfs.Default.Open(filepath.FromSlash(prebuiltSST))
 			require.NoError(t, err)
 
-			r, err := newReader(errorfs.WrapFile(f, errorfs.OnIndex(int32(i))), ReaderOptions{})
+			r, err := newReader(errorfs.WrapFile(f, errorfs.OnIndex(int32(i), errorfs.Always())), ReaderOptions{})
 			if err != nil {
 				return firstError(err, f.Close())
 			}