cache/s3: only request required upload checksums

Signed-off-by: CrazyMax <1951866+crazy-max@users.noreply.github.com>

cache/remotecache/s3/s3.go

@@ -481,8 +481,10 @@ func newS3Client(ctx context.Context, config Config) (*s3Client, error) {
 	})
 
 	return &s3Client{
-		Client:          client,
-		transferManager: transfermanager.New(client),
+		Client: client,
+		transferManager: transfermanager.New(client, func(options *transfermanager.Options) {
+			options.RequestChecksumCalculation = aws.RequestChecksumCalculationWhenRequired
+		}),
 		bucket:          config.Bucket,
 		prefix:          config.Prefix,
 		blobsPrefix:     config.BlobsPrefix,

client/client_cache_test.go

@@ -4,10 +4,13 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
+	"net/http"
 	"os"
 	"path"
 	"path/filepath"
 	"strconv"
+	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -366,6 +369,24 @@ func testBasicS3CacheImportExport(t *testing.T, sb integration.Sandbox) {
 		SecretAccessKey: "minioadmin",
 	}
 
+	var putRequests atomic.Int64
+	opts.RequestVerifier = func(r *http.Request) error {
+		if r.Method != http.MethodPut {
+			return nil
+		}
+		putRequests.Add(1)
+		if got := r.Header.Get("X-Amz-Sdk-Checksum-Algorithm"); got != "" {
+			return errors.Errorf("unexpected checksum algorithm header %q", got)
+		}
+		if got := r.Header.Get("X-Amz-Trailer"); got != "" {
+			return errors.Errorf("unexpected checksum trailer header %q", got)
+		}
+		if got := r.Header.Get("Content-Encoding"); strings.Contains(strings.ToLower(got), "aws-chunked") {
+			return errors.Errorf("unexpected aws-chunked content encoding %q", got)
+		}
+		return nil
+	}
+
 	s3Addr, s3Bucket, cleanup, err := helpers.NewMinioServer(t, sb, opts)
 	require.NoError(t, err)
 	defer cleanup()
@@ -393,6 +414,7 @@ func testBasicS3CacheImportExport(t *testing.T, sb integration.Sandbox) {
 		},
 	}
 	testBasicCacheImportExport(t, sb, []CacheOptionsEntry{im}, []CacheOptionsEntry{ex})
+	require.NotZero(t, putRequests.Load())
 }
 
 func testCacheExportCacheDeletedContent(t *testing.T, sb integration.Sandbox) {

util/testutil/helpers/minio.go

@@ -6,6 +6,9 @@ import (
 	"fmt"
 	"net"
 	"net/http"
+	"net/http/httptest"
+	"net/http/httputil"
+	"net/url"
 	"os"
 	"os/exec"
 	"testing"
@@ -24,6 +27,7 @@ type MinioOpts struct {
 	Region          string
 	AccessKeyID     string
 	SecretAccessKey string
+	RequestVerifier func(*http.Request) error
 }
 
 func NewMinioServer(t *testing.T, sb integration.Sandbox, opts MinioOpts) (address string, bucket string, cl func() error, err error) {
@@ -99,9 +103,37 @@ func NewMinioServer(t *testing.T, sb integration.Sandbox, opts MinioOpts) (addre
 	}
 	deferF.Append(traceStop)
 
+	if opts.RequestVerifier != nil {
+		proxyAddr, proxyStop, err := newMinioProxy(address, opts.RequestVerifier)
+		if err != nil {
+			return "", "", nil, err
+		}
+		deferF.Append(proxyStop)
+		address = proxyAddr
+	}
+
 	return
 }
 
+func newMinioProxy(target string, verify func(*http.Request) error) (string, func() error, error) {
+	u, err := url.Parse(target)
+	if err != nil {
+		return "", nil, err
+	}
+	proxy := httputil.NewSingleHostReverseProxy(u)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if err := verify(r); err != nil {
+			http.Error(w, err.Error(), http.StatusNotImplemented)
+			return
+		}
+		proxy.ServeHTTP(w, r)
+	}))
+	return server.URL, func() error {
+		server.Close()
+		return nil
+	}, nil
+}
+
 func waitMinio(ctx context.Context, address string, d time.Duration) error {
 	step := 1 * time.Second
 	i := 0