8422 list multipart uploads

esti/s3_gateway_test.go

@@ -289,6 +289,140 @@ func TestS3IfNoneMatch(t *testing.T) {
 	}
 }
 
+func TestListMultipartUploads(t *testing.T) {
+	blockStoreType := viper.GetString(ViperBlockstoreType)
+	if blockStoreType != "s3" {
+		return
+	}
+	ctx, logger, repo := setupTest(t)
+	defer tearDownTest(repo)
+	s3Endpoint := viper.GetString("s3_endpoint")
+	s3Client := createS3Client(s3Endpoint, t)
+	multipartNumberOfParts := 3
+	multipartPartSize := 5 * 1024 * 1024
+
+	// create two objects for two mpus
+	obj1 := "object1"
+	obj2 := "object2"
+	keysPrefix := "main/"
+	key1 := keysPrefix + obj1
+	key2 := keysPrefix + obj2
+
+	input1 := &s3.CreateMultipartUploadInput{
+		Bucket: aws.String(repo),
+		Key:    aws.String(key1),
+	}
+	input2 := &s3.CreateMultipartUploadInput{
+		Bucket: aws.String(repo),
+		Key:    aws.String(key2),
+	}
+	// create first mpu
+	resp1, err := s3Client.CreateMultipartUpload(ctx, input1)
+	require.NoError(t, err, "failed to create multipart upload")
+	parts := make([][]byte, multipartNumberOfParts)
+	for i := 0; i < multipartNumberOfParts; i++ {
+		parts[i] = randstr.Bytes(multipartPartSize + i)
+	}
+
+	completedParts1 := uploadMultipartParts(t, ctx, s3Client, logger, resp1, parts, 0)
+
+	completeInput1 := &s3.CompleteMultipartUploadInput{
+		Bucket:   resp1.Bucket,
+		Key:      resp1.Key,
+		UploadId: resp1.UploadId,
+		MultipartUpload: &types.CompletedMultipartUpload{
+			Parts: completedParts1,
+		},
+	}
+	// check first mpu appears
+	output, err := s3Client.ListMultipartUploads(ctx, &s3.ListMultipartUploadsInput{Bucket: resp1.Bucket})
+	require.NoError(t, err, "error listing multiparts")
+	keys := extractUploadKeys(output)
+	require.Contains(t, keys, obj1)
+
+	// create second mpu check both appear
+	_, err = s3Client.CreateMultipartUpload(ctx, input2)
+	require.NoError(t, err, "failed to create multipart upload")
+	output, err = s3Client.ListMultipartUploads(ctx, &s3.ListMultipartUploadsInput{Bucket: resp1.Bucket})
+	keys = extractUploadKeys(output)
+	require.Contains(t, keys, obj1)
+	require.Contains(t, keys, obj2)
+
+	// testing maxuploads - only first upload should return
+	maxUploads := aws.Int32(1)
+	output, err = s3Client.ListMultipartUploads(ctx, &s3.ListMultipartUploadsInput{Bucket: resp1.Bucket, MaxUploads: maxUploads})
+	require.NoError(t, err, "failed to list multipart uploads")
+	keys = extractUploadKeys(output)
+	require.Contains(t, keys, obj1)
+	require.NotContains(t, keys, obj2)
+
+	// testing key marker and upload id marker for pagination. only records after marker should return
+	keyMarker := output.NextKeyMarker
+	uploadIDMarker := output.NextUploadIdMarker
+	output, err = s3Client.ListMultipartUploads(ctx, &s3.ListMultipartUploadsInput{Bucket: resp1.Bucket, MaxUploads: maxUploads, KeyMarker: keyMarker, UploadIdMarker: uploadIDMarker})
+	require.NoError(t, err, "failed to list multipart uploads")
+	keys = extractUploadKeys(output)
+	require.NotContains(t, keys, obj1)
+	require.Contains(t, keys, obj2)
+
+	// finish first mpu check only second appear
+	_, err = s3Client.CompleteMultipartUpload(ctx, completeInput1)
+	require.NoError(t, err, "failed to complete multipart upload")
+	output, err = s3Client.ListMultipartUploads(ctx, &s3.ListMultipartUploadsInput{Bucket: resp1.Bucket})
+	require.NoError(t, err, "error listing multiparts")
+	keys = extractUploadKeys(output)
+	require.NotContains(t, keys, obj1)
+	require.Contains(t, keys, obj2)
+
+}
+
+func TestListMultipartUploadsUnsupported(t *testing.T) {
+	blockStoreType := viper.GetString(ViperBlockstoreType)
+	if blockStoreType != "s3" {
+		return
+	}
+	ctx, _, repo := setupTest(t)
+	defer tearDownTest(repo)
+	s3Endpoint := viper.GetString("s3_endpoint")
+	s3Client := createS3Client(s3Endpoint, t)
+	Bucket := aws.String(repo)
+
+	delimiter := aws.String("/")
+	prefix := aws.String("prefix")
+	encodingType := types.EncodingTypeUrl
+
+	t.Run("Delimiter", func(t *testing.T) {
+		_, err := s3Client.ListMultipartUploads(ctx, &s3.ListMultipartUploadsInput{Bucket: Bucket, Delimiter: delimiter})
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "NotImplemented")
+	})
+
+	t.Run("Prefix", func(t *testing.T) {
+		_, err := s3Client.ListMultipartUploads(ctx, &s3.ListMultipartUploadsInput{Bucket: Bucket, Prefix: prefix})
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "NotImplemented")
+	})
+
+	t.Run("EncodingType", func(t *testing.T) {
+		_, err := s3Client.ListMultipartUploads(ctx, &s3.ListMultipartUploadsInput{Bucket: Bucket, EncodingType: encodingType})
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "NotImplemented")
+	})
+}
+
+func extractUploadKeys(output *s3.ListMultipartUploadsOutput) []string {
+	if output == nil {
+		return nil
+	}
+	keys := make([]string, 0, len(output.Uploads))
+	for _, upload := range output.Uploads {
+		if upload.Key != nil {
+			keys = append(keys, *upload.Key)
+		}
+	}
+	return keys
+}
+
 func verifyObjectInfo(t *testing.T, got minio.ObjectInfo, expectedSize int) {
 	if got.Err != nil {
 		t.Errorf("%s: %s", got.Key, got.Err)

pkg/block/adapter.go

@@ -6,6 +6,8 @@ import (
 	"net/http"
 	"net/url"
 	"time"
+
+	"github.com/aws/aws-sdk-go-v2/service/s3/types"
 )
 
 // MultipartPart single multipart information
@@ -121,6 +123,14 @@ type ListPartsResponse struct {
 	IsTruncated          bool
 }
 
+type ListMultipartUploadsResponse struct {
+	Uploads            []types.MultipartUpload
+	NextUploadIDMarker *string
+	NextKeyMarker      *string
+	IsTruncated        bool
+	MaxUploads         *int32
+}
+
 // CreateMultiPartUploadOpts contains optional arguments for
 // CreateMultiPartUpload.  These should be analogous to options on
 // some underlying storage layer.  Missing arguments are mapped to the
@@ -139,6 +149,12 @@ type ListPartsOpts struct {
 	PartNumberMarker *string
 }
 
+type ListMultipartUploadsOpts struct {
+	MaxUploads     *int32
+	UploadIDMarker *string
+	KeyMarker      *string
+}
+
 // Properties of an object stored on the underlying block store.
 // Refer to the actual underlying Adapter for which properties are
 // actually reported.
@@ -187,8 +203,9 @@ type Adapter interface {
 
 	CreateMultiPartUpload(ctx context.Context, obj ObjectPointer, r *http.Request, opts CreateMultiPartUploadOpts) (*CreateMultiPartUploadResponse, error)
 	UploadPart(ctx context.Context, obj ObjectPointer, sizeBytes int64, reader io.Reader, uploadID string, partNumber int) (*UploadPartResponse, error)
-	ListParts(ctx context.Context, obj ObjectPointer, uploadID string, opts ListPartsOpts) (*ListPartsResponse, error)
 	UploadCopyPart(ctx context.Context, sourceObj, destinationObj ObjectPointer, uploadID string, partNumber int) (*UploadPartResponse, error)
+	ListParts(ctx context.Context, obj ObjectPointer, uploadID string, opts ListPartsOpts) (*ListPartsResponse, error)
+	ListMultipartUploads(ctx context.Context, obj ObjectPointer, opts ListMultipartUploadsOpts) (*ListMultipartUploadsResponse, error)
 	UploadCopyPartRange(ctx context.Context, sourceObj, destinationObj ObjectPointer, uploadID string, partNumber int, startPosition, endPosition int64) (*UploadPartResponse, error)
 	AbortMultiPartUpload(ctx context.Context, obj ObjectPointer, uploadID string) error
 	CompleteMultiPartUpload(ctx context.Context, obj ObjectPointer, uploadID string, multipartList *MultipartUploadCompletion) (*CompleteMultiPartUploadResponse, error)

pkg/block/azure/adapter.go

@@ -644,6 +644,9 @@ func (a *Adapter) GetPresignUploadPartURL(_ context.Context, _ block.ObjectPoint
 func (a *Adapter) ListParts(_ context.Context, _ block.ObjectPointer, _ string, _ block.ListPartsOpts) (*block.ListPartsResponse, error) {
 	return nil, block.ErrOperationNotSupported
 }
+func (a *Adapter) ListMultipartUploads(_ context.Context, _ block.ObjectPointer, _ block.ListMultipartUploadsOpts) (*block.ListMultipartUploadsResponse, error) {
+	return nil, block.ErrOperationNotSupported
+}
 
 // ParseURL - parses url and extracts account name and domain. If either are not found returns an error
 func ParseURL(uri *url.URL) (accountName string, domain string, err error) {

pkg/block/gs/adapter.go

@@ -717,3 +717,6 @@ func (a *Adapter) GetPresignUploadPartURL(_ context.Context, _ block.ObjectPoint
 func (a *Adapter) ListParts(_ context.Context, _ block.ObjectPointer, _ string, _ block.ListPartsOpts) (*block.ListPartsResponse, error) {
 	return nil, block.ErrOperationNotSupported
 }
+func (a *Adapter) ListMultipartUploads(_ context.Context, _ block.ObjectPointer, _ block.ListMultipartUploadsOpts) (*block.ListMultipartUploadsResponse, error) {
+	return nil, block.ErrOperationNotSupported
+}

pkg/block/local/adapter.go

@@ -600,3 +600,6 @@ func (l *Adapter) GetPresignUploadPartURL(_ context.Context, _ block.ObjectPoint
 func (l *Adapter) ListParts(_ context.Context, _ block.ObjectPointer, _ string, _ block.ListPartsOpts) (*block.ListPartsResponse, error) {
 	return nil, block.ErrOperationNotSupported
 }
+func (l *Adapter) ListMultipartUploads(_ context.Context, _ block.ObjectPointer, _ block.ListMultipartUploadsOpts) (*block.ListMultipartUploadsResponse, error) {
+	return nil, block.ErrOperationNotSupported
+}

pkg/block/mem/adapter.go

@@ -369,3 +369,6 @@ func (a *Adapter) GetPresignUploadPartURL(_ context.Context, _ block.ObjectPoint
 func (a *Adapter) ListParts(_ context.Context, _ block.ObjectPointer, _ string, _ block.ListPartsOpts) (*block.ListPartsResponse, error) {
 	return nil, block.ErrOperationNotSupported
 }
+func (a *Adapter) ListMultipartUploads(_ context.Context, _ block.ObjectPointer, _ block.ListMultipartUploadsOpts) (*block.ListMultipartUploadsResponse, error) {
+	return nil, block.ErrOperationNotSupported
+}

pkg/block/metrics.go

@@ -85,6 +85,10 @@ func (m *MetricsAdapter) ListParts(ctx context.Context, obj ObjectPointer, uploa
 	ctx = httputil.SetClientTrace(ctx, m.adapter.BlockstoreType())
 	return m.adapter.ListParts(ctx, obj, uploadID, opts)
 }
+func (m *MetricsAdapter) ListMultipartUploads(ctx context.Context, obj ObjectPointer, opts ListMultipartUploadsOpts) (*ListMultipartUploadsResponse, error) {
+	ctx = httputil.SetClientTrace(ctx, m.adapter.BlockstoreType())
+	return m.adapter.ListMultipartUploads(ctx, obj, opts)
+}
 
 func (m *MetricsAdapter) UploadCopyPart(ctx context.Context, sourceObj, destinationObj ObjectPointer, uploadID string, partNumber int) (*UploadPartResponse, error) {
 	ctx = httputil.SetClientTrace(ctx, m.adapter.BlockstoreType())

pkg/block/s3/adapter.go

@@ -849,6 +849,44 @@ func (a *Adapter) ListParts(ctx context.Context, obj block.ObjectPointer, upload
 	return &partsResp, nil
 }
 
+func (a *Adapter) ListMultipartUploads(ctx context.Context, obj block.ObjectPointer, opts block.ListMultipartUploadsOpts) (*block.ListMultipartUploadsResponse, error) {
+	var err error
+	defer reportMetrics("ListMultipartUploads", time.Now(), nil, &err)
+	bucket, key, qualifiedKey, err := a.extractParamsFromObj(obj)
+	if err != nil {
+		return nil, err
+	}
+	input := &s3.ListMultipartUploadsInput{
+		Bucket:         aws.String(bucket),
+		Prefix:         aws.String(key),
+		MaxUploads:     opts.MaxUploads,
+		UploadIdMarker: opts.UploadIDMarker,
+		KeyMarker:      opts.KeyMarker,
+	}
+
+	lg := a.log(ctx).WithFields(logging.Fields{
+		"qualified_ns":  qualifiedKey.GetStorageNamespace(),
+		"qualified_key": qualifiedKey.GetKey(),
+		"key":           obj.Identifier,
+	})
+
+	client := a.clients.Get(ctx, bucket)
+	resp, err := client.ListMultipartUploads(ctx, input)
+	if err != nil {
+		lg.WithError(err).Error("List multipart uploads failed")
+		return nil, err
+	}
+
+	mpuResp := block.ListMultipartUploadsResponse{
+		Uploads:            resp.Uploads,
+		NextUploadIDMarker: resp.NextUploadIdMarker,
+		NextKeyMarker:      resp.NextKeyMarker,
+		IsTruncated:        aws.ToBool(resp.IsTruncated),
+		MaxUploads:         resp.MaxUploads,
+	}
+	return &mpuResp, nil
+}
+
 func (a *Adapter) BlockstoreType() string {
 	return block.BlockstoreTypeS3
 }

pkg/block/transient/adapter.go

@@ -178,3 +178,6 @@ func (a *Adapter) GetPresignUploadPartURL(_ context.Context, _ block.ObjectPoint
 func (a *Adapter) ListParts(_ context.Context, _ block.ObjectPointer, _ string, _ block.ListPartsOpts) (*block.ListPartsResponse, error) {
 	return nil, block.ErrOperationNotSupported
 }
+func (a *Adapter) ListMultipartUploads(_ context.Context, _ block.ObjectPointer, _ block.ListMultipartUploadsOpts) (*block.ListMultipartUploadsResponse, error) {
+	return nil, block.ErrOperationNotSupported
+}