bucket logging - adapt s3 conn to aws sdk v3 (dfs 3887)

Signed-off-by: Amit Prinz Setter <[email protected]>

Author: alphaprinz

package-lock.json

@@ -76,6 +76,7 @@
     "@aws-sdk/client-sts": "3.864.0",
     "@aws-sdk/credential-providers": "3.864.0",
     "@aws-sdk/s3-request-presigner": "3.864.0",
+    "@aws-sdk/lib-storage": "3.864.0",
     "@azure/identity": "4.10.2",
     "@azure/monitor-query": "1.3.2",
     "@azure/storage-blob": "12.27.0",
@@ -126,7 +127,6 @@
   },
   "devDependencies": {
     "@aws-sdk/client-iam": "3.864.0",
-    "@aws-sdk/lib-storage": "3.864.0",
     "@stylistic/eslint-plugin-js": "1.8.1",
     "@types/jest": "30.0.0",
     "@types/lodash": "4.17.20",

package.json

@@ -1,9 +1,9 @@
 /* Copyright (C) 2024 NooBaa */
 'use strict';
 
-const AWS = require('aws-sdk');
+const { S3 } = require('@aws-sdk/client-s3');
+const noobaa_s3_client = require('../sdk/noobaa_s3_client/noobaa_s3_client');
 const config = require('../../config');
-const http_utils = require('../util/http_utils');
 const { account_id_cache } = require('../sdk/accountspace_fs');
 const { export_logs_to_target } = require('../util/bucket_logs_utils');
 const ManageCLIError = require('../manage_nsfs/manage_nsfs_cli_errors').ManageCLIError;
@@ -16,23 +16,36 @@ let config_fs;
 */
 async function export_bucket_logging(shared_config_fs) {
     config_fs = shared_config_fs;
-    const endpoint = `https://127.0.0.1:${config.ENDPOINT_SSL_PORT}`;
-    const noobaa_con = new AWS.S3({
-        endpoint,
-        s3ForcePathStyle: true,
-        sslEnabled: false,
-        httpOptions: {
-            agent: http_utils.get_unsecured_agent(endpoint)
-        }
-    });
-    const success = await export_logs_to_target(config_fs.fs_context, noobaa_con, get_bucket_owner_keys);
+    const success = await export_logs_to_target(config_fs.fs_context, noobaa_con_func, get_bucket_owner_keys);
     if (success) {
         write_stdout_response(ManageCLIResponse.LoggingExported);
     } else {
         throw_cli_error(ManageCLIError.LoggingExportFailed);
     }
 }
 
+/**
+ * A CB for bucket_logs_utils to get a v3 S3 connection.
+ * @param {Object} credentials for the target bucket
+ * @returns {S3} An S3 connection
+ */
+
+function noobaa_con_func(credentials) {
+    const endpoint = `https://127.0.0.1:${config.ENDPOINT_SSL_PORT}`;
+
+    return new S3({
+            endpoint,
+            forcePathStyle: true,
+            tls: false,
+            region: config.DEFAULT_REGION,
+            requestHandler: noobaa_s3_client.get_requestHandler_with_suitable_agent(endpoint),
+            credentials: {
+                accessKeyId: credentials[0].access_key,
+                secretAccessKey: credentials[0].secret_key
+            }
+    });
+}
+
 /**
  * return bucket owner's access and secret key
  * @param {string} log_bucket_name

src/manage_nsfs/manage_nsfs_logging.js

@@ -40,7 +40,7 @@ class BucketLogUploader {
         }
         if (config.BUCKET_LOG_TYPE === 'PERSISTENT') {
             const fs_context = get_process_fs_context();
-            const success = await export_logs_to_target(fs_context, this.noobaa_connection, this.get_bucket_owner_keys);
+            const success = await export_logs_to_target(fs_context, () => this.noobaa_connection, this.get_bucket_owner_keys);
             if (success) {
                 dbg.log0('Logs were uploaded succesfully to their target buckets');
             } else {
@@ -148,7 +148,7 @@ class BucketLogUploader {
         };
 
         try {
-            await noobaa_con.putObject(params).promise();
+            await noobaa_con.putObject(params);
         } catch (err) {
             dbg.error('Failed to upload bucket log object: ', log_object.log_object_name, ' to bucket: ', log_object.log_bucket_name, ' :', err);
         }

src/server/bg_services/bucket_logs_upload.js

@@ -9,12 +9,12 @@ const path = require('path');
 const { PersistentLogger } = require('../util/persistent_logger');
 const { format_aws_date } = require('../util/time_utils');
 const nsfs_schema_utils = require('../manage_nsfs/nsfs_schema_utils');
-const semaphore = require('../util/semaphore');
 const P = require('../util/promise');
 const nb_native = require('../util/nb_native');
-const AWS = require('aws-sdk');
+const { Upload } = require("@aws-sdk/lib-storage");
+const { once } = require('events');
 
-const sem = new semaphore.Semaphore(config.BUCKET_LOG_CONCURRENCY);
+//const sem = new semaphore.Semaphore(config.BUCKET_LOG_CONCURRENCY);
 
 // delimiter between bucket name and log object name.
 // Assuming noobaa bucket name follows the s3 bucket
@@ -27,16 +27,16 @@ const BUCKET_NAME_DEL = "_";
  * This function will process the persistent log of bucket logging
  * and will upload the log files in using provided noobaa connection
  * @param {nb.NativeFSContext} fs_context
- * @param {AWS.S3} s3_connection
+ * @param {function(Object): import("@aws-sdk/client-s3").S3} s3_conn_func
  * @param {function} bucket_to_owner_keys_func
  */
-async function export_logs_to_target(fs_context, s3_connection, bucket_to_owner_keys_func) {
+async function export_logs_to_target(fs_context, s3_conn_func, bucket_to_owner_keys_func) {
     const entries = await nb_native().fs.readdir(fs_context, config.PERSISTENT_BUCKET_LOG_DIR);
     const results = await P.map_with_concurrency(5, entries, async entry => {
         if (!entry.name.endsWith('.log')) return;
         const log = new PersistentLogger(config.PERSISTENT_BUCKET_LOG_DIR, path.parse(entry.name).name, { locking: 'EXCLUSIVE' });
         try {
-            return log.process(async file => _upload_to_targets(s3_connection, file, bucket_to_owner_keys_func));
+            return log.process(async file => _upload_to_targets(file, s3_conn_func, bucket_to_owner_keys_func));
         } catch (err) {
             dbg.error('processing log file failed', log.file);
             throw err;
@@ -51,12 +51,12 @@ async function export_logs_to_target(fs_context, s3_connection, bucket_to_owner_
  * This function gets a persistent log file, will go over it's entries one by one,
  * and will upload the entry to the target_bucket using the provided s3 connection
  * in order to know which user to use to upload to each bucket we will need to provide bucket_to_owner_keys_func
- * @param {AWS.S3} s3_connection
+ * @param {function(Object): import("@aws-sdk/client-s3").S3} s3_conn_func
  * @param {import('../util/persistent_logger').LogFile} log_file
  * @param {function} bucket_to_owner_keys_func
  * @returns {Promise<Boolean>}
  */
-async function _upload_to_targets(s3_connection, log_file, bucket_to_owner_keys_func) {
+async function _upload_to_targets(log_file, s3_conn_func, bucket_to_owner_keys_func) {
     const bucket_streams = {};
     const promises = [];
     try {
@@ -67,33 +67,44 @@ async function _upload_to_targets(s3_connection, log_file, bucket_to_owner_keys_
             const target_bucket = log_entry.log_bucket;
             const log_prefix = log_entry.log_prefix;
             const source_bucket = log_entry.source_bucket;
-            if (!bucket_streams[source_bucket + BUCKET_NAME_DEL + target_bucket]) {
+            let upload_stream = bucket_streams[source_bucket + BUCKET_NAME_DEL + target_bucket];
+            if (!upload_stream) {
                 /* new stream is needed for each target bucket, but also for each source bucket
                            - as mulitple buckets can't be written to the same object */
                 const date = new Date();
-                const upload_stream = new stream.PassThrough();
+                upload_stream = new stream.PassThrough();
                 let access_keys;
                 try {
                     access_keys = await bucket_to_owner_keys_func(target_bucket);
                 } catch (err) {
                     dbg.warn('Error when trying to resolve bucket keys', err);
                     if (err.rpc_code === 'NO_SUCH_BUCKET') return; // If the log_bucket doesn't exist any more - nowhere to upload - just skip
                 }
-                s3_connection.config.credentials = new AWS.Credentials(access_keys[0].access_key, access_keys[0].secret_key);
+                //target bucket can be in either cloud or an NC noobaa bucket
+                //use the s3_conn_func CB to get suitable connection
+                const s3_connection = s3_conn_func(access_keys);
                 const sha = crypto.createHash('sha512').update(target_bucket + date.getTime()).digest('hex');
-                promises.push(sem.surround(() => P.retry({
-                    attempts: 3,
-                    delay_ms: 1000,
-                    func: () => s3_connection.upload({
+
+                const upload = new Upload({
+                    client: s3_connection,
+                    params: {
                         Bucket: target_bucket,
                         Key: `${log_prefix}${format_aws_date(date)}-${sha.slice(0, 16).toUpperCase()}`,
-                        Body: upload_stream,
-                    }).promise()
-                })));
+                        Body: upload_stream
+                    },
+                    queueSize: 1,
+                });
+
                 bucket_streams[source_bucket + BUCKET_NAME_DEL + target_bucket] = upload_stream;
+                promises.push(upload.done());
             }
             dbg.log2(`uploading entry: ${entry} to target bucket: ${target_bucket}`);
-            bucket_streams[source_bucket + BUCKET_NAME_DEL + target_bucket].write(entry + '\n');
+            const can_write = upload_stream.write(entry + '\n');
+            if (!can_write) {
+                dbg.warn(`waiting for target bucket: ${target_bucket} to drain`);
+                await once(upload_stream, 'drain');
+                dbg.warn(`target bucket: ${target_bucket} drained, resume writing`);
+            }
         });
         Object.values(bucket_streams).forEach(st => st.end());
         await Promise.all(promises);