nodejs express4 cluster graceful restart / shutdown

app.js

...

app.use(function(req, res, next){
  if (app.get('graceful_shutdown') === true) {
    res.set('Connection', 'close');
  }
  next();
});

app.set('graceful_shutdown_start', function() {
  // Closing long-live http connections. WebSocket, EventStream, Comet..
});

app.set('graceful_shutdown_exit', function(callback) {
  // Clear timer, Closing database connections.
  return callback();
});

...

www

#!/usr/bin/env node
var cluster = require('cluster');
var fs = require('fs');
var util = require('util');
var app = require('../app');
var moment = require('moment');
var config = require(__dirname + '/../config/config');

var logger = function() {
  var timestamp = moment().format('YYYY-MM-DDTHH:mm:ss.SSSZZ');
  var type = (cluster.isMaster) ? 'master: ' : 'worker: ';
  process.stdout.write(timestamp + ' [' + process.pid + '] ' + type + util.format.apply(this, arguments) + '\n');
};

var pid_filepath = config.server.pid_filepath;

if (cluster.isMaster) {
  try {
    if (config.server.check_pid_exists) {
      fs.statSync(pid_filepath);
      logger('already running, pid file exists');
      process.exit(1);
    }
  }
  catch (err) {
    if (err.code != 'ENOENT') {
      logger(err.stack);
      process.exit(1);
    }
  }

  process.on('exit', function(code) {
    logger('process exited with code %s', code);
    fs.unlinkSync(pid_filepath);
  });

  process.on('SIGTERM', function() {
    process.exit(143);
  });

  process.on('SIGINT', function() {
    setTimeout(function() {
      process.exit(130);
    }, 0);
  });

  try {
    fs.writeFileSync(pid_filepath, process.pid + '\n');
  }
  catch (err) {
    logger(err.stack);
    process.exit(1);
  }

  var worker_processes = config.server.worker_processes || require('os').cpus().length;
  var shutdown_processing = false;

  cluster.on('exit', function(worker, code, signal) {
    if (worker.suicide !== true && (signal || code !== 0)) {
      logger('worker %d died (%s). restarting...', worker.process.pid, signal || code);
      var next_worker = cluster.fork().on('error', function(err) {
        logger(err.stack);
      });

      if (worker.listeners('listening').length == 1) {
        next_worker.once('listening', worker.listeners('listening')[0]);
      }
    }
  });

  for (var i = 0; i < worker_processes; i++) {
    cluster.fork().on('error', function(err) {
      logger(err.stack);
    });
  }

  process.on('SIGHUP', function() {
    if (shutdown_processing) {
      logger('could not interrupt, reload in progress');
      return;
    }

    shutdown_processing = true;
    logger('SIGHUP received, reloading workers...');

    var index = 0;
    var workers = Object.keys(cluster.workers);

    var killer = function() {
      if (index == workers.length) {
        logger('finished, SIGHUP processing');
        shutdown_processing = false;
        return;
      }

      if (cluster.workers[workers[index]] === undefined) {
        logger('already dead process (%s)', workers[index]);
        index++;
        killer();
      }

      logger('killing worker %d', cluster.workers[workers[index]].process.pid);
      var shutdown_timer;

      cluster.workers[workers[index]].once('disconnect', function() {
        logger('the worker %d has disconnected', this.process.pid);
        clearTimeout(shutdown_timer);
      });
      cluster.workers[workers[index]].send('graceful_shutdown');
      cluster.workers[workers[index]].disconnect();

      shutdown_timer = setTimeout(function(timeout_worker) {
        logger('could not close connections in time, forcefully shutting down (master)');
        timeout_worker.kill();
      }, config.server.shutdown_timeout, cluster.workers[workers[index]]);

      var worker = cluster.fork().on('error', function(err) {
        logger(err.stack);
      });

      worker.once('listening', function() {
        logger('finished, replacement worker online');
        index++;
        killer();
      });
    };

    killer();
  });

  logger('master starting');
} else if (cluster.isWorker) {
  app.set('port', config.server.port);

  var server = app.listen(app.get('port'), function() {
    logger('express server listening on port ' + server.address().port);
  });

  var gs_start_call = false;
  var gs_exit_call = false;

  process.on('message', function(msg) {
    if (msg === 'graceful_shutdown') {
      app.set('graceful_shutdown', true);
      if (gs_start_call === false && typeof app.get('graceful_shutdown_start') === 'function') {
        gs_start_call = true;
        logger('run graceful_shutdown_start');

        app.get('graceful_shutdown_start')();
      }

      server.close(function() {
        if (gs_exit_call === false && typeof app.get('graceful_shutdown_exit') === 'function') {
          gs_exit_call = true;
          logger('run graceful_shutdown_exit');

          app.get('graceful_shutdown_exit')(function() {
            logger('finished, graceful shutdown');
            process.exit(0);
          });
        }
        else {
          logger('finished, graceful shutdown');
          process.exit(0);
        }
      });

      setTimeout(function() {
        logger('could not close connections in time, forcefully shutting down (worker)');
        process.exit(0);
      }, config.server.shutdown_timeout);
    }
  });
}