Use smaller per part size to use lower memory

pkg/objstore/s3/s3.go

@@ -66,9 +66,9 @@ var DefaultConfig = Config{
 		MaxIdleConnsPerHost:   100,
 		MaxConnsPerHost:       0,
 	},
-	// Minimum file size after which an HTTP multipart request should be used to upload objects to storage.
-	// Set to 128 MiB as in the minio client.
-	PartSize: 1024 * 1024 * 128,
+	// TODO(bwplotka): This means 32MB will be always allocated for every Upload.
+	// See: https://github.com/thanos-io/thanos/issues/3917
+	PartSize: 1024 * 1024 * 8, // 8MB.
 }
 
 // Config stores the configuration for s3 bucket.
@@ -84,7 +84,7 @@ type Config struct {
 	HTTPConfig         HTTPConfig        `yaml:"http_config"`
 	TraceConfig        TraceConfig       `yaml:"trace"`
 	ListObjectsVersion string            `yaml:"list_objects_version"`
-	// PartSize used for multipart upload. Only used if uploaded object size is known and larger than configured PartSize.
+	// PartSize used for multipart upload. Only used if uploaded object size is known and 4x larger than configured PartSize.
 	PartSize  uint64    `yaml:"part_size"`
 	SSEConfig SSEConfig `yaml:"sse_config"`
 }
@@ -449,9 +449,8 @@ func (b *Bucket) Upload(ctx context.Context, name string, r io.Reader) error {
 		size = -1
 	}
 
-	// partSize cannot be larger than object size.
 	partSize := b.partSize
-	if size < int64(partSize) {
+	if size < 4*int64(partSize) {
 		partSize = 0
 	}
 	if _, err := b.client.PutObject(
@@ -461,6 +460,7 @@ func (b *Bucket) Upload(ctx context.Context, name string, r io.Reader) error {
 		r,
 		size,
 		minio.PutObjectOptions{
+			NumThreads:           4, // Minio default.
 			PartSize:             partSize,
 			ServerSideEncryption: sse,
 			UserMetadata:         b.putUserMetadata,

pkg/objstore/s3/s3_e2e_test.go

@@ -0,0 +1,55 @@
+// Copyright (c) The Thanos Authors.
+// Licensed under the Apache License 2.0.
+
+package s3_test
+
+import (
+	"bytes"
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/cortexproject/cortex/integration/e2e"
+	e2edb "github.com/cortexproject/cortex/integration/e2e/db"
+	"github.com/go-kit/kit/log"
+	"github.com/thanos-io/thanos/pkg/objstore/s3"
+	"github.com/thanos-io/thanos/test/e2e/e2ethanos"
+
+	"github.com/thanos-io/thanos/pkg/testutil"
+)
+
+// Regression benchmark for https://github.com/thanos-io/thanos/issues/3917.
+func BenchmarkUpload(b *testing.B) {
+	b.ReportAllocs()
+	ctx := context.Background()
+
+	s, err := e2e.NewScenario("e2e_bench_mino_client")
+	testutil.Ok(b, err)
+	b.Cleanup(e2ethanos.CleanScenario(b, s))
+
+	const bucket = "test"
+	m := e2edb.NewMinio(8080, bucket)
+	testutil.Ok(b, s.StartAndWaitReady(m))
+
+	bkt, err := s3.NewBucketWithConfig(log.NewNopLogger(), s3.Config{
+		Bucket:    bucket,
+		AccessKey: e2edb.MinioAccessKey,
+		SecretKey: e2edb.MinioSecretKey,
+		Endpoint:  m.HTTPEndpoint(),
+		Insecure:  true,
+	}, "test-feed")
+	testutil.Ok(b, err)
+
+	buf := bytes.Buffer{}
+	buf.Grow(1028 * 1028 * 100) // 100MB.
+	word := "abcdefghij"
+	for i := 0; i < buf.Cap()/len(word); i++ {
+		_, _ = buf.WriteString(word)
+	}
+	str := buf.String()
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		testutil.Ok(b, bkt.Upload(ctx, "test", strings.NewReader(str)))
+	}
+}

test/e2e/e2ethanos/helpers.go

@@ -16,7 +16,7 @@ import (
 	"github.com/thanos-io/thanos/pkg/testutil"
 )
 
-func CleanScenario(t *testing.T, s *e2e.Scenario) func() {
+func CleanScenario(t testing.TB, s *e2e.Scenario) func() {
 	return func() {
 		// Make sure Clean can properly delete everything.
 		testutil.Ok(t, exec.Command("chmod", "-R", "777", s.SharedDir()).Run())