From 3d246a2aeac4180b7b65a3e112542a5fe0b33679 Mon Sep 17 00:00:00 2001 From: Nick Craig-Wood Date: Tue, 9 Jun 2026 10:24:53 +0100 Subject: [PATCH] serve s3: stream multipart uploads to the backend instead of buffering in memory Previously serve s3 buffered every part of a multipart upload in memory (in the gofakes3 S3 library) and concatenated them when the upload completed, so memory use grew with the size of the upload. serve s3 now streams the parts, in part-number order, into a single PutStream upload to the underlying remote, which performs its own upload with bounded memory. The whole file is never held in memory - memory use is bounded by the parts in flight. This works for any remote that supports PutStream (nearly all, including crypt) and for any part size, so clients that don't produce uniform-sized parts (for example PostgreSQL backup tools such as pgBarman and pgBackRest) work too. Parts must arrive in ascending, contiguous part-number order; parts uploaded out of order are buffered until their turn, and there is no per-part retry (a failure aborts the whole upload). These trade-offs are documented. Passing --disable-multipart-streaming, or using a remote without PutStream, reverts to buffering the parts in memory (the previous behaviour); a one-off NOTICE is logged the first time this happens. Fixes #7453 --- cmd/serve/s3/backend.go | 12 +- cmd/serve/s3/multipart.go | 388 +++++++++++++++++++++++++++++++++ cmd/serve/s3/multipart_test.go | 198 +++++++++++++++++ cmd/serve/s3/s3.go | 17 +- cmd/serve/s3/s3_test.go | 27 ++- cmd/serve/s3/serve_s3.md | 62 +++++- go.mod | 2 +- go.sum | 4 +- 8 files changed, 686 insertions(+), 24 deletions(-) create mode 100644 cmd/serve/s3/multipart.go create mode 100644 cmd/serve/s3/multipart_test.go 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=