Migrated cleanupBuckets to sdk v3

Issue : S3UTILS-203

Author: SylvainSenechal

cleanupBuckets.js

@@ -1,7 +1,14 @@
-const { http } = require('httpagent');
-
+const { http, https } = require('httpagent');
 const async = require('async');
-const AWS = require('aws-sdk');
+const crypto = require('crypto');
+const { 
+    S3Client, 
+    ListObjectVersionsCommand, 
+    DeleteObjectsCommand, 
+    ListMultipartUploadsCommand,
+    AbortMultipartUploadCommand, 
+} = require('@aws-sdk/client-s3');
+const { NodeHttpHandler } = require('@aws-sdk/node-http-handler');
 const { Logger } = require('werelogs');
 
 const log = new Logger('s3utils::emptyBucket');
@@ -29,34 +36,33 @@ if (!SECRET_KEY) {
 }
 const LISTING_LIMIT = 1000;
 
-AWS.config.update({
-    accessKeyId: ACCESS_KEY,
-    secretAccessKey: SECRET_KEY,
+const s3 = new S3Client({
+    credentials: {
+        accessKeyId: ACCESS_KEY,
+        secretAccessKey: SECRET_KEY,
+    },
     endpoint: ENDPOINT,
     region: 'us-east-1',
-    sslEnabled: false,
-    s3ForcePathStyle: true,
-    apiVersions: { s3: '2006-03-01' },
-    signatureVersion: 'v4',
-    signatureCache: false,
-});
-
-const s3 = new AWS.S3({
-    httpOptions: {
-        maxRetries: 0,
-        timeout: 0,
-        agent: new http.Agent({ keepAlive: true }),
-    },
+    forcePathStyle: true,
+    requestHandler: new NodeHttpHandler({
+        httpAgent: new http.Agent({ keepAlive: true }),
+        httpsAgent: new https.Agent({ 
+            keepAlive: true,
+            rejectUnauthorized: false
+        }),
+    }),
+    tls: false,
 });
 
 // list object versions
-function _listObjectVersions(bucket, VersionIdMarker, KeyMarker, cb) {
-    return s3.listObjectVersions({
+async function _listObjectVersions(bucket, VersionIdMarker, KeyMarker) {
+    const params = {
         Bucket: bucket,
         MaxKeys: LISTING_LIMIT,
         VersionIdMarker,
         KeyMarker,
-    }, cb);
+    };
+    return await s3.send(new ListObjectVersionsCommand(params));
 }
 
 // return object with key and version_id
@@ -68,97 +74,103 @@ function _getKeys(keys) {
 }
 
 // delete all versions of an object
-function _deleteVersions(bucket, objectsToDelete, cb) {
-    // multi object delete can delete max 1000 objects
+async function _deleteVersions(bucket, objectsToDelete) {
+    if (!objectsToDelete || objectsToDelete.length === 0) {
+        log.info(`No objects to delete for ${bucket}, skipping deleteObjects call`);
+        return;
+    }
+    
     const params = {
         Bucket: bucket,
         Delete: { Objects: objectsToDelete },
     };
-    s3.deleteObjects(params, err => {
-        if (err) {
-            log.error('batch delete err', err);
-            return cb(err);
+    const command = new DeleteObjectsCommand(params);
+    command.middlewareStack.add(
+        next => async args => {
+            if (args.request.body) {
+                const bodyContent = Buffer.from(args.request.body);
+                const md5Hash = crypto.createHash('md5').update(bodyContent).digest('base64');
+                // eslint-disable-next-line no-param-reassign
+                args.request.headers['Content-MD5'] = md5Hash;
+            }
+            return await next(args);
+        },
+        {
+            step: 'build',
         }
+    );
+    
+    try {
+        await s3.send(command);
         objectsToDelete.forEach(v => log.info(`deleted key: ${v.Key}`));
-        return cb();
-    });
+    } catch (err) {
+        log.error('batch delete err', err);
+        throw err;
+    }
 }
 
-function cleanupVersions(bucket, cb) {
+async function cleanupVersions(bucket) {
     let VersionIdMarker = null;
     let KeyMarker = null;
-    async.doWhilst(
-        done => _listObjectVersions(
-            bucket,
-            VersionIdMarker,
-            KeyMarker,
-            (err, data) => {
-                if (err) {
-                    return done(err);
-                }
-                VersionIdMarker = data.NextVersionIdMarker;
-                KeyMarker = data.NextKeyMarker;
-                const keysToDelete = _getKeys(data.Versions);
-                const markersToDelete = _getKeys(data.DeleteMarkers);
-                return _deleteVersions(
-                    bucket,
-                    keysToDelete.concat(markersToDelete),
-                    done,
-                );
-            },
-        ),
-        () => {
-            if (VersionIdMarker || KeyMarker) {
-                return true;
-            }
-            return false;
-        },
-        cb,
-    );
+    let IsTruncated = true;
+    
+    while (IsTruncated) {
+        const data = await _listObjectVersions(bucket, VersionIdMarker, KeyMarker);
+        VersionIdMarker = data.NextVersionIdMarker;
+        KeyMarker = data.NextKeyMarker;
+        IsTruncated = data.IsTruncated;
+        
+        const keysToDelete = _getKeys(data.Versions || []);
+        const markersToDelete = _getKeys(data.DeleteMarkers || []);
+        const allObjectsToDelete = keysToDelete.concat(markersToDelete);
+        
+        if (allObjectsToDelete.length > 0) {
+            await _deleteVersions(bucket, allObjectsToDelete);
+        } else {
+            log.info(`No objects to delete for bucket ${bucket}`);
+        }
+    }
 }
 
-function abortAllMultipartUploads(bucket, cb) {
-    s3.listMultipartUploads({ Bucket: bucket }, (err, res) => {
-        if (err) {
-            return cb(err);
-        }
-        if (!res || !res.Uploads) {
-            return cb();
-        }
-        return async.mapLimit(
-            res.Uploads,
-            10,
-            (item, done) => {
-                const { Key, UploadId } = item;
-                const params = { Bucket: bucket, Key, UploadId };
-                s3.abortMultipartUpload(params, done);
-            },
-            cb,
-        );
+async function abortAllMultipartUploads(bucket) {
+    const res = await s3.send(new ListMultipartUploadsCommand({ Bucket: bucket }));
+    log.info(`Found ${res.Uploads ? res.Uploads.length : 0} multipart uploads to abort`);
+
+    if (!res || !res.Uploads || res.Uploads.length === 0) {
+        return;
+    }
+    
+    const deleteMpuPromises = res.Uploads.map(async item => {
+        const { Key, UploadId } = item;
+        const params = { Bucket: bucket, Key, UploadId };
+        return await s3.send(new AbortMultipartUploadCommand(params));
     });
+    
+    await Promise.all(deleteMpuPromises);
 }
 
-function _cleanupBucket(bucket, cb) {
-    async.parallel([
-        done => cleanupVersions(bucket, done),
-        done => abortAllMultipartUploads(bucket, done),
-    ], err => {
-        if (err) {
-            log.error('error occured deleting objects', err);
-            return cb(err);
-        }
+async function _cleanupBucket(bucket) {
+    try {
+        await Promise.all([
+            cleanupVersions(bucket),
+            abortAllMultipartUploads(bucket),
+        ]);
         log.info(`completed cleaning up of bucket: ${bucket}`);
-        return cb();
-    });
+    } catch (err) {
+        log.error('error occured deleting objects', err);
+        throw err;
+    }
 }
 
-function cleanupBuckets(buckets) {
-    async.mapLimit(buckets, 1, _cleanupBucket, err => {
-        if (err) {
-            return log.error('error occured deleting objects', err);
+async function cleanupBuckets(buckets) {
+    try {
+        for (const bucket of buckets) {
+            await _cleanupBucket(bucket);
         }
-        return log.info('completed cleaning up the given buckets');
-    });
+        log.info('completed cleaning all buckets');
+    } catch (err) {
+        log.error('error occured deleting objects', err);
+    }
 }
 
 cleanupBuckets(BUCKETS);