metamorphic: use FS-based remote storage

The metamorphic test uses in-mem remote storage, which doesn't work
when starting with an initial state (as in the crossversion tests).

Note that there is code to save the remote storage contents to disk
when the store is in-memory, but they are not read back when used as
initial state.

This commit changes to using FS-based remote storage in `shared`
and `external` subdirs inside the data dir. Note that the store itself
can still be in-memory; the data gets saved automatically with the
store.

We improve the FS-based Storage implementation to sync data and list
objects. We can now allow simulating crashes in the metamorphic test
when shared storage is enabled.

Author: RaduBerinde

metamorphic/options.go

@@ -124,14 +124,12 @@ func parseOptions(
 				return true
 			case "TestOptions.shared_storage_enabled":
 				opts.sharedStorageEnabled = true
-				opts.sharedStorageFS = remote.NewInMem()
 				if opts.Opts.Experimental.CreateOnShared == remote.CreateOnSharedNone {
 					opts.Opts.Experimental.CreateOnShared = remote.CreateOnSharedAll
 				}
 				return true
 			case "TestOptions.external_storage_enabled":
 				opts.externalStorageEnabled = true
-				opts.externalStorageFS = remote.NewInMem()
 				return true
 			case "TestOptions.secondary_cache_enabled":
 				opts.secondaryCacheEnabled = true
@@ -226,7 +224,6 @@ func parseOptions(
 	if opts.Opts.WALFailover != nil {
 		opts.Opts.WALFailover.Secondary.FS = opts.Opts.FS
 	}
-	opts.InitRemoteStorageFactory()
 	opts.Opts.EnsureDefaults()
 	return err
 }
@@ -435,10 +432,8 @@ type TestOptions struct {
 	asyncApplyToDB bool
 	// Enable the use of shared storage.
 	sharedStorageEnabled bool
-	sharedStorageFS      remote.Storage
 	// Enable the use of shared storage for external file ingestion.
 	externalStorageEnabled bool
-	externalStorageFS      remote.Storage
 	// Enables the use of shared replication in TestOptions.
 	useSharedReplicate bool
 	// Enables the use of external replication in TestOptions.
@@ -473,20 +468,6 @@ type TestOptions struct {
 	disableDownloads bool
 }
 
-// InitRemoteStorageFactory initializes Opts.Experimental.RemoteStorage.
-func (testOpts *TestOptions) InitRemoteStorageFactory() {
-	if testOpts.sharedStorageEnabled || testOpts.externalStorageEnabled {
-		m := make(map[remote.Locator]remote.Storage)
-		if testOpts.sharedStorageEnabled {
-			m[""] = testOpts.sharedStorageFS
-		}
-		if testOpts.externalStorageEnabled {
-			m["external"] = testOpts.externalStorageFS
-		}
-		testOpts.Opts.Experimental.RemoteStorage = remote.MakeSimpleFactory(m)
-	}
-}
-
 // CustomOption defines a custom option that configures the behavior of an
 // individual test run. Like all test options, custom options are serialized to
 // the OPTIONS file even if they're not options ordinarily understood by Pebble.
@@ -883,7 +864,6 @@ func RandomOptions(
 		if testOpts.Opts.FormatMajorVersion < pebble.FormatMinForSharedObjects {
 			testOpts.Opts.FormatMajorVersion = pebble.FormatMinForSharedObjects
 		}
-		testOpts.sharedStorageFS = remote.NewInMem()
 		// If shared storage is enabled, pick between writing all files on shared
 		// vs. lower levels only, 50% of the time.
 		testOpts.Opts.Experimental.CreateOnShared = remote.CreateOnSharedAll
@@ -906,7 +886,6 @@ func RandomOptions(
 		if testOpts.Opts.FormatMajorVersion < pebble.FormatSyntheticPrefixSuffix {
 			testOpts.Opts.FormatMajorVersion = pebble.FormatSyntheticPrefixSuffix
 		}
-		testOpts.externalStorageFS = remote.NewInMem()
 	}
 
 	// 75% of the time, randomize value separation parameters.
@@ -939,7 +918,6 @@ func RandomOptions(
 		return testOpts.useDeleteOnlyCompactionExcises
 	}
 	testOpts.disableDownloads = rng.IntN(2) == 0
-	testOpts.InitRemoteStorageFactory()
 	testOpts.Opts.EnsureDefaults()
 	return testOpts
 }

metamorphic/test.go

@@ -5,12 +5,10 @@
 package metamorphic
 
 import (
-	"context"
 	"fmt"
 	"io"
 	"os"
 	"path"
-	"path/filepath"
 	"sort"
 	"strings"
 	"time"
@@ -108,12 +106,6 @@ func (t *Test) init(
 	if numInstances < 1 {
 		numInstances = 1
 	}
-	if t.testOpts.externalStorageEnabled {
-		t.externalStorage = t.testOpts.externalStorageFS
-	} else {
-		t.externalStorage = remote.NewInMem()
-	}
-
 	t.opsWaitOn, t.opsDone = computeSynchronizationPoints(t.ops)
 
 	if t.opts.Cache != nil {
@@ -192,10 +184,7 @@ func (t *Test) init(
 			dir = path.Join(t.dir, fmt.Sprintf("db%d", i+1))
 		}
 		err = t.withRetries(func() error {
-			// Give each DB its own CompactionScheduler.
-			o := *t.opts
-			o.Experimental.CompactionScheduler =
-				pebble.NewConcurrencyLimitSchedulerWithNoPeriodicGrantingForTest()
+			o := t.finalizeOptions(dir)
 			db, err = pebble.Open(dir, &o)
 			return err
 		})
@@ -252,6 +241,45 @@ func (t *Test) init(
 	return nil
 }
 
+// finalizeOptions returns the options that need to be passed to pebble.Open().
+//
+// It initializes t.externalStorage and creates the compaction scheduler and
+// remote storage factory.
+func (t *Test) finalizeOptions(dataDir string) pebble.Options {
+	o := *t.opts
+	// Give each DB its own CompactionScheduler.
+	o.Experimental.CompactionScheduler =
+		pebble.NewConcurrencyLimitSchedulerWithNoPeriodicGrantingForTest()
+
+	// Set up external/shared storage.
+	externalDir := o.FS.PathJoin(dataDir, "external")
+	if err := o.FS.MkdirAll(externalDir, 0755); err != nil {
+		panic(fmt.Sprintf("failed to create directory %q: %s", externalDir, err))
+	}
+	// Even if externalStorageEnabled is false, the test uses externalStorage to
+	// emulate external ingestion.
+	t.externalStorage = remote.NewLocalFS(externalDir, o.FS)
+
+	m := make(map[remote.Locator]remote.Storage)
+	// If we are starting from an initial state (initialStatePath != ""), the
+	// existing store might use shared or external storage, so we set them up
+	// unconditionally.
+	if t.testOpts.sharedStorageEnabled || t.testOpts.initialStatePath != "" {
+		sharedDir := o.FS.PathJoin(dataDir, "shared")
+		if err := o.FS.MkdirAll(sharedDir, 0755); err != nil {
+			panic(fmt.Sprintf("failed to create directory %q: %s", sharedDir, err))
+		}
+		m[""] = remote.NewLocalFS(sharedDir, o.FS)
+	}
+	if t.testOpts.externalStorageEnabled || t.testOpts.initialStatePath != "" {
+		m["external"] = t.externalStorage
+	}
+	if len(m) > 0 {
+		o.Experimental.RemoteStorage = remote.MakeSimpleFactory(m)
+	}
+	return o
+}
+
 func (t *Test) withRetries(fn func() error) error {
 	return withRetries(fn, t.testOpts.RetryPolicy)
 }
@@ -279,13 +307,6 @@ func (t *Test) restartDB(dbID objID) error {
 	if !t.testOpts.strictFS {
 		return nil
 	}
-	if t.testOpts.sharedStorageEnabled {
-		// We simulate a crash by essentially ignoring writes to disk after a
-		// certain point. However, we cannot prevent the process (which didn't
-		// actually crash) from deleting an external object before we call Close().
-		// TODO(radu): perhaps we want all syncs to fail after the "crash" point?
-		return nil
-	}
 	// We can't do this if we have more than one database since they share the
 	// same FS (and we only close/reopen one of them).
 	// TODO(radu): have each database use its own MemFS.
@@ -326,10 +347,7 @@ func (t *Test) restartDB(dbID objID) error {
 		if len(t.dbs) > 1 {
 			dir = path.Join(dir, fmt.Sprintf("db%d", dbID.slot()))
 		}
-		// Give each DB its own CompactionScheduler.
-		o := *t.opts
-		o.Experimental.CompactionScheduler =
-			pebble.NewConcurrencyLimitSchedulerWithNoPeriodicGrantingForTest()
+		o := t.finalizeOptions(dir)
 		t.dbs[dbID.slot()-1], err = pebble.Open(dir, &o)
 		if err != nil {
 			return err
@@ -349,49 +367,6 @@ func (t *Test) saveInMemoryDataInternal() error {
 			return err
 		}
 	}
-	if t.testOpts.sharedStorageEnabled {
-		if err := copyRemoteStorage(t.testOpts.sharedStorageFS, filepath.Join(t.dir, "shared")); err != nil {
-			return err
-		}
-	}
-	if t.testOpts.externalStorageEnabled {
-		if err := copyRemoteStorage(t.testOpts.externalStorageFS, filepath.Join(t.dir, "external")); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-func copyRemoteStorage(fs remote.Storage, outputDir string) error {
-	if err := vfs.Default.MkdirAll(outputDir, 0755); err != nil {
-		return err
-	}
-	objs, err := fs.List("", "")
-	if err != nil {
-		return err
-	}
-	for i := range objs {
-		reader, readSize, err := fs.ReadObject(context.TODO(), objs[i])
-		if err != nil {
-			return err
-		}
-		buf := make([]byte, readSize)
-		if err := reader.ReadAt(context.TODO(), buf, 0); err != nil {
-			return err
-		}
-		outputPath := vfs.Default.PathJoin(outputDir, objs[i])
-		outputFile, err := vfs.Default.Create(outputPath, vfs.WriteCategoryUnspecified)
-		if err != nil {
-			return err
-		}
-		if _, err := outputFile.Write(buf); err != nil {
-			outputFile.Close()
-			return err
-		}
-		if err := outputFile.Close(); err != nil {
-			return err
-		}
-	}
 	return nil
 }
 

objstorage/remote/localfs.go

@@ -7,9 +7,11 @@ package remote
 import (
 	"context"
 	"io"
-	"os"
 	"path"
+	"strings"
 
+	"github.com/cockroachdb/errors"
+	"github.com/cockroachdb/errors/oserror"
 	"github.com/cockroachdb/pebble/vfs"
 )
 
@@ -78,24 +80,66 @@ func (r *localFSReader) Close() error {
 	return nil
 }
 
+func (f *localFSStore) sync() error {
+	file, err := f.vfs.OpenDir(f.dirname)
+	if err != nil {
+		return err
+	}
+	return errors.CombineErrors(file.Sync(), file.Close())
+}
+
+type objWriter struct {
+	vfs.File
+	store *localFSStore
+}
+
+func (w *objWriter) Close() error {
+	if w.File == nil {
+		return nil
+	}
+	err := w.File.Sync()
+	err = errors.CombineErrors(err, w.File.Close())
+	err = errors.CombineErrors(err, w.store.sync())
+	*w = objWriter{}
+	return err
+}
+
 // CreateObject is part of the remote.Storage interface.
 func (s *localFSStore) CreateObject(objName string) (io.WriteCloser, error) {
 	file, err := s.vfs.Create(path.Join(s.dirname, objName), vfs.WriteCategoryUnspecified)
-	return file, err
+	if err != nil {
+		return nil, err
+	}
+	return &objWriter{
+		File:  file,
+		store: s,
+	}, nil
 }
 
 // List is part of the remote.Storage interface.
 func (s *localFSStore) List(prefix, delimiter string) ([]string, error) {
-	// TODO(josh): For the intended use case of localfs.go of running 'pebble bench',
-	// List can always return <nil, nil>, since this indicates a file has only one ref,
-	// and since `pebble bench` implies running in a single-pebble-instance context.
-	// https://github.com/cockroachdb/pebble/blob/a9a079d4fb6bf4a9ebc52e4d83a76ad4cbf676cb/objstorage/objstorageprovider/shared.go#L292
-	return nil, nil
+	if delimiter != "" {
+		panic("delimiter unimplemented")
+	}
+	files, err := s.vfs.List(s.dirname)
+	if err != nil {
+		return nil, err
+	}
+	res := make([]string, 0, len(files))
+	for _, name := range files {
+		if strings.HasPrefix(name, prefix) {
+			res = append(res, name)
+		}
+	}
+	return res, nil
 }
 
 // Delete is part of the remote.Storage interface.
 func (s *localFSStore) Delete(objName string) error {
-	return s.vfs.Remove(path.Join(s.dirname, objName))
+	if err := s.vfs.Remove(path.Join(s.dirname, objName)); err != nil {
+		return err
+	}
+	return s.sync()
 }
 
 // Size is part of the remote.Storage interface.
@@ -114,5 +158,5 @@ func (s *localFSStore) Size(objName string) (int64, error) {
 
 // IsNotExistError is part of the remote.Storage interface.
 func (s *localFSStore) IsNotExistError(err error) bool {
-	return err == os.ErrNotExist
+	return oserror.IsNotExist(err)
 }