Repository: turbot/steampipe-plugin-aws (language: go)

Issue:
Add table aws_s3_multipart_upload
Hi guys,

Can we please add a feature that can track AWS s3 multi-part uploads ?

For objects bigger than 100MB, AWS provides a feature to chunk an object before uploading it to s3.

An incomplete multipart upload occurs when a user starts uploading a large object using multipart upload but does not finish the process. 

The problem with multi-part uploads is that they are a silent cost. 

Reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListMultipartUploads.html 

Regards

   -Alex


Discussion / hints:

Fix patch:
diff --git a/aws/plugin.go b/aws/plugin.go
index c8d556d45..637168035 100644
--- a/aws/plugin.go
+++ b/aws/plugin.go
@@ -462,6 +462,7 @@ func Plugin(ctx context.Context) *plugin.Plugin {
 			"aws_s3_bucket":                                                tableAwsS3Bucket(ctx),
 			"aws_s3_bucket_intelligent_tiering_configuration":              tableAwsS3BucketIntelligentTieringConfiguration(ctx),
 			"aws_s3_multi_region_access_point":                             tableAwsS3MultiRegionAccessPoint(ctx),
+			"aws_s3_multipart_upload":                                      tableAwsS3MultipartUpload(ctx),
 			"aws_s3_object":                                                tableAwsS3Object(ctx),
 			"aws_s3_object_version":                                        tableAwsS3ObjectVersion(ctx),
 			"aws_sagemaker_app":                                            tableAwsSageMakerApp(ctx),
diff --git a/aws/table_aws_s3_multipart_upload.go b/aws/table_aws_s3_multipart_upload.go
new file mode 100644
index 000000000..fb01db9af
--- /dev/null
+++ b/aws/table_aws_s3_multipart_upload.go
@@ -0,0 +1,156 @@
+package aws
+
+import (
+	"context"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/turbot/steampipe-plugin-sdk/v5/grpc/proto"
+	"github.com/turbot/steampipe-plugin-sdk/v5/plugin"
+	"github.com/turbot/steampipe-plugin-sdk/v5/plugin/transform"
+	"github.com/turbot/steampipe-plugin-sdk/v5/query_cache"
+)
+
+func tableAwsS3MultipartUpload(_ context.Context) *plugin.Table {
+	return &plugin.Table{
+		Name:        "aws_s3_multipart_upload",
+		Description: "AWS S3 Multipart Upload",
+		List: &plugin.ListConfig{
+			Hydrate: listS3MultipartUploads,
+			KeyColumns: []*plugin.KeyColumn{
+				{Name: "bucket_name", Require: plugin.Required, CacheMatch: query_cache.CacheMatchExact},
+			},
+			Tags: map[string]string{"service": "s3", "action": "ListMultipartUploads"},
+		},
+		Columns: awsAccountColumns([]*plugin.Column{
+			{
+				Name:        "bucket_name",
+				Description: "The name of the bucket to which the multipart upload was initiated.",
+				Type:        proto.ColumnType_STRING,
+				Transform:   transform.FromQual("bucket_name"),
+			},
+			{
+				Name:        "key",
+				Description: "The object key for which the multipart upload was initiated.",
+				Type:        proto.ColumnType_STRING,
+			},
+			{
+				Name:        "upload_id",
+				Description: "Upload ID that identifies the multipart upload.",
+				Type:        proto.ColumnType_STRING,
+			},
+			{
+				Name:        "initiated",
+				Description: "Date and time at which the multipart upload was initiated.",
+				Type:        proto.ColumnType_TIMESTAMP,
+			},
+			{
+				Name:        "storage_class",
+				Description: "The class of storage used to store the object.",
+				Type:        proto.ColumnType_STRING,
+			},
+			{
+				Name:        "initiator_id",
+				Description: "The ID of the initiator.",
+				Type:        proto.ColumnType_STRING,
+				Transform:   transform.FromField("Initiator.ID"),
+			},
+			{
+				Name:        "initiator_display_name",
+				Description: "Display name of the initiator.",
+				Type:        proto.ColumnType_STRING,
+				Transform:   transform.FromField("Initiator.DisplayName"),
+			},
+			{
+				Name:        "owner_id",
+				Description: "The ID of the owner.",
+				Type:        proto.ColumnType_STRING,
+				Transform:   transform.FromField("Owner.ID"),
+			},
+			{
+				Name:        "owner_display_name",
+				Description: "Display name of the owner.",
+				Type:        proto.ColumnType_STRING,
+				Transform:   transform.FromField("Owner.DisplayName"),
+			},
+
+			// Steampipe standard columns
+			{
+				Name:        "region",
+				Description: "The AWS Region in which the object is located.",
+				Type:        proto.ColumnType_STRING,
+				Hydrate:     getBucketRegionForObjects,
+				Transform:   transform.FromValue(),
+			},
+			{
+				Name:        "title",
+				Description: "Title of the resource.",
+				Type:        proto.ColumnType_STRING,
+				Transform:   transform.FromField("UploadId"),
+			},
+		}),
+	}
+}
+
+//// LIST FUNCTION
+
+func listS3MultipartUploads(ctx context.Context, d *plugin.QueryData, h *plugin.HydrateData) (interface{}, error) {
+	bucketName := d.EqualsQuals["bucket_name"].GetStringValue()
+
+	// Bucket location will be nil if getBucketLocationForObjects returned an error but
+	// was ignored through ignore_error_codes config arg
+	bucketRegion, err := doGetBucketRegion(ctx, d, h, bucketName)
+	if err != nil {
+		return nil, err
+	} else if bucketRegion == "" {
+		return nil, nil
+	}
+
+	// Create Session
+	svc, err := S3Client(ctx, d, bucketRegion)
+	if err != nil {
+		plugin.Logger(ctx).Error("aws_s3_multipart_upload.listS3MultipartUploads", "connection_error", err)
+		return nil, err
+	}
+
+	// Reduce the basic request limit down if the user has only requested a small number of rows
+	maxLimit := int32(1000)
+	if d.QueryContext.Limit != nil {
+		limit := int32(*d.QueryContext.Limit)
+		if limit < maxLimit {
+			maxLimit = limit
+		}
+	}
+
+	input := &s3.ListMultipartUploadsInput{
+		Bucket:     aws.String(bucketName),
+		MaxUploads: aws.Int32(maxLimit),
+	}
+
+	paginator := s3.NewListMultipartUploadsPaginator(svc, input, func(o *s3.ListMultipartUploadsPaginatorOptions) {
+		o.Limit = maxLimit
+		o.StopOnDuplicateToken = true
+	})
+
+	for paginator.HasMorePages() {
+		// apply rate limiting
+		d.WaitForListRateLimit(ctx)
+
+		output, err := paginator.NextPage(ctx)
+		if err != nil {
+			plugin.Logger(ctx).Error("aws_s3_multipart_upload.listS3MultipartUploads", "api_error", err)
+			return nil, err
+		}
+
+		for _, upload := range output.Uploads {
+			d.StreamListItem(ctx, upload)
+
+			// Context may get cancelled due to manual cancellation or if the limit has been reached
+			if d.RowsRemaining(ctx) == 0 {
+				return nil, nil
+			}
+		}
+	}
+
+	return nil, nil
+}
diff --git a/docs/tables/aws_s3_multipart_upload.md b/docs/tables/aws_s3_multipart_upload.md
new file mode 100644
index 000000000..fdb0bf74e
--- /dev/null
+++ b/docs/tables/aws_s3_multipart_upload.md
@@ -0,0 +1,224 @@
+---
+title: "Steampipe Table: aws_s3_multipart_upload - Query AWS S3 Multipart Uploads using SQL"
+description: "Allows users to query AWS S3 Multipart Uploads, providing information about in-progress multipart uploads across S3 buckets."
+folder: "S3"
+---
+
+# Table: aws_s3_multipart_upload - Query AWS S3 Multipart Uploads using SQL
+
+AWS S3 Multipart Upload is a feature that allows you to upload large objects in parts. This is useful when you're uploading large objects over a network with unreliable connectivity, as it allows you to retry uploading individual parts if any part of the upload fails. Incomplete multipart uploads can incur storage costs, so it's important to monitor and manage them effectively.
+
+## Table Usage Guide
+
+The `aws_s3_multipart_upload` table in Steampipe provides you with information about in-progress multipart uploads within AWS Simple Storage Service (S3). This table enables you, as a DevOps engineer or system administrator, to query upload-specific details, including the upload ID, initiated date, storage class, and associated metadata. You can utilize this table to gather insights on multipart uploads, such as identifying incomplete uploads that might be incurring costs, monitoring upload progress, and managing storage utilization. The schema outlines the various attributes of the S3 multipart upload, including the bucket name, key, upload ID, and associated metadata.
+
+**Important Notes**
+
+- You must specify a `bucket_name` in a where or join clause in order to use this table.
+- The table lists all in-progress multipart uploads for the specified bucket.
+
+## Examples
+
+### Basic info
+
+Explore the status of multipart uploads in your S3 buckets to track ongoing file transfers and manage storage costs.
+
+```sql+postgres
+select
+  bucket_name,
+  key,
+  upload_id,
+  initiated,
+  storage_class
+from
+  aws_s3_multipart_upload
+where
+  bucket_name = 'my-bucket-name';
+```
+
+```sql+sqlite
+select
+  bucket_name,
+  key,
+  upload_id,
+  initiated,
+  storage_class
+from
+  aws_s3_multipart_upload
+where
+  bucket_name = 'my-bucket-name';
+```
+
+### List all multipart uploads older than 7 days
+
+Identify potentially abandoned multipart uploads that might be unnecessarily incurring storage costs.
+
+```sql+postgres
+select
+  bucket_name,
+  key,
+  upload_id,
+  initiated,
+  storage_class
+from
+  aws_s3_multipart_upload
+where
+  bucket_name = 'my-bucket-name'
+  and initiated < current_timestamp - interval '7 days';
+```
+
+```sql+sqlite
+select
+  bucket_name,
+  key,
+  upload_id,
+  initiated,
+  storage_class
+from
+  aws_s3_multipart_upload
+where
+  bucket_name = 'my-bucket-name'
+  and initiated < datetime('now', '-7 days');
+```
+
+### Get multipart upload details by bucket and key prefix
+
+Analyze the details of multipart uploads within a specific bucket and key prefix to monitor upload activities and manage storage.
+
+```sql+postgres
+select
+  bucket_name,
+  key,
+  upload_id,
+  initiated,
+  storage_class,
+  initiator_id,
+  initiator_display_name
+from
+  aws_s3_multipart_upload
+where
+  bucket_name = 'my-bucket-name'
+  and key like 'prefix/%';
+```
+
+```sql+sqlite
+select
+  bucket_name,
+  key,
+  upload_id,
+  initiated,
+  storage_class,
+  initiator_id,
+  initiator_display_name
+from
+  aws_s3_multipart_upload
+where
+  bucket_name = 'my-bucket-name'
+  and key like 'prefix/%';
+```
+
+### List multipart uploads by storage class
+
+Analyze the distribution of multipart uploads across different storage classes to optimize storage costs and performance.
+
+```sql+postgres
+select
+  storage_class,
+  count(*) as upload_count,
+  bucket_name
+from
+  aws_s3_multipart_upload
+where
+  bucket_name = 'my-bucket-name'
+group by
+  storage_class,
+  bucket_name
+order by
+  upload_count desc;
+```
+
+```sql+sqlite
+select
+  storage_class,
+  count(*) as upload_count,
+  bucket_name
+from
+  aws_s3_multipart_upload
+where
+  bucket_name = 'my-bucket-name'
+group by
+  storage_class,
+  bucket_name
+order by
+  upload_count desc;
+```
+
+### Get multipart uploads with bucket details
+
+Analyze multipart uploads alongside their corresponding bucket information to better understand storage utilization and permissions.
+
+```sql+postgres
+select
+  u.bucket_name,
+  u.key,
+  u.upload_id,
+  u.initiated,
+  u.storage_class,
+  b.versioning_enabled,
+  b.region
+from
+  aws_s3_multipart_upload as u
+  join aws_s3_bucket as b on u.bucket_name = b.name;
+```
+
+```sql+sqlite
+select
+  u.bucket_name,
+  u.key,
+  u.upload_id,
+  u.initiated,
+  u.storage_class,
+  b.versioning_enabled,
+  b.region
+from
+  aws_s3_multipart_upload as u
+  join aws_s3_bucket as b on u.bucket_name = b.name;
+```
+
+### List multipart uploads by initiator
+
+Identify and analyze multipart uploads based on who initiated them, which can be useful for access auditing and resource management.
+
+```sql+postgres
+select
+  initiator_display_name,
+  initiator_id,
+  count(*) as upload_count,
+  array_agg(distinct bucket_name) as buckets
+from
+  aws_s3_multipart_upload
+where
+  bucket_name = 'my-bucket-name'
+group by
+  initiator_display_name,
+  initiator_id
+order by
+  upload_count desc;
+```
+
+```sql+sqlite
+select
+  initiator_display_name,
+  initiator_id,
+  count(*) as upload_count,
+  group_concat(distinct bucket_name) as buckets
+from
+  aws_s3_multipart_upload
+where
+  bucket_name = 'my-bucket-name'
+group by
+  initiator_display_name,
+  initiator_id
+order by
+  upload_count desc;
+```


Test patch: