Add list_objects_version to S3 config (#3312)

Signed-off-by: Sevag Hanssian <sevagh@protonmail.com>

CHANGELOG.md

@@ -19,6 +19,7 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 - [#3315](https://github.com/thanos-io/thanos/pull/3315) Query Frontend: Support results caching for label names, label values and series requests.
 - [#3346](https://github.com/thanos-io/thanos/pull/3346) Ruler UI: Fix a bug preventing the /rules endpoint from loading.
 - [#3115](https://github.com/thanos-io/thanos/pull/3115) compact: now deletes partially uploaded and blocks with deletion marks concurrently. It does that at the beginning and then every `--compact.cleanup-interval` time period. By default it is 5 minutes.
+- [#3312](https://github.com/thanos-io/thanos/pull/3312) s3: add list_objects_version config option for compatibility.
 
 ### Fixed
 - [#3257](https://github.com/thanos-io/thanos/pull/3257) Ruler: Prevent Ruler from crashing when using default DNS to lookup hosts that results in "No such hosts" errors.

docs/storage.md

@@ -90,6 +90,7 @@ config:
     insecure_skip_verify: false
   trace:
     enable: false
+  list_objects_version: ""
   part_size: 134217728
   sse_config:
     type: ""
@@ -110,6 +111,8 @@ Please refer to the documentation of [the Transport type](https://golang.org/pkg
 
 `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.
 
+Set `list_objects_version: "v1"` for S3 compatible APIs that don't support ListObjectsV2 (e.g. some versions of Ceph). Default value (`""`) is equivalent to `"v2"`.
+
 For debug and testing purposes you can set
 
 * `insecure: true` to switch to plain insecure HTTP instead of HTTPS

pkg/objstore/s3/s3.go

@@ -59,16 +59,17 @@ var DefaultConfig = Config{
 
 // Config stores the configuration for s3 bucket.
 type Config struct {
-	Bucket          string            `yaml:"bucket"`
-	Endpoint        string            `yaml:"endpoint"`
-	Region          string            `yaml:"region"`
-	AccessKey       string            `yaml:"access_key"`
-	Insecure        bool              `yaml:"insecure"`
-	SignatureV2     bool              `yaml:"signature_version2"`
-	SecretKey       string            `yaml:"secret_key"`
-	PutUserMetadata map[string]string `yaml:"put_user_metadata"`
-	HTTPConfig      HTTPConfig        `yaml:"http_config"`
-	TraceConfig     TraceConfig       `yaml:"trace"`
+	Bucket             string            `yaml:"bucket"`
+	Endpoint           string            `yaml:"endpoint"`
+	Region             string            `yaml:"region"`
+	AccessKey          string            `yaml:"access_key"`
+	Insecure           bool              `yaml:"insecure"`
+	SignatureV2        bool              `yaml:"signature_version2"`
+	SecretKey          string            `yaml:"secret_key"`
+	PutUserMetadata    map[string]string `yaml:"put_user_metadata"`
+	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  uint64    `yaml:"part_size"`
 	SSEConfig SSEConfig `yaml:"sse_config"`
@@ -137,6 +138,7 @@ type Bucket struct {
 	sse             encrypt.ServerSide
 	putUserMetadata map[string]string
 	partSize        uint64
+	listObjectsV1   bool
 }
 
 // parseConfig unmarshals a buffer into a Config with default HTTPConfig values.
@@ -246,13 +248,18 @@ func NewBucketWithConfig(logger log.Logger, config Config, component string) (*B
 		client.TraceOn(logWriter)
 	}
 
+	if config.ListObjectsVersion != "" && config.ListObjectsVersion != "v1" && config.ListObjectsVersion != "v2" {
+		return nil, errors.Errorf("Initialize s3 client list objects version: Unsupported version %q was provided. Supported values are v1, v2", config.ListObjectsVersion)
+	}
+
 	bkt := &Bucket{
 		logger:          logger,
 		name:            config.Bucket,
 		client:          client,
 		sse:             sse,
 		putUserMetadata: config.PutUserMetadata,
 		partSize:        config.PartSize,
+		listObjectsV1:   config.ListObjectsVersion == "v1",
 	}
 	return bkt, nil
 }
@@ -309,6 +316,7 @@ func (b *Bucket) Iter(ctx context.Context, dir string, f func(string) error) err
 	opts := minio.ListObjectsOptions{
 		Prefix:    dir,
 		Recursive: false,
+		UseV1:     b.listObjectsV1,
 	}
 
 	for object := range b.client.ListObjects(ctx, b.name, opts) {

pkg/objstore/s3/s3_test.go

@@ -233,3 +233,26 @@ http_config:
 	_, err := parseConfig(input)
 	testutil.NotOk(t, err)
 }
+
+func TestParseConfig_ListObjectsV1(t *testing.T) {
+	input := []byte(`bucket: "bucket-name"
+endpoint: "s3-endpoint"`)
+
+	cfg, err := parseConfig(input)
+	testutil.Ok(t, err)
+
+	if cfg.ListObjectsVersion != "" {
+		t.Errorf("when list_objects_version not set, it should default to empty")
+	}
+
+	input2 := []byte(`bucket: "bucket-name"
+endpoint: "s3-endpoint"
+list_objects_version: "abcd"`)
+
+	cfg2, err := parseConfig(input2)
+	testutil.Ok(t, err)
+
+	if cfg2.ListObjectsVersion != "abcd" {
+		t.Errorf("parsing of list_objects_version failed: got %v, expected %v", cfg.ListObjectsVersion, "abcd")
+	}
+}