mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-09-08 09:07:18 +00:00
S3: fold large chunk lists into manifest chunks on the direct write path (#10383)
s3: fold large chunk lists into manifest chunks on the direct write path The S3 gateway uploads chunks itself and hands the filer a fully prepared entry. On the routed write path (ObjectTransaction) the filer stores that entry as-is, so a large PutObject or CompleteMultipartUpload persisted its whole flat chunk list - a 900GB object carries 120k chunk references in one entry. Manifestize on the gateway before the entry is written, the same way mount, WebDAV, and filer.copy prepare theirs. Multipart part boundaries also record byte offsets now: the stored chunk indexes stop matching the entry once the list is folded, and partNumber reads plus GetObjectAttributes prefer the offsets. Legacy index-only records still work, with bounds checks instead of a possible panic. Copy paths resolve a manifested source into data chunks before their per-chunk copy loops - copying a manifest chunk raw would store its blob as object data still pointing at the source - and a large copied list is folded again on the destination. Completion likewise resolves manifest chunks a part entry may carry (the filer folds an oversized UploadPartCopy range) before rebasing part offsets.
This commit is contained in:
Chris Lu
@@ -205,6 +205,11 @@ type multipartPartBoundary struct {
|
||||
StartChunk int `json:"start"`
|
||||
EndChunk int `json:"end"`
|
||||
ETag string `json:"etag"`
|
||||
// Byte offsets of the part within the object. Readers prefer these over
|
||||
// the chunk indexes above, which stop matching the entry's chunk list
|
||||
// once large completions fold it into manifest chunks.
|
||||
StartOffset int64 `json:"startOffset,omitempty"`
|
||||
EndOffset int64 `json:"endOffset,omitempty"` // exclusive
|
||||
}
|
||||
|
||||
type multipartSSES3Info struct {
|
||||
@@ -492,7 +497,17 @@ func (s3a *S3ApiServer) prepareMultipartCompletionState(r *http.Request, input *
|
||||
continue
|
||||
}
|
||||
|
||||
// A part entry can itself carry manifest chunks (the filer folds an
|
||||
// oversized part, e.g. a large UploadPartCopy range). Resolve them
|
||||
// first: the rebase below shifts chunk offsets, which a manifest
|
||||
// chunk cannot express.
|
||||
if flattenErr := s3a.flattenManifestChunks(r.Context(), entry); flattenErr != nil {
|
||||
glog.Errorf("completeMultipartUpload %s %s part %d resolve manifest chunks: %v", *input.Bucket, *input.UploadId, partNumber, flattenErr)
|
||||
return nil, nil, s3err.ErrInternalError
|
||||
}
|
||||
|
||||
partStartChunk := len(finalParts)
|
||||
partStartOffset := offset
|
||||
partETag := getEtagFromEntry(entry)
|
||||
|
||||
for _, chunk := range entry.GetChunks() {
|
||||
@@ -507,10 +522,12 @@ func (s3a *S3ApiServer) prepareMultipartCompletionState(r *http.Request, input *
|
||||
|
||||
partEndChunk := len(finalParts)
|
||||
partBoundaries = append(partBoundaries, multipartPartBoundary{
|
||||
PartNumber: partNumber,
|
||||
StartChunk: partStartChunk,
|
||||
EndChunk: partEndChunk,
|
||||
ETag: partETag,
|
||||
PartNumber: partNumber,
|
||||
StartChunk: partStartChunk,
|
||||
EndChunk: partEndChunk,
|
||||
ETag: partETag,
|
||||
StartOffset: partStartOffset,
|
||||
EndOffset: offset,
|
||||
})
|
||||
|
||||
found = true
|
||||
@@ -554,6 +571,11 @@ func (s3a *S3ApiServer) prepareMultipartCompletionState(r *http.Request, input *
|
||||
}
|
||||
}
|
||||
|
||||
// Fold huge completions (>filer.ManifestBatch chunks) into manifest chunks
|
||||
// so the object entry stays small. Runs after the part boundaries above
|
||||
// captured their byte offsets against the flat list.
|
||||
finalParts = s3a.manifestizeChunks(dirName+"/"+entryName, *input.Bucket, 0, finalParts)
|
||||
|
||||
return &multipartCompletionState{
|
||||
deleteEntries: deleteEntries,
|
||||
partEntries: partEntries,
|
||||
|
||||
@@ -0,0 +1,61 @@
|
||||
package s3api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"math"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/filer"
|
||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
)
|
||||
|
||||
// saveManifestChunk returns the save function MaybeManifestize uses to store
|
||||
// manifest blobs, assigning volumes against the object's real filer path so
|
||||
// placement follows the bucket's storage rules.
|
||||
func (s3a *S3ApiServer) saveManifestChunk(filePath string, bucket string, ttlSec int32) filer.SaveDataAsChunkFunctionType {
|
||||
collection := ""
|
||||
if s3a.option.FilerGroup != "" {
|
||||
collection = s3a.getCollectionName(bucket)
|
||||
}
|
||||
return func(reader io.Reader, name string, offset int64, tsNs int64, expectedDataSize uint64) (*filer_pb.FileChunk, error) {
|
||||
return filer.SaveGatewayDataAsChunk(filer.GatewayChunkUploadRequest{
|
||||
FilerClient: s3a,
|
||||
Reader: reader,
|
||||
FullPath: filePath,
|
||||
Offset: offset,
|
||||
TsNs: tsNs,
|
||||
Collection: collection,
|
||||
TtlSec: ttlSec,
|
||||
DataCenter: s3a.option.DataCenter,
|
||||
Cipher: s3a.cipher,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// manifestizeChunks folds a large flat chunk list into manifest chunks
|
||||
// (filer.ManifestBatch data chunks per manifest). On failure the flat list is
|
||||
// returned so the write can still proceed, matching the filer's soft-fail.
|
||||
func (s3a *S3ApiServer) manifestizeChunks(filePath string, bucket string, ttlSec int32, chunks []*filer_pb.FileChunk) []*filer_pb.FileChunk {
|
||||
manifested, err := filer.MaybeManifestize(s3a.saveManifestChunk(filePath, bucket, ttlSec), chunks)
|
||||
if err != nil {
|
||||
glog.V(0).Infof("MaybeManifestize %s: %v", filePath, err)
|
||||
return chunks
|
||||
}
|
||||
return manifested
|
||||
}
|
||||
|
||||
// flattenManifestChunks resolves any manifest chunks on a copy source into the
|
||||
// flat data-chunk list, so per-chunk copy logic reads real data chunks instead
|
||||
// of raw manifest blobs.
|
||||
func (s3a *S3ApiServer) flattenManifestChunks(ctx context.Context, entry *filer_pb.Entry) error {
|
||||
if entry == nil || !filer.HasChunkManifest(entry.GetChunks()) {
|
||||
return nil
|
||||
}
|
||||
dataChunks, _, err := filer.ResolveChunkManifest(ctx, s3a.createLookupFileIdFunction(), entry.GetChunks(), 0, math.MaxInt64)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
entry.Chunks = dataChunks
|
||||
return nil
|
||||
}
|
||||
@@ -853,8 +853,18 @@ func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request)
|
||||
// Note: ETag is NOT overridden - AWS S3 returns the complete object's ETag
|
||||
// even when requesting a specific part via PartNumber
|
||||
var startOffset, endOffset int64
|
||||
if partInfo != nil {
|
||||
// Use part boundaries from metadata (accurate for multi-chunk parts)
|
||||
if partInfo != nil && partInfo.EndOffset > partInfo.StartOffset {
|
||||
// Byte-offset boundaries: valid regardless of how the entry's
|
||||
// chunk list is laid out (flat or manifest chunks)
|
||||
startOffset = partInfo.StartOffset
|
||||
endOffset = partInfo.EndOffset - 1
|
||||
} else if partInfo != nil {
|
||||
// Legacy boundaries carry chunk indexes into the flat chunk list
|
||||
if partInfo.StartChunk < 0 || partInfo.EndChunk <= partInfo.StartChunk || partInfo.EndChunk > len(objectEntryForSSE.Chunks) {
|
||||
glog.Errorf("GetObject: part %d boundary chunks [%d,%d) out of range (chunks: %d)", partNumber, partInfo.StartChunk, partInfo.EndChunk, len(objectEntryForSSE.Chunks))
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||
return
|
||||
}
|
||||
startOffset = objectEntryForSSE.Chunks[partInfo.StartChunk].Offset
|
||||
lastChunk := objectEntryForSSE.Chunks[partInfo.EndChunk-1]
|
||||
endOffset = lastChunk.Offset + int64(lastChunk.Size) - 1
|
||||
@@ -3037,6 +3047,11 @@ type PartBoundaryInfo struct {
|
||||
StartChunk int `json:"start"`
|
||||
EndChunk int `json:"end"` // exclusive
|
||||
ETag string `json:"etag"`
|
||||
// Byte offsets of the part within the object; preferred over the chunk
|
||||
// indexes, which stop matching once the entry's chunk list is folded
|
||||
// into manifest chunks. Zero EndOffset means a legacy boundary record.
|
||||
StartOffset int64 `json:"startOffset,omitempty"`
|
||||
EndOffset int64 `json:"endOffset,omitempty"` // exclusive
|
||||
}
|
||||
|
||||
// rc is a helper type that wraps a Reader and Closer for proper resource cleanup
|
||||
|
||||
@@ -309,7 +309,9 @@ func (s3a *S3ApiServer) buildObjectAttributesParts(entry *filer_pb.Entry, maxPar
|
||||
}
|
||||
|
||||
var partSize int64
|
||||
if b.StartChunk >= 0 && b.EndChunk >= 0 && b.StartChunk < len(chunks) && b.EndChunk <= len(chunks) && b.StartChunk < b.EndChunk {
|
||||
if b.EndOffset > b.StartOffset {
|
||||
partSize = b.EndOffset - b.StartOffset
|
||||
} else if b.StartChunk >= 0 && b.EndChunk >= 0 && b.StartChunk < len(chunks) && b.EndChunk <= len(chunks) && b.StartChunk < b.EndChunk {
|
||||
for ci := b.StartChunk; ci < b.EndChunk; ci++ {
|
||||
partSize += int64(chunks[ci].Size)
|
||||
}
|
||||
|
||||
@@ -423,7 +423,9 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
|
||||
return
|
||||
}
|
||||
|
||||
dstEntry.Chunks = dstChunks
|
||||
// Re-fold a large copied chunk list into manifest chunks, mirroring the
|
||||
// PutObject path (no-op below filer.ManifestBatch or for SSE chunks).
|
||||
dstEntry.Chunks = s3a.manifestizeChunks(fmt.Sprintf("%s/%s", s3a.bucketDir(dstBucket), dstObject), dstBucket, 0, dstChunks)
|
||||
|
||||
// Apply destination-specific metadata (e.g., SSE-C IV and headers)
|
||||
if dstMetadata != nil {
|
||||
@@ -917,6 +919,14 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req
|
||||
entry = cachedEntry
|
||||
}
|
||||
|
||||
// The part-copy paths below iterate entry.GetChunks() per chunk, so a
|
||||
// manifested source must be resolved into its data chunks first.
|
||||
if err := s3a.flattenManifestChunks(r.Context(), entry); err != nil {
|
||||
glog.Errorf("CopyObjectPartHandler: resolve source manifest chunks %s/%s: %v", srcBucket, srcObject, err)
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||
return
|
||||
}
|
||||
|
||||
// Validate conditional copy headers
|
||||
if err := s3a.validateConditionalCopyHeaders(r, entry); err != s3err.ErrNone {
|
||||
s3err.WriteErrorResponse(w, r, err)
|
||||
|
||||
@@ -14,6 +14,13 @@ import (
|
||||
// executeUnifiedCopyStrategy executes the appropriate copy strategy based on encryption state
|
||||
// Returns chunks and destination metadata that should be applied to the destination entry
|
||||
func (s3a *S3ApiServer) executeUnifiedCopyStrategy(entry *filer_pb.Entry, r *http.Request, srcBucket, dstBucket, srcObject, dstObject string) ([]*filer_pb.FileChunk, map[string][]byte, error) {
|
||||
// The per-chunk copy paths below must see real data chunks; copying a
|
||||
// manifest chunk raw would store its blob as object data pointing at the
|
||||
// source's chunks.
|
||||
if err := s3a.flattenManifestChunks(r.Context(), entry); err != nil {
|
||||
return nil, nil, fmt.Errorf("resolve source manifest chunks: %w", err)
|
||||
}
|
||||
|
||||
// Detect encryption state (using entry-aware detection for multipart objects)
|
||||
srcPath := fmt.Sprintf("%s/%s", s3a.bucketDir(srcBucket), srcObject)
|
||||
dstPath := fmt.Sprintf("%s/%s", s3a.bucketDir(dstBucket), dstObject)
|
||||
|
||||
@@ -807,6 +807,13 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
|
||||
glog.V(3).Infof("putToFiler: storing SSE-S3 metadata - keyID=%s, raw len=%d", sseS3Key.KeyID, len(sseS3Metadata))
|
||||
}
|
||||
|
||||
// Fold large flat chunk lists into manifest chunks before creating the
|
||||
// entry. Part uploads (object == "") stay flat: completion rebases their
|
||||
// offsets into the final object, which manifest chunks cannot express.
|
||||
if object != "" {
|
||||
entry.Chunks = s3a.manifestizeChunks(filePath, bucket, lifecycleTTLSec, entry.GetChunks())
|
||||
}
|
||||
|
||||
// Step 4: Save metadata to filer via gRPC
|
||||
// Use context.Background() to ensure metadata save completes even if HTTP request is cancelled
|
||||
// This matches the chunk upload behavior and prevents orphaned chunks
|
||||
@@ -892,9 +899,15 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
|
||||
}
|
||||
|
||||
// If the entry was never created, the uploaded chunks are orphaned and must be deleted.
|
||||
if !entryCreated && len(chunkResult.FileChunks) > 0 {
|
||||
glog.Warningf("putToFiler: finalization failed, attempting to cleanup %d orphaned chunks", len(chunkResult.FileChunks))
|
||||
s3a.deleteOrphanedChunks(chunkResult.FileChunks)
|
||||
if !entryCreated {
|
||||
orphaned := chunkResult.FileChunks
|
||||
if manifestChunks, _ := filer.SeparateManifestChunks(entry.GetChunks()); len(manifestChunks) > 0 {
|
||||
orphaned = append(manifestChunks, orphaned...)
|
||||
}
|
||||
if len(orphaned) > 0 {
|
||||
glog.Warningf("putToFiler: finalization failed, attempting to cleanup %d orphaned chunks", len(orphaned))
|
||||
s3a.deleteOrphanedChunks(orphaned)
|
||||
}
|
||||
}
|
||||
|
||||
return "", createCode, SSEResponseMetadata{}
|
||||
|
||||
@@ -0,0 +1,120 @@
|
||||
package s3api
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"testing"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
||||
)
|
||||
|
||||
// The writer (multipartPartBoundary) and reader (PartBoundaryInfo) structs are
|
||||
// kept separate; their JSON keys must stay in sync, including the offset
|
||||
// fields added for manifest-chunked entries.
|
||||
func TestPartBoundaryJsonCompatibility(t *testing.T) {
|
||||
written, err := json.Marshal([]multipartPartBoundary{
|
||||
{PartNumber: 1, StartChunk: 0, EndChunk: 2, ETag: "abc", StartOffset: 0, EndOffset: 16},
|
||||
{PartNumber: 2, StartChunk: 2, EndChunk: 3, ETag: "def", StartOffset: 16, EndOffset: 24},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
var read []PartBoundaryInfo
|
||||
if err := json.Unmarshal(written, &read); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(read) != 2 {
|
||||
t.Fatalf("expected 2 boundaries, got %d", len(read))
|
||||
}
|
||||
if read[0].EndChunk != 2 || read[0].ETag != "abc" || read[0].StartOffset != 0 || read[0].EndOffset != 16 {
|
||||
t.Errorf("boundary 1 mismatch: %+v", read[0])
|
||||
}
|
||||
if read[1].StartChunk != 2 || read[1].StartOffset != 16 || read[1].EndOffset != 24 {
|
||||
t.Errorf("boundary 2 mismatch: %+v", read[1])
|
||||
}
|
||||
|
||||
// Legacy records carry no offset fields; they must unmarshal to zero so
|
||||
// readers fall back to the chunk-index path.
|
||||
var legacy []PartBoundaryInfo
|
||||
if err := json.Unmarshal([]byte(`[{"part":1,"start":0,"end":2,"etag":"abc"}]`), &legacy); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if legacy[0].StartOffset != 0 || legacy[0].EndOffset != 0 {
|
||||
t.Errorf("legacy boundary should have zero offsets: %+v", legacy[0])
|
||||
}
|
||||
}
|
||||
|
||||
func partsEntry(t *testing.T, boundaries []multipartPartBoundary, chunks []*filer_pb.FileChunk) *filer_pb.Entry {
|
||||
t.Helper()
|
||||
boundariesJSON, err := json.Marshal(boundaries)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return &filer_pb.Entry{
|
||||
Chunks: chunks,
|
||||
Extended: map[string][]byte{
|
||||
s3_constants.SeaweedFSMultipartPartBoundaries: boundariesJSON,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildObjectAttributesPartsPrefersOffsets(t *testing.T) {
|
||||
s3a := &S3ApiServer{}
|
||||
|
||||
// Entry whose chunk list was folded into a single manifest chunk: the
|
||||
// stored chunk indexes no longer address the flat list, but the byte
|
||||
// offsets still describe each part.
|
||||
entry := partsEntry(t, []multipartPartBoundary{
|
||||
{PartNumber: 1, StartChunk: 0, EndChunk: 2, StartOffset: 0, EndOffset: 16},
|
||||
{PartNumber: 2, StartChunk: 2, EndChunk: 4, StartOffset: 16, EndOffset: 40},
|
||||
}, []*filer_pb.FileChunk{
|
||||
{FileId: "1,ab", Offset: 0, Size: 40, IsChunkManifest: true},
|
||||
})
|
||||
|
||||
parts := s3a.buildObjectAttributesParts(entry, 1000, 0)
|
||||
if parts == nil || len(parts.Parts) != 2 {
|
||||
t.Fatalf("expected 2 parts, got %+v", parts)
|
||||
}
|
||||
if parts.Parts[0].Size != 16 {
|
||||
t.Errorf("part 1 size = %d, want 16", parts.Parts[0].Size)
|
||||
}
|
||||
if parts.Parts[1].Size != 24 {
|
||||
t.Errorf("part 2 size = %d, want 24", parts.Parts[1].Size)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildObjectAttributesPartsLegacyChunkIndexes(t *testing.T) {
|
||||
s3a := &S3ApiServer{}
|
||||
|
||||
chunks := []*filer_pb.FileChunk{
|
||||
{FileId: "1,ab", Offset: 0, Size: 8},
|
||||
{FileId: "1,ac", Offset: 8, Size: 8},
|
||||
{FileId: "1,ad", Offset: 16, Size: 24},
|
||||
}
|
||||
entry := partsEntry(t, []multipartPartBoundary{
|
||||
{PartNumber: 1, StartChunk: 0, EndChunk: 2},
|
||||
{PartNumber: 2, StartChunk: 2, EndChunk: 3},
|
||||
}, chunks)
|
||||
|
||||
parts := s3a.buildObjectAttributesParts(entry, 1000, 0)
|
||||
if parts == nil || len(parts.Parts) != 2 {
|
||||
t.Fatalf("expected 2 parts, got %+v", parts)
|
||||
}
|
||||
if parts.Parts[0].Size != 16 {
|
||||
t.Errorf("part 1 size = %d, want 16", parts.Parts[0].Size)
|
||||
}
|
||||
if parts.Parts[1].Size != 24 {
|
||||
t.Errorf("part 2 size = %d, want 24", parts.Parts[1].Size)
|
||||
}
|
||||
|
||||
// Out-of-range legacy indexes (e.g. metadata from a differently shaped
|
||||
// entry) must not panic; the part is reported with size 0.
|
||||
badEntry := partsEntry(t, []multipartPartBoundary{
|
||||
{PartNumber: 1, StartChunk: 5, EndChunk: 9},
|
||||
}, chunks)
|
||||
parts = s3a.buildObjectAttributesParts(badEntry, 1000, 0)
|
||||
if parts == nil || len(parts.Parts) != 1 || parts.Parts[0].Size != 0 {
|
||||
t.Fatalf("expected 1 part with size 0, got %+v", parts)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user