Write blob parts directly to blobstore

Signed-off-by: Jörn Friedrich Dreyer <[email protected]>

Author: butonic

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

@@ -20,6 +20,8 @@ package blobstore
 
 import (
 	"bufio"
+	"context"
+	"fmt"
 	"io"
 	"os"
 	"path/filepath"
@@ -49,8 +51,7 @@ func New(root string) (*Blobstore, error) {
 
 // Upload stores some data in the blobstore under the given key
 func (bs *Blobstore) Upload(node *node.Node, source string) error {
-
-	dest := bs.path(node)
+	dest := bs.pathFromNode(node)
 	// ensure parent path exists
 	if err := os.MkdirAll(filepath.Dir(dest), 0700); err != nil {
 		return errors.Wrap(err, "Decomposedfs: oCIS blobstore: error creating parent folders for blob")
@@ -69,40 +70,82 @@ func (bs *Blobstore) Upload(node *node.Node, source string) error {
 
 	f, err := os.OpenFile(dest, os.O_CREATE|os.O_WRONLY, 0700)
 	if err != nil {
-		return errors.Wrapf(err, "could not open blob '%s' for writing", bs.path(node))
+		return errors.Wrapf(err, "could not open blob '%s' for writing", dest)
 	}
 
 	w := bufio.NewWriter(f)
 	_, err = w.ReadFrom(file)
 	if err != nil {
-		return errors.Wrapf(err, "could not write blob '%s'", bs.path(node))
+		return errors.Wrapf(err, "could not write blob '%s'", dest)
+	}
+
+	return w.Flush()
+}
+
+func (bs *Blobstore) StreamingUpload(ctx context.Context, spaceid, blobid string, offset, objectSize int64, reader io.Reader, userMetadata map[string]string) error {
+	dest := bs.path(spaceid, blobid)
+	if offset > 0 {
+		// write a part
+		dest = fmt.Sprintf("%s:%d-%d", dest, offset, offset+objectSize)
+	}
+
+	// ensure parent path exists
+	if err := os.MkdirAll(filepath.Dir(dest), 0700); err != nil {
+		return errors.Wrap(err, "Decomposedfs: oCIS blobstore: error creating parent folders for blob")
+	}
+
+	f, err := os.OpenFile(dest, os.O_CREATE|os.O_WRONLY, 0700)
+	if err != nil {
+		return errors.Wrapf(err, "could not open blob '%s' for stream writing", dest)
+	}
+
+	w := bufio.NewWriter(f)
+	_, err = w.ReadFrom(reader)
+	if err != nil {
+		return errors.Wrapf(err, "could not stream write blob '%s'", dest)
 	}
 
 	return w.Flush()
 }
+func (bs *Blobstore) StreamingDownload(ctx context.Context, spaceid, blobid string, offset, objectSize int64) (io.ReadCloser, error) {
+	dest := bs.path(spaceid, blobid)
+	if offset > 0 {
+		// read a part
+		dest = fmt.Sprintf("%s:%d-%d", dest, offset, offset+objectSize)
+	}
+	file, err := os.Open(dest)
+	if err != nil {
+		return nil, errors.Wrapf(err, "could not read blob '%s'", dest)
+	}
+	return file, nil
+}
 
 // Download retrieves a blob from the blobstore for reading
 func (bs *Blobstore) Download(node *node.Node) (io.ReadCloser, error) {
-	file, err := os.Open(bs.path(node))
+	file, err := os.Open(bs.pathFromNode(node))
 	if err != nil {
-		return nil, errors.Wrapf(err, "could not read blob '%s'", bs.path(node))
+		return nil, errors.Wrapf(err, "could not read blob '%s'", bs.pathFromNode(node))
 	}
 	return file, nil
 }
 
 // Delete deletes a blob from the blobstore
 func (bs *Blobstore) Delete(node *node.Node) error {
-	if err := utils.RemoveItem(bs.path(node)); err != nil {
-		return errors.Wrapf(err, "could not delete blob '%s'", bs.path(node))
+	if err := utils.RemoveItem(bs.pathFromNode(node)); err != nil {
+		return errors.Wrapf(err, "could not delete blob '%s'", bs.pathFromNode(node))
 	}
 	return nil
 }
 
-func (bs *Blobstore) path(node *node.Node) string {
+func (bs *Blobstore) pathFromNode(node *node.Node) string {
+	return bs.path(node.SpaceID, node.BlobID)
+}
+
+func (bs *Blobstore) path(spaceid, blobid string) string {
 	return filepath.Join(
 		bs.root,
 		filepath.Clean(filepath.Join(
-			"/", "spaces", lookup.Pathify(node.SpaceID, 1, 2), "blobs", lookup.Pathify(node.BlobID, 4, 2)),
+			"/", "spaces", lookup.Pathify(spaceid, 1, 2), "blobs", lookup.Pathify(blobid, 4, 2)),
 		),
 	)
 }

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

@@ -71,24 +71,47 @@ func (bs *Blobstore) Upload(node *node.Node, source string) error {
 	}
 	defer reader.Close()
 
-	_, err = bs.client.PutObject(context.Background(), bs.bucket, bs.path(node), reader, node.Blobsize, minio.PutObjectOptions{ContentType: "application/octet-stream", SendContentMd5: true})
+	_, err = bs.client.PutObject(context.Background(), bs.bucket, bs.pathFromNode(node), reader, node.Blobsize, minio.PutObjectOptions{ContentType: "application/octet-stream", SendContentMd5: true})
 
 	if err != nil {
-		return errors.Wrapf(err, "could not store object '%s' into bucket '%s'", bs.path(node), bs.bucket)
+		return errors.Wrapf(err, "could not store object '%s' into bucket '%s'", bs.pathFromNode(node), bs.bucket)
 	}
 	return nil
 }
 
+func (bs *Blobstore) StreamingUpload(ctx context.Context, spaceid, blobid string, offset, objectSize int64, reader io.Reader, userMetadata map[string]string) error {
+	key := bs.path(spaceid, blobid)
+	if offset > 0 {
+		// write a part
+		key = fmt.Sprintf("%s:%d-%d", key, offset, offset+objectSize)
+	}
+	_, err := bs.client.PutObject(ctx, bs.bucket, key, reader, objectSize, minio.PutObjectOptions{ContentType: "application/octet-stream", SendContentMd5: true, UserMetadata: userMetadata})
+
+	if err != nil {
+		return errors.Wrapf(err, "could not stream object '%s' into bucket '%s'", key, bs.bucket)
+	}
+	return nil
+}
+
+func (bs *Blobstore) StreamingDownload(ctx context.Context, spaceid, blobid string, offset, objectSize int64) (io.ReadCloser, error) {
+	key := bs.path(spaceid, blobid)
+	if offset > 0 {
+		// read a part
+		key = fmt.Sprintf("%s:%d-%d", key, offset, offset+objectSize)
+	}
+	return bs.client.GetObject(ctx, bs.bucket, key, minio.GetObjectOptions{})
+}
+
 // Download retrieves a blob from the blobstore for reading
 func (bs *Blobstore) Download(node *node.Node) (io.ReadCloser, error) {
-	reader, err := bs.client.GetObject(context.Background(), bs.bucket, bs.path(node), minio.GetObjectOptions{})
+	reader, err := bs.client.GetObject(context.Background(), bs.bucket, bs.pathFromNode(node), minio.GetObjectOptions{})
 	if err != nil {
-		return nil, errors.Wrapf(err, "could not download object '%s' from bucket '%s'", bs.path(node), bs.bucket)
+		return nil, errors.Wrapf(err, "could not download object '%s' from bucket '%s'", bs.pathFromNode(node), bs.bucket)
 	}
 
 	stat, err := reader.Stat()
 	if err != nil {
-		return nil, errors.Wrapf(err, "blob path: %s", bs.path(node))
+		return nil, errors.Wrapf(err, "blob path: %s", bs.pathFromNode(node))
 	}
 
 	if node.Blobsize != stat.Size {
@@ -100,14 +123,18 @@ func (bs *Blobstore) Download(node *node.Node) (io.ReadCloser, error) {
 
 // Delete deletes a blob from the blobstore
 func (bs *Blobstore) Delete(node *node.Node) error {
-	err := bs.client.RemoveObject(context.Background(), bs.bucket, bs.path(node), minio.RemoveObjectOptions{})
+	err := bs.client.RemoveObject(context.Background(), bs.bucket, bs.pathFromNode(node), minio.RemoveObjectOptions{})
 	if err != nil {
-		return errors.Wrapf(err, "could not delete object '%s' from bucket '%s'", bs.path(node), bs.bucket)
+		return errors.Wrapf(err, "could not delete object '%s' from bucket '%s'", bs.pathFromNode(node), bs.bucket)
 	}
 	return nil
 }
 
-func (bs *Blobstore) path(node *node.Node) string {
+func (bs *Blobstore) pathFromNode(node *node.Node) string {
+	return bs.path(node.SpaceID, node.BlobID)
+}
+
+func (bs *Blobstore) path(spaceid, blobid string) 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.
@@ -116,5 +143,5 @@ func (bs *Blobstore) path(node *node.Node) string {
 	//
 	// 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)))
+	return filepath.Clean(filepath.Join(spaceid, lookup.Pathify(blobid, 4, 2)))
 }

pkg/storage/utils/decomposedfs/decomposedfs.go

@@ -48,6 +48,7 @@ import (
 	"github.com/cs3org/reva/v2/pkg/storage/utils/chunking"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/lookup"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/metadata"
+	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/metadata/prefixes"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/migrator"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/node"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/options"
@@ -100,6 +101,9 @@ type Tree interface {
 	ReadBlob(node *node.Node) (io.ReadCloser, error)
 	DeleteBlob(node *node.Node) error
 
+	StreamBlob(ctx context.Context, spaceid, blobid string, offset, objectSize int64, reader io.Reader, userMetadata map[string]string) error
+	BlobReader(ctx context.Context, spaceid, blobid string, offset, objectSize int64) (io.ReadCloser, error)
+
 	Propagate(ctx context.Context, node *node.Node, sizeDiff int64) (err error)
 }
 
@@ -1027,11 +1031,40 @@ func (fs *Decomposedfs) Download(ctx context.Context, ref *provider.Reference) (
 	if currentEtag != expectedEtag {
 		return nil, errtypes.Aborted(fmt.Sprintf("file changed from etag %s to %s", expectedEtag, currentEtag))
 	}
-	reader, err := fs.tp.ReadBlob(n)
+
+	val, err := n.XattrString(ctx, prefixes.BlobOffsetsAttr)
+	var blobOffsets []int64
 	if err != nil {
-		return nil, errors.Wrap(err, "Decomposedfs: error download blob '"+n.ID+"'")
+		// not set, single blob
+		blobOffsets = []int64{0}
+	} else {
+		offsetStrings := strings.Split(val, ",")
+		blobOffsets = make([]int64, 0, len(offsetStrings))
+		for _, s := range offsetStrings {
+			offset, err := strconv.ParseInt(s, 10, 64)
+			if err != nil {
+				return nil, errors.Wrap(err, "Decomposedfs: malformed part offset '"+s+"' in node '"+n.ID+"'")
+			}
+			blobOffsets = append(blobOffsets, offset)
+		}
+	}
+
+	blobparts := len(blobOffsets)
+	readers := make([]io.Reader, 0, blobparts)
+	for i, offset := range blobOffsets {
+		var partSize int64
+		if i < blobparts && blobparts > 1 {
+			partSize = upload.OptimalPartSize
+		} else {
+			partSize = n.Blobsize - offset
+		}
+		blobReader, err := fs.tp.BlobReader(ctx, n.SpaceID, n.BlobID, offset, partSize)
+		if err != nil {
+			return nil, err
+		}
+		readers = append(readers, upload.CloseOnEOFReader(blobReader))
 	}
-	return reader, nil
+	return io.NopCloser(io.MultiReader(readers...)), nil
 }
 
 // GetLock returns an existing lock on the given reference

pkg/storage/utils/decomposedfs/metadata/prefixes/prefixes.go

@@ -36,8 +36,9 @@ const (
 	// updated when the file is renamed or moved
 	NameAttr string = OcisPrefix + "name"
 
-	BlobIDAttr   string = OcisPrefix + "blobid"
-	BlobsizeAttr string = OcisPrefix + "blobsize"
+	BlobIDAttr      string = OcisPrefix + "blobid"
+	BlobsizeAttr    string = OcisPrefix + "blobsize"
+	BlobOffsetsAttr string = OcisPrefix + "bloboffsets"
 
 	// statusPrefix is the prefix for the node status
 	StatusPrefix string = OcisPrefix + "nodestatus"

pkg/storage/utils/decomposedfs/revisions.go

@@ -247,6 +247,7 @@ func (fs *Decomposedfs) RestoreRevision(ctx context.Context, ref *provider.Refer
 			attributeName == prefixes.TypeAttr ||
 			attributeName == prefixes.BlobIDAttr ||
 			attributeName == prefixes.BlobsizeAttr ||
+			attributeName == prefixes.BlobOffsetsAttr ||
 			attributeName == prefixes.MTimeAttr // FIXME somewhere I mix up the revision time and the mtime, causing the restore to overwrite the other existing revisien
 	}, f, true)
 	if err != nil {
@@ -270,7 +271,8 @@ func (fs *Decomposedfs) RestoreRevision(ctx context.Context, ref *provider.Refer
 		return value, strings.HasPrefix(attributeName, prefixes.ChecksumPrefix) ||
 			attributeName == prefixes.TypeAttr ||
 			attributeName == prefixes.BlobIDAttr ||
-			attributeName == prefixes.BlobsizeAttr
+			attributeName == prefixes.BlobsizeAttr ||
+			attributeName == prefixes.BlobOffsetsAttr
 	}, false)
 	if err != nil {
 		return errtypes.InternalError("failed to copy blob xattrs to old revision to node: " + err.Error())

pkg/storage/utils/decomposedfs/tree/tree.go

@@ -62,6 +62,9 @@ type Blobstore interface {
 	Upload(node *node.Node, source string) error
 	Download(node *node.Node) (io.ReadCloser, error)
 	Delete(node *node.Node) error
+
+	StreamingUpload(ctx context.Context, spaceid, blobid string, offset, objectSize int64, reader io.Reader, userMetadata map[string]string) error
+	StreamingDownload(ctx context.Context, spaceid, blobid string, offset, objectSize int64) (io.ReadCloser, error)
 }
 
 // Tree manages a hierarchical tree
@@ -712,6 +715,14 @@ func (t *Tree) WriteBlob(node *node.Node, source string) error {
 	return t.blobstore.Upload(node, source)
 }
 
+// StreamBlob streams a blob to the blobstore
+func (t *Tree) StreamBlob(ctx context.Context, spaceid, blobid string, offset, objectSize int64, reader io.Reader, userMetadata map[string]string) error {
+	return t.blobstore.StreamingUpload(ctx, spaceid, blobid, offset, objectSize, reader, userMetadata)
+}
+func (t *Tree) BlobReader(ctx context.Context, spaceid, blobid string, offset, objectSize int64) (io.ReadCloser, error) {
+	return t.blobstore.StreamingDownload(ctx, spaceid, blobid, offset, objectSize)
+}
+
 // ReadBlob reads a blob from the blobstore
 func (t *Tree) ReadBlob(node *node.Node) (io.ReadCloser, error) {
 	if node.BlobID == "" {

pkg/storage/utils/decomposedfs/upload/part_producer.go

@@ -0,0 +1,54 @@
+package upload
+
+import (
+	"bytes"
+	"io"
+)
+
+// partProducer converts a stream of bytes from the reader into a stream memory of buffers
+type partProducer struct {
+	parts chan<- *bytes.Buffer
+	done  chan struct{}
+	err   error
+	r     io.Reader
+}
+
+func (spp *partProducer) produce(partSize int64) {
+	for {
+		file, err := spp.nextPart(partSize)
+		if err != nil {
+			spp.err = err
+			close(spp.parts)
+			return
+		}
+		if file == nil {
+			close(spp.parts)
+			return
+		}
+		select {
+		case spp.parts <- file:
+		case <-spp.done:
+			close(spp.parts)
+			return
+		}
+	}
+}
+
+func (spp *partProducer) nextPart(size int64) (*bytes.Buffer, error) {
+	buf := new(bytes.Buffer)
+
+	limitedReader := io.LimitReader(spp.r, size)
+	n, err := buf.ReadFrom(limitedReader)
+	if err != nil {
+		return nil, err
+	}
+
+	// If the entire request body is read and no more data is available,
+	// buf.ReadFrom returns 0 since it is unable to read any bytes. In that
+	// case, we can close the partProducer.
+	if n == 0 {
+		return nil, nil
+	}
+
+	return buf, nil
+}