Multipart features

docs/storage.md

@@ -62,6 +62,7 @@ config:
     insecure_skip_verify: false
   trace:
     enable: false
+  part_size: 0
 ```
 
 At a minimum, you will need to provide a value for the `bucket`, `endpoint`, `access_key`, and `secret_key` keys. The rest of the keys are optional.
@@ -74,6 +75,8 @@ You can configure the timeout settings for the HTTP client by setting the `http_
 
 Please refer to the documentation of [the Transport type](https://golang.org/pkg/net/http/#Transport) in the `net/http` package for detailed information on what each option does.
 
+`part_size` is specified in bytes and refers to the minimum file size used for multipart uploads, as some custom S3 implementations may have different requirements. A value of `0` means to use a default 128 MiB size.
+
 For debug and testing purposes you can set
 
 * `insecure: true` to switch to plain insecure HTTP instead of HTTPS

pkg/objstore/s3/s3.go

@@ -32,6 +32,10 @@ import (
 // DirDelim is the delimiter used to model a directory structure in an object store bucket.
 const DirDelim = "/"
 
+// 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.
+const minPartSize = 1024 * 1024 * 128
+
 // Config stores the configuration for s3 bucket.
 type Config struct {
 	Bucket          string            `yaml:"bucket"`
@@ -45,6 +49,7 @@ type Config struct {
 	PutUserMetadata map[string]string `yaml:"put_user_metadata"`
 	HTTPConfig      HTTPConfig        `yaml:"http_config"`
 	TraceConfig     TraceConfig       `yaml:"trace"`
+	PartSize        uint64            `yaml:"part_size"`
 }
 
 type TraceConfig struct {
@@ -65,6 +70,7 @@ type Bucket struct {
 	client          *minio.Client
 	sse             encrypt.ServerSide
 	putUserMetadata map[string]string
+	partSize        uint64
 }
 
 // parseConfig unmarshals a buffer into a Config with default HTTPConfig values.
@@ -81,6 +87,12 @@ func parseConfig(conf []byte) (Config, error) {
 	if config.PutUserMetadata == nil {
 		config.PutUserMetadata = make(map[string]string)
 	}
+
+	// Use the default minPartSize if not set.
+	if config.PartSize == 0 {
+		config.PartSize = minPartSize
+	}
+
 	return config, nil
 }
 
@@ -174,6 +186,7 @@ func NewBucketWithConfig(logger log.Logger, config Config, component string) (*B
 		client:          client,
 		sse:             sse,
 		putUserMetadata: config.PutUserMetadata,
+		partSize:        config.PartSize,
 	}
 	return bkt, nil
 }
@@ -308,6 +321,7 @@ func (b *Bucket) Upload(ctx context.Context, name string, r io.Reader) error {
 		r,
 		fileSize,
 		minio.PutObjectOptions{
+			PartSize:             b.partSize,
 			ServerSideEncryption: b.sse,
 			UserMetadata:         b.putUserMetadata,
 		},

pkg/objstore/s3/s3_test.go

@@ -100,3 +100,36 @@ http_config:
 
 	testutil.Equals(t, "bucket-owner-full-control", cfg2.PutUserMetadata["X-Amz-Acl"])
 }
+
+func TestParseConfig_PartSize(t *testing.T) {
+	input := []byte(`bucket: "bucket-name"
+endpoint: "s3-endpoint"
+access_key: "access_key"
+insecure: false
+signature_version2: false
+encrypt_sse: false
+secret_key: "secret_key"
+http_config:
+  insecure_skip_verify: false
+  idle_conn_timeout: 50s`)
+
+	cfg, err := parseConfig(input)
+	testutil.Ok(t, err)
+	testutil.Assert(t, cfg.PartSize == 1024*1024*128, "when part size not set it should default to 128MiB")
+
+	input2 := []byte(`bucket: "bucket-name"
+endpoint: "s3-endpoint"
+access_key: "access_key"
+insecure: false
+signature_version2: false
+encrypt_sse: false
+secret_key: "secret_key"
+part_size: 104857600
+http_config:
+  insecure_skip_verify: false
+  idle_conn_timeout: 50s`)
+
+	cfg2, err := parseConfig(input2)
+	testutil.Ok(t, err)
+	testutil.Assert(t, cfg2.PartSize == 1024*1024*100, "when part size should be set to 100MiB")
+}