diff --git a/cmd/serve/s3/backend.go b/cmd/serve/s3/backend.go index a7310c5b5..c4e959a96 100644 --- a/cmd/serve/s3/backend.go +++ b/cmd/serve/s3/backend.go @@ -9,6 +9,7 @@ import ( "math" "os" "path" + "slices" "strings" "sync" "sync/atomic" @@ -42,13 +43,12 @@ const putObjectPrefix = tempObjectPrefix + "put_" // PutStream, instead of being buffered in memory by gofakes3. type s3Backend struct { s *Server - meta *lru.Cache[string, map[string]string] // object metadata by path + meta *lru.Cache[metaKey, map[string]string] // object metadata of all tenants - // multipartUploads tracks in-flight streaming multipart uploads, - // keyed by gofakes3.UploadID. - multipartUploads sync.Map + tenantsMu sync.Mutex + tenants map[string]*tenant // per-user state by tenant ID - // uploads counts the entries in multipartUploads. + // uploads counts the in-flight uploads of all tenants. uploads atomic.Int64 // warnMaxUploadsOnce logs a single NOTICE the first time a multipart @@ -66,14 +66,34 @@ type s3Backend struct { reaperStop sync.Once } +// tenant is the state kept between requests for one user of the server. +// +// Anything remembered about a user's objects or uploads must be reached +// through their tenant, as different users may be serving different +// backends. +type tenant struct { + b *s3Backend + id string + + // uploads tracks in-flight streaming multipart uploads, + // keyed by gofakes3.UploadID. + uploads sync.Map +} + +// metaKey is the key of an object's metadata in s3Backend.meta. +type metaKey struct { + tenant string // tenant ID + fp string // object path +} + // newMetadataStore makes a store for the metadata of up to maxObjects // objects, forgetting that of the least recently used after that. If // maxObjects <= 0 there is no limit. -func newMetadataStore(maxObjects int) *lru.Cache[string, map[string]string] { +func newMetadataStore(maxObjects int) *lru.Cache[metaKey, map[string]string] { if maxObjects <= 0 { maxObjects = math.MaxInt } - meta, err := lru.New[string, map[string]string](maxObjects) + meta, err := lru.New[metaKey, map[string]string](maxObjects) if err != nil { panic(err) // only fails if maxObjects <= 0 } @@ -85,11 +105,47 @@ func newBackend(s *Server) *s3Backend { return &s3Backend{ s: s, meta: newMetadataStore(s.opt.MetadataMaxObjects), + tenants: map[string]*tenant{}, budget: newBufferBudget(int64(s.opt.MultipartStreamingBufferTotal)), reaperQuit: make(chan struct{}), } } +// tenant returns the state of the user making the request in ctx. +func (b *s3Backend) tenant(ctx context.Context) *tenant { + id := "" + b.tenantsMu.Lock() + defer b.tenantsMu.Unlock() + t, ok := b.tenants[id] + if !ok { + t = &tenant{b: b, id: id} + b.tenants[id] = t + } + return t +} + +// allTenants returns the state of every user seen so far. +func (b *s3Backend) allTenants() []*tenant { + b.tenantsMu.Lock() + defer b.tenantsMu.Unlock() + return slices.Collect(maps.Values(b.tenants)) +} + +// getMeta returns the stored metadata of the object at fp. +func (t *tenant) getMeta(fp string) (meta map[string]string, ok bool) { + return t.b.meta.Get(metaKey{tenant: t.id, fp: fp}) +} + +// setMeta stores meta as the metadata of the object at fp. +func (t *tenant) setMeta(fp string, meta map[string]string) { + t.b.meta.Add(metaKey{tenant: t.id, fp: fp}, meta) +} + +// removeMeta forgets the metadata of the object at fp. +func (t *tenant) removeMeta(fp string) { + t.b.meta.Remove(metaKey{tenant: t.id, fp: fp}) +} + // ListBuckets always returns the default bucket. func (b *s3Backend) ListBuckets(ctx context.Context) ([]gofakes3.BucketInfo, error) { _vfs, err := b.s.getVFS(ctx) @@ -203,7 +259,7 @@ func (b *s3Backend) HeadObject(ctx context.Context, bucketName, objectName strin "Content-Type": mimeType, } - if metaMap, ok := b.meta.Get(fp); ok { + if metaMap, ok := b.tenant(ctx).getMeta(fp); ok { maps.Copy(meta, metaMap) } @@ -283,7 +339,7 @@ func (b *s3Backend) GetObject(ctx context.Context, bucketName, objectName string "Content-Type": mimeType, } - if metaMap, ok := b.meta.Get(fp); ok { + if metaMap, ok := b.tenant(ctx).getMeta(fp); ok { maps.Copy(meta, metaMap) } @@ -297,12 +353,12 @@ func (b *s3Backend) GetObject(ctx context.Context, bucketName, objectName string }, nil } -// storeModtime sets both "mtime" and "X-Amz-Meta-Mtime" to val in b.meta. -// Call this whenever modtime is updated. -func (b *s3Backend) storeModtime(fp string, meta map[string]string, val string) { +// storeModtime sets both "mtime" and "X-Amz-Meta-Mtime" to val in the +// stored metadata. Call this whenever modtime is updated. +func (t *tenant) storeModtime(fp string, meta map[string]string, val string) { meta["X-Amz-Meta-Mtime"] = val meta["mtime"] = val - b.meta.Add(fp, meta) + t.setMeta(fp, meta) } // TouchObject creates or updates meta on specified object. @@ -328,12 +384,12 @@ func (b *s3Backend) TouchObject(ctx context.Context, fp string, meta map[string] return result, err } - b.meta.Add(fp, meta) + b.tenant(ctx).setMeta(fp, meta) if val, ok := meta["X-Amz-Meta-Mtime"]; ok { ti, err := swift.FloatStringToTime(val) if err == nil { - b.storeModtime(fp, meta, val) + b.tenant(ctx).storeModtime(fp, meta, val) return result, _vfs.Chtimes(fp, ti, ti) } // ignore error since the file is successfully created @@ -342,7 +398,7 @@ func (b *s3Backend) TouchObject(ctx context.Context, fp string, meta map[string] if val, ok := meta["mtime"]; ok { ti, err := swift.FloatStringToTime(val) if err == nil { - b.storeModtime(fp, meta, val) + b.tenant(ctx).storeModtime(fp, meta, val) return result, _vfs.Chtimes(fp, ti, ti) } // ignore error since the file is successfully created @@ -449,12 +505,12 @@ func (b *s3Backend) PutObject( return result, err } - b.meta.Add(fp, meta) + b.tenant(ctx).setMeta(fp, meta) if val, ok := meta["X-Amz-Meta-Mtime"]; ok { ti, err := swift.FloatStringToTime(val) if err == nil { - b.storeModtime(fp, meta, val) + b.tenant(ctx).storeModtime(fp, meta, val) return result, _vfs.Chtimes(fp, ti, ti) } // ignore error since the file is successfully created @@ -463,7 +519,7 @@ func (b *s3Backend) PutObject( if val, ok := meta["mtime"]; ok { ti, err := swift.FloatStringToTime(val) if err == nil { - b.storeModtime(fp, meta, val) + b.tenant(ctx).storeModtime(fp, meta, val) return result, _vfs.Chtimes(fp, ti, ti) } // ignore error since the file is successfully created @@ -517,7 +573,7 @@ func (b *s3Backend) deleteObject(ctx context.Context, bucketName, objectName str if err := _vfs.Remove(fp); err != nil && !os.IsNotExist(err) { return err } - b.meta.Remove(fp) + b.tenant(ctx).removeMeta(fp) // FIXME: unsafe operation rmdirRecursive(fp, _vfs) @@ -592,7 +648,7 @@ func (b *s3Backend) CopyObject(ctx context.Context, srcBucket, srcKey, dstBucket return result, gofakes3.KeyNotFound(srcKey) } if srcBucket == dstBucket && srcKey == dstKey { - b.meta.Add(fp, meta) + b.tenant(ctx).setMeta(fp, meta) val, ok := meta["X-Amz-Meta-Mtime"] if !ok { @@ -605,7 +661,7 @@ func (b *s3Backend) CopyObject(ctx context.Context, srcBucket, srcKey, dstBucket if err != nil { return result, nil } - b.storeModtime(fp, meta, val) + b.tenant(ctx).storeModtime(fp, meta, val) return result, _vfs.Chtimes(fp, ti, ti) } diff --git a/cmd/serve/s3/multipart.go b/cmd/serve/s3/multipart.go index 47f863e67..1a833a677 100644 --- a/cmd/serve/s3/multipart.go +++ b/cmd/serve/s3/multipart.go @@ -61,8 +61,9 @@ type multipartUpload struct { streamFp string // path the parts are written to (fp when the remote has no server-side move or copy) meta map[string]string - fh io.WriteCloser // sink the in-order parts are written to, opened by sink (guarded by mu) - vfs *vfs.VFS // the VFS fh is created on, used for all later operations + fh io.WriteCloser // sink the in-order parts are written to, opened by sink (guarded by mu) + vfs *vfs.VFS // the VFS fh is created on, used for all later operations + tenant *tenant // the user the upload belongs to mu sync.Mutex changed chan struct{} // closed when buffered shrinks, nextPart advances or the upload closes @@ -177,9 +178,10 @@ func (up *multipartUpload) endActivity() { up.mu.Unlock() } -// loadUpload looks up an in-flight upload by ID. -func (b *s3Backend) loadUpload(uploadID gofakes3.UploadID) (*multipartUpload, error) { - v, ok := b.multipartUploads.Load(uploadID) +// loadUpload looks up an in-flight upload of the user making the request +// in ctx by ID. +func (b *s3Backend) loadUpload(ctx context.Context, uploadID gofakes3.UploadID) (*multipartUpload, error) { + v, ok := b.tenant(ctx).uploads.Load(uploadID) if !ok { return nil, gofakes3.ErrNoSuchUpload } @@ -207,15 +209,16 @@ func (b *s3Backend) releaseUpload() { b.uploads.Add(-1) } -// addUpload records a new in-flight upload, which must hold a place +// addUpload records a new in-flight upload of t, which must hold a place // reserved by reserveUpload. -func (b *s3Backend) addUpload(uploadID gofakes3.UploadID, up *multipartUpload) { - b.multipartUploads.Store(uploadID, up) +func (b *s3Backend) addUpload(t *tenant, uploadID gofakes3.UploadID, up *multipartUpload) { + up.tenant = t + t.uploads.Store(uploadID, up) } // deleteUpload removes the record of an in-flight upload, if present. -func (b *s3Backend) deleteUpload(uploadID gofakes3.UploadID) { - if _, ok := b.multipartUploads.LoadAndDelete(uploadID); ok { +func (b *s3Backend) deleteUpload(uploadID gofakes3.UploadID, up *multipartUpload) { + if _, ok := up.tenant.uploads.LoadAndDelete(uploadID); ok { b.uploads.Add(-1) } } @@ -296,7 +299,7 @@ func (b *s3Backend) CreateMultipartUpload(ctx context.Context, bucketName, objec up.budget = b.budget up.vfs = _vfs - b.addUpload(uploadID, up) + b.addUpload(b.tenant(ctx), uploadID, up) return uploadID, nil } @@ -307,7 +310,7 @@ func (b *s3Backend) CreateMultipartUpload(ctx context.Context, bucketName, objec // the stream reaches it, and a re-upload of a part already streamed is just // hashed to check it matches. 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) + up, err := b.loadUpload(ctx, uploadID) if err != nil { return "", err } @@ -832,7 +835,7 @@ func pipePart(w io.Writer, rw *pool.RW) error { // 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) + up, err := b.loadUpload(ctx, uploadID) if err != nil { return "", "", err } @@ -866,17 +869,17 @@ func (b *s3Backend) CompleteMultipartUpload(ctx context.Context, bucketName, obj return "", "", err } } - b.deleteUpload(uploadID) + b.deleteUpload(uploadID, up) - b.meta.Add(up.fp, up.meta) + up.tenant.setMeta(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) + up.tenant.storeModtime(up.fp, up.meta, val) _ = up.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) + up.tenant.storeModtime(up.fp, up.meta, val) _ = up.vfs.Chtimes(up.fp, ti, ti) } } @@ -887,11 +890,11 @@ func (b *s3Backend) CompleteMultipartUpload(ctx context.Context, bucketName, obj // AbortMultipartUpload tears down an in-progress upload, discarding any data // already received. func (b *s3Backend) AbortMultipartUpload(ctx context.Context, bucketName, objectName string, uploadID gofakes3.UploadID) error { - up, err := b.loadUpload(uploadID) + up, err := b.loadUpload(ctx, uploadID) if err != nil { return err } - defer b.deleteUpload(uploadID) + defer b.deleteUpload(uploadID, up) if err := up.abort(); err != nil { fs.Errorf(up.fp, "aborting multipart upload: %v", err) } @@ -910,7 +913,7 @@ func (b *s3Backend) failUpload(uploadID gofakes3.UploadID, up *multipartUpload) // forgetUpload aborts up, discards what it wrote and removes every record // of it, ours and gofakes3's. func (b *s3Backend) forgetUpload(uploadID gofakes3.UploadID, up *multipartUpload) { - b.deleteUpload(uploadID) + b.deleteUpload(uploadID, up) if err := up.abort(); err != nil { fs.Errorf(up.fp, "aborting multipart upload: %v", err) } @@ -947,7 +950,15 @@ func (b *s3Backend) stopReaper() { // reapExpiredUploads aborts and cleans up multipart uploads which have had // no request activity for longer than expiry. func (b *s3Backend) reapExpiredUploads(now time.Time, expiry time.Duration) { - b.multipartUploads.Range(func(key, value any) bool { + for _, t := range b.allTenants() { + b.reapExpiredTenantUploads(t, now, expiry) + } +} + +// reapExpiredTenantUploads aborts and cleans up the multipart uploads of +// t which have had no request activity for longer than expiry. +func (b *s3Backend) reapExpiredTenantUploads(t *tenant, now time.Time, expiry time.Duration) { + t.uploads.Range(func(key, value any) bool { uploadID := key.(gofakes3.UploadID) up := value.(*multipartUpload) up.mu.Lock() diff --git a/cmd/serve/s3/multipart_test.go b/cmd/serve/s3/multipart_test.go index 89258f627..3085aac1f 100644 --- a/cmd/serve/s3/multipart_test.go +++ b/cmd/serve/s3/multipart_test.go @@ -476,7 +476,7 @@ func TestMultipartBufferLimit(t *testing.T) { // CreateMultipartUpload does. func storeTestUpload(t *testing.T, b *s3Backend, uploadID gofakes3.UploadID, up *multipartUpload) { require.NoError(t, b.reserveUpload()) - b.addUpload(uploadID, up) + b.addUpload(b.tenant(context.Background()), uploadID, up) } // stubSink is a multipartUpload sink which records how it was closed. @@ -627,7 +627,7 @@ func TestMultipartCompleteRenameFailureKeepsUpload(t *testing.T) { _, _, err = b.CompleteMultipartUpload(ctx, bucket, "key", uploadID, &gofakes3.CompleteMultipartUploadRequest{}) require.Error(t, err) - _, err = b.loadUpload(uploadID) + _, err = b.loadUpload(context.Background(), uploadID) require.NoError(t, err, "the upload record must survive a retryable Complete failure") } @@ -728,11 +728,11 @@ func TestMultipartReapExpiredUploads(t *testing.T) { b.reapExpiredUploads(now, expiry) - _, err = b.loadUpload("fresh") + _, err = b.loadUpload(context.Background(), "fresh") assert.NoError(t, err, "a fresh upload must not be reaped") - _, err = b.loadUpload("busy") + _, err = b.loadUpload(context.Background(), "busy") assert.NoError(t, err, "an upload with a request in flight must not be reaped") - _, err = b.loadUpload("idle") + _, err = b.loadUpload(context.Background(), "idle") assert.ErrorIs(t, err, gofakes3.ErrNoSuchUpload, "an idle upload past the expiry must be reaped") // The reaped upload was aborted, not committed. @@ -1274,7 +1274,7 @@ func (s *bufferSink) Bytes() []byte { func newBufferSinkUpload(t *testing.T, b *s3Backend, bucket, object string) (gofakes3.UploadID, *bufferSink) { uploadID, err := b.CreateMultipartUpload(context.Background(), bucket, object, nil) require.NoError(t, err) - up, err := b.loadUpload(uploadID) + up, err := b.loadUpload(context.Background(), uploadID) require.NoError(t, err) if up.fh != nil { // This fails the VFS upload, returning errBoom @@ -1467,7 +1467,7 @@ func TestMultipartPartFailsAfterWholeBody(t *testing.T) { // a part which could never be accepted. _, err = b.UploadPart(ctx, bucket, object, uploadID, 1, int64(len(part1)), bytes.NewReader(part1)) require.ErrorIs(t, err, gofakes3.ErrNoSuchUpload) - _, err = b.loadUpload(uploadID) + _, err = b.loadUpload(context.Background(), uploadID) require.ErrorIs(t, err, gofakes3.ErrNoSuchUpload) } @@ -1582,7 +1582,7 @@ func TestMultipartSinkOpenedOnFirstPart(t *testing.T) { ctx := context.Background() sinkOpen := func(uploadID gofakes3.UploadID) bool { - up, err := b.loadUpload(uploadID) + up, err := b.loadUpload(context.Background(), uploadID) require.NoError(t, err) up.mu.Lock() defer up.mu.Unlock() @@ -1661,7 +1661,7 @@ func TestMultipartMaxUploads(t *testing.T) { assert.ErrorIs(t, err, vfs.ENOENT, "a refused upload created directories") // And so does expiring one. - up, err := b.loadUpload(id3) + up, err := b.loadUpload(context.Background(), id3) require.NoError(t, err) up.mu.Lock() up.lastUsed = time.Now().Add(-2 * time.Hour)