diff --git a/cmd/serve/s3/backend.go b/cmd/serve/s3/backend.go index d4531e987..54019c7fb 100644 --- a/cmd/serve/s3/backend.go +++ b/cmd/serve/s3/backend.go @@ -23,10 +23,20 @@ var ( ) // s3Backend implements the gofacess3.Backend interface to make an S3 -// backend for gofakes3 +// backend for gofakes3. It also implements gofakes3.MultipartBackend so that +// multipart uploads stream straight through to the underlying Fs via +// PutStream, instead of being buffered in memory by gofakes3. type s3Backend struct { s *Server meta *sync.Map + + // multipartUploads tracks in-flight streaming multipart uploads, + // keyed by gofakes3.UploadID. + multipartUploads sync.Map + + // warnInMemoryOnce logs a single NOTICE the first time a multipart + // upload falls back to being buffered in memory. + warnInMemoryOnce sync.Once } // newBackend creates a new SimpleBucketBackend. diff --git a/cmd/serve/s3/multipart.go b/cmd/serve/s3/multipart.go new file mode 100644 index 000000000..e9207879a --- /dev/null +++ b/cmd/serve/s3/multipart.go @@ -0,0 +1,388 @@ +// Multipart upload support for serve s3. +// +// Multipart uploads received by serve s3 are streamed, in part-number order, +// into a single PutStream upload to the underlying Fs, so the whole file is +// never buffered in memory. This implements the gofakes3.MultipartBackend +// interface on s3Backend. +// +// When streaming is disabled (--disable-multipart-streaming) or the Fs has no +// PutStream, ErrMultipartUploadNotSupported is returned so that gofakes3 falls +// back to buffering the parts in memory. + +package s3 + +import ( + "context" + "crypto/md5" + "encoding/hex" + "errors" + "fmt" + "io" + "path" + "sort" + "strings" + "sync" + "time" + + "github.com/google/uuid" + "github.com/ncw/swift/v2" + "github.com/rclone/gofakes3" + "github.com/rclone/rclone/fs" + "github.com/rclone/rclone/fs/object" + "github.com/rclone/rclone/lib/multipart" + "github.com/rclone/rclone/lib/pool" +) + +// multipartUpload tracks one in-flight S3 multipart upload that is being +// streamed, in part order, into a single PutStream upload to the underlying Fs. +type multipartUpload struct { + bucket, key string + fp string // = path.Join(bucket, key) + meta map[string]string + + pipeW *io.PipeWriter // parts are streamed here, in part-number order + + mu sync.Mutex + partMD5s map[int][]byte // raw MD5 sums per part (for the final S3 multipart ETag) + partSizes map[int]int64 // observed part sizes + closed bool + + putCancel context.CancelFunc // cancels the background PutStream + putDone chan struct{} // closed when the background PutStream returns + putErr error // PutStream result (read only after putDone is closed) + nextPart int // next part number to stream (1-based) + streamBuf map[int]*pool.RW // parts received ahead of nextPart, awaiting their turn + pumping bool // a goroutine is currently writing to the pipe +} + +// newMultipartUpload allocates an upload struct. +func newMultipartUpload(bucket, key, fp string, meta map[string]string) *multipartUpload { + return &multipartUpload{ + bucket: bucket, + key: key, + fp: fp, + meta: meta, + partMD5s: map[int][]byte{}, + partSizes: map[int]int64{}, + nextPart: 1, + streamBuf: map[int]*pool.RW{}, + } +} + +// loadUpload looks up an in-flight upload by ID. +func (b *s3Backend) loadUpload(uploadID gofakes3.UploadID) (*multipartUpload, error) { + v, ok := b.multipartUploads.Load(uploadID) + if !ok { + return nil, gofakes3.ErrNoSuchUpload + } + return v.(*multipartUpload), nil +} + +// CreateMultipartUpload begins a new multipart upload that streams the parts, +// in part-number order, into a single PutStream upload to the underlying Fs. +// +// If streaming is disabled (--disable-multipart-streaming) or the Fs has no +// PutStream, ErrMultipartUploadNotSupported is returned so that gofakes3 falls +// back to buffering the whole upload in memory; a one-off NOTICE warns about +// the memory use. +func (b *s3Backend) CreateMultipartUpload(ctx context.Context, bucketName, objectName string, meta map[string]string) (gofakes3.UploadID, error) { + _vfs, err := b.s.getVFS(ctx) + if err != nil { + return "", err + } + if _, err := _vfs.Stat(bucketName); err != nil { + return "", gofakes3.BucketNotFound(bucketName) + } + + f := _vfs.Fs() + features := f.Features() + if b.s.opt.DisableMultipartStreaming || features.PutStream == nil { + b.warnInMemoryOnce.Do(func() { + reason := "this backend doesn't support streaming uploads" + if b.s.opt.DisableMultipartStreaming { + reason = "--disable-multipart-streaming is set" + } + fs.Logf(nil, "serve s3: buffering multipart uploads in memory because %s - this may use a lot of memory", reason) + }) + return "", gofakes3.ErrMultipartUploadNotSupported + } + + fp := path.Join(bucketName, objectName) + objectDir := path.Dir(fp) + if objectDir != "." { + if err := mkdirRecursive(objectDir, _vfs); err != nil { + return "", err + } + } + + up := newMultipartUpload(bucketName, objectName, fp, meta) + + src := object.NewStaticObjectInfo(fp, time.Now(), -1, true, nil, f) + pr, pw := io.Pipe() + // Use a context that outlives this request (it's cancelled on abort) but + // keeps its values. + putCtx, cancel := context.WithCancel(context.WithoutCancel(ctx)) + up.pipeW = pw + up.putCancel = cancel + up.putDone = make(chan struct{}) + go func() { + _, err := features.PutStream(putCtx, pr, src) + up.putErr = err + _ = pr.CloseWithError(err) + close(up.putDone) + }() + + uploadID := gofakes3.UploadID(uuid.New().String()) + b.multipartUploads.Store(uploadID, up) + return uploadID, nil +} + +// UploadPart writes a single part from the S3 client into the streaming upload. +func (b *s3Backend) UploadPart(ctx context.Context, bucketName, objectName string, uploadID gofakes3.UploadID, partNumber int, contentLength int64, body io.Reader) (string, error) { + up, err := b.loadUpload(uploadID) + if err != nil { + return "", err + } + + // Buffer the part in a pool-backed RW so we can MD5 it (for the ETag) and + // stream it once it is this part's turn. + rw := multipart.NewRW().Reserve(contentLength) + hasher := md5.New() + n, err := io.Copy(rw, io.TeeReader(body, hasher)) + if err != nil { + _ = rw.Close() + return "", err + } + if n != contentLength { + _ = rw.Close() + return "", gofakes3.ErrIncompleteBody + } + md5Sum := hasher.Sum(nil) + etag := fmt.Sprintf("%q", hex.EncodeToString(md5Sum)) + + if err := up.streamPart(partNumber, n, md5Sum, rw); err != nil { + return "", err + } + return etag, nil +} + +// streamPart records a part and streams the parts into the pipe in order. +// +// Parts must be uploaded in ascending, contiguous part-number order. A part +// that arrives ahead of the next expected one is buffered until its turn; the +// parts are then pumped into the pipe in order. Whichever goroutine finds the +// next part available does the pumping, so concurrent (but in-order) clients +// are tolerated with buffering bounded by how far ahead they run. +func (up *multipartUpload) streamPart(partNumber int, size int64, md5Sum []byte, rw *pool.RW) error { + up.mu.Lock() + up.partMD5s[partNumber] = md5Sum + up.partSizes[partNumber] = size + up.streamBuf[partNumber] = rw + + if up.pumping { + // Another goroutine owns the pipe and will pump this part in turn. + up.mu.Unlock() + return nil + } + up.pumping = true + + for { + prw, ok := up.streamBuf[up.nextPart] + if !ok { + up.pumping = false + up.mu.Unlock() + return nil + } + delete(up.streamBuf, up.nextPart) + up.mu.Unlock() + + err := pipePart(up.pipeW, prw) + _ = prw.Close() + if err != nil { + up.mu.Lock() + up.pumping = false + up.mu.Unlock() + return err + } + + up.mu.Lock() + up.nextPart++ + } +} + +// pipePart writes the whole of rw into w (the pipe). +func pipePart(w io.Writer, rw *pool.RW) error { + if _, err := rw.Seek(0, io.SeekStart); err != nil { + return err + } + _, err := io.Copy(w, rw) + return err +} + +// CompleteMultipartUpload finalises a streamed multipart upload. It closes the +// pipe (so PutStream finishes), registers the new file with the VFS, computes +// the S3-style multipart ETag, and stores the user metadata so HeadObject and +// GetObject see the same fields the in-memory PutObject path produces. +func (b *s3Backend) CompleteMultipartUpload(ctx context.Context, bucketName, objectName string, uploadID gofakes3.UploadID, input *gofakes3.CompleteMultipartUploadRequest) (gofakes3.VersionID, string, error) { + up, err := b.loadUpload(uploadID) + if err != nil { + return "", "", err + } + defer b.multipartUploads.Delete(uploadID) + + if err := up.validate(input); err != nil { + _ = up.abort(ctx) + return "", "", err + } + + // All parts must have been streamed: contiguous part numbers from 1 with + // nothing left buffered. A leftover means the client used non-contiguous + // part numbers, which the in-order stream can't place. + up.mu.Lock() + streamed := up.nextPart - 1 + total := len(up.partSizes) + leftover := len(up.streamBuf) + up.mu.Unlock() + if leftover != 0 || streamed != total { + _ = up.abort(ctx) + return "", "", gofakes3.ErrInvalidPart + } + + if err := up.close(ctx); err != nil { + return "", "", err + } + + _vfs, err := b.s.getVFS(ctx) + if err != nil { + return "", "", err + } + + // Invalidate the parent directory's cached listing so subsequent VFS + // Stat / List calls pick up the newly-written object from the + // underlying Fs (we wrote to the Fs directly, bypassing VFS). + if root, err := _vfs.Root(); err == nil { + root.ForgetPath(up.fp, fs.EntryObject) + } + + b.meta.Store(up.fp, up.meta) + if val, ok := up.meta["X-Amz-Meta-Mtime"]; ok { + if ti, err := swift.FloatStringToTime(val); err == nil { + b.storeModtime(up.fp, up.meta, val) + _ = _vfs.Chtimes(up.fp, ti, ti) + } + } else if val, ok := up.meta["mtime"]; ok { + if ti, err := swift.FloatStringToTime(val); err == nil { + b.storeModtime(up.fp, up.meta, val) + _ = _vfs.Chtimes(up.fp, ti, ti) + } + } + + return "", up.multipartETag(input), nil +} + +// AbortMultipartUpload tears down an in-progress upload, asking the background +// PutStream to discard any data already sent. +func (b *s3Backend) AbortMultipartUpload(ctx context.Context, bucketName, objectName string, uploadID gofakes3.UploadID) error { + up, err := b.loadUpload(uploadID) + if err != nil { + return err + } + defer b.multipartUploads.Delete(uploadID) + return up.abort(ctx) +} + +// validate cross-checks the part list supplied by the client against the +// parts we actually received. +func (up *multipartUpload) validate(input *gofakes3.CompleteMultipartUploadRequest) error { + up.mu.Lock() + defer up.mu.Unlock() + + for i := 1; i < len(input.Parts); i++ { + if input.Parts[i].PartNumber <= input.Parts[i-1].PartNumber { + return gofakes3.ErrInvalidPartOrder + } + } + if len(input.Parts) != len(up.partSizes) { + return gofakes3.ErrInvalidPart + } + for _, p := range input.Parts { + md5Sum, ok := up.partMD5s[p.PartNumber] + if !ok { + return gofakes3.ErrInvalidPart + } + clientETag := strings.Trim(p.ETag, `"`) + if clientETag != hex.EncodeToString(md5Sum) { + return gofakes3.ErrInvalidPart + } + } + return nil +} + +// close finalises the upload by signalling EOF to the background PutStream and +// waiting for it to finish. +func (up *multipartUpload) close(ctx context.Context) error { + up.mu.Lock() + if up.closed { + up.mu.Unlock() + return nil + } + up.closed = true + up.mu.Unlock() + + err := up.pipeW.Close() + <-up.putDone + up.putCancel() + if up.putErr != nil { + return up.putErr + } + return err +} + +// errMultipartAborted makes the background PutStream fail when an upload is +// aborted, so it tears down its partial object instead of completing. +var errMultipartAborted = errors.New("serve s3: multipart upload aborted") + +// abort cancels the background PutStream and releases any buffered parts. +func (up *multipartUpload) abort(ctx context.Context) error { + up.mu.Lock() + if up.closed { + up.mu.Unlock() + return nil + } + up.closed = true + streamBuf := up.streamBuf + up.streamBuf = nil + up.mu.Unlock() + + for _, rw := range streamBuf { + _ = rw.Close() + } + + // Fail the background PutStream (so it discards its partial object) and + // wait for it to return. + up.putCancel() + _ = up.pipeW.CloseWithError(errMultipartAborted) + <-up.putDone + return nil +} + +// multipartETag computes the S3 multipart ETag for the assembled object: +// +// hex(md5(concat(part_md5s_in_order))) + "-" + N +func (up *multipartUpload) multipartETag(input *gofakes3.CompleteMultipartUploadRequest) string { + partNumbers := make([]int, 0, len(input.Parts)) + for _, p := range input.Parts { + partNumbers = append(partNumbers, p.PartNumber) + } + sort.Ints(partNumbers) + + up.mu.Lock() + concat := make([]byte, 0, len(partNumbers)*md5.Size) + for _, n := range partNumbers { + concat = append(concat, up.partMD5s[n]...) + } + up.mu.Unlock() + + sum := md5.Sum(concat) + return fmt.Sprintf("%q", fmt.Sprintf("%s-%d", hex.EncodeToString(sum[:]), len(partNumbers))) +} diff --git a/cmd/serve/s3/multipart_test.go b/cmd/serve/s3/multipart_test.go new file mode 100644 index 000000000..48b907166 --- /dev/null +++ b/cmd/serve/s3/multipart_test.go @@ -0,0 +1,198 @@ +// Multipart upload tests for serve s3. + +package s3 + +import ( + "bytes" + "context" + "fmt" + "io" + "net/url" + "path" + "sync" + "testing" + + "github.com/minio/minio-go/v7" + "github.com/minio/minio-go/v7/pkg/credentials" + "github.com/rclone/rclone/cmd/serve/proxy" + "github.com/rclone/rclone/fs" + "github.com/rclone/rclone/fstest" + "github.com/rclone/rclone/lib/random" + "github.com/rclone/rclone/vfs/vfscommon" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// newMultipartTestServer starts a serve s3 server backed by a fresh local temp +// directory and returns a low-level minio Core client (for explicit control of +// the multipart parts), the backing Fs and the bucket name. The server and +// client are torn down via t.Cleanup. +func newMultipartTestServer(t *testing.T, disableStreaming bool) (*minio.Core, fs.Fs, string) { + fstest.Initialise() + ctx := context.Background() + f, err := fs.NewFs(ctx, t.TempDir()) + require.NoError(t, err) + const bucket = "test" + require.NoError(t, f.Mkdir(ctx, bucket)) + + keyid := random.String(16) + keysec := random.String(16) + opt := Opt + opt.DisableMultipartStreaming = disableStreaming + opt.AuthKey = []string{fmt.Sprintf("%s,%s", keyid, keysec)} + opt.HTTP.ListenAddr = []string{endpoint} + w, err := newServer(ctx, f, &opt, &vfscommon.Opt, &proxy.Opt) + require.NoError(t, err) + go func() { _ = w.Serve() }() + t.Cleanup(func() { _ = w.Shutdown() }) + + u, err := url.Parse(w.server.URLs()[0]) + require.NoError(t, err) + core, err := minio.NewCore(u.Host, &minio.Options{ + Creds: credentials.NewStaticV4(keyid, keysec, ""), + Secure: false, + }) + require.NoError(t, err) + return core, f, bucket +} + +// readObject reads bucket/object back from the backing Fs. +func readObject(t *testing.T, f fs.Fs, bucket, object string) []byte { + ctx := context.Background() + o, err := f.NewObject(ctx, path.Join(bucket, object)) + require.NoError(t, err) + rc, err := o.Open(ctx) + require.NoError(t, err) + got, err := io.ReadAll(rc) + require.NoError(t, err) + require.NoError(t, rc.Close()) + return got +} + +// multipartUploadParts uploads object to bucket as a multipart upload with the +// given (in-order) part sizes and returns the assembled contents plus the +// first error encountered. +func multipartUploadParts(t *testing.T, core *minio.Core, bucket, object string, partSizes []int) ([]byte, error) { + ctx := context.Background() + uploadID, err := core.NewMultipartUpload(ctx, bucket, object, minio.PutObjectOptions{}) + if err != nil { + return nil, err + } + var want []byte + var parts []minio.CompletePart + for i, sz := range partSizes { + data := []byte(random.String(sz)) + want = append(want, data...) + p, err := core.PutObjectPart(ctx, bucket, object, uploadID, i+1, bytes.NewReader(data), int64(sz), minio.PutObjectPartOptions{}) + if err != nil { + _ = core.AbortMultipartUpload(ctx, bucket, object, uploadID) + return want, err + } + parts = append(parts, minio.CompletePart{PartNumber: i + 1, ETag: p.ETag}) + } + _, err = core.CompleteMultipartUpload(ctx, bucket, object, uploadID, parts, minio.PutObjectOptions{}) + return want, err +} + +// TestMultipartNonUniform checks that a multipart upload whose parts are NOT a +// uniform size round-trips correctly, both with the default streaming path and +// with the in-memory fallback (--disable-multipart-streaming). +func TestMultipartNonUniform(t *testing.T) { + // Non-uniform parts, last one smaller. + partSizes := []int{120 * 1024, 100 * 1024, 53 * 1024} + const object = "non-uniform.bin" + + for _, tc := range []struct { + name string + disableStreaming bool + }{ + {"Streaming", false}, + {"InMemory", true}, + } { + t.Run(tc.name, func(t *testing.T) { + core, f, bucket := newMultipartTestServer(t, tc.disableStreaming) + want, err := multipartUploadParts(t, core, bucket, object, partSizes) + require.NoError(t, err) + assert.Equal(t, want, readObject(t, f, bucket, object)) + }) + } +} + +// TestMultipartOutOfOrder uploads the parts concurrently and out of order, +// exercising the reorder buffer and the in-order pump handoff. +func TestMultipartOutOfOrder(t *testing.T) { + core, f, bucket := newMultipartTestServer(t, false) + ctx := context.Background() + const object = "out-of-order.bin" + + sizes := []int{70 * 1024, 90 * 1024, 50 * 1024, 33 * 1024} + datas := make([][]byte, len(sizes)) + var want []byte + for i, sz := range sizes { + datas[i] = []byte(random.String(sz)) + want = append(want, datas[i]...) + } + + uploadID, err := core.NewMultipartUpload(ctx, bucket, object, minio.PutObjectOptions{}) + require.NoError(t, err) + + parts := make([]minio.CompletePart, len(sizes)) + errs := make([]error, len(sizes)) + var wg sync.WaitGroup + for _, i := range []int{2, 0, 3, 1} { // shuffled upload order + wg.Add(1) + go func(i int) { + defer wg.Done() + p, err := core.PutObjectPart(ctx, bucket, object, uploadID, i+1, bytes.NewReader(datas[i]), int64(sizes[i]), minio.PutObjectPartOptions{}) + errs[i] = err + parts[i] = minio.CompletePart{PartNumber: i + 1, ETag: p.ETag} + }(i) + } + wg.Wait() + for _, err := range errs { + require.NoError(t, err) + } + + _, err = core.CompleteMultipartUpload(ctx, bucket, object, uploadID, parts, minio.PutObjectOptions{}) + require.NoError(t, err) + assert.Equal(t, want, readObject(t, f, bucket, object)) +} + +// TestMultipartNonContiguous checks that a multipart upload with a gap in the +// part numbers (which the in-order stream can't place) is rejected. +func TestMultipartNonContiguous(t *testing.T) { + core, _, bucket := newMultipartTestServer(t, false) + ctx := context.Background() + const object = "gap.bin" + + uploadID, err := core.NewMultipartUpload(ctx, bucket, object, minio.PutObjectOptions{}) + require.NoError(t, err) + + var parts []minio.CompletePart + for _, pn := range []int{1, 2, 4} { // part 3 missing + data := []byte(random.String(40 * 1024)) + p, err := core.PutObjectPart(ctx, bucket, object, uploadID, pn, bytes.NewReader(data), int64(len(data)), minio.PutObjectPartOptions{}) + require.NoError(t, err) + parts = append(parts, minio.CompletePart{PartNumber: pn, ETag: p.ETag}) + } + _, err = core.CompleteMultipartUpload(ctx, bucket, object, uploadID, parts, minio.PutObjectOptions{}) + require.Error(t, err) +} + +// TestMultipartAbort checks that aborting an upload tears down the streamed +// PutStream so no object is left behind. +func TestMultipartAbort(t *testing.T) { + core, f, bucket := newMultipartTestServer(t, false) + ctx := context.Background() + const object = "aborted.bin" + + uploadID, err := core.NewMultipartUpload(ctx, bucket, object, minio.PutObjectOptions{}) + require.NoError(t, err) + data := []byte(random.String(50 * 1024)) + _, err = core.PutObjectPart(ctx, bucket, object, uploadID, 1, bytes.NewReader(data), int64(len(data)), minio.PutObjectPartOptions{}) + require.NoError(t, err) + require.NoError(t, core.AbortMultipartUpload(ctx, bucket, object, uploadID)) + + _, err = f.NewObject(ctx, path.Join(bucket, object)) + require.ErrorIs(t, err, fs.ErrorObjectNotFound) +} diff --git a/cmd/serve/s3/s3.go b/cmd/serve/s3/s3.go index d4eb711c2..70d18ce99 100644 --- a/cmd/serve/s3/s3.go +++ b/cmd/serve/s3/s3.go @@ -37,6 +37,10 @@ var OptionsInfo = fs.Options{{ Name: "no_cleanup", Default: false, Help: "Not to cleanup empty folder after object is deleted", +}, { + Name: "disable_multipart_streaming", + Default: false, + Help: "Buffer multipart uploads in memory instead of streaming them to the backend (see the Multipart uploads docs section)", }}. Add(httplib.ConfigInfo). Add(httplib.AuthConfigInfo) @@ -44,12 +48,13 @@ var OptionsInfo = fs.Options{{ // Options contains options for the s3 Server type Options struct { //TODO add more options - ForcePathStyle bool `config:"force_path_style"` - EtagHash string `config:"etag_hash"` - AuthKey []string `config:"auth_key"` - NoCleanup bool `config:"no_cleanup"` - Auth httplib.AuthConfig - HTTP httplib.Config + ForcePathStyle bool `config:"force_path_style"` + EtagHash string `config:"etag_hash"` + AuthKey []string `config:"auth_key"` + NoCleanup bool `config:"no_cleanup"` + DisableMultipartStreaming bool `config:"disable_multipart_streaming"` + Auth httplib.AuthConfig + HTTP httplib.Config } // Opt is options set by command line flags diff --git a/cmd/serve/s3/s3_test.go b/cmd/serve/s3/s3_test.go index 15b19d36d..1ebf6bf56 100644 --- a/cmd/serve/s3/s3_test.go +++ b/cmd/serve/s3/s3_test.go @@ -18,6 +18,7 @@ import ( "github.com/minio/minio-go/v7" "github.com/minio/minio-go/v7/pkg/credentials" _ "github.com/rclone/rclone/backend/local" + _ "github.com/rclone/rclone/backend/s3" // for TestS3Minio backing remote "github.com/rclone/rclone/cmd/serve/proxy" "github.com/rclone/rclone/cmd/serve/servetest" "github.com/rclone/rclone/fs" @@ -52,10 +53,10 @@ func serveS3(t *testing.T, f fs.Fs) (testURL string, keyid string, keysec string return } -// TestS3 runs the s3 server then runs the unit tests for the -// s3 remote against it. -func TestS3(t *testing.T) { - start := func(f fs.Fs) (configmap.Simple, func()) { +// startS3 builds the start callback that brings up a serve s3 server +// wrapping f and returns the client config for connecting to it. +func startS3(t *testing.T) servetest.StartFn { + return func(f fs.Fs) (configmap.Simple, func()) { testURL, keyid, keysec, _ := serveS3(t, f) // Config for the backend we'll use to connect to the server config := configmap.Simple{ @@ -65,11 +66,25 @@ func TestS3(t *testing.T) { "access_key_id": keyid, "secret_access_key": keysec, } - return config, func() {} } +} - servetest.Run(t, "s3", start) +// TestS3 runs the s3 server backed by a local directory then runs the +// s3 backend integration tests against it. The local backend only +// supports OpenWriterAt, so this exercises that streaming path in +// serve s3. +func TestS3(t *testing.T) { + servetest.Run(t, "s3", startS3(t)) +} + +// TestS3Minio runs the s3 server backed by a minio docker container +// (via fstest/testserver/init.d/TestS3Minio) then runs the s3 backend +// integration tests against it. Minio supports OpenChunkWriter, so +// this exercises that streaming path in serve s3 - the path the local +// backing in TestS3 cannot cover. +func TestS3Minio(t *testing.T) { + servetest.RunWithBackend(t, "s3", startS3(t), "TestS3Minio:") } // tests using the minio client diff --git a/cmd/serve/s3/serve_s3.md b/cmd/serve/s3/serve_s3.md index a1281b515..321d144f6 100644 --- a/cmd/serve/s3/serve_s3.md +++ b/cmd/serve/s3/serve_s3.md @@ -86,19 +86,65 @@ provider = Rclone endpoint = http://127.0.0.1:8080/ access_key_id = ACCESS_KEY_ID secret_access_key = SECRET_ACCESS_KEY -use_multipart_uploads = false ``` -Note that setting `use_multipart_uploads = false` is to work around -[a bug](#bugs) which will be fixed in due course. +### Multipart uploads + +By default `serve s3` **streams** each multipart upload, in part-number +order, into a single `PutStream` upload to the underlying remote, so the +whole file is never buffered in memory - memory use stays bounded by the +parts in flight. The remote then performs its own internal upload (for +example its own multipart upload, still with bounded memory). This works +for any remote that supports `PutStream`, which is nearly all of them, +including through `crypt`. + +**Advantages** + +- The whole object is never buffered in memory; memory use is bounded by + the parts in flight, not the upload size. +- Parts can be any size. Clients that don't produce uniform-sized parts + work fine - for example PostgreSQL backup tools such as **pgBarman** + and **pgBackRest**, which flush an upload buffer once it grows past + the chunk size, so each part is the chunk size plus a variable + overshoot. +- Works through `crypt` for any part size, since the object is encrypted + as one continuous stream. +- Backend-agnostic - it only needs the remote to support `PutStream`. + +**Limitations** + +- Parts must arrive in ascending, contiguous part-number order + (1, 2, 3, ...). Parts the client uploads concurrently or out of order + are buffered until their turn, so higher client upload concurrency + uses more memory; non-contiguous part numbers are rejected. Configure + the client to upload in part order, ideally with low concurrency, for + the lowest memory use. +- No per-part retry. Once a part has been streamed it is committed, so a + failure partway through aborts the whole upload and the client must + start it again, rather than retrying a single part. (The remote's own + upload still retries its internal chunks.) +- Parts are serialised into one stream, so ingest from the client is + effectively single-threaded, although the remote's own upload still + runs concurrently. + +#### Disabling streaming + +If you pass `--disable-multipart-streaming`, or the remote doesn't +support `PutStream`, multipart uploads are instead **buffered in memory** +by the underlying S3 library: every part is held in memory and the whole +object is written out in one go when the upload completes (the previous +behaviour). This removes the in-order/contiguous-part restriction above, +so parts can be uploaded in any order, but **memory use grows with the +size of the upload**, so it is only suitable for small objects. A one-off +`NOTICE` is logged the first time this happens. + +Alternatively, if the client is an rclone `s3` remote (like the +`[serves3]` example above), you can set `use_multipart_uploads = false` +on it so it uploads each object as a single stream and skips multipart +uploads altogether. ### Bugs -When uploading multipart files `serve s3` holds all the parts in -memory (see [#7453](https://github.com/rclone/rclone/issues/7453)). -This is a limitaton of the library rclone uses for serving S3 and will -hopefully be fixed at some point. - Multipart server side copies do not work (see [#7454](https://github.com/rclone/rclone/issues/7454)). These take a very long time and eventually fail. The default threshold for diff --git a/go.mod b/go.mod index 3e1a2e70d..9ffd492c3 100644 --- a/go.mod +++ b/go.mod @@ -73,7 +73,7 @@ require ( github.com/quasilyte/go-ruleguard/dsl v0.3.23 github.com/rclone/Proton-API-Bridge v1.0.3 github.com/rclone/go-proton-api v1.0.2 - github.com/rclone/gofakes3 v0.0.6 + github.com/rclone/gofakes3 v0.0.7 github.com/rfjakob/eme v1.2.0 github.com/rivo/uniseg v0.4.7 github.com/rogpeppe/go-internal v1.14.1 diff --git a/go.sum b/go.sum index ba18403fe..ebd27549c 100644 --- a/go.sum +++ b/go.sum @@ -598,8 +598,8 @@ github.com/rclone/Proton-API-Bridge v1.0.3 h1:Bs7RC4xCFSN0BPIYVda/BNxp0qo3NV0gB2 github.com/rclone/Proton-API-Bridge v1.0.3/go.mod h1:26RAest751Ofk+F/d8xtl4UyWXrZvMQwn39U8rm/WKM= github.com/rclone/go-proton-api v1.0.2 h1:cJtJUab0MGJ3C6q5kiEJs3pbyhSLnOKMyYOQehA0PBc= github.com/rclone/go-proton-api v1.0.2/go.mod h1:LB2kCEaZMzNn3ocdz+qYfxXmuLxxN0ka62KJd2x53Bc= -github.com/rclone/gofakes3 v0.0.6 h1:gTW5Dq04EouaoKyyTSuSL+WS3VZU3k2QNOBn9ghX7kE= -github.com/rclone/gofakes3 v0.0.6/go.mod h1:M/uFc52GcEFuIP4stgrSR99SqqqxMR0ncgiTbj0eszc= +github.com/rclone/gofakes3 v0.0.7 h1:5AeROIAFbhp/7+f7BSsc8LzDbnx/WxgjV22SAUKmHp4= +github.com/rclone/gofakes3 v0.0.7/go.mod h1:M/uFc52GcEFuIP4stgrSR99SqqqxMR0ncgiTbj0eszc= github.com/relvacode/iso8601 v1.7.0 h1:BXy+V60stMP6cpswc+a93Mq3e65PfXCgDFfhvNNGrdo= github.com/relvacode/iso8601 v1.7.0/go.mod h1:FlNp+jz+TXpyRqgmM7tnzHHzBnz776kmAH2h3sZCn0I= github.com/rfjakob/eme v1.2.0 h1:8dAHL+WVAw06+7DkRKnRiFp1JL3QjcJEZFqDnndUaSI=