s3: Moved bucket-owner-full-control to be optional behind the flag. (t…

…hanos-io#710)

Fixes thanos-io#708
Fixed potential nil pointer panic - fixed wrong casting handling.

CHANGELOG.md

@@ -19,7 +19,13 @@ We use *breaking* word for marking changes that are not backward compatible (rel
 ### Fixed
 
 - [#649](https://github.com/improbable-eng/thanos/issues/649) - Fixed store label values api to add also external label values.
+- [#708](https://github.com/improbable-eng/thanos/issues/708) - `"X-Amz-Acl": "bucket-owner-full-control"` metadata for s3 upload operation is no longer set by default which was breaking some providers handled by minio client.
 
+### Changed
+
+- S3 provider:
+  - Added `put_user_metadata` option to config.
+
 ## [v0.2.1](https://github.com/improbable-eng/thanos/releases/tag/v0.2.1) - 2018.12.27
 
 ### Added

docs/storage.md

@@ -47,6 +47,7 @@ config:
   signature_version2: false
   encrypt_sse: false
   secret_key: ""
+  put_user_metadata: {}
   http_config:
     idle_conn_timeout: 0s
 ```

pkg/objstore/s3/s3.go

@@ -34,14 +34,15 @@ const DirDelim = "/"
 
 // Config stores the configuration for s3 bucket.
 type Config struct {
-	Bucket        string     `yaml:"bucket"`
-	Endpoint      string     `yaml:"endpoint"`
-	AccessKey     string     `yaml:"access_key"`
-	Insecure      bool       `yaml:"insecure"`
-	SignatureV2   bool       `yaml:"signature_version2"`
-	SSEEncryption bool       `yaml:"encrypt_sse"`
-	SecretKey     string     `yaml:"secret_key"`
-	HTTPConfig    HTTPConfig `yaml:"http_config"`
+	Bucket          string            `yaml:"bucket"`
+	Endpoint        string            `yaml:"endpoint"`
+	AccessKey       string            `yaml:"access_key"`
+	Insecure        bool              `yaml:"insecure"`
+	SignatureV2     bool              `yaml:"signature_version2"`
+	SSEEncryption   bool              `yaml:"encrypt_sse"`
+	SecretKey       string            `yaml:"secret_key"`
+	PutUserMetadata map[string]string `yaml:"put_user_metadata"`
+	HTTPConfig      HTTPConfig        `yaml:"http_config"`
 }
 
 // HTTPConfig stores the http.Transport configuration for the s3 minio client.
@@ -51,10 +52,11 @@ type HTTPConfig struct {
 
 // Bucket implements the store.Bucket interface against s3-compatible APIs.
 type Bucket struct {
-	logger log.Logger
-	name   string
-	client *minio.Client
-	sse    encrypt.ServerSide
+	logger          log.Logger
+	name            string
+	client          *minio.Client
+	sse             encrypt.ServerSide
+	putUserMetadata map[string]string
 }
 
 // parseConfig unmarshals a buffer into a Config with default HTTPConfig values.
@@ -64,6 +66,10 @@ func parseConfig(conf []byte) (Config, error) {
 	if err := yaml.Unmarshal(conf, &config); err != nil {
 		return Config{}, err
 	}
+
+	if config.PutUserMetadata == nil {
+		config.PutUserMetadata = make(map[string]string)
+	}
 	return config, nil
 }
 
@@ -144,10 +150,11 @@ func NewBucketWithConfig(logger log.Logger, config Config, component string) (*B
 	}
 
 	bkt := &Bucket{
-		logger: logger,
-		name:   config.Bucket,
-		client: client,
-		sse:    sse,
+		logger:          logger,
+		name:            config.Bucket,
+		client:          client,
+		sse:             sse,
+		putUserMetadata: config.PutUserMetadata,
 	}
 	return bkt, nil
 }
@@ -256,22 +263,40 @@ func (b *Bucket) Exists(ctx context.Context, name string) (bool, error) {
 	return true, nil
 }
 
+func (b *Bucket) guessFileSize(name string, r io.Reader) int64 {
+	if f, ok := r.(*os.File); ok {
+		fileInfo, err := f.Stat()
+		if err == nil {
+			return fileInfo.Size()
+		}
+		level.Warn(b.logger).Log("msg", "could not stat file for multipart upload", "name", name, "err", err)
+		return -1
+	}
+
+	level.Warn(b.logger).Log("msg", "could not guess file size for multipart upload", "name", name)
+	return -1
+}
+
 // Upload the contents of the reader as an object into the bucket.
 func (b *Bucket) Upload(ctx context.Context, name string, r io.Reader) error {
-	// TODO(PR #617): Consider removing requirement on pre-known length when all providers will support this.
-	fileSize := int64(-1)
-	fileInfo, err := r.(*os.File).Stat()
-	if err != nil {
-		level.Warn(b.logger).Log("msg", "could not guess file size for multipart upload", "name", name)
-	} else {
-		fileSize = fileInfo.Size()
+	// TODO(https://github.com/improbable-eng/thanos/issues/678): Remove guessing length when minio provider will support multipart upload without this.
+	fileSize := b.guessFileSize(name, r)
+
+	if _, err := b.client.PutObjectWithContext(
+		ctx,
+		b.name,
+		name,
+		r,
+		fileSize,
+		minio.PutObjectOptions{
+			ServerSideEncryption: b.sse,
+			UserMetadata:         b.putUserMetadata,
+		},
+	); err != nil {
+		return errors.Wrap(err, "upload s3 object")
 	}
 
-	_, err = b.client.PutObjectWithContext(ctx, b.name, name, r, fileSize,
-		minio.PutObjectOptions{ServerSideEncryption: b.sse, UserMetadata: map[string]string{"X-Amz-Acl": "bucket-owner-full-control"}},
-	)
-
-	return errors.Wrap(err, "upload s3 object")
+	return nil
 }
 
 // Delete removes the object with the given name.

pkg/objstore/s3/s3_test.go

@@ -41,6 +41,23 @@ http_config:
   idle_conn_timeout: 0s`)
 	cfg, err := parseConfig(input)
 	testutil.Ok(t, err)
-
 	testutil.Ok(t, validate(cfg))
+	testutil.Assert(t, cfg.PutUserMetadata != nil, "map should not be nil")
+
+	input2 := []byte(`bucket: "bucket-name"
+endpoint: "s3-endpoint"
+access_key: "access_key"
+insecure: false
+signature_version2: false
+encrypt_sse: false
+secret_key: "secret_key"
+put_user_metadata: 
+  "X-Amz-Acl": "bucket-owner-full-control"
+http_config:
+  idle_conn_timeout: 0s`)
+	cfg2, err := parseConfig(input2)
+	testutil.Ok(t, err)
+	testutil.Ok(t, validate(cfg2))
+
+	testutil.Equals(t, "bucket-owner-full-control", cfg2.PutUserMetadata["X-Amz-Acl"])
 }