diff --git a/weed/s3api/s3api_object_handlers_copy.go b/weed/s3api/s3api_object_handlers_copy.go index 887219145..2fd32e337 100644 --- a/weed/s3api/s3api_object_handlers_copy.go +++ b/weed/s3api/s3api_object_handlers_copy.go @@ -1055,9 +1055,11 @@ func (s3a *S3ApiServer) copySingleChunk(chunk *filer_pb.FileChunk, dstPath strin // 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. + // chunk-sized buffers the buffered path needs. isFullChunk=true asks + // the source volume for compressed bytes, so a gzipped chunk is + // forwarded to the destination without anyone having to decompress. if canStreamCopyChunk(chunk) { - if err := s3a.streamCopyChunkRange(context.Background(), srcUrl, fileId, 0, int64(chunk.Size), assignResult, chunk.IsCompressed); err != nil { + if err := s3a.streamCopyChunkRange(context.Background(), srcUrl, fileId, 0, int64(chunk.Size), true /*isFullChunk*/, assignResult); err != nil { return nil, fmt.Errorf("stream chunk: %w", err) } return dstChunk, nil @@ -1103,8 +1105,16 @@ func (s3a *S3ApiServer) copySingleChunkForRange(originalChunk, rangeChunk *filer // 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. + // + // When the requested range happens to cover the entire source chunk + // exactly, switch to the full-chunk fetch mode: that asks the source + // volume for compressed bytes (Accept-Encoding: gzip) and forwards + // them as-is, avoiding the volume-side decompression that a Range + // fetch on a gzipped chunk would otherwise pay. Harbor's typical + // part-size = chunk-size assemble pattern hits this branch. if canStreamCopyChunk(originalChunk) { - if err := s3a.streamCopyChunkRange(context.Background(), srcUrl, fileId, offsetInChunk, int64(rangeChunk.Size), assignResult, originalChunk.IsCompressed); err != nil { + isFullChunk := offsetInChunk == 0 && rangeChunk.Size == originalChunk.Size + if err := s3a.streamCopyChunkRange(context.Background(), srcUrl, fileId, offsetInChunk, int64(rangeChunk.Size), isFullChunk, assignResult); err != nil { return nil, fmt.Errorf("stream chunk range: %w", err) } return dstChunk, nil diff --git a/weed/s3api/s3api_object_handlers_copy_bench_test.go b/weed/s3api/s3api_object_handlers_copy_bench_test.go index 047585cc1..31fb77383 100644 --- a/weed/s3api/s3api_object_handlers_copy_bench_test.go +++ b/weed/s3api/s3api_object_handlers_copy_bench_test.go @@ -160,7 +160,7 @@ func BenchmarkCopyChunk_Streamed(b *testing.B) { 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 { + true /*isFullChunk*/, env.assign); err != nil { b.Fatalf("stream: %v", err) } } diff --git a/weed/s3api/s3api_object_handlers_copy_stream.go b/weed/s3api/s3api_object_handlers_copy_stream.go index 19adef011..98ea6da7f 100644 --- a/weed/s3api/s3api_object_handlers_copy_stream.go +++ b/weed/s3api/s3api_object_handlers_copy_stream.go @@ -31,7 +31,8 @@ import ( // 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. +// just forward the wire bytes (with the destination Content-Encoding +// header set to whatever the source actually returned). func canStreamCopyChunk(chunk *filer_pb.FileChunk) bool { if len(chunk.CipherKey) > 0 { return false @@ -48,35 +49,96 @@ func canStreamCopyChunk(chunk *filer_pb.FileChunk) bool { // 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. +// +// isFullChunk is the load-bearing distinction: +// +// - false (partial range): a Range header is sent to the source volume, +// which serves the requested byte range. For compressed chunks the +// volume internally decompresses to satisfy the range, then returns +// raw bytes (no Content-Encoding); we forward those raw bytes. +// - true (full chunk): we ask the source volume to send compressed +// bytes via Accept-Encoding: gzip, bypass Go's auto-decompression on +// resp.Body, and forward whatever wire bytes arrived to the +// destination. When the chunk is stored gzipped that means we never +// decompress on either the source-volume or s3-server side; the +// destination receives gzipped bytes and stores them as-is. +// +// For the Harbor-style assemble repro (where each UploadPartCopy range +// matches a source chunk boundary), the full-chunk path eliminates the +// "bytes.growSlice from util.DecompressData" cost that dominates the +// volume-side heap profile after the s3-side fixes. +// +// The destination's Content-Encoding header is set from the source's +// response Content-Encoding only — never from a caller-supplied hint. +// Trusting a hint that disagrees with the wire bytes (e.g. caller says +// "compressed" but the volume returned raw bytes for a partial range) +// would label raw bytes as gzip and corrupt reads on the destination. func (s3a *S3ApiServer) streamCopyChunkRange( ctx context.Context, srcUrl, srcFileId string, offset, size int64, + isFullChunk bool, 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. + // Range header takes int64, but mirror the int32 bound from + // downloadChunkData so the two paths reject the same inputs. 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. + // Child context so a terminal error here unblocks both legs + // immediately. Without this, a failed POST closes pipeReader + // (which only fails the producer's writes), but the source GET's + // read loop would keep draining srcResp.Body in the background + // until EOF — wasting source-volume bandwidth and CPU on a copy + // that's already failed. Cancelling streamCtx tears down both the + // source GET and the destination POST when this function returns. 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. + // Source GET: built directly (not through ReadUrlAsStream) because that + // helper auto-decompresses gzipped responses, which would defeat the + // whole-chunk pass-through optimization. We want the wire bytes. + srcReq, err := http.NewRequestWithContext(streamCtx, http.MethodGet, srcUrl, nil) + if err != nil { + return fmt.Errorf("create source GET: %w", err) + } + if srcJwt != "" { + srcReq.Header.Set("Authorization", "BEARER "+srcJwt) + } + if isFullChunk { + // Manually setting Accept-Encoding tells Go's http.Transport that + // the user wants raw bytes back — Go's automatic decompression is + // only applied when it adds the header itself. The server may or + // may not actually gzip; the response Content-Encoding header is + // authoritative. + srcReq.Header.Set("Accept-Encoding", "gzip") + } else { + srcReq.Header.Set("Range", fmt.Sprintf("bytes=%d-%d", offset, offset+size-1)) + } + + srcResp, err := util_http.GetGlobalHttpClient().Do(srcReq) + if err != nil { + return fmt.Errorf("source GET: %w", err) + } + // CloseResponse drains and closes resp.Body for keepalive. + defer util_http.CloseResponse(srcResp) + + if srcResp.StatusCode >= 400 { + return fmt.Errorf("source GET %s: %s", srcUrl, srcResp.Status) + } + + // What's actually on the wire — trust this over caller hints. For range + // fetches the volume always returns raw bytes (no Content-Encoding); for + // full-chunk fetches it depends on whether the chunk is stored gzipped. + bodyIsGzipped := srcResp.Header.Get("Content-Encoding") == "gzip" + + // 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 hand-off buffer — not the chunk size. pipeReader, pipeWriter := io.Pipe() mw := multipart.NewWriter(pipeWriter) contentType := mw.FormDataContentType() @@ -94,7 +156,7 @@ func (s3a *S3ApiServer) streamCopyChunkRange( h.Set("Content-Disposition", mime.FormatMediaType("form-data", map[string]string{"name": "file", "filename": ""})) h.Set("Idempotency-Key", dstUrl) - if isCompressed { + if bodyIsGzipped { h.Set("Content-Encoding", "gzip") } @@ -104,40 +166,12 @@ func (s3a *S3ApiServer) streamCopyChunkRange( 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 := io.Copy(fw, srcResp.Body); err != nil { + if shouldLogStreamError(err) { + glog.V(2).Infof("stream copy %s offset=%d size=%d: %v", + srcUrl, offset, size, err) } - 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) + producerErr = fmt.Errorf("stream copy: %w", err) return } @@ -149,8 +183,6 @@ func (s3a *S3ApiServer) streamCopyChunkRange( 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) } @@ -166,8 +198,6 @@ func (s3a *S3ApiServer) streamCopyChunkRange( 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 { @@ -175,3 +205,10 @@ func (s3a *S3ApiServer) streamCopyChunkRange( } return nil } + +func shouldLogStreamError(err error) bool { + // io.ErrClosedPipe shows up when the HTTP transport closed pipeReader + // after the POST failed; the underlying error is reported via the POST + // path, no need to log twice. + return err != nil && err != io.ErrClosedPipe +}