#2995 nginx/pm2 integration

Unitech · Aug 23, 2017 · f8b7bf3 · f8b7bf3
1 parent 7408d0c
commit f8b7bf3
Show file tree

Hide file tree

Showing 11 changed files with 184 additions and 34 deletions.
diff --git a/TODO-LB.md b/TODO-LB.md
@@ -0,0 +1,18 @@
+
+- [X] Send balancer packet via spiderlink to the nginx agent
+- [X] Allow to remove port fwding when deleting process
+- [ ] DRY lb system
+
+- [~] Bug when multiple delete (multi edit of the same file is fucking it)
+- [X] How to handle restart? (Currently does not work)
+- [X] How to handle unexpected restart?
+- [ ] Find a way to list process that being run in sudo mode
+- [ ] Find a way to keep alive nginx node.js interface
+- [X] Auto delete empty application (not any upstream left)
+- [ ] Verify that port assigned to worker process is available
+- [ ] Pass parameters of ecosystem.config.js to parameter nginx (SSL cert path)
+
+## Notifs
+
+- WORKS Running app on port < 1024
+- WARN pm2 must run first, then nginx-agent connect to it and finally start app
diff --git a/bin/pm2 b/bin/pm2
@@ -16,6 +16,12 @@ var PM2       = require('../lib/API.js');
 var pkg       = require('../package.json');
 var tabtab    = require('../lib/completion.js');
 
+var spm2 = require('spiderlink')('pm2');
+
+spm2.longCall('connector', function(data) {
+  console.log(data);
+});
+
 var pm2 = new PM2();
 
 commander.version(pkg.version)
@@ -60,6 +66,8 @@ commander.version(pkg.version)
   .option('--ignore-watch <folders|files>', 'folder/files to be ignored watching, chould be a specific name or regex - e.g. --ignore-watch="test node_modules \"some scripts\""')
   .option('--node-args <node_args>', 'space delimited arguments to pass to node in cluster mode - e.g. --node-args="--debug=7001 --trace-deprecation"')
   .option('--no-color', 'skip colors')
+  .option('-t --external-lb', 'configure external lb to listen on app port and lb to <instances> apps')
+  .option('--lb-mode <sticky|rr|least_conn>', 'load balancing mode')
   .option('--no-vizion', 'start an app without vizion feature (versioning control)')
   .option('--no-autorestart', 'start an app without automatic restart')
   .option('--no-treekill', 'Only kill the main process, not detached children')
@@ -122,6 +130,7 @@ function beginCommandProcessing() {
       console.log('');
     }
   });
+
   commander.parse(process.argv);
 }
 

diff --git a/examples/http.js b/examples/http.js
@@ -1,11 +1,13 @@
 
 var http = require('http');
+var port = 0;
 
 var server = http.createServer(function(req, res) {
   res.writeHead(200);
-  res.end('hey');
+  res.end(port + '');
 }).listen(process.env.PORT || 8000, function() {
-  console.log('App listening on port %d in env %s', process.env.PORT || 8000, process.env.NODE_ENV);
+  port = server.address().port;
+  console.log('App listening on port %d in env %s', server.address().port, process.env.NODE_ENV);
 
   // 1# Notify application ready
   setTimeout(function() {

diff --git a/lib/API.js b/lib/API.js
@@ -22,7 +22,7 @@ var Modularizer = require('./API/Modules/Modularizer.js');
 var path_structure = require('../paths.js');
 var UX          = require('./API/CliUx');
 
-var IMMUTABLE_MSG = chalk.bold.blue('Use --update-env to update environment variables');
+var IMMUTABLE_MSG = chalk.bold.magenta('>>> Use --update-env to update environment variables');
 
 /**
  * Main Function to be imported

diff --git a/lib/API/CliUx.js b/lib/API/CliUx.js
@@ -10,6 +10,7 @@ var Common        = require('../Common');
 var Spinner       = require('./Spinner.js');
 var os = require('os');
 var UX = module.exports = {};
+var spm2 = require('spiderlink')('pm2');
 
 /**
  * Description
@@ -320,7 +321,9 @@ UX.dispAsTable = function(list, interact_infos) {
 
   });
 
+  /** Print Tables **/
   console.log(app_table.toString());
+
   if (module_table.length > 0) {
     console.log(chalk.bold(' Module activated'));
     console.log(module_table.toString());

diff --git a/lib/API/schema.json b/lib/API/schema.json
@@ -17,6 +17,12 @@
     ],
     "alias": ["interpreterArgs", "interpreter_args"]
   },
+  "external_lb" : {
+    "type" : "boolean"
+  },
+  "lb_mode" : {
+    "type" : "string"
+  },
   "name": {
     "type": "string"
   },

diff --git a/lib/God.js b/lib/God.js
@@ -22,10 +22,19 @@ var EventEmitter2 = require('eventemitter2').EventEmitter2;
 var fs            = require('fs');
 var pidusage      = require('pidusage');
 var vizion        = require('vizion');
+var Spiderlink    = require('spiderlink');
+var async         = require('async');
 var debug         = require('debug')('pm2:god');
 var Utility       = require('./Utility');
 var cst           = require('../constants.js');
-var async = require('async');
+var PortPool      = require('./PortPool.js');
+
+/**
+ * Initialize Spiderlink
+ */
+Spiderlink({
+  server : true
+})
 
 /**
  * Override cluster module configuration
@@ -41,6 +50,7 @@ var God = module.exports = {
   next_id : 0,
   clusters_db : {},
   started_at : Date.now(),
+  spiderlink : Spiderlink('pm2'),
   bus : new EventEmitter2({
     wildcard: true,
     delimiter: ':',
@@ -72,6 +82,10 @@ require('./Watcher')(God);
 God.executeApp = function executeApp(env, cb) {
   var env_copy = Utility.clone(env);
 
+  if (env_copy.external_lb == true) {
+    env_copy.env.NEW_PORT = PortPool.getAvailablePort();
+  }
+
   Utility.extend(env_copy, env_copy.env);
 
   env_copy['status']         = cst.LAUNCHING_STATUS;
@@ -141,14 +155,51 @@ God.executeApp = function executeApp(env, cb) {
 
   /** Callback when application is launched */
   var readyCb = function ready(proc) {
-      if (proc.pm2_env.vizion !== false && proc.pm2_env.vizion !== "false")
+    if (proc.pm2_env.vizion !== false && proc.pm2_env.vizion !== "false")
         God.finalizeProcedure(proc);
-      else
+    else
         God.notify('online', proc);
 
-      proc.pm2_env.status = cst.ONLINE_STATUS;
-      console.log('App name:%s id:%s online', proc.pm2_env.name, proc.pm2_env.pm_id);
-      if (cb) cb(null, proc);
+    proc.pm2_env.status = cst.ONLINE_STATUS;
+
+    if (proc.pm2_env.external_lb == true) {
+      var p = God.clusters_db[proc.pm2_env.pm_id];
+
+      if (p.pm2_env.prev_listening_port) {
+        return God.spiderlink.call('addPortRouting', {
+          app_name : p.pm2_env.name,
+          opts : {
+            in_port : p.pm2_env.prev_listening_port,
+            out_port : p.pm2_env.env.NEW_PORT
+          }
+        }, function(err) {
+          if (err) console.error(err);
+          if (cb) return cb(null, proc);
+        });
+      }
+
+      var listener = function (packet) {
+        God.bus.removeListener('prev_listening_port', listener);
+        God.clusters_db[proc.pm2_env.pm_id].pm2_env.prev_listening_port = packet.data.port;
+
+        return God.spiderlink.call('addPortRouting', {
+          app_name : p.pm2_env.name,
+          opts : {
+            in_port : p.pm2_env.prev_listening_port,
+            out_port : p.pm2_env.env.NEW_PORT
+          }
+        }, function(err) {
+          if (err) console.error(err);
+          return cb(null, proc);
+        });
+        return false;
+      }
+
+      return God.bus.on('prev_listening_port', listener);
+    }
+
+    console.log('App name:%s id:%s online', proc.pm2_env.name, proc.pm2_env.pm_id);
+    if (cb) cb(null, proc);
   }
 
   if (env_copy.exec_mode === 'cluster_mode') {
@@ -290,11 +341,23 @@ God.handleExit = function handleExit(clu, exit_code, kill_signal) {
     return false;
   }
 
+  if (proc.pm2_env.external_lb == true) {
+    // External load balancer reconfiguration
+    var balancer = {
+      app_name : proc.pm2_env.name,
+      port : proc.pm2_env.env.NEW_PORT
+    }
+
+    God.spiderlink.call('deletePortRouting', balancer, function(err, packet) {
+      console.log('Succesfully deleted routing to port %s', proc.pm2_env.env.NEW_PORT);
+    });
+  }
+
   if (proc.process.pid)
     pidusage.unmonitor(proc.process.pid);
 
   var stopping    = (proc.pm2_env.status == cst.STOPPING_STATUS
-                     || proc.pm2_env.status == cst.STOPPED_STATUS
+                    || proc.pm2_env.status == cst.STOPPED_STATUS
                      || proc.pm2_env.status == cst.ERRORED_STATUS) || (proc.pm2_env.autorestart === false ||
                                                                        proc.pm2_env.autorestart === "false");
 
@@ -382,38 +445,36 @@ God.handleExit = function handleExit(clu, exit_code, kill_signal) {
 /**
  * Init new process
  */
-God.prepare = function prepare (env, cb) {
-  // if the app is standalone, no multiple instance
-  if (typeof env.instances === 'undefined') {
-    env.vizion_running = false;
-    if (env.env && env.env.vizion_running) env.env.vizion_running = false;
+God.prepare = function prepare(env, cb) {
+  // var external_lb = false;
 
-    return God.executeApp(env, function (err, clu) {
-      if (err) return cb(err);
-      God.notify('start', clu, true);
-      return cb(null, [ Utility.clone(clu) ]);
-    });
-  }
+  // if (env.external_lb)
+  //   external_lb = true;
+
+  // Determine number of instances to prepare
+  if (typeof(env.instances) === 'undefined')
+    env.instances = 1;
 
-  // find how many replicate the user want
   env.instances = parseInt(env.instances);
-  if (env.instances === 0) {
+
+  if (env.instances === 0)
     env.instances = numCPUs;
-  } else if (env.instances < 0) {
+  else if (env.instances < 0)
     env.instances += numCPUs;
-  }
-  if (env.instances <= 0) {
+  if (env.instances <= 0)
     env.instances = 1;
-  }
 
+  // Start each instance
   async.timesLimit(env.instances, 1, function (n, next) {
     env.vizion_running = false;
-    if (env.env && env.env.vizion_running) {
-      env.env.vizion_running = false;
-    }
 
     God.injectVariables(env, function inject (err, _env) {
       if (err) return next(err);
+
+      // if (external_lb === true) {
+      //   _env.env.NEW_PORT = PortPool.getAvailablePort();
+      // }
+
       return God.executeApp(Utility.clone(_env), function (err, clu) {
         if (err) return next(err);
         God.notify('start', clu, true);
@@ -422,7 +483,36 @@ God.prepare = function prepare (env, cb) {
         return next(null, Utility.clone(clu));
       });
     });
-  }, cb);
+  }, function(err, procs) {
+    // if (!err && external_lb === true) {
+    //   var ret = procs.map(function(proc) {
+    //     return proc.pm2_env.env.NEW_PORT;
+    //   });
+
+    //   var listener = function (packet) {
+    //     God.bus.removeListener('prev_listening_port', listener);
+    //     God.clusters_db[procs[0].pm2_env.pm_id].pm2_env.prev_listening_port = packet.data.port;
+
+    //     var balancer = {
+    //       app_name : procs[0].pm2_env.name,
+    //       routing : {
+    //         mode : 'http',
+    //         in_port : packet.data.port,
+    //         out_ports : ret
+    //       }
+    //     };
+
+    //     God.spiderlink.call('addOrUpdateAppRouting', balancer, function() {
+    //       return cb(err, procs);
+    //     });
+    //     return false;
+    //   }
+
+    //   return God.bus.on('prev_listening_port', listener);
+    // }
+
+    return cb(err, procs);
+  });
 };
 
 /**

diff --git a/lib/God/ForkMode.js b/lib/God/ForkMode.js
@@ -192,6 +192,12 @@ module.exports = function ForkMode(God) {
          * Do the same in ClusterMode.js !
          *********************************/
         if (msg.data && msg.type) {
+
+          // For external lb mode, we keep the default listening port
+          if (msg.type == 'prev_listening_port') {
+            cspr.pm2_env.prev_listening_port = msg.data.port;
+          }
+
           process.nextTick(function() {
             return God.bus.emit(msg.type ? msg.type : 'process:msg', {
               at      : Utility.getDate(),
@@ -205,7 +211,6 @@ module.exports = function ForkMode(God) {
           });
         }
         else {
-
           if (typeof msg == 'object' && 'node_version' in msg) {
             cspr.pm2_env.node_version = msg.node_version;
             return false;

diff --git a/lib/PortPool.js b/lib/PortPool.js
@@ -0,0 +1,14 @@
+
+var port = 20000;
+
+var PortPool = module.exports = {};
+
+PortPool.getAvailablePort = function() {
+  return port++;
+}
+
+PortPool.startWorker = function() {
+  setInterval(function() {
+
+  }, 2000);
+}
diff --git a/lib/ProcessContainerFork.js b/lib/ProcessContainerFork.js
@@ -5,9 +5,10 @@
  */
 // Inject custom modules
 if (process.env.pmx !== "false") {
-  require('pmx').init({
+  var pmx = require('pmx').init({
     transactions: process.env.km_link == 'true' && process.env.trace == 'true' || false,
-    http: process.env.km_link == 'true' || false
+    http: process.env.km_link == 'true' || false,
+    new_port : process.env.NEW_PORT || null
   });
 }
 

diff --git a/package.json b/package.json
@@ -176,6 +176,8 @@
     "pm2-multimeter": "^0.1.2",
     "pmx": "keymetrics/pmx#development",
     "semver": "^5.3",
+    "spiderlink": "latest",
+    "ws": "*",
     "shelljs": "0.7.8",
     "source-map-support": "^0.4.15",
     "sprintf-js": "1.1.1",