NSFS | add dedicated servers for each forks for health checks

Signed-off-by: nadav mizrahi <nadav.mizrahi16@gmail.com>

Author: nadavMiz

config.js

@@ -979,6 +979,8 @@ config.ENDPOINT_SSL_PORT = Number(process.env.ENDPOINT_SSL_PORT) || 6443;
 // Remove the NSFS condition when NSFS starts to support STS.
 config.ENDPOINT_SSL_STS_PORT = Number(process.env.ENDPOINT_SSL_STS_PORT) || (process.env.NC_NSFS_NO_DB_ENV === 'true' ? -1 : 7443);
 config.ENDPOINT_SSL_IAM_PORT = Number(process.env.ENDPOINT_SSL_IAM_PORT) || -1;
+// each fork will get port in range [ENDPOINT_FORK_PORT_BASE, ENDPOINT_FORK_PORT_BASE + number of forks - 1)]
+config.ENDPOINT_FORK_PORT_BASE = Number(process.env.ENDPOINT_FORK_PORT_BASE) || 6002;
 config.ALLOW_HTTP = false;
 config.ALLOW_HTTP_METRICS = true;
 config.ALLOW_HTTPS_METRICS = true;

src/endpoint/endpoint.js

@@ -43,6 +43,7 @@ const { SemaphoreMonitor } = require('../server/bg_services/semaphore_monitor');
 const prom_reporting = require('../server/analytic_services/prometheus_reporting');
 const { PersistentLogger } = require('../util/persistent_logger');
 const { get_notification_logger } = require('../util/notifications_util');
+const { is_nc_enviorment } = require('../nc/nc_utils');
 const NoobaaEvent = require('../manage_nsfs/manage_nsfs_events_utils').NoobaaEvent;
 const cluster = /** @type {import('node:cluster').Cluster} */ (
     /** @type {unknown} */ (require('node:cluster'))
@@ -57,7 +58,8 @@ const SERVICES_TYPES_ENUM = Object.freeze({
     S3: 'S3',
     STS: 'STS',
     IAM: 'IAM',
-    METRICS: 'METRICS'
+    METRICS: 'METRICS',
+    FORK_HEALTH: 'FORK_HEALTH',
 });
 
 const new_umask = process.env.NOOBAA_ENDPOINT_UMASK || 0o000;
@@ -117,11 +119,11 @@ async function main(options = {}) {
         const https_metrics_port = options.https_metrics_port || config.EP_METRICS_SERVER_SSL_PORT;
         /**
         * Please notice that we can run the main in 2 states:
-        * 1. Only the primary process runs the main (fork is 0 or undefined) - everything that 
+        * 1. Only the primary process runs the main (fork is 0 or undefined) - everything that
         *    is implemented here would be run by this process.
-        * 2. A primary process with multiple forks (IMPORTANT) - if there is implementation that 
-        *    in only relevant to the primary process it should be implemented in 
-        *    fork_utils.start_workers because the primary process returns after start_workers 
+        * 2. A primary process with multiple forks (IMPORTANT) - if there is implementation that
+        *    in only relevant to the primary process it should be implemented in
+        *    fork_utils.start_workers because the primary process returns after start_workers
         *    and the forks will continue executing the code lines in this function
         *  */
         const is_workers_started_from_primary = await fork_utils.start_workers(http_metrics_port, https_metrics_port,
@@ -202,14 +204,29 @@ async function main(options = {}) {
             { ...options, https_port: https_port_s3, http_port: http_port_s3, virtual_hosts, bucket_logger, notification_logger });
         await start_endpoint_server_and_cert(SERVICES_TYPES_ENUM.STS, init_request_sdk, { https_port: https_port_sts, virtual_hosts });
         await start_endpoint_server_and_cert(SERVICES_TYPES_ENUM.IAM, init_request_sdk, { https_port: https_port_iam });
-
+        const is_nc = is_nc_enviorment();
+        // fork health server currently runs only on non containerized enviorment
+        if (is_nc) {
+            // current process is the primary and only fork. start the fork server directly with the base port
+            if (cluster.isPrimary) {
+                await fork_message_request_handler({
+                    nsfs_config_root: options.nsfs_config_root,
+                    health_port: config.ENDPOINT_FORK_PORT_BASE
+                });
+            // current process is a worker so we listen to get the port from the primary process.
+            } else {
+                process.on('message', fork_message_request_handler);
+                //send a message to the primary process that we are ready to receive messages
+                process.send('ready');
+            }
+        }
 
         // START METRICS SERVER
         if ((http_metrics_port > 0 || https_metrics_port > 0) && cluster.isPrimary) {
             await prom_reporting.start_server(http_metrics_port, https_metrics_port, false, options.nsfs_config_root);
         }
 
-        // TODO: currently NC NSFS deployments don't have internal_rpc_client nor db, 
+        // TODO: currently NC NSFS deployments don't have internal_rpc_client nor db,
         // there for namespace monitor won't be registered
         if (internal_rpc_client && config.NAMESPACE_MONITOR_ENABLED) {
             endpoint_stats_collector.instance().set_rpc_client(internal_rpc_client);
@@ -289,8 +306,6 @@ function create_endpoint_handler(server_type, init_request_sdk, { virtual_hosts,
                 return blob_rest_handler(req, res);
             } else if (req.url.startsWith('/total_fork_count')) {
                 return fork_count_handler(req, res);
-            } else if (req.url.startsWith('/endpoint_fork_id')) {
-                return endpoint_fork_id_handler(req, res);
             } else if (req.url.startsWith('/_/')) {
                 // internals non S3 requests
                 const api = req.url.slice('/_/'.length);
@@ -531,8 +546,37 @@ function unavailable_handler(req, res) {
     res.end(reply);
 }
 
+/**
+ * handler for the inidivdual fork server. used to handle requests the get the worker id
+ * currently used to check if fork is alive by the health script
+ * @type {EndpointHandler}
+ */
+function fork_main_handler(req, res) {
+    endpoint_utils.set_noobaa_server_header(res);
+    endpoint_utils.prepare_rest_request(req);
+    if (req.url.startsWith('/endpoint_fork_id')) {
+        return endpoint_fork_id_handler(req, res);
+    } else {
+        return internal_api_error(req, res, `Unknown API call ${req.url}`);
+    }
+}
+
+/**
+ * fork_message_request_handler is used to handle messages from the primary process.
+ * the primary process sends a message with the designated port to start the fork server.
+ * @param {Object} msg
+ */
+async function fork_message_request_handler(msg) {
+    await http_utils.start_https_server(msg.health_port,
+        SERVICES_TYPES_ENUM.FORK_HEALTH,
+        fork_main_handler,
+        msg.nsfs_config_root
+    );
+}
+
 exports.main = main;
 exports.create_endpoint_handler = create_endpoint_handler;
 exports.create_init_request_sdk = create_init_request_sdk;
+exports.endpoint_fork_id_handler = endpoint_fork_id_handler;
 
 if (require.main === module) main();

src/manage_nsfs/health.js

@@ -101,12 +101,14 @@ process.env.AWS_SDK_JS_SUPPRESS_MAINTENANCE_MODE_MESSAGE = '1';
 class NSFSHealth {
     constructor(options) {
         this.https_port = options.https_port;
+        this.fork_base_port = options.fork_base_port;
         this.all_account_details = options.all_account_details;
         this.all_bucket_details = options.all_bucket_details;
         this.all_connection_details = options.all_connection_details;
         this.notif_storage_threshold = options.notif_storage_threshold;
         this.lifecycle = options.lifecycle;
         this.config_fs = options.config_fs;
+        this.disable_service_validation = options.disable_service_validation;
     }
 
     /**
@@ -121,7 +123,13 @@ class NSFSHealth {
     async nc_nsfs_health() {
         let endpoint_state;
         let memory;
-        const noobaa_service_state = await this.get_service_state(NOOBAA_SERVICE_NAME);
+        let noobaa_service_state;
+        if (this.disable_service_validation) {
+            // if service validation is disabled, we assume the service is running, set dummy pid. should be used only for testing
+            noobaa_service_state = { name: NOOBAA_SERVICE_NAME, service_status: 'active', pid: '100' };
+        } else {
+            noobaa_service_state = await this.get_service_state(NOOBAA_SERVICE_NAME);
+        }
         const { service_status, pid } = noobaa_service_state;
         if (pid !== '0') {
             endpoint_state = await this.get_endpoint_response();
@@ -241,10 +249,10 @@ class NSFSHealth {
         return service_health;
     }
 
-    async make_endpoint_health_request(url_path) {
+    async make_endpoint_health_request(url_path, port = this.https_port) {
         const response = await make_https_request({
-            HOSTNAME,
-            port: this.https_port,
+            hostname: HOSTNAME,
+            port,
             path: url_path,
             method: 'GET',
             rejectUnauthorized: false,
@@ -260,43 +268,37 @@ class NSFSHealth {
         let url_path = '/total_fork_count';
         const worker_ids = [];
         let total_fork_count = 0;
+        let fork_count_response;
         let response;
         try {
-            const fork_count_response = await this.make_endpoint_health_request(url_path);
-            if (!fork_count_response) {
-                return {
-                    response: fork_response_code.NOT_RUNNING,
-                    total_fork_count: total_fork_count,
-                    running_workers: worker_ids,
-                };
-            }
-            total_fork_count = fork_count_response.fork_count;
-            if (total_fork_count > 0) {
-                url_path = '/endpoint_fork_id';
-                await P.retry({
-                    attempts: total_fork_count * 2,
-                    delay_ms: 1,
-                    func: async () => {
-                        const fork_id_response = await this.make_endpoint_health_request(url_path);
-                        if (fork_id_response.worker_id && !worker_ids.includes(fork_id_response.worker_id)) {
-                            worker_ids.push(fork_id_response.worker_id);
-                        }
-                        if (worker_ids.length < total_fork_count) {
-                            throw new Error('Number of running forks is less than the expected fork count.');
-                        }
-                    }
-                });
-                if (worker_ids.length === total_fork_count) {
-                    response = fork_response_code.RUNNING;
-                } else {
-                    response = fork_response_code.MISSING_FORKS;
-                }
-            } else {
-                response = fork_response_code.RUNNING;
-            }
+            fork_count_response = await this.make_endpoint_health_request(url_path);
         } catch (err) {
-            dbg.log1('Error while pinging endpoint host :' + HOSTNAME + ', port ' + this.https_port, err);
-            response = fork_response_code.NOT_RUNNING;
+            dbg.log0('Error while pinging endpoint host :' + HOSTNAME, err);
+        }
+        if (!fork_count_response) {
+            return {
+                response: fork_response_code.NOT_RUNNING,
+                total_fork_count: total_fork_count,
+                running_workers: worker_ids,
+            };
+        }
+
+        total_fork_count = fork_count_response.fork_count;
+        url_path = '/endpoint_fork_id';
+        for (let i = 0; i < total_fork_count; i++) {
+            const port = this.fork_base_port + i;
+            try {
+                const fork_id_response = await this.make_endpoint_health_request(url_path, port);
+                worker_ids.push(fork_id_response.worker_id);
+            } catch (err) {
+                dbg.log0('Error while pinging fork :' + HOSTNAME + ', port ' + port, err);
+            }
+        }
+        if (worker_ids.length < total_fork_count) {
+            dbg.log0('Number of running forks is less than the expected fork count.');
+            response = fork_response_code.MISSING_FORKS;
+        } else {
+            response = fork_response_code.RUNNING;
         }
         return {
             response: response,
@@ -635,16 +637,19 @@ class NSFSHealth {
 async function get_health_status(argv, config_fs) {
     try {
         const https_port = Number(argv.https_port) || config.ENDPOINT_SSL_PORT;
+        const fork_base_port = Number(argv.fork_base_port) || config.ENDPOINT_FORK_PORT_BASE;
         const deployment_type = argv.deployment_type || 'nc';
         const all_account_details = get_boolean_or_string_value(argv.all_account_details);
         const all_bucket_details = get_boolean_or_string_value(argv.all_bucket_details);
         const all_connection_details = get_boolean_or_string_value(argv.all_connection_details);
         const notif_storage_threshold = get_boolean_or_string_value(argv.notif_storage_threshold);
         const lifecycle = get_boolean_or_string_value(argv.lifecycle);
+        const disable_service_validation = get_boolean_or_string_value(argv.disable_service_validation);
 
         if (deployment_type === 'nc') {
-            const health = new NSFSHealth({ https_port,
-                all_account_details, all_bucket_details, all_connection_details, notif_storage_threshold, lifecycle, config_fs });
+            const health = new NSFSHealth({ https_port, fork_base_port,
+                all_account_details, all_bucket_details, all_connection_details,
+                notif_storage_threshold, lifecycle, config_fs, disable_service_validation });
             const health_status = await health.nc_nsfs_health();
             write_stdout_response(ManageCLIResponse.HealthStatus, health_status);
         } else {

src/manage_nsfs/manage_nsfs_constants.js

@@ -75,7 +75,7 @@ const VALID_OPTIONS_GLACIER = {
 };
 
 const VALID_OPTIONS_DIAGNOSE = {
-    'health': new Set([ 'https_port', 'deployment_type', 'all_account_details', 'all_bucket_details', 'all_connection_details', 'notif_storage_threshold', 'lifecycle', ...CLI_MUTUAL_OPTIONS]),
+    'health': new Set([ 'https_port', 'deployment_type', 'all_account_details', 'all_bucket_details', 'all_connection_details', 'notif_storage_threshold', 'lifecycle', 'disable_service_validation', ...CLI_MUTUAL_OPTIONS]),
     'gather-logs': new Set([ CONFIG_ROOT_FLAG]),
     'metrics': new Set([CONFIG_ROOT_FLAG])
 };

src/nc/nc_utils.js

@@ -24,7 +24,15 @@ function check_root_account_owns_user(root_account, account) {
     return root_account._id === account.owner;
 }
 
+/**
+ * @returns {boolean} true if the current environment is a NooBaa non containerized environment
+ */
+function is_nc_enviorment() {
+    return process.env.NC_NSFS_NO_DB_ENV && process.env.NC_NSFS_NO_DB_ENV === 'true';
+}
+
 // EXPORTS
 exports.generate_id = generate_id;
 exports.check_root_account_owns_user = check_root_account_owns_user;
+exports.is_nc_enviorment = is_nc_enviorment;
 

src/test/unit_tests/test_nc_with_a_couple_of_forks.js

@@ -223,4 +223,27 @@ const s3_uid6001 = generate_s3_client(access_details.access_key,
         await s3_uid6001.deleteBucket({ Bucket: bucket_name4 });
         await fs.promises.rm(new_bucket_path_param2, { recursive: true });
     });
+
+    mocha.it('health concurrency with load - ping forks', async function() {
+        this.timeout(100000); // eslint-disable-line no-invalid-this
+
+        for (let i = 0; i < 10; i++) {
+            // a couple of requests for health check
+            const failed_operations = [];
+            const successful_operations = [];
+            const num_of_concurrency = 10;
+            for (let j = 0; j < num_of_concurrency; j++) {
+                exec_manage_cli(TYPES.DIAGNOSE, 'health', {disable_service_validation: true})
+                    .catch(err => failed_operations.push(err))
+                    .then(res => successful_operations.push(res));
+            }
+            await P.delay(7000);
+            assert.equal(successful_operations.length, num_of_concurrency);
+            assert.equal(failed_operations.length, 0);
+            for (const res of successful_operations) {
+                const parsed_res = JSON.parse(res);
+                assert.strictEqual(parsed_res.response.reply.status, 'OK');
+            }
+        }
+    });
 });

src/util/fork_utils.js

@@ -10,6 +10,8 @@ const prom_reporting = require('../server/analytic_services/prometheus_reporting
 const NoobaaEvent = require('../manage_nsfs/manage_nsfs_events_utils').NoobaaEvent;
 const config = require('../../config');
 const stats_collector_utils = require('./stats_collector_utils');
+const { once } = require('node:events');
+const { is_nc_enviorment } = require('../nc/nc_utils');
 
 
 const io_stats = {
@@ -27,7 +29,7 @@ const fs_workers_stats = {};
  * When count > 0 the primary process will fork worker processes to process incoming http requests.
  * In case of any worker exit, also the entire process group will exit.
  * @see https://nodejs.org/api/cluster.html
- * 
+ *
  * @param {number} [metrics_port]
  * @param {number} [https_metrics_port]
  * @param {string} [nsfs_config_root] nsfs configuration path
@@ -36,15 +38,19 @@ const fs_workers_stats = {};
  */
 async function start_workers(metrics_port, https_metrics_port, nsfs_config_root, count = 0) {
     const exit_events = [];
+    const fork_port_offsets = [];
     if (cluster.isPrimary && count > 0) {
         for (let i = 0; i < count; ++i) {
             const worker = cluster.fork();
             console.warn('WORKER started', { id: worker.id, pid: worker.process.pid });
+            // no need to await. will run once the worker is ready to receive messages
+            _send_fork_server_message(worker, i, nsfs_config_root);
+            fork_port_offsets.push(worker.id);
         }
 
         // We don't want to leave the process with a partial set of workers,
         // so if any worker exits, we will print an error message in the logs and start a new one.
-        cluster.on('exit', (worker, code, signal) => {
+        cluster.on('exit', async (worker, code, signal) => {
             console.warn('WORKER exit', { id: worker.id, pid: worker.process.pid, code, signal });
             new NoobaaEvent(NoobaaEvent.FORK_EXIT).create_event(undefined, { id: worker.id, pid: worker.process.pid,
                 code: code, signal: signal}, undefined);
@@ -66,7 +72,11 @@ async function start_workers(metrics_port, https_metrics_port, nsfs_config_root,
             console.warn(`${exit_events.length} exit events in the last ${config.NSFS_EXIT_EVENTS_TIME_FRAME_MIN} minutes,` +
                 ` max allowed are: ${config.NSFS_MAX_EXIT_EVENTS_PER_TIME_FRAME}`);
             const new_worker = cluster.fork();
-            console.warn('WORKER re-started', { id: new_worker.id, pid: new_worker.process.pid });
+            const offset = fork_port_offsets.findIndex(id => id === worker.id);
+            _send_fork_server_message(new_worker, offset, nsfs_config_root);
+            fork_port_offsets[offset] = new_worker.id;
+            const port = is_nc_enviorment() ? {port: config.ENDPOINT_FORK_PORT_BASE + offset} : {};
+            console.warn('WORKER re-started', { id: new_worker.id, pid: new_worker.process.pid, ...port});
         });
         for (const id in cluster.workers) {
             if (id) {
@@ -101,6 +111,29 @@ function nsfs_io_stats_handler(msg) {
     }
 }
 
+/**
+ * Sends a message to the worker to start the fork server with the giver port
+ * NOTE - currently runs only on non containerized enviorment
+ * @param {*} worker
+ * @param {*} offset
+ * @param {*} nsfs_config_root
+ */
+async function _send_fork_server_message(worker, offset, nsfs_config_root) {
+    const is_nc = is_nc_enviorment();
+    if (is_nc && offset >= 0) {
+        //wait for the worker to be ready to receive messages
+        try {
+            await once(worker, 'message', { signal: AbortSignal.timeout(5 * 60000) });
+            worker.send({
+                nsfs_config_root: nsfs_config_root,
+                health_port: config.ENDPOINT_FORK_PORT_BASE + offset
+            });
+        } catch (err) {
+            dbg.warn(`Timeout: It took more than 5 minute to get a message from worker ${worker.id} do not send start server message`);
+        }
+    }
+}
+
 function _update_ops_stats(stats) {
     //Go over the op_stats
     for (const op_name of stats_collector_utils.op_names) {