diff --git a/cli/storage_filesystem.go b/cli/storage_filesystem.go index 38cc5bcc7..1c770ce5e 100644 --- a/cli/storage_filesystem.go +++ b/cli/storage_filesystem.go @@ -36,6 +36,7 @@ func (c *storageFilesystemFlags) setup(_ storageProviderServices, cmd *kingpin.C cmd.Flag("file-mode", "File mode for newly created files (0600)").PlaceHolder("MODE").StringVar(&c.connectFileMode) cmd.Flag("dir-mode", "Mode of newly directory files (0700)").PlaceHolder("MODE").StringVar(&c.connectDirMode) cmd.Flag("flat", "Use flat directory structure").BoolVar(&c.connectFlat) + cmd.Flag("list-parallelism", "Set list parallelism").Hidden().IntVar(&c.options.ListParallelism) } func (c *storageFilesystemFlags) connect(ctx context.Context, isNew bool) (blob.Storage, error) { diff --git a/cli/storage_rclone.go b/cli/storage_rclone.go index 384d93a05..7e1515ef4 100644 --- a/cli/storage_rclone.go +++ b/cli/storage_rclone.go @@ -26,6 +26,7 @@ func (c *storageRcloneFlags) setup(_ storageProviderServices, cmd *kingpin.CmdCl cmd.Flag("embed-rclone-config", "Embed the provider RClone config").ExistingFileVar(&c.embedRCloneConfigFile) cmd.Flag("rclone-debug", "Log rclone output").Hidden().BoolVar(&c.opt.Debug) cmd.Flag("rclone-nowait-for-transfers", "Don't wait for transfers when closing storage").Hidden().BoolVar(&c.opt.NoWaitForTransfers) + cmd.Flag("list-parallelism", "Set list parallelism").Hidden().IntVar(&c.opt.ListParallelism) } func (c *storageRcloneFlags) connect(ctx context.Context, isNew bool) (blob.Storage, error) { diff --git a/cli/storage_sftp.go b/cli/storage_sftp.go index 875ccb948..7f6bb564d 100644 --- a/cli/storage_sftp.go +++ b/cli/storage_sftp.go @@ -35,6 +35,7 @@ func (c *storageSFTPFlags) setup(_ storageProviderServices, cmd *kingpin.CmdClau cmd.Flag("flat", "Use flat directory structure").BoolVar(&c.connectFlat) cmd.Flag("max-connections", "Maximum number of SFTP server connections to establish").Default("1").IntVar(&c.options.MaxConnections) + cmd.Flag("list-parallelism", "Set list parallelism").Hidden().IntVar(&c.options.ListParallelism) } func (c *storageSFTPFlags) getOptions() (*sftp.Options, error) { diff --git a/cli/storage_webdav.go b/cli/storage_webdav.go index 621ff1ea5..2bf40d60f 100644 --- a/cli/storage_webdav.go +++ b/cli/storage_webdav.go @@ -20,6 +20,7 @@ func (c *storageWebDAVFlags) setup(_ storageProviderServices, cmd *kingpin.CmdCl cmd.Flag("flat", "Use flat directory structure").BoolVar(&c.connectFlat) cmd.Flag("webdav-username", "WebDAV username").Envar("KOPIA_WEBDAV_USERNAME").StringVar(&c.options.Username) cmd.Flag("webdav-password", "WebDAV password").Envar("KOPIA_WEBDAV_PASSWORD").StringVar(&c.options.Password) + cmd.Flag("list-parallelism", "Set list parallelism").Hidden().IntVar(&c.options.ListParallelism) } func (c *storageWebDAVFlags) connect(ctx context.Context, isNew bool) (blob.Storage, error) { diff --git a/internal/blobtesting/asserts.go b/internal/blobtesting/asserts.go index 52cd7e490..219735132 100644 --- a/internal/blobtesting/asserts.go +++ b/internal/blobtesting/asserts.go @@ -21,13 +21,10 @@ func AssertGetBlob(ctx context.Context, t *testing.T, s blob.Storage, blobID blo t.Helper() b, err := s.GetBlob(ctx, blobID, 0, -1) - if err != nil { - t.Errorf("GetBlob(%v) returned error %v, expected data: %v", blobID, err, expected) - return - } + require.NoErrorf(t, err, "GetBlob(%v)", blobID) if !bytes.Equal(b, expected) { - t.Errorf("GetBlob(%v) returned %x, but expected %x", blobID, b, expected) + t.Fatalf("GetBlob(%v) returned %x, but expected %x", blobID, b, expected) } half := int64(len(expected) / 2) @@ -37,33 +34,33 @@ func AssertGetBlob(ctx context.Context, t *testing.T, s blob.Storage, blobID blo b, err = s.GetBlob(ctx, blobID, 0, 0) if err != nil { - t.Errorf("GetBlob(%v) returned error %v, expected data: %v", blobID, err, expected) + t.Fatalf("GetBlob(%v) returned error %v, expected data: %v", blobID, err, expected) return } if len(b) != 0 { - t.Errorf("GetBlob(%v) returned non-zero length: %v", blobID, len(b)) + t.Fatalf("GetBlob(%v) returned non-zero length: %v", blobID, len(b)) return } b, err = s.GetBlob(ctx, blobID, 0, half) if err != nil { - t.Errorf("GetBlob(%v) returned error %v, expected data: %v", blobID, err, expected) + t.Fatalf("GetBlob(%v) returned error %v, expected data: %v", blobID, err, expected) return } if !bytes.Equal(b, expected[0:half]) { - t.Errorf("GetBlob(%v) returned %x, but expected %x", blobID, b, expected[0:half]) + t.Fatalf("GetBlob(%v) returned %x, but expected %x", blobID, b, expected[0:half]) } b, err = s.GetBlob(ctx, blobID, half, int64(len(expected))-half) if err != nil { - t.Errorf("GetBlob(%v) returned error %v, expected data: %v", blobID, err, expected) + t.Fatalf("GetBlob(%v) returned error %v, expected data: %v", blobID, err, expected) return } if !bytes.Equal(b, expected[len(expected)-int(half):]) { - t.Errorf("GetBlob(%v) returned %x, but expected %x", blobID, b, expected[len(expected)-int(half):]) + t.Fatalf("GetBlob(%v) returned %x, but expected %x", blobID, b, expected[len(expected)-int(half):]) } AssertInvalidOffsetLength(ctx, t, s, blobID, -3, 1) @@ -77,7 +74,7 @@ func AssertInvalidOffsetLength(ctx context.Context, t *testing.T, s blob.Storage t.Helper() if _, err := s.GetBlob(ctx, blobID, offset, length); err == nil { - t.Errorf("GetBlob(%v,%v,%v) did not return error for invalid offset/length", blobID, offset, length) + t.Fatalf("GetBlob(%v,%v,%v) did not return error for invalid offset/length", blobID, offset, length) } } @@ -87,7 +84,7 @@ func AssertGetBlobNotFound(ctx context.Context, t *testing.T, s blob.Storage, bl b, err := s.GetBlob(ctx, blobID, 0, -1) if !errors.Is(err, blob.ErrBlobNotFound) || b != nil { - t.Errorf("GetBlob(%v) returned %v, %v but expected ErrNotFound", blobID, b, err) + t.Fatalf("GetBlob(%v) returned %v, %v but expected ErrNotFound", blobID, b, err) } } @@ -97,7 +94,7 @@ func AssertGetMetadataNotFound(ctx context.Context, t *testing.T, s blob.Storage _, err := s.GetMetadata(ctx, blobID) if !errors.Is(err, blob.ErrBlobNotFound) { - t.Errorf("GetMetadata(%v) returned %v but expected ErrNotFound", blobID, err) + t.Fatalf("GetMetadata(%v) returned %v but expected ErrNotFound", blobID, err) } } @@ -112,15 +109,15 @@ func AssertListResults(ctx context.Context, t *testing.T, s blob.Storage, prefix m2, err := s.GetMetadata(ctx, m.BlobID) if err != nil { - t.Errorf("GetMetadata() failed: %v", err) + t.Fatalf("GetMetadata() failed: %v", err) } if got, want := m2.BlobID, m.BlobID; got != want { - t.Errorf("invalid blob ID on %v: %v, want %v", m.BlobID, got, want) + t.Fatalf("invalid blob ID on %v: %v, want %v", m.BlobID, got, want) } if got, want := m2.Length, m.Length; got != want { - t.Errorf("invalid length on %v: %v, want %v", m.BlobID, got, want) + t.Fatalf("invalid length on %v: %v, want %v", m.BlobID, got, want) } timeDiff := m2.Timestamp.Sub(m.Timestamp) @@ -130,19 +127,19 @@ func AssertListResults(ctx context.Context, t *testing.T, s blob.Storage, prefix // truncated time comparison, because some providers return different precision of time in list vs get if timeDiff > maxTimeDiffBetweenGetAndList { - t.Errorf("invalid timestamp on %v: getmetadata returned %v, list returned %v", m.BlobID, m2.Timestamp, m.Timestamp) + t.Fatalf("invalid timestamp on %v: getmetadata returned %v, list returned %v", m.BlobID, m2.Timestamp, m.Timestamp) } return nil }); err != nil { - t.Errorf("err: %v", err) + t.Fatalf("err: %v", err) } names = sorted(names) want = sorted(want) if !reflect.DeepEqual(names, want) { - t.Errorf("ListBlobs(%v) returned %v, but wanted %v", prefix, names, want) + t.Fatalf("ListBlobs(%v) returned %v, but wanted %v", prefix, names, want) } } @@ -156,7 +153,7 @@ func AssertListResultsIDs(ctx context.Context, t *testing.T, s blob.Storage, pre names = append(names, m.BlobID) return nil }); err != nil { - t.Errorf("err: %v", err) + t.Fatalf("err: %v", err) } require.ElementsMatch(t, names, want) diff --git a/internal/blobtesting/verify.go b/internal/blobtesting/verify.go index e8309977d..7137a731d 100644 --- a/internal/blobtesting/verify.go +++ b/internal/blobtesting/verify.go @@ -5,11 +5,11 @@ import ( "context" "fmt" "os" - "reflect" "testing" "time" "github.com/pkg/errors" + "github.com/stretchr/testify/require" "github.com/kopia/kopia/internal/gather" "github.com/kopia/kopia/repo/blob" @@ -81,8 +81,30 @@ func VerifyStorage(ctx context.Context, t *testing.T, r blob.Storage) { } }) - AssertListResults(ctx, t, r, "", blocks[0].blk, blocks[1].blk, blocks[2].blk, blocks[3].blk, blocks[4].blk) - AssertListResults(ctx, t, r, "ab", blocks[0].blk, blocks[2].blk, blocks[3].blk) + t.Run("ListBlobs", func(t *testing.T) { + errExpected := errors.New("expected error") + + t.Run("ListErrorNoPrefix", func(t *testing.T) { + t.Parallel() + require.ErrorIs(t, r.ListBlobs(ctx, "", func(bm blob.Metadata) error { + return errExpected + }), errExpected) + }) + t.Run("ListErrorWithPrefix", func(t *testing.T) { + t.Parallel() + require.ErrorIs(t, r.ListBlobs(ctx, "ab", func(bm blob.Metadata) error { + return errExpected + }), errExpected) + }) + t.Run("ListNoPrefix", func(t *testing.T) { + t.Parallel() + AssertListResults(ctx, t, r, "", blocks[0].blk, blocks[1].blk, blocks[2].blk, blocks[3].blk, blocks[4].blk) + }) + t.Run("ListWithPrefix", func(t *testing.T) { + t.Parallel() + AssertListResults(ctx, t, r, "ab", blocks[0].blk, blocks[2].blk, blocks[3].blk) + }) + }) t.Run("OverwriteBlobs", func(t *testing.T) { for _, b := range blocks { @@ -91,10 +113,7 @@ func VerifyStorage(ctx context.Context, t *testing.T, r blob.Storage) { t.Run(string(b.blk), func(t *testing.T) { t.Parallel() - if err := r.PutBlob(ctx, b.blk, gather.FromSlice(b.contents)); err != nil { - t.Errorf("can't put blob: %v", err) - } - + require.NoErrorf(t, r.PutBlob(ctx, b.blk, gather.FromSlice(b.contents)), "can't put blob: %v", b) AssertGetBlob(ctx, t, r, b.blk, b.contents) }) } @@ -109,29 +128,25 @@ func VerifyStorage(ctx context.Context, t *testing.T, r blob.Storage) { t.Run(string(b.blk), func(t *testing.T) { t.Parallel() - if err := r.SetTime(ctx, b.blk, ts); errors.Is(err, blob.ErrSetTimeUnsupported) { + err := r.SetTime(ctx, b.blk, ts) + if errors.Is(err, blob.ErrSetTimeUnsupported) { return } + require.NoError(t, err) + md, err := r.GetMetadata(ctx, b.blk) if err != nil { t.Errorf("unable to get blob metadata") } - if got, want := md.Timestamp, ts; !got.Equal(want) { - t.Errorf("invalid time after SetTme(): %vm want %v", got, want) - } + require.True(t, md.Timestamp.Equal(ts), "invalid time after SetTme(): %vm want %v", md.Timestamp, ts) }) } }) - if err := r.DeleteBlob(ctx, blocks[0].blk); err != nil { - t.Errorf("unable to delete block: %v", err) - } - - if err := r.DeleteBlob(ctx, blocks[0].blk); err != nil { - t.Errorf("invalid error when deleting deleted block: %v", err) - } + require.NoError(t, r.DeleteBlob(ctx, blocks[0].blk)) + require.NoError(t, r.DeleteBlob(ctx, blocks[0].blk)) AssertListResults(ctx, t, r, "ab", blocks[2].blk, blocks[3].blk) AssertListResults(ctx, t, r, "", blocks[1].blk, blocks[2].blk, blocks[3].blk, blocks[4].blk) @@ -149,11 +164,7 @@ func AssertConnectionInfoRoundTrips(ctx context.Context, t *testing.T, s blob.St } ci2 := s2.ConnectionInfo() - if !reflect.DeepEqual(ci, ci2) { - t.Errorf("connection info does not round-trip: %v vs %v", ci, ci2) - } + require.Equal(t, ci, ci2) - if err := s2.Close(ctx); err != nil { - t.Errorf("unable to close storage: %v", err) - } + require.NoError(t, s2.Close(ctx)) } diff --git a/internal/retry/retry.go b/internal/retry/retry.go index ec6149f38..d04139df8 100644 --- a/internal/retry/retry.go +++ b/internal/retry/retry.go @@ -14,7 +14,7 @@ var log = logging.GetContextLoggerFunc("retry") var ( maxAttempts = 10 - retryInitialSleepAmount = 1 * time.Second + retryInitialSleepAmount = 100 * time.Millisecond retryMaxSleepAmount = 32 * time.Second ) diff --git a/repo/blob/filesystem/filesystem_options.go b/repo/blob/filesystem/filesystem_options.go index 2ca3cf385..114c7d6b3 100644 --- a/repo/blob/filesystem/filesystem_options.go +++ b/repo/blob/filesystem/filesystem_options.go @@ -13,6 +13,8 @@ type Options struct { FileUID *int `json:"uid,omitempty"` FileGID *int `json:"gid,omitempty"` + + ListParallelism int `json:"listParallelism,omitempty"` } func (fso *Options) fileMode() os.FileMode { diff --git a/repo/blob/filesystem/filesystem_storage.go b/repo/blob/filesystem/filesystem_storage.go index 6c52af17f..b3815be17 100644 --- a/repo/blob/filesystem/filesystem_storage.go +++ b/repo/blob/filesystem/filesystem_storage.go @@ -292,10 +292,11 @@ func New(ctx context.Context, opts *Options) (blob.Storage, error) { return &fsStorage{ sharded.Storage{ - Impl: &fsImpl{Options: *opts}, - RootPath: opts.Path, - Suffix: fsStorageChunkSuffix, - Shards: opts.shards(), + Impl: &fsImpl{Options: *opts}, + RootPath: opts.Path, + Suffix: fsStorageChunkSuffix, + Shards: opts.shards(), + ListParallelism: opts.ListParallelism, }, }, nil } diff --git a/repo/blob/rclone/rclone_options.go b/repo/blob/rclone/rclone_options.go index fb78ff015..54754a5ea 100644 --- a/repo/blob/rclone/rclone_options.go +++ b/repo/blob/rclone/rclone_options.go @@ -11,4 +11,5 @@ type Options struct { NoWaitForTransfers bool `json:"noWaitForTransfers,omitempty"` // when set to true, don't wait for transfers to finish when closing DirectoryShards []int `json:"dirShards"` EmbeddedConfig string `json:"embeddedConfig,omitempty"` + ListParallelism int `json:"listParallelism,omitempty"` } diff --git a/repo/blob/rclone/rclone_storage.go b/repo/blob/rclone/rclone_storage.go index cb2949e56..644547d65 100644 --- a/repo/blob/rclone/rclone_storage.go +++ b/repo/blob/rclone/rclone_storage.go @@ -53,12 +53,6 @@ func (r *rcloneStorage) PutBlob(ctx context.Context, b blob.ID, data blob.Bytes) return errors.Wrap(r.Storage.PutBlob(ctx, b, data), "error writing blob using WebDAV") } -func (r *rcloneStorage) DeleteBlob(ctx context.Context, b blob.ID) error { - atomic.StoreInt32(r.changeCount, 1) - - return errors.Wrap(r.Storage.DeleteBlob(ctx, b), "error deleting blob using WebDAV") -} - func (r *rcloneStorage) ConnectionInfo() blob.ConnectionInfo { return blob.ConnectionInfo{ Type: rcloneStorageType, @@ -290,6 +284,7 @@ func New(ctx context.Context, opt *Options) (blob.Storage, error) { Username: webdavUsername, Password: webdavPassword, TrustedServerCertificateFingerprint: hex.EncodeToString(fingerprintBytes[:]), + ListParallelism: opt.ListParallelism, }) if err != nil { return nil, errors.Wrap(err, "error connecting to webdav storage") diff --git a/repo/blob/rclone/rclone_storage_test.go b/repo/blob/rclone/rclone_storage_test.go index 3f24c89c3..9640e6c51 100644 --- a/repo/blob/rclone/rclone_storage_test.go +++ b/repo/blob/rclone/rclone_storage_test.go @@ -26,7 +26,7 @@ const defaultCleanupAge = time.Hour var rcloneExternalProviders = map[string]string{ "GoogleDrive": "gdrive:/kopia", - // "OneDrive": "onedrive:/kopia", broken + "OneDrive": "onedrive:/kopia", } func mustGetRcloneExeOrSkip(t *testing.T) string { @@ -129,7 +129,6 @@ func TestRCloneStorageInvalidFlags(t *testing.T) { } func TestRCloneProviders(t *testing.T) { - t.Parallel() testutil.ProviderTest(t) var ( @@ -150,6 +149,11 @@ func TestRCloneProviders(t *testing.T) { rcloneArgs = append(rcloneArgs, "--config="+cfg) } + rcloneArgs = append(rcloneArgs, + "--vfs-cache-max-size=100M", + "--vfs-cache-mode=full", + ) + if len(rcloneArgs)+len(embeddedConfig) == 0 { t.Skipf("Either KOPIA_RCLONE_EMBEDDED_CONFIG_B64 or KOPIA_RCLONE_CONFIG_FILE must be provided") } @@ -160,18 +164,26 @@ func TestRCloneProviders(t *testing.T) { rp := rp opt := &rclone.Options{ - RemotePath: rp, - RCloneExe: rcloneExe, - RCloneArgs: rcloneArgs, - EmbeddedConfig: embeddedConfig, + RemotePath: rp, + RCloneExe: rcloneExe, + RCloneArgs: rcloneArgs, + EmbeddedConfig: embeddedConfig, + Debug: true, + ListParallelism: 16, } - t.Run(name, func(t *testing.T) { + t.Run("Cleanup-"+name, func(t *testing.T) { t.Parallel() ctx := testlogging.Context(t) cleanupOldData(ctx, t, opt, defaultCleanupAge) + }) + + t.Run(name, func(t *testing.T) { + t.Parallel() + + ctx := testlogging.Context(t) // we are using shared storage, append a guid so that tests don't collide opt.RemotePath += "/" + uuid.NewString() diff --git a/repo/blob/sftp/sftp_options.go b/repo/blob/sftp/sftp_options.go index d2d0457e6..2ee7a54ac 100644 --- a/repo/blob/sftp/sftp_options.go +++ b/repo/blob/sftp/sftp_options.go @@ -23,6 +23,7 @@ type Options struct { SSHArguments string `json:"sshArguments,omitempty"` DirectoryShards []int `json:"dirShards"` + ListParallelism int `json:"listParallelism,omitempty"` } func (sftpo *Options) shards() []int { diff --git a/repo/blob/sftp/sftp_storage.go b/repo/blob/sftp/sftp_storage.go index 66f9f2513..1a0135d88 100644 --- a/repo/blob/sftp/sftp_storage.go +++ b/repo/blob/sftp/sftp_storage.go @@ -591,10 +591,11 @@ func New(ctx context.Context, opts *Options) (blob.Storage, error) { r := &sftpStorage{ sharded.Storage{ - Impl: impl, - RootPath: opts.Path, - Suffix: fsStorageChunkSuffix, - Shards: opts.shards(), + Impl: impl, + RootPath: opts.Path, + Suffix: fsStorageChunkSuffix, + Shards: opts.shards(), + ListParallelism: opts.ListParallelism, }, } diff --git a/repo/blob/sharded/sharded.go b/repo/blob/sharded/sharded.go index 46a0030a9..5f99dbeda 100644 --- a/repo/blob/sharded/sharded.go +++ b/repo/blob/sharded/sharded.go @@ -9,7 +9,9 @@ import ( "time" "github.com/pkg/errors" + "golang.org/x/sync/errgroup" + "github.com/kopia/kopia/internal/parallelwork" "github.com/kopia/kopia/repo/blob" ) @@ -29,9 +31,10 @@ type Impl interface { type Storage struct { Impl Impl - RootPath string - Suffix string - Shards []int + RootPath string + Suffix string + Shards []int + ListParallelism int } // GetBlob implements blob.Storage. @@ -56,16 +59,29 @@ func (s Storage) makeFileName(blobID blob.ID) string { // ListBlobs implements blob.Storage. func (s Storage) ListBlobs(ctx context.Context, prefix blob.ID, callback func(blob.Metadata) error) error { + pw := parallelwork.NewQueue() + + // channel to which pw will write blob.Metadata, some buf + result := make(chan blob.Metadata, 128) // nolint:gomnd + + finished := make(chan struct{}) + defer close(finished) + var walkDir func(string, string) error walkDir = func(directory string, currentPrefix string) error { + select { + case <-finished: // already finished + return nil + default: + } + entries, err := s.Impl.ReadDir(ctx, directory) if err != nil { return errors.Wrap(err, "error reading directory") } for _, e := range entries { - // nolint:nestif if e.IsDir() { var match bool @@ -77,27 +93,66 @@ func (s Storage) ListBlobs(ctx context.Context, prefix blob.ID, callback func(bl } if match { - if err := walkDir(directory+"/"+e.Name(), currentPrefix+e.Name()); err != nil { - return err - } - } - } else if fullID, ok := s.getBlobIDFromFileName(currentPrefix + e.Name()); ok { - if strings.HasPrefix(string(fullID), string(prefix)) { - if err := callback(blob.Metadata{ - BlobID: fullID, - Length: e.Size(), - Timestamp: e.ModTime(), - }); err != nil { - return err - } + subdir := directory + "/" + e.Name() + subprefix := currentPrefix + e.Name() + + pw.EnqueueFront(ctx, func() error { + return walkDir(subdir, subprefix) + }) } + + continue + } + + fullID, ok := s.getBlobIDFromFileName(currentPrefix + e.Name()) + if !ok { + continue + } + + if !strings.HasPrefix(string(fullID), string(prefix)) { + continue + } + + select { + case result <- blob.Metadata{ + BlobID: fullID, + Length: e.Size(), + Timestamp: e.ModTime(), + }: + case <-finished: } } return nil } - return walkDir(s.RootPath, "") + pw.EnqueueFront(ctx, func() error { + return walkDir(s.RootPath, "") + }) + + par := s.ListParallelism + if par == 0 { + par = 1 + } + + var eg errgroup.Group + + // start populating the channel in parallel + eg.Go(func() error { + defer close(result) + + return errors.Wrap(pw.Process(ctx, par), "error processing directory shards") + }) + + // invoke the callback on the current goroutine until it fails + for bm := range result { + if err := callback(bm); err != nil { + return err + } + } + + // nolint:wrapcheck + return eg.Wait() } // GetMetadata implements blob.Storage. diff --git a/repo/blob/sharded/sharded_test.go b/repo/blob/sharded/sharded_test.go new file mode 100644 index 000000000..281150b74 --- /dev/null +++ b/repo/blob/sharded/sharded_test.go @@ -0,0 +1,53 @@ +package sharded_test + +import ( + "fmt" + "io/ioutil" + "path/filepath" + "testing" + + "github.com/kopia/kopia/internal/blobtesting" + "github.com/kopia/kopia/internal/testlogging" + "github.com/kopia/kopia/internal/testutil" + "github.com/kopia/kopia/repo/blob/filesystem" +) + +func TestShardedFileStorage(t *testing.T) { + t.Parallel() + + ctx := testlogging.Context(t) + + for _, parallel := range []int{0, 1, 4, 16} { + t.Run(fmt.Sprintf("parallel-%v", parallel), func(t *testing.T) { + for _, shardSpec := range [][]int{ + {0}, + {1}, + {3, 3}, + {2}, + {1, 1}, + {1, 2}, + {2, 2, 2}, + } { + path := testutil.TempDirectory(t) + + r, err := filesystem.New(ctx, &filesystem.Options{ + Path: path, + DirectoryShards: shardSpec, + }) + + ioutil.WriteFile(filepath.Join(path, "foreign-file"), []byte{1, 2, 3}, 0600) + + if r == nil || err != nil { + t.Errorf("unexpected result: %v %v", r, err) + } + + blobtesting.VerifyStorage(ctx, t, r) + blobtesting.AssertConnectionInfoRoundTrips(ctx, t, r) + + if err := r.Close(ctx); err != nil { + t.Fatalf("err: %v", err) + } + } + }) + } +} diff --git a/repo/blob/webdav/webdav_options.go b/repo/blob/webdav/webdav_options.go index ea5010bdf..bb5c674dc 100644 --- a/repo/blob/webdav/webdav_options.go +++ b/repo/blob/webdav/webdav_options.go @@ -7,6 +7,7 @@ type Options struct { Username string `json:"username,omitempty"` Password string `json:"password,omitempty" kopia:"sensitive"` TrustedServerCertificateFingerprint string `json:"trustedServerCertificateFingerprint,omitempty"` + ListParallelism int `json:"listParallelism,omitempty"` } func (fso *Options) shards() []int { diff --git a/repo/blob/webdav/webdav_storage.go b/repo/blob/webdav/webdav_storage.go index e908d1168..ee0bce0b6 100644 --- a/repo/blob/webdav/webdav_storage.go +++ b/repo/blob/webdav/webdav_storage.go @@ -214,9 +214,10 @@ func New(ctx context.Context, opts *Options) (blob.Storage, error) { Options: *opts, cli: cli, }, - RootPath: "", - Suffix: fsStorageChunkSuffix, - Shards: opts.shards(), + RootPath: "", + Suffix: fsStorageChunkSuffix, + Shards: opts.shards(), + ListParallelism: opts.ListParallelism, }, })