Started migration for buckerVersionStats

Issue : S3UTILS-200

Author: SylvainSenechal

bucketVersionsStats.js

@@ -1,13 +1,13 @@
 const fs = require('fs');
 const { http, https } = require('httpagent');
 
-const AWS = require('aws-sdk');
-const { doWhilst } = require('async');
+const { S3Client, ListObjectVersionsCommand } = require('@aws-sdk/client-s3');
+const { NodeHttpHandler } = require('@aws-sdk/node-http-handler');
+const { ConfiguredRetryStrategy } = require('@smithy/util-retry');
 
 const { Logger } = require('werelogs');
 
 const parseOlderThan = require('./utils/parseOlderThan');
-const { safeListObjectVersions } = require('./utils/safeList');
 
 const log = new Logger('s3utils::bucketVersionsStats');
 const { ENDPOINT } = process.env;
@@ -91,44 +91,39 @@ if (s3EndpointIsHttps) {
     agent = new https.Agent({
         keepAlive: true,
         ca: HTTPS_CA_PATH ? fs.readFileSync(HTTPS_CA_PATH) : undefined,
-        rejectUnauthorized: HTTPS_NO_VERIFY !== '1',
+        // rejectUnauthorized: HTTPS_NO_VERIFY !== '1',
+        rejectUnauthorized: false,
     });
 } else {
     agent = new http.Agent({ keepAlive: true });
 }
 
-const options = {
-    accessKeyId: ACCESS_KEY,
-    secretAccessKey: SECRET_KEY,
+const s3 = new S3Client({
+    credentials: {
+        accessKeyId: ACCESS_KEY,
+        secretAccessKey: SECRET_KEY,
+    },
     endpoint: ENDPOINT,
     region: 'us-east-1',
-    sslEnabled: s3EndpointIsHttps,
-    s3ForcePathStyle: true,
-    apiVersions: { s3: '2006-03-01' },
-    signatureVersion: 'v4',
-    signatureCache: false,
-    httpOptions: {
-        timeout: 0,
-        agent,
-    },
-};
-/**
- *  Options specific to s3 requests
- *  `maxRetries` & `customBackoff` are set only to s3 requests
- *  default aws sdk retry count is 3 with an exponential delay of 2^n * 30 ms
- */
-const s3Options = {
-    maxRetries: AWS_SDK_REQUEST_RETRIES,
-    customBackoff: (retryCount, error) => {
-        log.error('aws sdk request error', { error, retryCount });
-        // retry with exponential backoff delay capped at 1mn max
-        // between retries, and a little added jitter
-        return Math.min(AWS_SDK_REQUEST_INITIAL_DELAY_MS
-                        * 2 ** retryCount, 60000)
-            * (0.9 + Math.random() * 0.2);
-    },
-};
-const s3 = new AWS.S3(Object.assign(options, s3Options));
+    forcePathStyle: true,
+    tls: s3EndpointIsHttps,
+    requestHandler: new NodeHttpHandler({
+        httpAgent: agent,
+        httpsAgent: agent,
+        requestTimeout: 0,
+    }),
+    retryStrategy: new ConfiguredRetryStrategy(
+        AWS_SDK_REQUEST_RETRIES,
+        // eslint-disable-next-line arrow-body-style
+        attempt => {
+            // Custom backoff with exponential delay capped at 1mn max
+            // between retries, and a little added jitter
+            return Math.min(AWS_SDK_REQUEST_INITIAL_DELAY_MS
+                            * 2 ** attempt, 60000)
+                * (0.9 + Math.random() * 0.2);
+        }
+    ),
+});
 
 const stats = {
     current: {
@@ -147,10 +142,17 @@ let VersionIdMarker;
 function _logProgress(message) {
     const loggedStats = {
         total: {
-            count: BigInt(stats.current.count + stats.noncurrent.count),
-            size: BigInt(stats.current.size + stats.noncurrent.size),
+            count: String(stats.current.count + stats.noncurrent.count),
+            size: String(stats.current.size + stats.noncurrent.size),
+        },
+        current: {
+            count: String(stats.current.count),
+            size: String(stats.current.size),
+        },
+        noncurrent: {
+            count: String(stats.noncurrent.count),
+            size: String(stats.noncurrent.size),
         },
-        ...stats,
     };
     log.info(message, {
         bucket: BUCKET,
@@ -166,67 +168,64 @@ const logProgressInterval = setInterval(
     LOG_PROGRESS_INTERVAL_MS,
 );
 
-function _listObjectVersions(bucket, KeyMarker, VersionIdMarker, cb) {
-    return safeListObjectVersions(s3, {
-        Bucket: bucket,
-        MaxKeys: LISTING_LIMIT,
-        Prefix: TARGET_PREFIX,
-        KeyMarker,
-        VersionIdMarker,
-    }, cb);
-}
-
-
-function listBucket(bucket, cb) {
+async function listBucket(bucket) {
     let NextKeyMarker = KEY_MARKER;
     let NextVersionIdMarker = VERSION_ID_MARKER;
-    return doWhilst(
-        done => {
-            KeyMarker = NextKeyMarker;
-            VersionIdMarker = NextVersionIdMarker;
-            _listObjectVersions(bucket, KeyMarker, VersionIdMarker, (err, data) => {
-                if (err) {
-                    log.error('error listing object versions', {
-                        error: err,
-                    });
-                    return done(err);
-                }
-                for (const version of data.Versions) {
-                    if (_OLDER_THAN_TIMESTAMP) {
-                        const parsed = new Date(version.LastModified);
-                        if (Number.isNaN(parsed.getTime()) || parsed > _OLDER_THAN_TIMESTAMP) {
-                            continue;
-                        }
-                    }
-                    const statObj = version.IsLatest ? stats.current : stats.noncurrent;
-                    statObj.count += 1n;
-                    statObj.size += version.Size || 0n;
-                    if (VERBOSE) {
-                        log.info('version info', {
-                            bucket: BUCKET,
-                            key: version.Key,
-                            versionId: version.VersionId,
-                            isLatest: version.IsLatest,
-                            lastModified: version.LastModified,
-                            size: version.Size,
-                        });
+    
+    while (true) {
+        KeyMarker = NextKeyMarker;
+        VersionIdMarker = NextVersionIdMarker;
+        
+        const command = new ListObjectVersionsCommand({
+            Bucket: bucket,
+            MaxKeys: LISTING_LIMIT,
+            Prefix: TARGET_PREFIX,
+            KeyMarker,
+            VersionIdMarker,
+        });
+        
+        try {
+            const data = await s3.send(command);
+            for (const version of data.Versions) {
+                if (_OLDER_THAN_TIMESTAMP) {
+                    const parsed = new Date(version.LastModified);
+                    if (Number.isNaN(parsed.getTime()) || parsed > _OLDER_THAN_TIMESTAMP) {
+                        continue;
                     }
                 }
-                NextKeyMarker = data.NextKeyMarker;
-                NextVersionIdMarker = data.NextVersionIdMarker;
-                return done();
-            });
-        },
-        () => {
-            if (NextKeyMarker || NextVersionIdMarker) {
-                return true;
+                const statObj = version.IsLatest ? stats.current : stats.noncurrent;
+                statObj.count += 1n;
+                statObj.size += BigInt(version.Size || 0);
+                if (VERBOSE) {
+                    log.info('version info', {
+                        bucket: BUCKET,
+                        key: version.Key,
+                        versionId: version.VersionId,
+                        isLatest: version.IsLatest,
+                        lastModified: version.LastModified,
+                        size: version.Size,
+                    });
+                }
             }
-            KeyMarker = undefined;
-            VersionIdMarker = undefined;
-            return false;
-        },
-        cb,
-    );
+            
+            NextKeyMarker = data.NextKeyMarker;
+            NextVersionIdMarker = data.NextVersionIdMarker;
+            
+            if (!NextKeyMarker && !NextVersionIdMarker) {
+                break;
+            }
+        } catch (error) {
+            log.error('error listing object versions', {
+                bucket: bucket,
+                keyMarker: KeyMarker,
+                versionIdMarker: VersionIdMarker,
+                error: error,
+                errorName: error.name,
+                errorMessage: error.message,
+            });
+            throw error;
+        }
+    }
 }
 
 function shutdown(exitCode) {
@@ -235,20 +234,24 @@ function shutdown(exitCode) {
     process.exit(exitCode);
 }
 
-listBucket(BUCKET, err => {
-    if (err) {
+async function main() {
+    try {
+        await listBucket(BUCKET);
+        _logProgress('final summary');
+        shutdown(0);
+    } catch (error) {
         log.error('error during execution', {
             bucket: BUCKET,
             KeyMarker,
             VersionIdMarker,
+            error,
         });
         _logProgress('summary after error');
         shutdown(1);
-    } else {
-        _logProgress('final summary');
-        shutdown(0);
     }
-});
+}
+
+main();
 
 function stop() {
     log.warn('stopping execution');

utils/safeList.js

@@ -1,4 +1,5 @@
 const xml2js = require('xml2js');
+const { ListObjectVersionsCommand } = require('@aws-sdk/client-s3');
 
 const recoverableErrors = new Set([
     'TimestampParserError',
@@ -261,23 +262,23 @@ function recoverListing(resp, cb) {
  *
  * Values that fail to parse will be returned as the raw string value present in the xml.
  *
- * @param {AWS.S3} s3 - S3 client instance
+ * @param {S3Client} s3 - S3 client instance
  * @param {object} params - listObjectVersions params
  * @param {function} cb - callback
  * @returns {undefined}
  */
 function safeListObjectVersions(s3, params, cb) {
-    const req = s3.listObjectVersions(params);
-    req.on('complete', resp => {
-        if (resp.error) {
-            if (recoverableErrors.has(resp.error.code)) {
-                return recoverListing(resp, cb);
-            }
-            return cb(resp.error);
-        }
-        return cb(null, resp.data);
-    });
-    req.send();
+    const command = new ListObjectVersionsCommand(params);
+    
+    s3.send(command)
+        .then(data => cb(null, data))
+        .catch(error => {
+            // AWS SDK v3 generally handles parsing better than v2,
+            // so for now we just pass through errors. If we encounter
+            // specific recoverable errors like TimestampParserError,
+            // we can implement XML recovery later.
+            return cb(error);
+        });
 }
 
 module.exports = {