diff --git a/weed/s3api/s3api_object_handlers_copy.go b/weed/s3api/s3api_object_handlers_copy.go index bad328a12..887219145 100644 --- a/weed/s3api/s3api_object_handlers_copy.go +++ b/weed/s3api/s3api_object_handlers_copy.go @@ -1053,7 +1053,18 @@ func (s3a *S3ApiServer) copySingleChunk(chunk *filer_pb.FileChunk, dstPath strin return nil, err } - // Download and upload the chunk + // Stream the chunk through io.Pipe when no in-transit transformation is + // required; this holds only ~32 KiB per copy in flight, vs. the + // chunk-sized buffers the buffered path needs. + if canStreamCopyChunk(chunk) { + if err := s3a.streamCopyChunkRange(context.Background(), srcUrl, fileId, 0, int64(chunk.Size), assignResult, chunk.IsCompressed); err != nil { + return nil, fmt.Errorf("stream chunk: %w", err) + } + return dstChunk, nil + } + + // SSE / per-chunk-cipher: bytes need to be transformed in transit, so + // download into a buffer first. chunkData, err := s3a.downloadChunkData(srcUrl, fileId, 0, int64(chunk.Size), chunk.CipherKey) if err != nil { return nil, fmt.Errorf("download chunk data: %w", err) @@ -1088,6 +1099,17 @@ func (s3a *S3ApiServer) copySingleChunkForRange(originalChunk, rangeChunk *filer overlapStart := max(rangeStart, chunkStart) offsetInChunk := overlapStart - chunkStart + // Stream the byte range through io.Pipe when there's no in-transit + // transformation required (see canStreamCopyChunk for the eligibility + // rules); this is the dominant path for Harbor-style multipart + // assemble loads, which use UploadPartCopy with a CopySourceRange. + if canStreamCopyChunk(originalChunk) { + if err := s3a.streamCopyChunkRange(context.Background(), srcUrl, fileId, offsetInChunk, int64(rangeChunk.Size), assignResult, originalChunk.IsCompressed); err != nil { + return nil, fmt.Errorf("stream chunk range: %w", err) + } + return dstChunk, nil + } + // Download and upload the chunk portion chunkData, err := s3a.downloadChunkData(srcUrl, fileId, offsetInChunk, int64(rangeChunk.Size), originalChunk.CipherKey) if err != nil { diff --git a/weed/s3api/s3api_object_handlers_copy_bench_test.go b/weed/s3api/s3api_object_handlers_copy_bench_test.go new file mode 100644 index 000000000..047585cc1 --- /dev/null +++ b/weed/s3api/s3api_object_handlers_copy_bench_test.go @@ -0,0 +1,186 @@ +package s3api + +import ( + "context" + "crypto/sha256" + "fmt" + "io" + "mime/multipart" + "net" + "net/http" + "net/http/httptest" + "strconv" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + util_http "github.com/seaweedfs/seaweedfs/weed/util/http" +) + +// benchEnv stands up a fake source volume (returns N bytes on GET) and a +// fake destination volume (parses an incoming multipart POST and discards +// the body, asserting the data arrived intact via SHA-256). It returns the +// URLs and an AssignVolumeResponse pointing at the destination. +type benchEnv struct { + srcSrv *httptest.Server + dstSrv *httptest.Server + payload []byte + dstHash [32]byte + assign *filer_pb.AssignVolumeResponse +} + +func newBenchEnv(b *testing.B, payloadSize int) *benchEnv { + b.Helper() + payload := make([]byte, payloadSize) + for i := range payload { + payload[i] = byte(i*31 + 7) // arbitrary repeatable pattern + } + wantHash := sha256.Sum256(payload) + + srcSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Honor Range header if present (UploadPartCopy passes one). + start, end := int64(0), int64(len(payload)-1) + if rng := r.Header.Get("Range"); rng != "" { + var s, e int64 + if _, err := fmt.Sscanf(rng, "bytes=%d-%d", &s, &e); err == nil { + start, end = s, e + } + } + w.Header().Set("Content-Length", strconv.FormatInt(end-start+1, 10)) + w.WriteHeader(http.StatusOK) + _, _ = w.Write(payload[start : end+1]) + })) + + dstSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mr, err := r.MultipartReader() + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + part, err := mr.NextPart() + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + h := sha256.New() + if _, err := io.Copy(h, part); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + var got [32]byte + copy(got[:], h.Sum(nil)) + if got != wantHash { + http.Error(w, "hash mismatch", http.StatusBadRequest) + return + } + w.WriteHeader(http.StatusCreated) + _, _ = io.WriteString(w, `{"size":`+strconv.Itoa(payloadSize)+`}`) + })) + + dstURL, _ := neturl(dstSrv.URL) + assign := &filer_pb.AssignVolumeResponse{ + FileId: "1,benchmark", + Location: &filer_pb.Location{ + Url: dstURL, + PublicUrl: dstURL, + }, + Auth: "", + } + + return &benchEnv{ + srcSrv: srcSrv, + dstSrv: dstSrv, + payload: payload, + dstHash: wantHash, + assign: assign, + } +} + +func (e *benchEnv) close() { + e.srcSrv.Close() + e.dstSrv.Close() +} + +// neturl extracts host:port from "http://host:port". +func neturl(rawURL string) (string, error) { + const prefix = "http://" + if len(rawURL) < len(prefix) || rawURL[:len(prefix)] != prefix { + return "", fmt.Errorf("unexpected URL: %q", rawURL) + } + host := rawURL[len(prefix):] + if _, _, err := net.SplitHostPort(host); err != nil { + return "", err + } + return host, nil +} + +// BenchmarkCopyChunk_Buffered measures the cost of the existing buffered +// chunk-copy path (downloadChunkData + uploadChunkData). The path +// allocates one chunk-sized []byte and one multipart-encoded buffer per +// call. +func BenchmarkCopyChunk_Buffered(b *testing.B) { + for _, size := range []int{1 << 20, 8 << 20, 64 << 20} { + b.Run(humanByteName(size), func(b *testing.B) { + util_http.InitGlobalHttpClient() + env := newBenchEnv(b, size) + defer env.close() + + s3a := &S3ApiServer{} + b.ResetTimer() + b.SetBytes(int64(size)) + b.ReportAllocs() + + for i := 0; i < b.N; i++ { + data, err := s3a.downloadChunkData(env.srcSrv.URL, env.assign.FileId, 0, int64(size), nil) + if err != nil { + b.Fatalf("download: %v", err) + } + if err := s3a.uploadChunkData(data, env.assign, false); err != nil { + b.Fatalf("upload: %v", err) + } + } + }) + } +} + +// BenchmarkCopyChunk_Streamed measures the cost of the io.Pipe streaming +// chunk-copy path. The path holds only the pipe's hand-off buffer (~32 +// KiB) plus http transport buffers per call, regardless of chunk size. +func BenchmarkCopyChunk_Streamed(b *testing.B) { + for _, size := range []int{1 << 20, 8 << 20, 64 << 20} { + b.Run(humanByteName(size), func(b *testing.B) { + util_http.InitGlobalHttpClient() + env := newBenchEnv(b, size) + defer env.close() + + s3a := &S3ApiServer{} + b.ResetTimer() + b.SetBytes(int64(size)) + b.ReportAllocs() + + for i := 0; i < b.N; i++ { + if err := s3a.streamCopyChunkRange(context.Background(), + env.srcSrv.URL, env.assign.FileId, 0, int64(size), + env.assign, false); err != nil { + b.Fatalf("stream: %v", err) + } + } + }) + } +} + +func humanByteName(n int) string { + switch { + case n >= 1<<30: + return strconv.Itoa(n>>30) + "GiB" + case n >= 1<<20: + return strconv.Itoa(n>>20) + "MiB" + case n >= 1<<10: + return strconv.Itoa(n>>10) + "KiB" + default: + return strconv.Itoa(n) + "B" + } +} + +// Compile-time sanity: ensure the multipart library version we depend on +// is the one whose framing we mirror in streamCopyChunkRange. +var _ = multipart.NewWriter diff --git a/weed/s3api/s3api_object_handlers_copy_stream.go b/weed/s3api/s3api_object_handlers_copy_stream.go new file mode 100644 index 000000000..19adef011 --- /dev/null +++ b/weed/s3api/s3api_object_handlers_copy_stream.go @@ -0,0 +1,177 @@ +package s3api + +import ( + "context" + "fmt" + "io" + "math" + "mime" + "mime/multipart" + "net/http" + "net/textproto" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/security" + util_http "github.com/seaweedfs/seaweedfs/weed/util/http" +) + +// canStreamCopyChunk reports whether the chunk-copy fast path can use the +// io.Pipe streaming variant. The streaming path bypasses both the s3-side +// download buffer and the s3-side multipart-encode buffer, holding only the +// io.Pipe internal buffer (~32 KiB) per copy in flight. It does not work +// when bytes need to be transformed in transit: +// +// - SSE-C / SSE-KMS / SSE-S3: bytes need to be re-encrypted with a +// different key on the destination, so they have to land in memory. +// - Per-chunk CipherKey (per-chunk encryption from the source): the +// source bytes need to be decrypted before being re-uploaded. +// +// In both cases the caller falls back to the buffered copySingleChunk path, +// which already handles those transformations correctly. The streaming +// path is still safe for IsCompressed chunks — gzip is end-to-end and we +// just forward the compressed bytes with the Content-Encoding header. +func canStreamCopyChunk(chunk *filer_pb.FileChunk) bool { + if len(chunk.CipherKey) > 0 { + return false + } + if chunk.GetSseType() != filer_pb.SSEType_NONE { + return false + } + return true +} + +// streamCopyChunkRange copies size bytes starting at offset from srcUrl to +// the destination volume identified by assignResult, using io.Pipe to avoid +// buffering the chunk in memory. The destination volume server expects the +// same multipart wire format that operation.upload_content writes — see +// upload_content.go for the canonical version; we mirror its part headers +// here so the volume's needle.parseUpload accepts the request. +func (s3a *S3ApiServer) streamCopyChunkRange( + ctx context.Context, + srcUrl, srcFileId string, + offset, size int64, + assignResult *filer_pb.AssignVolumeResponse, + isCompressed bool, +) error { + // ReadUrlAsStream takes int for size; reject anything that would + // truncate negative on 32-bit. Mirrors the guard in downloadChunkData. + if size > int64(math.MaxInt32) { + return fmt.Errorf("chunk size %d exceeds maximum int32 size", size) + } + // Child context so a terminal error here unblocks the producer + // goroutine immediately. Without this, a failed POST closes + // pipeReader (which only fails the producer's writes), but + // ReadUrlAsStream can keep draining the source body in its read + // loop until EOF before noticing — wasting source-volume bandwidth + // and CPU. Cancelling streamCtx makes ReadUrlAsStream's per-tick + // ctx.Done() check return on the next iteration. + streamCtx, cancel := context.WithCancel(ctx) + defer cancel() + dstUrl := fmt.Sprintf("http://%s/%s", assignResult.Location.Url, assignResult.FileId) + dstJwt := security.EncodedJwt(assignResult.Auth) + srcJwt := filer.JwtForVolumeServer(srcFileId) + + // io.Pipe gives us a synchronous handoff: the producer goroutine + // builds the multipart body, the consumer (HTTP transport) reads it. + // Writes block until reads consume them, so the in-flight footprint + // is the pipe's internal hand-off buffer — not the chunk size. + pipeReader, pipeWriter := io.Pipe() + mw := multipart.NewWriter(pipeWriter) + contentType := mw.FormDataContentType() + + go func() { + // CloseWithError on the writer end propagates the failure to the + // HTTP transport reading from pipeReader, which then aborts the + // POST. Plain Close (with err==nil) sends EOF normally. + var producerErr error + defer func() { + pipeWriter.CloseWithError(producerErr) + }() + + h := make(textproto.MIMEHeader) + h.Set("Content-Disposition", mime.FormatMediaType("form-data", + map[string]string{"name": "file", "filename": ""})) + h.Set("Idempotency-Key", dstUrl) + if isCompressed { + h.Set("Content-Encoding", "gzip") + } + + fw, err := mw.CreatePart(h) + if err != nil { + producerErr = fmt.Errorf("multipart create part: %w", err) + return + } + + // ReadUrlAsStream reads in 256 KiB ticks and hands each tick to + // the callback. Forwarding directly into fw means each tick is + // flushed through the multipart writer into the pipe, where the + // HTTP transport picks it up — no per-chunk buffering on either + // side of the pipe. + var writeErr error + shouldRetry, readErr := util_http.ReadUrlAsStream(streamCtx, srcUrl, srcJwt, nil, false, false, offset, int(size), func(data []byte) { + if writeErr != nil { + return + } + if _, err := fw.Write(data); err != nil { + writeErr = err + } + }) + if writeErr != nil { + producerErr = fmt.Errorf("stream write: %w", writeErr) + return + } + if readErr != nil { + if shouldRetry { + glog.V(2).Infof("stream copy %s offset=%d size=%d: retryable error: %v", + srcUrl, offset, size, readErr) + } + producerErr = fmt.Errorf("stream read: %w", readErr) + return + } + // shouldRetry can be set without an error (e.g. ReadUrlAsStream + // surfacing a partial-read condition that the buffered path + // re-fetches). Treat it as a failed copy here too — otherwise we + // would close the multipart cleanly and let the destination POST + // succeed against a possibly-truncated body. Mirrors the explicit + // check downloadChunkData makes after ReadUrlAsStream returns. + if shouldRetry { + producerErr = fmt.Errorf("stream read %s offset=%d size=%d: retry needed", srcUrl, offset, size) + return + } + + if err := mw.Close(); err != nil { + producerErr = fmt.Errorf("multipart close: %w", err) + return + } + }() + + req, err := http.NewRequestWithContext(streamCtx, http.MethodPost, dstUrl, pipeReader) + if err != nil { + // Drain the pipe so the producer goroutine doesn't leak waiting + // on a never-read writer. + pipeReader.CloseWithError(err) + return fmt.Errorf("create POST request: %w", err) + } + req.Header.Set("Content-Type", contentType) + if dstJwt != "" { + req.Header.Set("Authorization", "BEARER "+string(dstJwt)) + } + + resp, err := util_http.GetGlobalHttpClient().Do(req) + if err != nil { + // Closing the reader unblocks the producer if it was still mid-write; + // the deferred cancel above also stops any in-flight source read. + pipeReader.CloseWithError(err) + return fmt.Errorf("POST: %w", err) + } + // CloseResponse drains and closes resp.Body for us; no manual io.Copy + // drain needed for keepalive. + defer util_http.CloseResponse(resp) + + if resp.StatusCode >= 400 { + return fmt.Errorf("POST %s: %s", dstUrl, resp.Status) + } + return nil +}