Batch delete files from S3

Author: rowanseymour

archives/archives.go

@@ -847,9 +847,9 @@ func DeleteArchivedOrgRecords(ctx context.Context, rt *runtime.Runtime, now time
 }
 
 const sqlSelectRolledUpDailyArchives = `
-  SELECT id, location
+  SELECT uuid, id, org_id, start_date::timestamp with time zone AS start_date, period, archive_type, hash, location, size, record_count, needs_deletion, rollup_id
     FROM archives_archive 
-   WHERE org_id = $1 AND archive_type = $2 AND period = $3 AND rollup_id IS NOT NULL AND deleted_on IS NOT NULL`
+   WHERE org_id = $1 AND archive_type = $2 AND period = 'D' AND rollup_id IS NOT NULL AND deleted_on IS NOT NULL`
 
 // DeleteRolledUpDailyArchives deletes daily archives that have been rolled up into monthlies and had their records deleted
 func DeleteRolledUpDailyArchives(ctx context.Context, rt *runtime.Runtime, org Org, archiveType ArchiveType) (int, error) {
@@ -858,39 +858,34 @@ func DeleteRolledUpDailyArchives(ctx context.Context, rt *runtime.Runtime, org O
 
 	log := slog.With("org_id", org.ID, "org_name", org.Name, "archive_type", archiveType)
 
-	// first, get the archives to delete so we can clean up their S3 files
-	type archiveToDelete struct {
-		ID       int         `db:"id"`
-		Location null.String `db:"location"`
-	}
-	var archivesToDelete []archiveToDelete
-	err := rt.DB.SelectContext(ctx, &archivesToDelete, sqlSelectRolledUpDailyArchives, org.ID, archiveType, DayPeriod)
-	if err != nil {
+	var toDelete []*Archive
+	if err := rt.DB.SelectContext(ctx, &toDelete, sqlSelectRolledUpDailyArchives, org.ID, archiveType); err != nil {
 		return 0, fmt.Errorf("error selecting rolled up daily archives: %w", err)
 	}
 
-	if len(archivesToDelete) == 0 {
+	if len(toDelete) == 0 {
 		return 0, nil
 	}
 
-	// collect IDs and delete S3 files for archives that were uploaded
-	ids := make([]int, 0, len(archivesToDelete))
+	// collect IDs and S3 keys grouped by bucket
+	ids := make([]int, 0, len(toDelete))
+	keysByBucket := make(map[string][]string)
+	for _, a := range toDelete {
+		ids = append(ids, a.ID)
+		if a.isUploaded() {
+			bucket, key := a.location()
+			keysByBucket[bucket] = append(keysByBucket[bucket], key)
+		}
+	}
+
+	// delete S3 files
 	s3DeletedCount := 0
-	for _, archive := range archivesToDelete {
-		ids = append(ids, archive.ID)
-
-		if archive.Location != "" {
-			// parse bucket:key from location (format is "bucket:key")
-			parts := strings.SplitN(string(archive.Location), ":", 2)
-			if len(parts) == 2 {
-				bucket, key := parts[0], parts[1]
-				if err := DeleteS3File(ctx, rt.S3, bucket, key); err != nil {
-					log.Error("error deleting S3 file for rolled up daily archive", "archive_id", archive.ID, "bucket", bucket, "key", key, "error", err)
-					// continue to try deleting other files and the database records
-				} else {
-					s3DeletedCount++
-				}
-			}
+	for bucket, keys := range keysByBucket {
+		deleted, err := DeleteS3Files(ctx, rt.S3, bucket, keys)
+		s3DeletedCount += deleted
+		if err != nil {
+			log.Error("error deleting S3 files for rolled up daily archives", "bucket", bucket, "error", err)
+			// continue to try deleting database records
 		}
 	}
 

archives/s3.go

@@ -8,6 +8,7 @@ import (
 	"io"
 	"log/slog"
 	"os"
+	"slices"
 	"strings"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
@@ -139,16 +140,45 @@ func GetS3File(ctx context.Context, s3Client *s3x.Service, bucket, key string) (
 	return output.Body, nil
 }
 
-// DeleteS3File deletes a file from S3
-func DeleteS3File(ctx context.Context, s3Client *s3x.Service, bucket, key string) error {
-	_, err := s3Client.Client.DeleteObject(ctx, &s3.DeleteObjectInput{
-		Bucket: aws.String(bucket),
-		Key:    aws.String(key),
-	})
-	if err != nil {
-		return fmt.Errorf("error deleting S3 object bucket=%s key=%s: %w", bucket, key, err)
+// DeleteS3Files deletes multiple files from S3, automatically batching into requests of 1000 keys
+func DeleteS3Files(ctx context.Context, s3Client *s3x.Service, bucket string, keys []string) (int, error) {
+	if len(keys) == 0 {
+		return 0, nil
 	}
 
-	slog.Debug("deleted S3 file", "bucket", bucket, "key", key)
-	return nil
+	totalDeleted := 0
+	var lastErr error
+
+	for batch := range slices.Chunk(keys, 1000) {
+		objects := make([]types.ObjectIdentifier, len(batch))
+		for i, key := range batch {
+			objects[i] = types.ObjectIdentifier{Key: aws.String(key)}
+		}
+
+		output, err := s3Client.Client.DeleteObjects(ctx, &s3.DeleteObjectsInput{
+			Bucket: aws.String(bucket),
+			Delete: &types.Delete{Objects: objects, Quiet: aws.Bool(true)},
+		})
+		if err != nil {
+			lastErr = fmt.Errorf("error batch deleting S3 objects bucket=%s: %w", bucket, err)
+			continue
+		}
+
+		// check for individual errors
+		if len(output.Errors) > 0 {
+			for _, e := range output.Errors {
+				slog.Error("error deleting S3 object in batch", "bucket", bucket, "key", *e.Key, "error", *e.Message)
+			}
+			totalDeleted += len(batch) - len(output.Errors)
+			lastErr = fmt.Errorf("%d errors deleting S3 objects", len(output.Errors))
+		} else {
+			totalDeleted += len(batch)
+		}
+	}
+
+	if totalDeleted > 0 {
+		slog.Debug("deleted S3 files", "bucket", bucket, "count", totalDeleted)
+	}
+
+	return totalDeleted, lastErr
 }