Merge pull request #57 from gdg-x/fixClusterResourceLeak

fix(cluster): fix server outages due to cluster resource leakage

Author: Splaktar

lib/.jshintrc

@@ -2,14 +2,21 @@
   "node": true,
   "esnext": true,
   "bitwise": true,
+  "camelcase": true,
+  "curly": true,
   "eqeqeq": true,
   "immed": true,
-  "latedef": true,
+  "indent": 2,
+  "latedef": "nofunc",
   "newcap": true,
   "noarg": true,
+  "quotmark": "single",
   "regexp": true,
   "undef": true,
+  "unused": true,
   "smarttabs": true,
+  "strict": true,
+  "trailing": true,
   "globals": {
     "next": false,
     "document": false

lib/config/keys.js

@@ -3,14 +3,13 @@
 module.exports = {
 	keys: {
 		google: {
-			simpleApiKey: process.env.GOOGLE_SIMPLE_API_KEY || "",
-			oauthClientId: process.env.GOOGLE_OAUTH_CLIENT_ID || "",
-			oauthClientSecret: process.env.GOOGLE_OAUTH_CLIENT_SECRET || ""
+			simpleApiKey: process.env.GOOGLE_SIMPLE_API_KEY || '',
+			oauthClientId: process.env.GOOGLE_OAUTH_CLIENT_ID || '',
+			oauthClientSecret: process.env.GOOGLE_OAUTH_CLIENT_SECRET || ''
 		},
-		newrelic: {
-			licenseKey: ""
+		frisbee: {
+			serverClientId: process.env.SERVER_KEY_SECRET || '',
+      androidClientIds: process.env.ANDROID_CLIENT_IDS || []
 		}
 	}
 };
-
-process.env.NEWRELIC_AGENT = module.exports.keys.newrelic.licenseKey;

lib/config/passport.js

@@ -3,13 +3,11 @@
 var mongoose = require('mongoose'),
   User = mongoose.model('User'),
   passport = require('passport'),
-  SimpleApiKey = mongoose.model('SimpleApiKey'),
   OauthConsumer = mongoose.model('OauthConsumer'),
   OauthClient = mongoose.model('OauthClient'),
   ConsumerStrategy = require('passport-http-oauth').ConsumerStrategy,
   TokenStrategy = require('passport-http-oauth').TokenStrategy,
   BearerStrategy = require('passport-http-bearer').Strategy,
-  request = require('superagent'),
   config = require('./config'),
   utils = require('../utils');
 
@@ -98,8 +96,10 @@ module.exports = function () {
     function (token, done) {
       utils.getGoogleCert(function (certs) {
         utils.decodeAndVerifyJwt(token, certs, function (err, claims) {
+
           if (claims.aud === config.keys.frisbee.serverClientId &&
-            config.keys.frisbee.androidClientIds.indexOf(claims.azp) !== -1) {
+              config.keys.frisbee.androidClientIds.indexOf(claims.azp) !== -1) {
+
             User.findOne({_id: claims.sub}, function (err, user) {
               if (err) {
                 return done(err);

lib/fixtures/countries.js

@@ -1,14 +1,13 @@
 'use strict';
 
-var config = require('../config/config'),
-  mongoose = require('mongoose'),
-  fs = require("fs"),
+var mongoose = require('mongoose'),
+  fs = require('fs'),
   Country = mongoose.model('Country');
 
 module.exports = function () {
   Country.count({}, function (err, count) {
     if (count === 0) {
-      fs.readFile(__dirname + "/countries.json", 'utf8', function (err, data) {
+      fs.readFile(__dirname + '/countries.json', 'utf8', function (err, data) {
         if (err) {
           console.log('Error: ' + err);
           return;

lib/fixtures/index.js

@@ -1,9 +1,9 @@
 'use strict';
 
 module.exports = function () {
-  require("fs").readdirSync(__dirname + '/').forEach(function (file) {
+  require('fs').readdirSync(__dirname + '/').forEach(function (file) {
     if (file.match(/.+\.js/g) !== null && file.match(/.+\.json/g) === null && file !== 'index.js') {
-      require("./" + file)();
+      require('./' + file)();
     }
   });
 };

lib/risky.js

@@ -27,7 +27,7 @@ module.exports = (function () {
       }
     }
 
-    console.log("was master: " + master + ", now master: " + iAmMaster);
+    console.log('was master: ' + master + ', now master: ' + iAmMaster);
     master = iAmMaster;
   };
 
@@ -37,11 +37,11 @@ module.exports = (function () {
 
     var logentry = new TaskLog();
     logentry._id = id;
-    logentry.task_type = type;
-    logentry.started_at = moment(started);
-    logentry.ended_at = ended;
-    logentry.requested_by = myId;
-    logentry.executed_by = executor;
+    logentry.task_type = type; // jshint ignore:line
+    logentry.started_at = moment(started); // jshint ignore:line
+    logentry.ended_at = ended; // jshint ignore:line
+    logentry.requested_by = myId; // jshint ignore:line
+    logentry.executed_by = executor; // jshint ignore:line
 
     if (err) {
       logentry.result = 1;
@@ -51,8 +51,9 @@ module.exports = (function () {
     }
     logentry.save();
 
-    if (tasks[id])
+    if (tasks[id]) {
       delete tasks[id];
+    }
 
     cb(err, result, elapsed);
   };
@@ -63,16 +64,16 @@ module.exports = (function () {
     readyCount++;
     if (readyCount === 2) {
       startup = moment().valueOf();
-      console.log("Risky is up. I'm " + myId);
+      console.log('Risky is up. I\'m ' + myId);
       self.on('group:hello', function (data) {
-        if (data.sender !== myId && data.type === "hi") {
+        if (data.sender !== myId && data.type === 'hi') {
           if (responder) {
-            console.log("Cancel masterResponder");
+            console.log('Cancel masterResponder');
             clearTimeout(responder);
             responder = null;
           }
 
-          console.log(data.sender + " just said hi. Replying.");
+          console.log(data.sender + ' just said hi. Replying.');
           cluster[data.sender] = {
             started: data.started,
             nextHeartbeat: moment().valueOf() + 10000
@@ -84,9 +85,9 @@ module.exports = (function () {
             type: 'hi_reply',
             started: startup
           });
-        } else if (data.sender !== myId && data.type === "hi_reply") {
+        } else if (data.sender !== myId && data.type === 'hi_reply') {
           if (responder) {
-            console.log("Cancel masterResponder");
+            console.log('Cancel masterResponder');
             clearTimeout(responder);
             responder = null;
           }
@@ -109,7 +110,7 @@ module.exports = (function () {
           var next = (data.nextHeartbeat - moment().valueOf()) + 2000;
 
           cluster[data.sender].timeout = setTimeout(function () {
-            console.log(data.sender + " has gone down...");
+            console.log(data.sender + ' has gone down...');
             delete cluster[data.sender];
             evaluateMaster();
           }, next);
@@ -120,24 +121,24 @@ module.exports = (function () {
         if (!master) {
           if (taskHandler[data.type]) {
             // Offer to execute task
-            console.log("Offering to execute task with type: " + data.type);
+            console.log('Offering to execute task with type: ' + data.type);
             self.emit('group:taskoffer', {
               sender: myId,
               type: data.type,
               id: data.taskId
             });
           } else {
-            console.log("Don't know how to execute task");
+            console.log('Don\'t know how to execute task');
           }
         } else {
-          console.log("I'm the master, not doing any tasks");
+          console.log('I\'m the master, not doing any tasks');
         }
       });
 
       self.on('group:taskoffer', function (data) {
         // First one to send an offer wins
-        if (tasks[data.id] && tasks[data.id].state === "open") {
-          tasks[data.id].state = "executing";
+        if (tasks[data.id] && tasks[data.id].state === 'open') {
+          tasks[data.id].state = 'executing';
           tasks[data.id].executor = data.sender;
 
           tasks[data.id].acceptCallback(null, data.id, data.type, moment());
@@ -154,7 +155,7 @@ module.exports = (function () {
 
       self.on('group:taskstart', function (data) {
         if (data.recipient === myId) {
-          console.log("Executing task... type: " + data.type + ", id: " + data.id);
+          console.log('Executing task... type: ' + data.type + ', id: ' + data.id);
           taskHandler[data.type](data.id, data.params, function (err, result) {
             self.emit('group:taskdone', {
               sender: myId,
@@ -238,11 +239,11 @@ module.exports = (function () {
 
       var id = uuid.v4();
       if ((master || force) && Object.keys(cluster).length > 0) {
-        console.log("Sending out task " + taskType + " with id " + id);
+        console.log('Sending out task ' + taskType + ' with id ' + id);
         tasks[id] = {
           type: taskType,
           started: moment().valueOf(),
-          state: "open",
+          state: 'open',
           params: params,
           acceptCallback: acceptCallback,
           cb: cb
@@ -254,15 +255,15 @@ module.exports = (function () {
           taskId: id
         });
       } else if (Object.keys(cluster).length === 0 && taskHandler[taskType]) {
-        console.log("Doing task myself...");
+        console.log('Doing task myself...');
         acceptCallback(null, id, taskType, moment());
         var started = moment().valueOf();
         taskHandler[taskType](id, params, function (err, result) {
           taskDone(id, taskType, started, myId, err, result, cb);
         });
       } else {
-        console.log("Not doing it...");
-        acceptCallback("Not doing it");
+        console.log('Not doing it...');
+        acceptCallback('Not doing it');
       }
     },
     on: function (channel, handler, cb) {

server.js

@@ -21,19 +21,39 @@ console.log = function () {
   }
 };
 
-if (cluster.isMaster) {
+function initWorker(worker) {
+  var listeners;
+
+  listeners = worker.process.listeners('exit')[0];
+  var exit = listeners[Object.keys(listeners)[0]];
+
+  listeners = worker.process.listeners('disconnect')[0];
+  var disconnect = listeners[Object.keys(listeners)[0]];
+
+  worker.process.removeListener('exit', exit);
+  worker.process.once('exit', function(exitCode, signalCode) {
+    if (worker.state !== 'disconnected') {
+      disconnect();
+    }
+    exit(exitCode, signalCode);
+  });
+}
 
+if (cluster.isMaster) {
+  var i;
   // Fork workers.
-  for (var i = 0; i < numCPUs; i++) {
+  for (i = 0; i < numCPUs; i++) {
     setTimeout(function () {
       var worker = cluster.fork();
+      initWorker(worker);
       console.log('worker started, PID ' + worker.process.pid);
     }, (i + 1) * 5000); // jshint ignore:line
   }
 
   cluster.on('exit', function (deadWorker, code, signal) {
     // Restart the worker
     var worker = cluster.fork();
+    initWorker(worker);
 
     // Note the process IDs
     var newPID = worker.process.pid;
@@ -43,17 +63,13 @@ if (cluster.isMaster) {
     console.log('worker ' + oldPID + ' died. Code: ' + code + ', Signal: ' + signal);
     console.log('worker ' + newPID + ' born.');
   });
-
 } else {
   // Default node environment to development
   process.env.NODE_ENV = process.env.NODE_ENV || 'development';
 
   // Application Config
   var config = require('./lib/config/config');
 
-  // Disable NewRelic for now.
-  //var newrelic = require('newrelic');
-
   // Connect to database
   var db = mongoose.connect(config.mongo.uri, config.mongo.options); // jshint ignore:line
 
@@ -64,6 +80,7 @@ if (cluster.isMaster) {
     require(modelsPath + '/' + file);
   });
 
+  // Import static data
   require('./lib/fixtures')();
 
   var risky = require('./lib/risky');
@@ -72,9 +89,9 @@ if (cluster.isMaster) {
   require('./lib/config/passport')();
 
   if (config.env === 'production' && config.redis) {
-    var myId = process.env.OPENSHIFT_GEAR_UUID + '';
+    var myId = 'workerId';
     if (cluster.isWorker) {
-      myId = process.env.OPENSHIFT_GEAR_UUID + '-' + cluster.worker.process.pid;
+      myId = 'worker-' + cluster.worker.process.pid;
     }
     risky.connect({
       port: config.redis.port,