feat(region): bucket list uploads

pkg/cloudprovider/objectstore.go

@@ -154,6 +154,16 @@ type SBucketPolicyStatementInput struct {
 	IpNotEquals []string
 }
 
+type SBucketMultipartUploads struct {
+	// object name
+	ObjectName string
+	UploadID   string
+	// 发起人
+	Initiator string
+	// 发起时间
+	Initiated time.Time
+}
+
 type SBaseCloudObject struct {
 	Key          string
 	SizeBytes    int64
@@ -264,6 +274,8 @@ type ICloudBucket interface {
 	GetTags() (map[string]string, error)
 	SetTags(tags map[string]string) error
 	DeleteTags() error
+
+	ListMultipartUploads() ([]SBucketMultipartUploads, error)
 }
 
 type ICloudObject interface {

pkg/multicloud/aliyun/bucket.go

@@ -778,3 +778,40 @@ func (b *SBucket) GetMetadata() *jsonutils.JSONDict {
 	}
 	return meta
 }
+
+func (b *SBucket) ListMultipartUploads() ([]cloudprovider.SBucketMultipartUploads, error) {
+	osscli, err := b.region.GetOssClient()
+	if err != nil {
+		return nil, errors.Wrap(err, "GetOssClient")
+	}
+	result := []cloudprovider.SBucketMultipartUploads{}
+
+	ossBucket, err := osscli.Bucket(b.Name)
+	if err != nil {
+		return nil, errors.Wrap(err, "osscli.Bucket(b.Name)")
+	}
+
+	keyMarker := oss.KeyMarker("")
+	uploadIDMarker := oss.UploadIDMarker("")
+	for {
+		output, err := ossBucket.ListMultipartUploads(keyMarker, uploadIDMarker)
+		if err != nil {
+			return nil, errors.Wrap(err, " coscli.Bucket.ListMultipartUploads(context.Background(), &input)")
+		}
+		for i := range output.Uploads {
+			temp := cloudprovider.SBucketMultipartUploads{
+				ObjectName: output.Uploads[i].Key,
+				UploadID:   output.Uploads[i].UploadID,
+				Initiated:  output.Uploads[i].Initiated,
+			}
+			result = append(result, temp)
+		}
+		keyMarker = oss.KeyMarker(output.NextKeyMarker)
+		uploadIDMarker = oss.UploadIDMarker(output.NextUploadIDMarker)
+		if !output.IsTruncated {
+			break
+		}
+	}
+
+	return result, nil
+}

pkg/multicloud/aws/bucket.go

@@ -754,3 +754,59 @@ func (b *SBucket) GetMetadata() *jsonutils.JSONDict {
 	}
 	return meta
 }
+
+func (b *SBucket) ListMultipartUploads() ([]cloudprovider.SBucketMultipartUploads, error) {
+	s3cli, err := b.region.GetS3Client()
+	if err != nil {
+		return nil, errors.Wrap(err, "GetS3Client")
+	}
+	result := []cloudprovider.SBucketMultipartUploads{}
+
+	input := s3.ListMultipartUploadsInput{}
+	input.SetBucket(b.Name)
+	keyMarker := ""
+	uploadIDMarker := ""
+	for {
+		if len(keyMarker) > 0 {
+			input.SetKeyMarker(keyMarker)
+		}
+		if len(uploadIDMarker) > 0 {
+			input.SetUploadIdMarker(uploadIDMarker)
+		}
+		output, err := s3cli.ListMultipartUploads(&input)
+		if err != nil {
+			return nil, errors.Wrap(err, " coscli.Bucket.ListMultipartUploads(context.Background(), &input)")
+		}
+		if output == nil {
+			return nil, nil
+		}
+		for i := range output.Uploads {
+			temp := cloudprovider.SBucketMultipartUploads{}
+			if output.Uploads[i].Key != nil {
+				temp.ObjectName = *output.Uploads[i].Key
+			}
+			if output.Uploads[i].Initiator != nil {
+				temp.Initiator = *output.Uploads[i].Initiator.DisplayName
+			}
+			if output.Uploads[i].Initiated != nil {
+				temp.Initiated = *output.Uploads[i].Initiated
+			}
+			if output.Uploads[i].UploadId != nil {
+				temp.UploadID = *output.Uploads[i].UploadId
+			}
+			result = append(result, temp)
+		}
+		if output.NextKeyMarker != nil {
+			keyMarker = *output.NextKeyMarker
+		}
+		if output.NextUploadIdMarker != nil {
+			uploadIDMarker = *output.NextUploadIdMarker
+		}
+
+		if output.IsTruncated == nil || !*output.IsTruncated {
+			break
+		}
+	}
+
+	return result, nil
+}

pkg/multicloud/bucket_base.go

@@ -133,3 +133,7 @@ func (b *SBaseBucket) SetTags(tags map[string]string) error {
 func (b *SBaseBucket) DeleteTags() error {
 	return cloudprovider.ErrNotImplemented
 }
+
+func (b *SBaseBucket) ListMultipartUploads() ([]cloudprovider.SBucketMultipartUploads, error) {
+	return nil, cloudprovider.ErrNotImplemented
+}

pkg/multicloud/huawei/bucket.go

@@ -737,3 +737,44 @@ func (b *SBucket) GetMetadata() *jsonutils.JSONDict {
 	}
 	return meta
 }
+
+func (b *SBucket) ListMultipartUploads() ([]cloudprovider.SBucketMultipartUploads, error) {
+	obscli, err := b.region.getOBSClient()
+	if err != nil {
+		return nil, errors.Wrap(err, "GetOBSClient")
+	}
+	result := []cloudprovider.SBucketMultipartUploads{}
+
+	input := obs.ListMultipartUploadsInput{Bucket: b.Name}
+	keyMarker := ""
+	uploadIDMarker := ""
+	for {
+		if len(keyMarker) > 0 {
+			input.KeyMarker = keyMarker
+		}
+		if len(uploadIDMarker) > 0 {
+			input.UploadIdMarker = uploadIDMarker
+		}
+
+		output, err := obscli.ListMultipartUploads(&input)
+		if err != nil {
+			return nil, errors.Wrap(err, " coscli.Bucket.ListMultipartUploads(context.Background(), &input)")
+		}
+		for i := range output.Uploads {
+			temp := cloudprovider.SBucketMultipartUploads{
+				ObjectName: output.Uploads[i].Key,
+				UploadID:   output.Uploads[i].UploadId,
+				Initiator:  output.Uploads[i].Initiator.DisplayName,
+				Initiated:  output.Uploads[i].Initiated,
+			}
+			result = append(result, temp)
+		}
+		keyMarker = output.NextKeyMarker
+		uploadIDMarker = output.NextUploadIdMarker
+		if !output.IsTruncated {
+			break
+		}
+	}
+
+	return result, nil
+}

pkg/multicloud/objectstore/shell.go

@@ -669,6 +669,19 @@ func S3Shell() {
 		return nil
 	})
 
+	type BucketGetUploads struct {
+		BUCKET string `help:"name of bucket to put object"`
+	}
+	shellutils.R(&BucketGetUploads{}, "bucket-get-uploads", "get bucket uploads", func(cli cloudprovider.ICloudRegion, args *BucketGetUploads) error {
+		bucket, err := cli.GetIBucketById(args.BUCKET)
+		if err != nil {
+			return err
+		}
+		uplaods, err := bucket.ListMultipartUploads()
+		printList(uplaods, len(uplaods), 0, len(uplaods), nil)
+		return nil
+	})
+
 	type BucketObjectDownloadOptions struct {
 		BUCKET string `help:"name of bucket"`
 		KEY    string `help:"Key of object"`

pkg/multicloud/qcloud/bucket.go

@@ -1135,3 +1135,39 @@ func (b *SBucket) GetMetadata() *jsonutils.JSONDict {
 	}
 	return meta
 }
+
+func (b *SBucket) ListMultipartUploads() ([]cloudprovider.SBucketMultipartUploads, error) {
+	coscli, err := b.region.GetCosClient(b)
+	if err != nil {
+		log.Errorf("GetCosClient fail %s", err)
+		return nil, errors.Wrap(err, "b.region.GetCosClient(b)")
+	}
+	result := []cloudprovider.SBucketMultipartUploads{}
+	input := cos.ListMultipartUploadsOptions{}
+	keyMarker := ""
+	uploadIDMarker := ""
+	for {
+		input.KeyMarker = keyMarker
+		input.UploadIDMarker = uploadIDMarker
+		output, _, err := coscli.Bucket.ListMultipartUploads(context.Background(), &input)
+		if err != nil {
+			return nil, errors.Wrap(err, " coscli.Bucket.ListMultipartUploads(context.Background(), &input)")
+		}
+		for i := range output.Uploads {
+			temp := cloudprovider.SBucketMultipartUploads{
+				ObjectName: output.Uploads[i].Key,
+				UploadID:   output.Uploads[i].UploadID,
+				Initiator:  output.Uploads[i].Initiator.DisplayName,
+			}
+			temp.Initiated, _ = timeutils.ParseTimeStr(output.Uploads[i].Initiated)
+			result = append(result, temp)
+		}
+		keyMarker = output.NextKeyMarker
+		uploadIDMarker = output.NextUploadIDMarker
+		if !output.IsTruncated {
+			break
+		}
+	}
+
+	return result, nil
+}