Merge pull request #8981 from liranmauda/liran-remove-p-resolve

Refactor and maintenance

Author: liranmauda

src/server/node_services/nodes_monitor.js

@@ -849,31 +849,31 @@ class NodesMonitor extends EventEmitter {
         });
     }
 
-    _run_node(item) {
-        if (!this._started) return P.reject(new Error('monitor has not started'));
+    async _run_node(item) {
+        if (!this._started) throw new Error('monitor has not started');
         item._run_node_serial = item._run_node_serial || new semaphore.Semaphore(1);
-        if (item.node.deleted) return P.reject(new Error(`node ${item.node.name} is deleted`));
-        return item._run_node_serial.surround(() =>
-            P.resolve()
-            .then(() => dbg.log1('_run_node:', item.node.name))
-            .then(() => this._get_agent_info(item))
-            .then(() => { //If internal or cloud resource, cut down initializing time (in update_rpc_config)
-                if (!item.node_from_store && (item.node.is_mongo_node || item.node.is_cloud_node)) {
-                    return this._update_nodes_store('force');
-                }
-            })
-            .then(() => this._uninstall_deleting_node(item))
-            .then(() => this._remove_hideable_nodes(item))
-            .then(() => this._update_node_service(item))
-            .then(() => this._update_create_node_token(item))
-            .then(() => this._update_rpc_config(item))
-            .then(() => this._test_nodes_validity(item))
-            .then(() => this._update_status(item))
-            .then(() => this._handle_issues(item))
-            .then(() => this._update_nodes_store())
-            .catch(err => {
+        if (item.node.deleted) throw new Error(`node ${item.node.name} is deleted`);
+        return item._run_node_serial.surround(async () => {
+            dbg.log1('_run_node:', item.node.name);
+            await this._get_agent_info(item);
+            //If internal or cloud resource, cut down initializing time (in update_rpc_config)
+            if (!item.node_from_store && (item.node.is_mongo_node || item.node.is_cloud_node)) {
+                return this._update_nodes_store('force');
+            }
+            try {
+                await this._uninstall_deleting_node(item);
+                this._remove_hideable_nodes(item);
+                await this._update_node_service(item);
+                await this._update_create_node_token(item);
+                await this._update_rpc_config(item);
+                await this._test_nodes_validity(item);
+                this._update_status(item);
+                this._handle_issues(item);
+                await this._update_nodes_store();
+            } catch (err) {
                 dbg.warn('_run_node: ERROR', err.stack || err, 'node', item.node);
-            }));
+            }
+        });
     }
 
     _handle_issues(item) {
@@ -1681,71 +1681,71 @@ class NodesMonitor extends EventEmitter {
     // This is why we are required to use a new variable by the name ready_to_be_deleted
     // In order to mark the nodes that wait for their processes to be removed (cloud/mongo resource)
     // If the node is not relevant to a cloud/mongo resouce it will be just marked as deleted
-    _update_deleted_nodes(deleted_nodes) {
+    async _update_deleted_nodes(deleted_nodes) {
         if (!deleted_nodes.length) return;
         const items_to_update = [];
-        return P.map_with_concurrency(10, deleted_nodes, item => {
-                dbg.log0('_update_nodes_store deleted_node:', item);
 
-                if (item.node.deleted) {
-                    if (!item.node_from_store.deleted) {
-                        items_to_update.push(item);
-                    }
-                    return;
-                }
+        await P.map_with_concurrency(10, deleted_nodes, async item => {
+            dbg.log0('_update_nodes_store deleted_node:', item);
 
-                // TODO handle deletion of normal nodes (uninstall?)
-                // Just mark the node as deleted and we will not scan it anymore
-                // This is done once the node's proccess is deleted (relevant to cloud/mongo resource)
-                // Or in a normal node it is done immediately
-                if (!item.node.is_cloud_node &&
-                    !item.node.is_mongo_node &&
-                    !item.node.is_internal_node) {
-                    item.node.deleted = Date.now();
+            if (item.node.deleted) {
+                if (!item.node_from_store.deleted) {
                     items_to_update.push(item);
-                    return;
                 }
+                return;
+            }
 
-                return P.resolve()
-                    .then(() => {
-                        if (item.node.is_internal_node) {
-                            return P.reject('Do not support internal_node deletion yet');
-                        }
-                        // Removing the internal node from the processes
-                        return server_rpc.client.hosted_agents.remove_pool_agent({
-                            node_name: item.node.name
-                        });
-                    })
-                    .then(() => {
-                        // Marking the node as deleted since we've removed it completely
-                        // If we did not succeed at removing the process we don't mark the deletion
-                        // This is done in order to cycle the node once again and attempt until
-                        // We succeed
-                        item.node.deleted = Date.now();
-                        items_to_update.push(item);
-                    })
-                    .catch(err => {
-                        // We will just wait another cycle and attempt to delete it fully again
-                        dbg.warn('delete_cloud_or_mongo_pool_node ERROR node', item.node, err);
-                    });
-            })
-            .then(() => NodesStore.instance().bulk_update(items_to_update))
-            .then(res => {
-                // mark failed updates to retry
-                if (res.failed) {
-                    for (const item of res.failed) {
-                        this._set_need_update.add(item);
-                    }
+            // TODO handle deletion of normal nodes (uninstall?)
+            // Just mark the node as deleted and we will not scan it anymore
+            // This is done once the node's proccess is deleted (relevant to cloud/mongo resource)
+            // Or in a normal node it is done immediately
+            if (!item.node.is_cloud_node &&
+                !item.node.is_mongo_node &&
+                !item.node.is_internal_node) {
+                item.node.deleted = Date.now();
+                items_to_update.push(item);
+                return;
+            }
+
+            try {
+                if (item.node.is_internal_node) {
+                    throw new Error('Do not support internal_node deletion yet');
                 }
-                if (res.updated) {
-                    for (const item of res.updated) {
-                        this._remove_node_from_maps(item);
-                    }
+                // Removing the internal node from the processes
+                await server_rpc.client.hosted_agents.remove_pool_agent({
+                    node_name: item.node.name
+                });
+
+                // Marking the node as deleted since we've removed it completely
+                // If we did not succeed at removing the process we don't mark the deletion
+                // This is done in order to cycle the node once again and attempt until
+                // We succeed
+                item.node.deleted = Date.now();
+                items_to_update.push(item);
+
+            } catch (err) {
+                // We will just wait another cycle and attempt to delete it fully again
+                dbg.warn('delete_cloud_or_mongo_pool_node ERROR node', item.node, err);
+            }
+        });
+
+        try {
+            const res = NodesStore.instance().bulk_update(items_to_update);
+
+            // mark failed updates to retry
+            if (res.failed) {
+                for (const item of res.failed) {
+                    this._set_need_update.add(item);
                 }
-            })
-            .catch(err => {
-                dbg.warn('_update_deleted_nodes: ERROR', err.stack || err);
-            });
+            }
+            if (res.updated) {
+                for (const item of res.updated) {
+                    this._remove_node_from_maps(item);
+                }
+            }
+        } catch (err) {
+            dbg.warn('_update_deleted_nodes: ERROR', err.stack || err);
+        }
     }
 
     _should_enable_agent(info, agent_config) {

src/test/system_tests/test_bucket_lambda_triggers.js

@@ -130,16 +130,13 @@ async function authenticate() {
     await client.create_auth_token(auth_params);
 }
 
-function prepare_func(fn) {
-    return P.resolve()
-        .then(() => zip_utils.zip_from_files(fn.Files))
-        .then(zipfile => zip_utils.zip_to_buffer(zipfile))
-        .then(zip_buffer => {
-            delete fn.Files;
-            fn.Code = {
-                ZipFile: zip_buffer
-            };
-        });
+async function prepare_func(fn) {
+    const zipfile = await zip_utils.zip_from_files(fn.Files);
+    const zip_buffer = await zip_utils.zip_to_buffer(zipfile);
+    delete fn.Files;
+    fn.Code = {
+        ZipFile: zip_buffer
+    };
 }
 
 function get_new_server(user) {
@@ -218,7 +215,7 @@ async function setup() {
         await client.bucket.create_bucket({
             name: 'ns.external.bucket1',
             namespace: {
-                read_resources: [ nsr1 ],
+                read_resources: [nsr1],
                 write_resource: nsr2
             }
         });

src/test/system_tests/test_bucket_placement.js

@@ -53,7 +53,7 @@ async function run_test() {
         await perform_placement_tests();
         await perform_quota_tests();
         rpc.disconnect_all();
-        return P.resolve("Test Passed! Everything Seems To Be Fine...");
+        return "Test Passed! Everything Seems To Be Fine...";
     } catch (err) {
         console.error('test_bucket_placement FAILED: ', err.stack || err);
         rpc.disconnect_all();
@@ -199,26 +199,24 @@ async function perform_quota_tests() {
     await update_quota_on_bucket();
 }
 
-function update_quota_on_bucket(limit_gb) {
-    return P.resolve()
-        .then(() => {
-            if (limit_gb) {
-                return client.bucket.update_bucket({
-                    name: TEST_QUOTA_BUCKET_NAME,
-                    quota: {
-                        size: limit_gb,
-                        unit: 'GIGABYTE'
-                    }
-                });
-            } else {
-                return client.bucket.update_bucket({
-                    name: TEST_QUOTA_BUCKET_NAME,
-                });
-            }
-        })
-        .catch(err => {
-            throw new Error(`Failed setting quota with ${limit_gb} - ${err}`);
-        });
+async function update_quota_on_bucket(limit_gb) {
+    try {
+        if (limit_gb) {
+            return await client.bucket.update_bucket({
+                name: TEST_QUOTA_BUCKET_NAME,
+                quota: {
+                    size: limit_gb,
+                    unit: 'GIGABYTE'
+                }
+            });
+        } else {
+            return await client.bucket.update_bucket({
+                name: TEST_QUOTA_BUCKET_NAME,
+            });
+        }
+    } catch (err) {
+        throw new Error(`Failed setting quota with ${limit_gb} - ${err}`);
+    }
 }
 
 function main() {

src/test/system_tests/test_md_aggregator.js

@@ -59,39 +59,31 @@ function control_services(command, services) {
 }
 
 // Does the Auth and returns the nodes in the system
-function create_bucket(bucket_name) {
-    return P.resolve()
-        .then(() => client.tier.create_tier({
-            name: `${bucket_name}tier`,
-            attached_pools: ['first-pool'],
-            data_placement: 'SPREAD'
-        }))
-        .then(() => client.tiering_policy.create_policy({
-            name: `${bucket_name}tiering`,
-            tiers: [{
-                order: 0,
-                tier: `${bucket_name}tier`,
-                spillover: false,
-                disabled: false
-            }]
-        }))
-        .then(() => client.bucket.create_bucket({
-            name: bucket_name,
-            tiering: `${bucket_name}tiering`,
-        }));
+async function create_bucket(bucket_name) {
+    await client.tier.create_tier({
+        name: `${bucket_name}tier`,
+        attached_pools: ['first-pool'],
+        data_placement: 'SPREAD'
+    });
+    await client.tiering_policy.create_policy({
+        name: `${bucket_name}tiering`,
+        tiers: [{
+            order: 0,
+            tier: `${bucket_name}tier`,
+            spillover: false,
+            disabled: false
+        }]
+    });
+    return await client.bucket.create_bucket({
+        name: bucket_name,
+        tiering: `${bucket_name}tiering`,
+    });
 }
 
-function upload_file_to_bucket(bucket_name) {
-    let fkey;
-    return P.resolve()
-        .then(() => basic_server_ops.generate_random_file(1))
-        .then(fl => {
-            fkey = fl;
-            return basic_server_ops.upload_file(argv.ip, fl, bucket_name, fl);
-        })
-        .then(function() {
-            return fkey;
-        });
+async function upload_file_to_bucket(bucket_name) {
+    const file_key = await basic_server_ops.generate_random_file(1);
+    await basic_server_ops.upload_file(argv.ip, file_key, bucket_name, file_key);
+    return file_key;
 }
 
 async function prepare_buckets_with_objects() {

src/test/system_tests/test_node_failure.js

@@ -231,22 +231,20 @@ function test_node_fail_replicate() {
         }));
 }
 
-function run_test() {
-    return P.resolve()
-        .then(authenticate)
-        .then(setup)
-        .then(upload_file)
-        .then(read_mappings)
-        .then(test_node_fail_replicate)
-        .then(remove_agents)
-        .then(() => {
-            console.log('test_node_failure PASSED');
-        })
-        .catch(err => {
-            remove_agents();
-            console.log('test_node_failure failed. err =', err);
-            throw err;
-        });
+async function run_test() {
+    try {
+        await authenticate();
+        await setup();
+        await upload_file();
+        await read_mappings();
+        await test_node_fail_replicate();
+        await remove_agents();
+        console.log('test_node_failure PASSED');
+    } catch (err) {
+        await remove_agents();
+        console.log('test_node_failure failed. err =', err);
+        throw err;
+    }
 }