feat(pkg/objstore/s3): support for prefixes

docs/storage.md

@@ -90,6 +90,7 @@ config:
     insecure_skip_verify: false
   trace:
     enable: false
+  prefix: ""
   part_size: 134217728
 ```
 

pkg/objstore/s3/s3.go

@@ -55,6 +55,7 @@ type Config struct {
 	PutUserMetadata map[string]string `yaml:"put_user_metadata"`
 	HTTPConfig      HTTPConfig        `yaml:"http_config"`
 	TraceConfig     TraceConfig       `yaml:"trace"`
+	Prefix          string            `yaml:"prefix"`
 	// PartSize used for multipart upload. Only used if uploaded object size is known and larger than configured PartSize.
 	PartSize uint64 `yaml:"part_size"`
 }
@@ -78,6 +79,7 @@ type Bucket struct {
 	sse             encrypt.ServerSide
 	putUserMetadata map[string]string
 	partSize        uint64
+	objectPrefix    string
 }
 
 // parseConfig unmarshals a buffer into a Config with default HTTPConfig values.
@@ -87,6 +89,11 @@ func parseConfig(conf []byte) (Config, error) {
 		return Config{}, err
 	}
 
+	// here for testing
+	if config.Prefix != "" {
+		config.Prefix = strings.TrimSuffix(config.Prefix, DirDelim) + DirDelim
+	}
+
 	return config, nil
 }
 
@@ -180,6 +187,7 @@ func NewBucketWithConfig(logger log.Logger, config Config, component string) (*B
 		sse:             sse,
 		putUserMetadata: config.PutUserMetadata,
 		partSize:        config.PartSize,
+		objectPrefix:    config.Prefix,
 	}
 	return bkt, nil
 }
@@ -202,6 +210,12 @@ func validate(conf Config) error {
 	if conf.AccessKey != "" && conf.SecretKey == "" {
 		return errors.New("no s3 secret_key specified while access_key is present in config file; either both should be present in config or envvars/IAM should be used.")
 	}
+
+	// trim trailing slash and insert one
+	if conf.Prefix != "" {
+		conf.Prefix = strings.TrimSuffix(conf.Prefix, DirDelim) + DirDelim
+	}
+
 	return nil
 }
 
@@ -224,20 +238,21 @@ func (b *Bucket) Iter(ctx context.Context, dir string, f func(string) error) err
 		dir = strings.TrimSuffix(dir, DirDelim) + DirDelim
 	}
 
-	for object := range b.client.ListObjects(b.name, dir, false, ctx.Done()) {
+	for object := range b.client.ListObjects(b.name, b.objectPrefix+dir, false, ctx.Done()) {
+		keyName := strings.TrimPrefix(object.Key, b.objectPrefix)
 		// Catch the error when failed to list objects.
 		if object.Err != nil {
 			return object.Err
 		}
 		// This sometimes happens with empty buckets.
-		if object.Key == "" {
+		if keyName == "" {
 			continue
 		}
 		// The s3 client can also return the directory itself in the ListObjects call above.
-		if object.Key == dir {
+		if keyName == dir {
 			continue
 		}
-		if err := f(object.Key); err != nil {
+		if err := f(keyName); err != nil {
 			return err
 		}
 	}
@@ -252,7 +267,7 @@ func (b *Bucket) getRange(ctx context.Context, name string, off, length int64) (
 			return nil, err
 		}
 	}
-	r, err := b.client.GetObjectWithContext(ctx, b.name, name, *opts)
+	r, err := b.client.GetObjectWithContext(ctx, b.name, b.objectPrefix+name, *opts)
 	if err != nil {
 		return nil, err
 	}
@@ -281,7 +296,7 @@ func (b *Bucket) GetRange(ctx context.Context, name string, off, length int64) (
 
 // Exists checks if the given object exists.
 func (b *Bucket) Exists(ctx context.Context, name string) (bool, error) {
-	_, err := b.client.StatObject(b.name, name, minio.StatObjectOptions{})
+	_, err := b.client.StatObject(b.name, b.objectPrefix+name, minio.StatObjectOptions{})
 	if err != nil {
 		if b.IsObjNotFoundErr(err) {
 			return false, nil
@@ -319,7 +334,7 @@ func (b *Bucket) Upload(ctx context.Context, name string, r io.Reader) error {
 	if _, err := b.client.PutObjectWithContext(
 		ctx,
 		b.name,
-		name,
+		b.objectPrefix+name,
 		r,
 		size,
 		minio.PutObjectOptions{
@@ -336,7 +351,7 @@ func (b *Bucket) Upload(ctx context.Context, name string, r io.Reader) error {
 
 // Delete removes the object with the given name.
 func (b *Bucket) Delete(ctx context.Context, name string) error {
-	return b.client.RemoveObject(b.name, name)
+	return b.client.RemoveObject(b.name, b.objectPrefix+name)
 }
 
 // IsObjNotFoundErr returns true if error means that object is not found. Relevant to Get operations.

pkg/objstore/s3/s3_test.go

@@ -67,6 +67,28 @@ http_config:
 	}
 }
 
+func TestParseConfig_PrefixNoSlash(t *testing.T) {
+	input := []byte(`bucket: abcd
+prefix: ssss`)
+	cfg, err := parseConfig(input)
+	testutil.Ok(t, err)
+
+	if cfg.Prefix != "ssss/" {
+		t.Errorf("parsing of prefix failed: got '%s', expected 'ssss/'", cfg.Prefix)
+	}
+}
+
+func TestParseConfig_PrefixSlash(t *testing.T) {
+	input := []byte(`bucket: abcd
+prefix: ssss/`)
+	cfg, err := parseConfig(input)
+	testutil.Ok(t, err)
+
+	if cfg.Prefix != "ssss/" {
+		t.Errorf("parsing of prefix failed: got '%s', expected 'ssss/'", cfg.Prefix)
+	}
+}
+
 func TestValidate_OK(t *testing.T) {
 	input := []byte(`bucket: "bucket-name"
 endpoint: "s3-endpoint"