mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-08-20 14:17:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

s3: place multipart part chunks by the destination object's storage rule (#10845)

Browse Source 
Multipart parts stage under /buckets/<bucket>/.uploads/<id>/, so the filer
resolved filer.conf storage rules against that path when the gateway assigned
volumes for them. A rule scoped to a key prefix - fs.configure
-locationPrefix=/buckets/b/data/ -ttl=30d - then matched a small object but not
the parts of a large one, so an object whose entry carried the rule's TTL had
its bytes spread over TTL-less volumes.

Assign part chunks against the destination object's filer path instead, the way
the x-seaweedfs-destination header made the filer resolve it before the S3 write
path moved off the filer proxy. Covers PutObjectPart and both UploadPartCopy
paths. The part entry itself is still written under .uploads, so a read-only
rule there still rejects it.

The lifecycle XML Expiration.Days TTL keeps passing 0 for parts: that rule
targets the user-visible object key and would start its clock before
CompleteMultipartUpload.
This commit is contained in:
Chris Lu
2026-08-19 20:07:57 -07:00
committed by GitHub
parent abd61de52c
commit e5edd8be3c
6 changed files with 71 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
weed/s3api/s3api_object_handlers_copy.go
+11 -9
View File
@@ -1003,7 +1003,7 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req
}
if uploadEntryHasSSE(uploadEntry) || sourceEntryHasSSE(entry) || uploadEntryHasChecksum(uploadEntry) {
etag, sseMetadata, errCode := s3a.copyObjectPartViaReencryption(r, entry, startOffset, endOffset, dstBucket, uploadID, partID, uploadEntry)
etag, sseMetadata, errCode := s3a.copyObjectPartViaReencryption(r, entry, startOffset, endOffset, dstBucket, dstObject, uploadID, partID, uploadEntry)
if errCode != s3err.ErrNone {
s3err.WriteErrorResponse(w, r, errCode)
return
@@ -1040,14 +1040,16 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req
Extended: make(map[string][]byte),
}
// The copied part lives under the destination bucket's .uploads folder.
// Assign destination volumes against that real filer path so they land in
// the destination bucket's collection. r.URL.Path is the S3 request URI
// (e.g. /bucket/key), not a filer path, so passing it would skip the
// filer's bucket-to-collection mapping and route the copied bytes to the
// default collection.
uploadDir, partName := s3a.copyPartLocation(dstBucket, uploadID, partID)
dstPartPath := uploadDir + "/" + partName
// The copied part entry lives under the destination bucket's .uploads
// folder, but its bytes become the object, so assign destination volumes
// against the destination object's filer path: that is the path filer.conf
// storage rules and the bucket-to-collection mapping are written for.
// r.URL.Path is the S3 request URI (e.g. /bucket/key), not a filer path, so
// passing it would skip the mapping and route the copied bytes to the
// default collection.
dstAssignPath := s3a.toFilerPath(dstBucket, dstObject)
// Handle zero-size files or empty ranges
if entry.Attributes.FileSize == 0 || endOffset < startOffset {
@@ -1056,7 +1058,7 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req
dstEntry.Chunks = nil
} else {
// Copy chunks that overlap with the range
dstChunks, err := s3a.copyChunksForRange(entry, startOffset, endOffset, dstPartPath)
dstChunks, err := s3a.copyChunksForRange(entry, startOffset, endOffset, dstAssignPath)
if err != nil {
glog.Errorf("CopyObjectPartHandler copy chunks error: %v", err)
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
weed/s3api/s3api_object_handlers_copy_collection_test.go
+39 -9
View File
@@ -5,6 +5,7 @@ import (
"testing"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/util"
)
@@ -29,17 +30,46 @@ func TestCopyDestinationPathResolvesBucketCollection(t *testing.T) {
t.Fatalf("S3 request URI unexpectedly mapped to collection %q; the test no longer reproduces the bug", got)
}
// UploadPartCopy assigns against the destination part path under .uploads.
uploadDir, partName := s3a.copyPartLocation(bucket, "uploadid", 1)
partPath := uploadDir + "/" + partName
if got := f.DetectBucket(util.FullPath(partPath)); got != bucket {
t.Fatalf("UploadPartCopy dst path %q resolved to collection %q, want %q", partPath, got, bucket)
}
// CopyObject (including the SSE-C paths) assigns against the destination
// object path.
// CopyObject (including the SSE-C paths) and UploadPartCopy both assign
// against the destination object path.
objPath := fmt.Sprintf("%s/%s", s3a.bucketDir(bucket), "blobs/data")
if got := f.DetectBucket(util.FullPath(objPath)); got != bucket {
t.Fatalf("CopyObject dst path %q resolved to collection %q, want %q", objPath, got, bucket)
}
// The part entry itself still lives under .uploads, which maps to the same
// collection, so the entry write is unaffected either way.
uploadDir, partName := s3a.copyPartLocation(bucket, "uploadid", 1)
partPath := uploadDir + "/" + partName
if got := f.DetectBucket(util.FullPath(partPath)); got != bucket {
t.Fatalf("UploadPartCopy part path %q resolved to collection %q, want %q", partPath, got, bucket)
}
}
// TestMultipartStorageRuleFollowsDestinationObject guards the filer.conf storage
// rule a multipart part's chunks are placed by.
//
// Parts stage under /buckets/<bucket>/.uploads/..., so a rule scoped to a key
// prefix ("place /buckets/b/data/ on a 30d TTL volume") matches the object but
// not the part. Assigning against the part path therefore scattered a large
// object's bytes onto TTL-less volumes while its entry carried the rule's TTL.
// The gateway assigns against the destination object instead, the way the
// x-seaweedfs-destination header made the filer resolve it before the S3 write
// path moved off the filer proxy.
func TestMultipartStorageRuleFollowsDestinationObject(t *testing.T) {
const bucket, object = "b", "/data/big.bin"
s3a := &S3ApiServer{option: &S3ApiServerOption{BucketsPath: "/buckets"}}
fc := filer.NewFilerConf()
fc.AddLocationConf(&filer_pb.FilerConf_PathConf{LocationPrefix: "/buckets/b/data/", Ttl: "30d"})
partPath := s3a.genPartUploadPath(bucket, "uploadid", 1)
if got := fc.MatchStorageRule(partPath).GetTtl(); got != "" {
t.Fatalf("part path %q matched ttl %q; the test no longer reproduces the gap", partPath, got)
}
// What PutObjectPart and UploadPartCopy now assign against.
dstPath := s3a.toFilerPath(bucket, object)
if got := fc.MatchStorageRule(dstPath).GetTtl(); got != "30d" {
t.Fatalf("destination path %q resolved ttl %q, want 30d", dstPath, got)
}
}
weed/s3api/s3api_object_handlers_copy_part_sse.go
+2 -2
View File
@@ -364,7 +364,7 @@ func (s3a *S3ApiServer) copyObjectPartViaReencryption(
r *http.Request,
srcEntry *filer_pb.Entry,
startOffset, endOffset int64,
dstBucket, uploadID string,
dstBucket, dstObject, uploadID string,
partID int,
uploadEntry *filer_pb.Entry,
) (etag string, sseMetadata SSEResponseMetadata, errCode s3err.ErrorCode) {
@@ -399,7 +399,7 @@ func (s3a *S3ApiServer) copyObjectPartViaReencryption(
applyDestChecksumHeaderToCopyRequest(cloned, uploadEntry)
filePath := s3a.genPartUploadPath(dstBucket, uploadID, partID)
tag, code, putSSE := s3a.putToFiler(cloned, filePath, srcReader, dstBucket, "", partID, 0, nil, false)
tag, code, putSSE := s3a.putToFiler(cloned, filePath, srcReader, dstBucket, "", partID, 0, nil, false, s3a.toFilerPath(dstBucket, dstObject))
if code != s3err.ErrNone {
return "", SSEResponseMetadata{}, code
}
weed/s3api/s3api_object_handlers_multipart.go
+3 -2
View File
@@ -459,8 +459,9 @@ func (s3a *S3ApiServer) PutObjectPartHandler(w http.ResponseWriter, r *http.Requ
// volume TTL: the rule targets the user-visible object, not the
// transient .uploads/<id>/<n> path, and a part write would otherwise
// start the TTL clock before CompleteMultipartUpload ever assembled
// the object.
etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, dataReader, bucket, "", partID, 0, nil, false)
// the object. filer.conf storage rules are the opposite case: they
// place the bytes the part becomes, so they resolve against the object.
etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, dataReader, bucket, "", partID, 0, nil, false, s3a.toFilerPath(bucket, object))
if errCode != s3err.ErrNone {
glog.Errorf("PutObjectPart: putToFiler failed with error code %v for bucket=%s, object=%s, partNumber=%d",
errCode, bucket, object, partID)
weed/s3api/s3api_object_handlers_postpolicy.go
+1 -1
View File
@@ -140,7 +140,7 @@ func (s3a *S3ApiServer) PostPolicyBucketHandler(w http.ResponseWriter, r *http.R
// fields and boundaries inflates ContentLength relative to the
// object body, which would mis-evaluate any size-filtered rule.
ttlSec := s3a.lifecycleTTLForObjectWrite(bucket, object, fileSize)
etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, fileBody, bucket, object, 1, ttlSec, nil, false)
etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, fileBody, bucket, object, 1, ttlSec, nil, false, "")
if errCode != s3err.ErrNone {
s3err.WriteErrorResponse(w, r, errCode)
weed/s3api/s3api_object_handlers_put.go
+15 -5
View File
@@ -308,7 +308,7 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
}
ttlSec := s3a.lifecycleTTLForObjectWrite(bucket, object, r.ContentLength)
etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, dataReader, bucket, object, 1, ttlSec, nil, false)
etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, dataReader, bucket, object, 1, ttlSec, nil, false, "")
if errCode != s3err.ErrNone {
s3err.WriteErrorResponse(w, r, errCode)
@@ -402,7 +402,12 @@ func (s3a *S3ApiServer) uploadChunkSize() int32 {
// pass 0 because their own keys aren't the user-visible object the rule
// targets and a part write would otherwise bind a TTL clock starting
// before CompleteMultipartUpload.
func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader io.Reader, bucket string, object string, partNumber int, lifecycleTTLSec int32, finalize *putFinalize, uniqueWritePath bool) (etag string, code s3err.ErrorCode, sseMetadata SSEResponseMetadata) {
//
// storageDestination overrides the path the filer resolves filer.conf storage
// rules against, empty meaning filePath. MPU parts stage under the bucket's
// .uploads folder but their bytes become the object, so they pass the object's
// path the way the x-seaweedfs-destination header used to carry it.
func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader io.Reader, bucket string, object string, partNumber int, lifecycleTTLSec int32, finalize *putFinalize, uniqueWritePath bool, storageDestination string) (etag string, code s3err.ErrorCode, sseMetadata SSEResponseMetadata) {
if !s3_constants.IsValidBucketName(bucket) || (object != "" && !s3_constants.IsValidObjectKey(object)) {
return "", s3err.ErrInvalidRequest, SSEResponseMetadata{}
}
@@ -506,6 +511,11 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
collection = s3a.getCollectionName(bucket)
}
assignPath := filePath
if storageDestination != "" {
assignPath = storageDestination
}
// Create assign function for chunked upload
assignFunc := func(ctx context.Context, count int, expectedDataSize uint64) (*operation.VolumeAssignRequest, *operation.AssignResult, error) {
var assignResult *filer_pb.AssignVolumeResponse
@@ -516,7 +526,7 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
Collection: collection,
DiskType: "",
DataCenter: s3a.option.DataCenter,
Path: filePath,
Path: assignPath,
ExpectedDataSize: expectedDataSize,
TtlSec: lifecycleTTLSec,
})
@@ -1382,7 +1392,7 @@ func (s3a *S3ApiServer) putSuspendedVersioningObject(r *http.Request, bucket, ob
}
return s3err.ErrNone
},
}, false)
}, false, "")
if errCode != s3err.ErrNone {
glog.Errorf("putSuspendedVersioningObject: failed to upload object: %v", errCode)
return "", errCode, SSEResponseMetadata{}
@@ -1572,7 +1582,7 @@ func (s3a *S3ApiServer) putVersionedObject(r *http.Request, bucket, object strin
// directly — versioned objects sit on regular volumes and the
// lifecycle worker handles their expiration.
etag, errCode, sseMetadata = s3a.putToFiler(r, versionFilePath, body, bucket, normalizedObject, 1, 0,
s3a.versionedFinalize(bucket, normalizedObject, versionId, versionFileName, useInvertedFormat), true)
s3a.versionedFinalize(bucket, normalizedObject, versionId, versionFileName, useInvertedFormat), true, "")
if errCode != s3err.ErrNone {
glog.Errorf("putVersionedObject: failed to upload version: %v", errCode)
return "", "", errCode, SSEResponseMetadata{}