decomposed fs: change blobstore layout (#2763)

* change decomposed fs blobstore layout for oCIS and S3NG

* update mocks

* add changelog

* fix blobstore tests

* fix sonarcloud code fence language

changelog/unreleased/change-ocis-s3ng-fs-blob-layout.md

@@ -0,0 +1,80 @@
+Change: Change the oCIS and S3NG storage driver blob store layout
+
+We've optimized the oCIS and S3NG storage driver blob store layout.
+
+For the oCIS storage driver, blobs will now be stored inside the folder
+of a space, next to the nodes. This allows admins to easily archive, backup and restore
+spaces as a whole with UNIX tooling. We also moved from a single folder for blobs to
+multiple folders for blobs, to make the filesystem interactions more performant for
+large numbers of files.
+
+The previous layout on disk looked like this:
+
+```markdown
+|-- spaces
+| |-- ..
+| | |-- ..
+| |-- xx
+| |-- xxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx <- partitioned space id
+| |-- nodes
+| |-- ..
+| |-- xx
+| |-- xx
+| |-- xx
+| |-- xx
+| |-- -xxxx-xxxx-xxxx-xxxxxxxxxxxx <- partitioned node id
+|-- blobs
+ |-- ..
+ |-- xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx <- blob id
+```
+
+Now it looks like this:
+
+```markdown
+|-- spaces
+| |-- ..
+| | |-- ..
+ |-- xx
+ |-- xxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx <- partitioned space id
+ |-- nodes
+ | |-- ..
+ | |-- xx
+ | |-- xx
+ | |-- xx
+ | |-- xx
+ | |-- -xxxx-xxxx-xxxx-xxxxxxxxxxxx <- partitioned node id
+ |-- blobs
+ |-- ..
+ |-- xx
+ |-- xx
+ |-- xx
+ |-- xx
+ |-- -xxxx-xxxx-xxxx-xxxxxxxxxxxx <- partitioned blob id
+```
+
+For the S3NG storage driver, blobs will now be prefixed with the space id and
+also a part of the blob id will be used as prefix. This creates a better prefix partitioning
+and mitigates S3 api performance drops for large buckets (https://aws.amazon.com/de/premiumsupport/knowledge-center/s3-prefix-nested-folders-difference/).
+
+The previous S3 bucket (blobs only looked like this):
+
+```markdown
+|-- ..
+|-- xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx <- blob id
+```
+
+Now it looks like this:
+
+```markdown
+|-- ..
+|-- xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx <- space id
+ |-- ..
+ |-- xx
+ |-- xx
+ |-- xx
+ |-- xx
+ |-- -xxxx-xxxx-xxxx-xxxxxxxxxxxx <- partitioned blob id
+```
+
+https://github.com/cs3org/reva/pull/2763
+https://github.com/owncloud/ocis/issues/3557

pkg/storage/fs/ocis/blobstore/blobstore.go

@@ -24,6 +24,8 @@ import (
  "os"
  "path/filepath"
 
+ "github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/lookup"
+ "github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/node"
  "github.com/pkg/errors"
 )
 
@@ -45,39 +47,50 @@ func New(root string) (*Blobstore, error) {
 }
 
 // Upload stores some data in the blobstore under the given key
-func (bs *Blobstore) Upload(key string, data io.Reader) error {
- f, err := os.OpenFile(bs.path(key), os.O_CREATE|os.O_WRONLY, 0700)
+func (bs *Blobstore) Upload(node *node.Node, data io.Reader) error {
+
+ // ensure parent path exists
+ if err := os.MkdirAll(filepath.Dir(bs.path(node)), 0700); err != nil {
+ return errors.Wrap(err, "Decomposedfs: oCIS blobstore: error creating parent folders for blob")
+ }
+
+ f, err := os.OpenFile(bs.path(node), os.O_CREATE|os.O_WRONLY, 0700)
  if err != nil {
- return errors.Wrapf(err, "could not open blob '%s' for writing", key)
+ return errors.Wrapf(err, "could not open blob '%s' for writing", bs.path(node))
  }
 
  w := bufio.NewWriter(f)
  _, err = w.ReadFrom(data)
  if err != nil {
- return errors.Wrapf(err, "could not write blob '%s'", key)
+ return errors.Wrapf(err, "could not write blob '%s'", bs.path(node))
  }
 
  return w.Flush()
 }
 
 // Download retrieves a blob from the blobstore for reading
-func (bs *Blobstore) Download(key string) (io.ReadCloser, error) {
- file, err := os.Open(bs.path(key))
+func (bs *Blobstore) Download(node *node.Node) (io.ReadCloser, error) {
+ file, err := os.Open(bs.path(node))
  if err != nil {
- return nil, errors.Wrapf(err, "could not read blob '%s'", key)
+ return nil, errors.Wrapf(err, "could not read blob '%s'", bs.path(node))
  }
  return file, nil
 }
 
 // Delete deletes a blob from the blobstore
-func (bs *Blobstore) Delete(key string) error {
- err := os.Remove(bs.path(key))
+func (bs *Blobstore) Delete(node *node.Node) error {
+ err := os.Remove(bs.path(node))
  if err != nil {
- return errors.Wrapf(err, "could not delete blob '%s'", key)
+ return errors.Wrapf(err, "could not delete blob '%s'", bs.path(node))
  }
  return nil
 }
 
-func (bs *Blobstore) path(key string) string {
- return filepath.Join(bs.root, filepath.Clean(filepath.Join("/", key)))
+func (bs *Blobstore) path(node *node.Node) string {
+ return filepath.Join(
+ bs.root,
+ filepath.Clean(filepath.Join(
+ "/", "spaces", lookup.Pathify(node.SpaceID, 1, 2), "blobs", lookup.Pathify(node.BlobID, 4, 2)),
+ ),
+ )
 }

pkg/storage/fs/ocis/blobstore/blobstore_test.go

@@ -25,6 +25,7 @@ import (
  "path"
 
  "github.com/cs3org/reva/v2/pkg/storage/fs/ocis/blobstore"
+ "github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/node"
  "github.com/cs3org/reva/v2/tests/helpers"
 
  . "github.com/onsi/ginkgo/v2"
@@ -34,7 +35,7 @@ import (
 var _ = Describe("Blobstore", func() {
  var (
  tmpRoot string
- key string
+ blobNode *node.Node
  blobPath string
  data []byte
 
@@ -47,10 +48,13 @@ var _ = Describe("Blobstore", func() {
  Expect(err).ToNot(HaveOccurred())
 
  data = []byte("1234567890")
- key = "foo"
- blobPath = path.Join(tmpRoot, "blobs", key)
+ blobNode = &node.Node{
+ SpaceID: "wonderfullspace",
+ BlobID: "huuuuugeblob",
+ }
+ blobPath = path.Join(tmpRoot, "spaces", "wo", "nderfullspace", "blobs", "hu", "uu", "uu", "ge", "blob")
 
- bs, err = blobstore.New(path.Join(tmpRoot, "blobs"))
+ bs, err = blobstore.New(path.Join(tmpRoot))
  Expect(err).ToNot(HaveOccurred())
  })
 
@@ -61,13 +65,13 @@ var _ = Describe("Blobstore", func() {
  })
 
  It("creates the root directory if it doesn't exist", func() {
- _, err := os.Stat(path.Join(tmpRoot, "blobs"))
+ _, err := os.Stat(path.Join(tmpRoot))
  Expect(err).ToNot(HaveOccurred())
  })
 
  Describe("Upload", func() {
  It("writes the blob", func() {
- err := bs.Upload(key, bytes.NewReader(data))
+ err := bs.Upload(blobNode, bytes.NewReader(data))
  Expect(err).ToNot(HaveOccurred())
 
  writtenBytes, err := ioutil.ReadFile(blobPath)
@@ -78,12 +82,13 @@ var _ = Describe("Blobstore", func() {
 
  Context("with an existing blob", func() {
  BeforeEach(func() {
+ Expect(os.MkdirAll(path.Dir(blobPath), 0700)).To(Succeed())
  Expect(ioutil.WriteFile(blobPath, data, 0700)).To(Succeed())
  })
 
  Describe("Download", func() {
  It("cleans the key", func() {
- reader, err := bs.Download("../" + key)
+ reader, err := bs.Download(blobNode)
  Expect(err).ToNot(HaveOccurred())
 
  readData, err := ioutil.ReadAll(reader)
@@ -92,7 +97,7 @@ var _ = Describe("Blobstore", func() {
  })
 
  It("returns a reader to the blob", func() {
- reader, err := bs.Download(key)
+ reader, err := bs.Download(blobNode)
  Expect(err).ToNot(HaveOccurred())
 
  readData, err := ioutil.ReadAll(reader)
@@ -106,7 +111,7 @@ var _ = Describe("Blobstore", func() {
  _, err := os.Stat(blobPath)
  Expect(err).ToNot(HaveOccurred())
 
- err = bs.Delete(key)
+ err = bs.Delete(blobNode)
  Expect(err).ToNot(HaveOccurred())
 
  _, err = os.Stat(blobPath)

pkg/storage/fs/ocis/ocis.go

@@ -40,7 +40,7 @@ func New(m map[string]interface{}) (storage.FS, error) {
  return nil, err
  }
 
- bs, err := blobstore.New(path.Join(o.Root, "blobs"))
+ bs, err := blobstore.New(path.Join(o.Root))
  if err != nil {
  return nil, err
  }

pkg/storage/fs/s3ng/blobstore/blobstore.go

@@ -23,7 +23,10 @@ import (
  "io"
  "net/url"
  "os"
+ "path/filepath"
 
+ "github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/lookup"
+ "github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/node"
  "github.com/minio/minio-go/v7"
  "github.com/minio/minio-go/v7/pkg/credentials"
  "github.com/pkg/errors"
@@ -60,38 +63,50 @@ func New(endpoint, region, bucket, accessKey, secretKey string) (*Blobstore, err
 }
 
 // Upload stores some data in the blobstore under the given key
-func (bs *Blobstore) Upload(key string, reader io.Reader) error {
+func (bs *Blobstore) Upload(node *node.Node, reader io.Reader) error {
  size := int64(-1)
  if file, ok := reader.(*os.File); ok {
  info, err := file.Stat()
  if err != nil {
- return errors.Wrapf(err, "could not determine file size for object '%s'", key)
+ return errors.Wrapf(err, "could not determine file size for object '%s'", bs.path(node))
  }
  size = info.Size()
  }
 
- _, err := bs.client.PutObject(context.Background(), bs.bucket, key, reader, size, minio.PutObjectOptions{ContentType: "application/octet-stream"})
+ _, err := bs.client.PutObject(context.Background(), bs.bucket, bs.path(node), reader, size, minio.PutObjectOptions{ContentType: "application/octet-stream"})
 
  if err != nil {
- return errors.Wrapf(err, "could not store object '%s' into bucket '%s'", key, bs.bucket)
+ return errors.Wrapf(err, "could not store object '%s' into bucket '%s'", bs.path(node), bs.bucket)
  }
  return nil
 }
 
 // Download retrieves a blob from the blobstore for reading
-func (bs *Blobstore) Download(key string) (io.ReadCloser, error) {
- reader, err := bs.client.GetObject(context.Background(), bs.bucket, key, minio.GetObjectOptions{})
+func (bs *Blobstore) Download(node *node.Node) (io.ReadCloser, error) {
+ reader, err := bs.client.GetObject(context.Background(), bs.bucket, bs.path(node), minio.GetObjectOptions{})
  if err != nil {
- return nil, errors.Wrapf(err, "could not download object '%s' from bucket '%s'", key, bs.bucket)
+ return nil, errors.Wrapf(err, "could not download object '%s' from bucket '%s'", bs.path(node), bs.bucket)
  }
  return reader, nil
 }
 
 // Delete deletes a blob from the blobstore
-func (bs *Blobstore) Delete(key string) error {
- err := bs.client.RemoveObject(context.Background(), bs.bucket, key, minio.RemoveObjectOptions{})
+func (bs *Blobstore) Delete(node *node.Node) error {
+ err := bs.client.RemoveObject(context.Background(), bs.bucket, bs.path(node), minio.RemoveObjectOptions{})
  if err != nil {
- return errors.Wrapf(err, "could not delete object '%s' from bucket '%s'", key, bs.bucket)
+ return errors.Wrapf(err, "could not delete object '%s' from bucket '%s'", bs.path(node), bs.bucket)
  }
  return nil
 }
+
+func (bs *Blobstore) path(node *node.Node) string {
+ // https://aws.amazon.com/de/premiumsupport/knowledge-center/s3-prefix-nested-folders-difference/
+ // Prefixes are used to partion a bucket. A prefix is everything except the filename.
+ // For a file `BucketName/foo/bar/lorem.ipsum`, `BucketName/foo/bar/` is the prefix.
+ // There are request limits per prefix, therefore you should have many prefixes.
+ // There are no limits to prefixes per bucket, so in general it's better to have more then less.
+ //
+ // Since the spaceID is always the same for a space, we don't need to pathify that, because it would
+ // not yield any performance gains
+ return filepath.Clean(filepath.Join(node.SpaceID, lookup.Pathify(node.BlobID, 4, 2)))
+}

pkg/storage/utils/decomposedfs/decomposedfs.go

@@ -82,9 +82,9 @@ type Tree interface {
  RestoreRecycleItemFunc(ctx context.Context, spaceid, key, trashPath string, target *node.Node) (*node.Node, *node.Node, func() error, error)
  PurgeRecycleItemFunc(ctx context.Context, spaceid, key, purgePath string) (*node.Node, func() error, error)
 
- WriteBlob(key string, reader io.Reader) error
- ReadBlob(key string) (io.ReadCloser, error)
- DeleteBlob(key string) error
+ WriteBlob(node *node.Node, reader io.Reader) error
+ ReadBlob(node *node.Node) (io.ReadCloser, error)
+ DeleteBlob(node *node.Node) error
 
  Propagate(ctx context.Context, node *node.Node) (err error)
 }
@@ -568,7 +568,7 @@ func (fs *Decomposedfs) Download(ctx context.Context, ref *provider.Reference) (
  return nil, errtypes.PermissionDenied(filepath.Join(node.ParentID, node.Name))
  }
 
- reader, err := fs.tp.ReadBlob(node.BlobID)
+ reader, err := fs.tp.ReadBlob(node)
  if err != nil {
  return nil, errors.Wrap(err, "Decomposedfs: error download blob '"+node.ID+"'")
  }