[#100] s3: Add a flag for permanent versioned object deletion

Signed-off-by: Nikita Zinkevich <n.zinkevich@yadro.com>

internal/s3/client.go

@@ -106,51 +106,8 @@ func (c *Client) Multipart(bucket, key string, objPartSize, concurrency int, pay
 	return PutResponse{Success: true}
 }
 
-func (c *Client) Delete(bucket, key string, allVersions bool) DeleteResponse {
+func (c *Client) Delete(bucket, key string) DeleteResponse {
 	start := time.Now()
-	if allVersions {
-		versions, err := c.cli.ListObjectVersions(c.vu.Context(), &s3.ListObjectVersionsInput{
-			Bucket: aws.String(bucket),
-			Prefix: aws.String(key),
-		})
-		if err != nil {
-			stats.Report(c.vu, objPutFails, 1)
-			return DeleteResponse{Success: false, Error: err.Error()}
-		}
-
-		var toDelete []types.ObjectIdentifier
-		for _, v := range versions.Versions {
-			if *v.Key == key {
-				toDelete = append(toDelete, types.ObjectIdentifier{
-					Key:       v.Key,
-					VersionId: v.VersionId,
-				})
-			}
-		}
-		for _, marker := range versions.DeleteMarkers {
-			if *marker.Key == key {
-				toDelete = append(toDelete, types.ObjectIdentifier{
-					Key:       marker.Key,
-					VersionId: marker.VersionId,
-				})
-			}
-		}
-		if len(toDelete) != 0 {
-			_, err := c.cli.DeleteObjects(c.vu.Context(), &s3.DeleteObjectsInput{
-				Bucket: aws.String(bucket),
-				Delete: &types.Delete{
-					Objects: toDelete,
-					Quiet:   true,
-				},
-			})
-			if err != nil {
-				stats.Report(c.vu, objPutFails, 1)
-				return DeleteResponse{Success: false, Error: err.Error()}
-			}
-
-			return DeleteResponse{Success: true}
-		}
-	}
 
 	_, err := c.cli.DeleteObject(c.vu.Context(), &s3.DeleteObjectInput{
 		Bucket: aws.String(bucket),
@@ -185,6 +142,76 @@ func (c *Client) Get(bucket, key string) GetResponse {
 	return GetResponse{Success: true}
 }
 
+func (c *Client) DeleteObjectVersion(bucket, key, version string) DeleteResponse {
+	var toDelete []types.ObjectIdentifier
+
+	if version != "" {
+		toDelete = append(toDelete, types.ObjectIdentifier{
+			Key:       aws.String(key),
+			VersionId: aws.String(version),
+		})
+	} else {
+		v, err := c.cli.GetBucketVersioning(c.vu.Context(), &s3.GetBucketVersioningInput{
+			Bucket: aws.String(bucket),
+		})
+		if err != nil {
+			stats.Report(c.vu, objDeleteFails, 1)
+			return DeleteResponse{Success: false, Error: err.Error()}
+		}
+		if v.Status == "" {
+			// delete non-versioned object
+			return c.Delete(bucket, key)
+		}
+
+		versions, err := c.cli.ListObjectVersions(c.vu.Context(), &s3.ListObjectVersionsInput{
+			Bucket: aws.String(bucket),
+			Prefix: aws.String(key),
+		})
+		if err != nil {
+			stats.Report(c.vu, objDeleteFails, 1)
+			return DeleteResponse{Success: false, Error: err.Error()}
+		}
+		toDelete = append(toDelete, filterObjectVersions(versions, key)...)
+	}
+	if len(toDelete) != 0 {
+		_, err := c.cli.DeleteObjects(c.vu.Context(), &s3.DeleteObjectsInput{
+			Bucket: aws.String(bucket),
+			Delete: &types.Delete{
+				Objects: toDelete,
+				Quiet:   true,
+			},
+		})
+		if err != nil {
+			stats.Report(c.vu, objDeleteFails, 1)
+			return DeleteResponse{Success: false, Error: err.Error()}
+		}
+	}
+
+	return DeleteResponse{Success: true}
+}
+
+func filterObjectVersions(versions *s3.ListObjectVersionsOutput, key string) []types.ObjectIdentifier {
+	var result []types.ObjectIdentifier
+
+	for _, v := range versions.Versions {
+		if *v.Key == key {
+			result = append(result, types.ObjectIdentifier{
+				Key:       v.Key,
+				VersionId: v.VersionId,
+			})
+		}
+	}
+	for _, marker := range versions.DeleteMarkers {
+		if *marker.Key == key {
+			result = append(result, types.ObjectIdentifier{
+				Key:       marker.Key,
+				VersionId: marker.VersionId,
+			})
+		}
+	}
+	return result
+}
+
 func get(
 	c *s3.Client,
 	bucket string,

scenarios/s3.js

@@ -215,7 +215,7 @@ export function obj_delete() {
     return;
   }
 
-  const resp = s3_client.delete(obj.s3_bucket, obj.s3_key, false);
+  const resp = s3_client.delete(obj.s3_bucket, obj.s3_key);
   if (!resp.success) {
     log.withFields({bucket : obj.s3_bucket, key : obj.s3_key, op : 'DELETE'})
         .error(resp.error);

scenarios/s3_car.js

@@ -228,7 +228,7 @@ export function obj_delete() {
     return;
   }
 
-  const resp = s3_client.delete(obj.s3_bucket, obj.s3_key, false);
+  const resp = s3_client.delete(obj.s3_bucket, obj.s3_key);
   if (!resp.success) {
     log.withFields({bucket: obj.s3_bucket, key: obj.s3_key, op: 'DELETE'})
         .error(resp.error);

scenarios/s3_dar.js

@@ -222,7 +222,7 @@ export function delete_object(obj) {
         return;
     }
 
-    const resp = s3_client.delete(obj.s3_bucket, obj.s3_key, false);
+    const resp = s3_client.delete(obj.s3_bucket, obj.s3_key);
     if (!resp.success) {
         log.withFields({bucket : obj.s3_bucket, key : obj.s3_key, op : 'DELETE'})
             .error(resp.error);