Merge pull request #1 from cubejs/master

rebasing

Author: dimichgh

.gitignore

@@ -12,3 +12,6 @@ logs
 cluster-cache-domain
 cluster-cache-persist
 ports
+/coverage/
+/lib-cov/
+/log/

lib/index.js

@@ -82,6 +82,7 @@ Cluster.prototype.listen = function(createApp, cb) {
             connThreshold: options.connThreshold || 10000, // recycle workers after this many connections
             uptimeThreshold: options.uptimeThreshold || 3600 * 24, // 24 hours (uptimeThreshold is in seconds)
             heartbeatInterval: options.heartbeatInterval,
+            maxHeartbeatDelay: options.maxHeartbeatDelay,
             emitter: self
         });
 

lib/process.js

@@ -49,6 +49,8 @@ var Process = module.exports = function Process(options) {
         workersKilled: 0
     };
 
+    this.options.maxHeartbeatDelay = options.maxHeartbeatDelay || 3*60000; //default 3 mins
+
     this._heartbeats = [];
 
     this.killall = function(signal) {
@@ -150,9 +152,12 @@ var Process = module.exports = function Process(options) {
                 self.emitter.emit('listening', message.pid);
             }
             if(message.type === 'heartbeat'){
-
                 if(message.pid != process.pid){
                     self._heartbeats.push(message);//must append to the tail
+                    // update the last heartbeat time for the worker
+                    var workerStats = self.stats.workers[message.pid];
+                    //console.log('heartbeat ' + process.pid);
+                    workerStats.lastHeartbeatAt= Date.now();
                 }
 
                 self._heartbeatScheduler = self._heartbeatScheduler || setInterval(function () {
@@ -194,6 +199,23 @@ var Process = module.exports = function Process(options) {
                     });
 
                     self.lastTime = Date.now();
+
+                    // Check the last heartbeat time of all the workers
+                    _.each(self.stats.workers, function (workerStats, pid) {
+                        var now = Date.now();
+                        if (now - workerStats.lastHeartbeatAt> self.options.maxHeartbeatDelay) {
+                            // this worker hasn't been sending heartbeat for maxHeartbeatDelay 
+                            log(util.format('[Cluster2] Detected worker%d is not responsive for %d', pid, now - workerStats.lastHeartbeatAt));
+                            var deathQueue = require('./misc').deathQueue;
+                            deathQueue(self.workers[pid], self.emitter, function () {
+                                // create a successor
+                                var successor = self.createWorker();
+                                self.workers[successor.pid] = successor;
+                                log(util.format('[Cluster2] Created a new worker with pid %d', successor.pid));
+                                return successor;
+                            });
+                        }
+                    });
 
                 }, self.options.heartbeatInterval || 60000);
             }
@@ -385,17 +407,19 @@ Process.prototype.listen = function() {
                     self.emitter.emit('died', worker.pid);
                     self.stats.workersKilled++;
                     self.stats.noWorkers--;
+                    delete self.workers[worker.pid + ''];
+                    delete self.stats.workers[worker.pid];
                     return;
                 }
 
                 self.emitter.emit('died', worker.pid);
                 self.stats.workersKilled++;
                 self.stats.noWorkers--;
+                delete self.workers[worker.pid + ''];
+                delete self.stats.workers[worker.pid];
                 //bugfix by huzhou@ebay.com, worker & replacement name collision
                 var replacement = self.createWorker();
                 self.workers[replacement.pid + ''] = replacement;
-                delete self.workers[worker.pid + ''];
-                delete self.stats.workers[worker.pid];
 
                 log('[cluster2] updated worker list:' + _.keys(self.workers));
             };
@@ -456,6 +480,8 @@ Process.prototype.listen = function() {
                 }
             });
 
+            // put the emitter in the process
+            process.emitter = self.emitter;
             self.emitter.emit(signal, {
                 pid: process.pid,
                 type: 'worker'
@@ -551,7 +577,7 @@ Process.prototype.listen = function() {
         // we'd like to have the threshold randomized in between [1, 1.5) of the given threshold
         // to avoid all workers die around the same time. This is in particular important for boxes of small number of cpu cores
         var connThreshold = self.options.connThreshold,
-            uptimeThreshod = self.options.uptimeThreshod;
+            uptimeThreshold = self.options.uptimeThreshold;
 
         var recycle = setInterval(function() {
 
@@ -637,6 +663,8 @@ Process.prototype.listen = function() {
             });
 
         }, self.options.heartbeatInterval || 60000);
+        // put the heartbeat interval id in the process context 
+        process.heartbeat = heartbeat;
 
         _.each(apps, function(app){
 

package.json

@@ -5,7 +5,7 @@
         "email": "subbu@ebaysf.com"
     }],
     "name": "cluster2",
-    "version": "0.4.15",
+    "version": "0.4.19",
     "repository": {
         "type": "git",
         "url": "https://github.com/ql-io/cluster2"

test/lib/server.js

@@ -26,6 +26,17 @@ server.get('/', function(req, res) {
     }
 });
 
+// test nanny feature
+server.get('/nanny-feature-test', function (req, res) {
+    // the first worker that handles the request will run away
+    res.send(process.pid + '');
+    console.log('worker ' + process.pid + ' will run away after 2s');
+    setTimeout(function () {
+        console.log('worker ' + process.pid + ' runs away');
+        clearInterval(process.heartbeat);
+    }, 2000);
+});
+
 server.on('close', function() {
     serving = false;
 })
@@ -40,12 +51,14 @@ var c = new Cluster({
     ecv: {
         control: true
     },
-    heartbeatInterval: process.env["heartbeatInterval"] || 1000
+    heartbeatInterval: process.env["heartbeatInterval"] || 1000,
+    maxHeartbeatDelay: process.env["maxHeartbeatDelay"] || 3000
 });
 
 c.on('died', function(pid) {
-    console.log('Worker ' + pid + ' died');
+    //console.log('Worker ' + pid + ' died');
     process.send({
+        pid: pid,
         dead: true
     })
 });

test/nanny-feature-test.js

@@ -0,0 +1,72 @@
+'use strict';
+
+var should = require('should'),
+    request = require('request'),
+    _ = require('underscore'),
+    spawn = require('child_process').spawn,
+    nodeunit = require('nodeunit'),
+    EventEmitter = require('events').EventEmitter,
+    emitter = new EventEmitter(),
+    childProc;
+
+exports['Nanny Feature Test'] = {
+    setUp: function (callback) {
+        var env = {};
+        _.extend(env, process.env);
+        _.extend(env, {
+            host: '127.0.0.1',
+            port: 3000,
+            monPort: 3001,
+            noWorkers: 2,
+            hearheatInterval: 500,
+            maxHeartbeatDelay: 1500
+        });
+
+        childProc = spawn('node', ['test/lib/server.js'], {
+            env: env,
+            stdio: ['pipe', 1, 2, 'ipc']
+        });
+
+        childProc.once('message', function (msg) {
+            if (msg.ready) {
+                //test.done();
+                callback();
+            }
+        });
+    },
+    
+    tearDown: function (callback) {
+        childProc.kill('SIGKILL');
+        callback();
+    },
+
+    'worker runs away and gets killed & replacement has been created': function (test) {
+        var pid;
+        request.get('http://127.0.0.1:3000/nanny-feature-test', function (err, res, body) {
+            test.equal(err, null);
+            test.ok(body);
+            pid = parseInt(body);
+            console.log(pid);
+        });
+        childProc.on('message', function (msg) {
+            if (msg.dead) {
+                console.log(msg.pid);
+                test.strictEqual(msg.pid, pid);
+                request.get('http://127.0.0.1:3001/ComponentStatus?component=worker', function (err, res, body) {
+                    console.log(body);
+                    var pids = body.substring(1, body.length - 1).split(',');
+                    test.strictEqual(pids.length, 3); // master & 2 workers
+                    test.done();
+                });
+            }
+        });
+    },
+
+    /*'New worker created to replace the killed worker': function (test) {
+        request.get('http://127.0.0.1:3001/ComponentStatus?component=worker', function (err, res, body) {
+            var pids = body.substring(1, body.length - 1).split(',');
+            test.strictEqual(pids.length, 3); // master & 2 workers
+            test.done();
+        });
+    }*/
+};