made assets service to sync with cluster master instantiation resolve… (

#587) * made assets service to sync with cluster master instantiation resolves #584 * encapsulated asset online logic in closure
terascope · Oct 30, 2017 · 73feeb2 · 73feeb2
1 parent 604fd97
commit 73feeb2
Show file tree

Hide file tree

Showing 4 changed files with 48 additions and 6 deletions.
diff --git a/lib/cluster/cluster_master.js b/lib/cluster/cluster_master.js
@@ -20,6 +20,8 @@ module.exports = function(context) {
 
     var services = [];
 
+    const assetServiceOnline = assetServicelistener();
+
     require('./services/cluster')(context, server)
         .then(function(cluster_service) {
             logger.trace('cluster_service has instantiated');
@@ -29,6 +31,9 @@ module.exports = function(context) {
         .then(function(jobs_service) {
             logger.trace('jobs_service has instantiated');
             services.push(jobs_service);
+            return assetServiceOnline();
+        })
+        .then(function() {
             return require('./services/api')(context, app, services);
         })
         .then(function(api_service) {
@@ -68,4 +73,33 @@ module.exports = function(context) {
             });
     });
 
+    function assetServicelistener() {
+        let assetsServiceIsReady = false;
+        let assetsServiceError = false;
+
+        messaging.register('assets:service:available', (assetServiceResponse) => {
+            if (assetServiceResponse.error) {
+                assetsServiceError = true;
+            } else {
+                assetsServiceIsReady = true;
+            }
+        });
+        return () => new Promise(((resolve, reject) => {
+            if (assetsServiceIsReady) {
+                resolve(true);
+            } else {
+                const assetReady = setInterval(() => {
+                    if (assetsServiceIsReady) {
+                        clearInterval(assetReady);
+                        resolve(true);
+                    }
+                    if (assetsServiceError) {
+                        clearInterval(assetReady);
+                        reject('An error occurred while instantiating the assets service');
+                    }
+                }, 100);
+            }
+        }));
+    }
+
 };
diff --git a/lib/cluster/node_master.js b/lib/cluster/node_master.js
@@ -103,8 +103,8 @@ module.exports = function(context) {
         deleteDir(`${assets_path}/${assetID}`)
             .catch(function(err) {
                 var errMsg = parseError(err);
-                logger.error(errMsg)
-            })
+                logger.error(errMsg);
+            });
     });
 
     messaging.register('assets:preloaded', function(assetsMsg) {
@@ -251,7 +251,13 @@ module.exports = function(context) {
 
     messaging.register('cluster:job:pause', function(data) {
         messageWorkers(context, data, {message: 'cluster:job:pause'}, function(worker) {
-            return worker.ex_id === data.ex_id && worker.assignment === 'slicer'
+            return worker.ex_id === data.ex_id && worker.assignment === 'slicer';
+        });
+    });
+
+    messaging.register('assets:service:available', function(assetsMsg) {
+        messageWorkers(context, assetsMsg, assetsMsg, function(worker) {
+            return worker.assignment === 'cluster_master';
         });
     });
 

diff --git a/lib/cluster/services/assets.js b/lib/cluster/services/assets.js
@@ -79,15 +79,16 @@ module.exports = function(context) {
             assets_store = store;
             var port = process.env.port;
             logger.info(`assets_service is listening on port ${port}`);
-
             app.listen(port);
+            messaging.send({message: 'assets:service:available'});
         })
         .catch(function(err) {
             var errMsg = parseError(err);
             logger.error(`Error while creating assets_service, error: ${errMsg}`);
+            messaging.send({ message: 'assets:service:available', error: errMsg });
             setTimeout(function() {
                 process.exit(0)
-            }, 10)
+            }, 100)
         });
 
     function createAssetTable(query, req, res) {

diff --git a/lib/cluster/services/messaging.js b/lib/cluster/services/messaging.js
@@ -43,7 +43,8 @@ var processMapping = {
     'shutdown': 'worker:shutdown',
     'assets:loaded': 'assets:loaded',
     'jobs_service:verify_assets': 'jobs_service:verify_assets',
-    'assets:preloaded': 'assets:preloaded'
+    'assets:preloaded': 'assets:preloaded',
+    'assets:service:available': 'assets:service:available'
 };
 
 //needed so slicer can listen for a process message from node master (originally a network call from cluster)