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

docs/storage.md

@@ -63,6 +63,7 @@ config:
   trace:
     enable: false
   part_size: 0
+  prefix: ""
 ```
 
 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.

pkg/objstore/s3/s3.go

@@ -50,6 +50,7 @@ type Config struct {
 	HTTPConfig      HTTPConfig        `yaml:"http_config"`
 	TraceConfig     TraceConfig       `yaml:"trace"`
 	PartSize        uint64            `yaml:"part_size"`
+	Prefix          string            `yaml:"prefix"`
 }
 
 type TraceConfig struct {
@@ -71,6 +72,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.
@@ -92,6 +94,11 @@ func parseConfig(conf []byte) (Config, error) {
 		config.PartSize = defaultMinPartSize
 	}
 
+	// here for testing
+	if config.Prefix != "" {
+		config.Prefix = strings.TrimSuffix(config.Prefix, DirDelim) + DirDelim
+	}
+
 	return config, nil
 }
 
@@ -105,7 +112,7 @@ func NewBucket(logger log.Logger, conf []byte, component string) (*Bucket, error
 	return NewBucketWithConfig(logger, config, component)
 }
 
-// NewBucket returns a new Bucket using the provided s3 config values.
+// NewBucketWithConfig returns a new Bucket using the provided s3 config values.
 func NewBucketWithConfig(logger log.Logger, config Config, component string) (*Bucket, error) {
 	var chain []credentials.Provider
 
@@ -186,6 +193,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
 }
@@ -208,6 +216,17 @@ 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.")
 	}
+
+	// ensure that the prefix isn't too long
+	if conf.Prefix != "" && len(conf.Prefix) >= 1024 {
+		return errors.New("prefix is too long (limited by Amazon at 1024 bytes long)")
+	}
+
+	// trim trailing slash and insert one
+	if conf.Prefix != "" {
+		conf.Prefix = strings.TrimSuffix(conf.Prefix, DirDelim) + DirDelim
+	}
+
 	return nil
 }
 
@@ -230,16 +249,17 @@ 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.Replace(object.Key, b.objectPrefix, "", 1)
 		// 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
 		}
-		if err := f(object.Key); err != nil {
+		if err := f(keyName); err != nil {
 			return err
 		}
 	}
@@ -254,7 +274,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
 	}
@@ -283,7 +303,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
@@ -316,7 +336,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,
 		fileSize,
 		minio.PutObjectOptions{
@@ -333,7 +353,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"