s3store: Add optional key prefix for metadata objects (#347)

* Add MetadataObjectPrefix field to S3Store

* Add metadataKeyWithPrefix helper function

* Use metadataKeyWithPrefix for .info and .part operations

* Add s3store tests for metadata object prefixes

* Clarify ObjectPrefix docs

pkg/s3store/s3store.go

@@ -101,10 +101,13 @@ var nonASCIIRegexp = regexp.MustCompile(`([^\x00-\x7F]|[\r\n])`)
 type S3Store struct {
 	// Bucket used to store the data in, e.g. "tusdstore.example.com"
 	Bucket string
-	// ObjectPrefix is prepended to the name of each S3 object that is created.
+	// ObjectPrefix is prepended to the name of each S3 object that is created
-	// It can be used to create a pseudo-directory structure in the bucket,
+	// to store uploaded files. It can be used to create a pseudo-directory
-	// e.g. "path/to/my/uploads".
+	// structure in the bucket, e.g. "path/to/my/uploads".
 	ObjectPrefix string
+	// MetadataObjectPrefix is prepended to the name of each .info and .part S3
+	// object that is created. If it is not set, then ObjectPrefix is used.
+	MetadataObjectPrefix string
 	// Service specifies an interface used to communicate with the S3 backend.
 	// Usually, this is an instance of github.com/aws/aws-sdk-go/service/s3.S3
 	// (http://docs.aws.amazon.com/sdk-for-go/api/service/s3/S3.html).
@@ -261,7 +264,7 @@ func (upload *s3Upload) writeInfo(ctx context.Context, info handler.FileInfo) er
 	// Create object on S3 containing information about the file
 	_, err = store.Service.PutObjectWithContext(ctx, &s3.PutObjectInput{
 		Bucket:        aws.String(store.Bucket),
-		Key:           store.keyWithPrefix(uploadId + ".info"),
+		Key:           store.metadataKeyWithPrefix(uploadId + ".info"),
 		Body:          bytes.NewReader(infoJson),
 		ContentLength: aws.Int64(int64(len(infoJson))),
 	})
@@ -395,7 +398,7 @@ func (upload s3Upload) fetchInfo(ctx context.Context) (info handler.FileInfo, er
 	// Get file info stored in separate object
 	res, err := store.Service.GetObjectWithContext(ctx, &s3.GetObjectInput{
 		Bucket: aws.String(store.Bucket),
-		Key:    store.keyWithPrefix(uploadId + ".info"),
+		Key:    store.metadataKeyWithPrefix(uploadId + ".info"),
 	})
 	if err != nil {
 		if isAwsError(err, "NoSuchKey") {
@@ -521,10 +524,10 @@ func (upload s3Upload) Terminate(ctx context.Context) error {
 						Key: store.keyWithPrefix(uploadId),
 					},
 					{
-						Key: store.keyWithPrefix(uploadId + ".part"),
+						Key: store.metadataKeyWithPrefix(uploadId + ".part"),
 					},
 					{
-						Key: store.keyWithPrefix(uploadId + ".info"),
+						Key: store.metadataKeyWithPrefix(uploadId + ".info"),
 					},
 				},
 				Quiet: aws.Bool(true),
@@ -702,7 +705,7 @@ func (store S3Store) downloadIncompletePartForUpload(ctx context.Context, upload
 func (store S3Store) getIncompletePartForUpload(ctx context.Context, uploadId string) (*s3.GetObjectOutput, error) {
 	obj, err := store.Service.GetObjectWithContext(ctx, &s3.GetObjectInput{
 		Bucket: aws.String(store.Bucket),
-		Key:    store.keyWithPrefix(uploadId + ".part"),
+		Key:    store.metadataKeyWithPrefix(uploadId + ".part"),
 	})
 
 	if err != nil && (isAwsError(err, s3.ErrCodeNoSuchKey) || isAwsError(err, "NotFound") || isAwsError(err, "AccessDenied")) {
@@ -715,7 +718,7 @@ func (store S3Store) getIncompletePartForUpload(ctx context.Context, uploadId st
 func (store S3Store) putIncompletePartForUpload(ctx context.Context, uploadId string, r io.ReadSeeker) error {
 	_, err := store.Service.PutObjectWithContext(ctx, &s3.PutObjectInput{
 		Bucket: aws.String(store.Bucket),
-		Key:    store.keyWithPrefix(uploadId + ".part"),
+		Key:    store.metadataKeyWithPrefix(uploadId + ".part"),
 		Body:   r,
 	})
 	return err
@@ -724,7 +727,7 @@ func (store S3Store) putIncompletePartForUpload(ctx context.Context, uploadId st
 func (store S3Store) deleteIncompletePartForUpload(ctx context.Context, uploadId string) error {
 	_, err := store.Service.DeleteObjectWithContext(ctx, &s3.DeleteObjectInput{
 		Bucket: aws.String(store.Bucket),
-		Key:    store.keyWithPrefix(uploadId + ".part"),
+		Key:    store.metadataKeyWithPrefix(uploadId + ".part"),
 	})
 	return err
 }
@@ -801,3 +804,15 @@ func (store S3Store) keyWithPrefix(key string) *string {
 
 	return aws.String(prefix + key)
 }
+
+func (store S3Store) metadataKeyWithPrefix(key string) *string {
+	prefix := store.MetadataObjectPrefix
+	if prefix == "" {
+		prefix = store.ObjectPrefix
+	}
+	if prefix != "" && !strings.HasSuffix(prefix, "/") {
+		prefix += "/"
+	}
+
+	return aws.String(prefix + key)
+}

pkg/s3store/s3store_test.go

@@ -120,6 +120,55 @@ func TestNewUploadWithObjectPrefix(t *testing.T) {
 	assert.NotNil(upload)
 }
 
+func TestNewUploadWithMetadataObjectPrefix(t *testing.T) {
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+	assert := assert.New(t)
+
+	s3obj := NewMockS3API(mockCtrl)
+	store := New("bucket", s3obj)
+	store.ObjectPrefix = "my/uploaded/files"
+	store.MetadataObjectPrefix = "my/metadata"
+
+	assert.Equal("bucket", store.Bucket)
+	assert.Equal(s3obj, store.Service)
+
+	s1 := "hello"
+	s2 := "men?"
+
+	gomock.InOrder(
+		s3obj.EXPECT().CreateMultipartUploadWithContext(context.Background(), &s3.CreateMultipartUploadInput{
+			Bucket: aws.String("bucket"),
+			Key:    aws.String("my/uploaded/files/uploadId"),
+			Metadata: map[string]*string{
+				"foo": &s1,
+				"bar": &s2,
+			},
+		}).Return(&s3.CreateMultipartUploadOutput{
+			UploadId: aws.String("multipartId"),
+		}, nil),
+		s3obj.EXPECT().PutObjectWithContext(context.Background(), &s3.PutObjectInput{
+			Bucket:        aws.String("bucket"),
+			Key:           aws.String("my/metadata/uploadId.info"),
+			Body:          bytes.NewReader([]byte(`{"ID":"uploadId+multipartId","Size":500,"SizeIsDeferred":false,"Offset":0,"MetaData":{"bar":"menü","foo":"hello"},"IsPartial":false,"IsFinal":false,"PartialUploads":null,"Storage":{"Bucket":"bucket","Key":"my/uploaded/files/uploadId","Type":"s3store"}}`)),
+			ContentLength: aws.Int64(int64(253)),
+		}),
+	)
+
+	info := handler.FileInfo{
+		ID:   "uploadId",
+		Size: 500,
+		MetaData: map[string]string{
+			"foo": "hello",
+			"bar": "menü",
+		},
+	}
+
+	upload, err := store.NewUpload(context.Background(), info)
+	assert.Nil(err)
+	assert.NotNil(upload)
+}
+
 func TestNewUploadLargerMaxObjectSize(t *testing.T) {
 	mockCtrl := gomock.NewController(t)
 	defer mockCtrl.Finish()
@@ -227,6 +276,72 @@ func TestGetInfo(t *testing.T) {
 	assert.Equal("my/uploaded/files/uploadId", info.Storage["Key"])
 }
 
+func TestGetInfoWithMetadataObjectPrefix(t *testing.T) {
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+	assert := assert.New(t)
+
+	s3obj := NewMockS3API(mockCtrl)
+	store := New("bucket", s3obj)
+	store.MetadataObjectPrefix = "my/metadata"
+
+	gomock.InOrder(
+		s3obj.EXPECT().GetObjectWithContext(context.Background(), &s3.GetObjectInput{
+			Bucket: aws.String("bucket"),
+			Key:    aws.String("my/metadata/uploadId.info"),
+		}).Return(&s3.GetObjectOutput{
+			Body: ioutil.NopCloser(bytes.NewReader([]byte(`{"ID":"uploadId+multipartId","Size":500,"Offset":0,"MetaData":{"bar":"menü","foo":"hello"},"IsPartial":false,"IsFinal":false,"PartialUploads":null,"Storage":{"Bucket":"bucket","Key":"my/uploaded/files/uploadId","Type":"s3store"}}`))),
+		}, nil),
+		s3obj.EXPECT().ListPartsWithContext(context.Background(), &s3.ListPartsInput{
+			Bucket:           aws.String("bucket"),
+			Key:              aws.String("uploadId"),
+			UploadId:         aws.String("multipartId"),
+			PartNumberMarker: aws.Int64(0),
+		}).Return(&s3.ListPartsOutput{
+			Parts: []*s3.Part{
+				{
+					Size: aws.Int64(100),
+				},
+				{
+					Size: aws.Int64(200),
+				},
+			},
+			NextPartNumberMarker: aws.Int64(2),
+			IsTruncated:          aws.Bool(true),
+		}, nil),
+		s3obj.EXPECT().ListPartsWithContext(context.Background(), &s3.ListPartsInput{
+			Bucket:           aws.String("bucket"),
+			Key:              aws.String("uploadId"),
+			UploadId:         aws.String("multipartId"),
+			PartNumberMarker: aws.Int64(2),
+		}).Return(&s3.ListPartsOutput{
+			Parts: []*s3.Part{
+				{
+					Size: aws.Int64(100),
+				},
+			},
+		}, nil),
+		s3obj.EXPECT().GetObjectWithContext(context.Background(), &s3.GetObjectInput{
+			Bucket: aws.String("bucket"),
+			Key:    aws.String("my/metadata/uploadId.part"),
+		}).Return(&s3.GetObjectOutput{}, awserr.New("NoSuchKey", "Not found", nil)),
+	)
+
+	upload, err := store.GetUpload(context.Background(), "uploadId+multipartId")
+	assert.Nil(err)
+
+	info, err := upload.GetInfo(context.Background())
+	assert.Nil(err)
+	assert.Equal(int64(500), info.Size)
+	assert.Equal(int64(400), info.Offset)
+	assert.Equal("uploadId+multipartId", info.ID)
+	assert.Equal("hello", info.MetaData["foo"])
+	assert.Equal("menü", info.MetaData["bar"])
+	assert.Equal("s3store", info.Storage["Type"])
+	assert.Equal("bucket", info.Storage["Bucket"])
+	assert.Equal("my/uploaded/files/uploadId", info.Storage["Key"])
+}
+
 func TestGetInfoWithIncompletePart(t *testing.T) {
 	mockCtrl := gomock.NewController(t)
 	defer mockCtrl.Finish()