Production Deployment with node.js Clusters

api.socketServer.gracefulShutdown = function (api, next, alreadyShutdown) {
  if (alreadyShutdown == null) {
    alreadyShutdown = false;
  }
  if (alreadyShutdown == false) {
    api.socketServer.server.close();
    alreadyShutdown = true;
  }
  for (var i in api.socketServer.connections) {
    var connection = api.socketServer.connections[i];
    if (connection.responsesWaitingCount == 0) {
      api.socketServer.connections[i].end("Server going down NOW\r\nBye!\r\n");
    }
  }
  if (api.socketServer.connections.length != 0) {
    api.log(
      "[socket] waiting on shutdown, there are still " +
        api.socketServer.connections.length +
        " connected clients waiting on a response",
    );
    setTimeout(function () {
      api.socketServer.gracefulShutdown(api, next, alreadyShutdown);
    }, 3000);
  } else {
    next();
  }
};

var reloadAWorker = function (next) {
  var count = 0;
  for (var i in cluster.workers) {
    count++;
  }
  if (workersExpected > count) {
    startAWorker();
  }
  if (workerRestartArray.length > 0) {
    var worker = workerRestartArray.pop();
    worker.send("stop");
  }
};

cluster.on("exit", function (worker, code, signal) {
  log("worker " + worker.process.pid + " (#" + worker.id + ") has exited");
  setTimeout(reloadAWorker, 1000); // to prevent CPU-splsions if crashing too fast
});

#!/usr/bin/env node

// load in the actionHero class
var actionHero = require(__dirname + "/../api.js").actionHero; // normally if installed by npm: var actionHero = require("actionHero").actionHero;
var cluster = require("cluster");

// if there is no config.js file in the application's root, then actionHero will load in a collection of default params.  You can overwrite them with params.configChanges
var params = {};
params.configChanges = {};

// any additional functions you might wish to define to be globally accessable can be added as part of params.initFunction.  The api object will be availalbe.
params.initFunction = function (api, next) {
  next();
};

// start the server!
var startServer = function (next) {
  if (cluster.isWorker) {
    process.send("starting");
  }
  actionHero.start(params, function (api_from_callback) {
    api = api_from_callback;
    api.log("Boot Sucessful @ worker #" + process.pid, "green");
    if (typeof next == "function") {
      if (cluster.isWorker) {
        process.send("started");
      }
      next(api);
    }
  });
};

// handle signals from leader if running in cluster
if (cluster.isWorker) {
  process.on("message", function (msg) {
    if (msg == "start") {
      process.send("starting");
      startServer(function () {
        process.send("started");
      });
    }
    if (msg == "stop") {
      process.send("stopping");
      actionHero.stop(function () {
        api = null;
        process.send("stopped");
        process.exit();
      });
    }
    if (msg == "restart") {
      process.send("restarting");
      actionHero.restart(function (success, api_from_callback) {
        api = api_from_callback;
        process.send("restarted");
      });
    }
  });
}

// start the server!
startServer(function (api) {
  api.log("Successfully Booted!", ["green", "bold"]);
});

#!/usr/bin/env node

//////////////////////////////////////////////////////////////////////////////////////////////////////
//
// TO START IN CONSOLE: `./scripts/actionHeroCluster`
// TO DAMEONIZE: `forever start scripts/actionHeroCluster`
//
// ** Producton-ready actionHero cluster example **
// - workers which die will be restarted
// - maser/manager specific logging
// - pidfile for leader
// - USR2 restarts (graceful reload of workers while handling requets)
//   -- Note, socket/websocket clients will be disconnected, but there will always be a worker to handle them
//   -- HTTP, HTTPS, and TCP clients will be allowed to finish the action they are working on before the server goes down
// - TTOU and TTIN signals to subtract/add workers
// - WINCH to stop all workers
// - TCP, HTTP(s), and Web-socket clients will all be shared across the cluster
// - Can be run as a daemon or in-console
//   -- Lazy Dameon: `nohup ./scripts/actionHeroCluster &`
//   -- you may want to explore `forever` as a dameonizing option
//
// * Setting process titles does not work on windows or OSX
//
// This example was heavily inspired by Ruby Unicorns [[ http://unicorn.bogomips.org/ ]]
//
//////////////////////////////////////////////////////////////////////////////////////////////////////

//////////////
// Includes //
//////////////

var fs = require("fs");
var cluster = require("cluster");
var colors = require("colors");

var numCPUs = require("os").cpus().length;
var numWorkers = numCPUs - 2;
if (numWorkers < 2) {
  numWorkers = 2;
}

////////////
// config //
////////////

var config = {
  // script for workers to run (You probably will be changing this)
  exec: __dirname + "/actionHero",
  workers: numWorkers,
  pidfile: "./cluster_pidfile",
  log: process.cwd() + "/log/cluster.log",
  title: "actionHero-leader",
  workerTitlePrefix: " actionHero-worker",
  silent: true, // don't pass stdout/err to the leader
};

/////////
// Log //
/////////

var logHandle = fs.createWriteStream(config.log, { flags: "a" });
var log = function (msg, col) {
  var sqlDateTime = function (time) {
    if (time == null) {
      time = new Date();
    }
    var dateStr =
      padDateDoubleStr(time.getFullYear()) +
      "-" +
      padDateDoubleStr(1 + time.getMonth()) +
      "-" +
      padDateDoubleStr(time.getDate()) +
      " " +
      padDateDoubleStr(time.getHours()) +
      ":" +
      padDateDoubleStr(time.getMinutes()) +
      ":" +
      padDateDoubleStr(time.getSeconds());
    return dateStr;
  };

  var padDateDoubleStr = function (i) {
    return i < 10 ? "0" + i : "" + i;
  };
  msg = sqlDateTime() + " | " + msg;
  logHandle.write(msg + "\r\n");
  if (typeof col == "string") {
    col = [col];
  }
  for (var i in col) {
    msg = colors[col[i]](msg);
  }
  console.log(msg);
};

////////// // Main // //////////
log(" - STARTING CLUSTER -", ["bold", "green"]);
// set pidFile
if (config.pidfile != null) {
  fs.writeFileSync(config.pidfile, process.pid.toString());
}
process.stdin.resume();
process.title = config.title;
var workerRestartArray = [];

// used to trask rolling restarts of workers
var workersExpected = 0;

// signals
process.on("SIGINT", function () {
  log("Signal: SIGINT");
  workersExpected = 0;
  setupShutdown();
});

process.on("SIGTERM", function () {
  log("Signal: SIGTERM");
  workersExpected = 0;
  setupShutdown();
});

process.on("SIGKILL", function () {
  log("Signal: SIGKILL");
  workersExpected = 0;
  setupShutdown();
});

process.on("SIGUSR2", function () {
  log("Signal: SIGUSR2");
  log("swap out new workers one-by-one");
  workerRestartArray = [];
  for (var i in cluster.workers) {
    workerRestartArray.push(cluster.workers[i]);
  }
  reloadAWorker();
});

process.on("SIGHUP", function () {
  log("Signal: SIGHUP");
  log("reload all workers now");
  for (var i in cluster.workers) {
    var worker = cluster.workers[i];
    worker.send("restart");
  }
});

process.on("SIGWINCH", function () {
  log("Signal: SIGWINCH");
  log("stop all workers");
  workersExpected = 0;
  for (var i in cluster.workers) {
    var worker = cluster.workers[i];
    worker.send("stop");
  }
});

process.on("SIGTTIN", function () {
  log("Signal: SIGTTIN");
  log("add a worker");
  workersExpected++;
  startAWorker();
});

process.on("SIGTTOU", function () {
  log("Signal: SIGTTOU");
  log("remove a worker");
  workersExpected--;
  for (var i in cluster.workers) {
    var worker = cluster.workers[i];
    worker.send("stop");
    break;
  }
});

process.on("exit", function () {
  workersExpected = 0;
  log("Bye!");
});

// signal helpers
var startAWorker = function () {
  worker = cluster.fork();
  log("starting worker #" + worker.id);
  worker.on("message", function (message) {
    if (worker.state != "none") {
      log("Message [" + worker.process.pid + "]: " + message);
    }
  });
};

var setupShutdown = function () {
  log("Cluster manager quitting", "red");
  log("Stopping each worker...");
  for (var i in cluster.workers) {
    cluster.workers[i].send("stop");
  }
  setTimeout(loopUntilNoWorkers, 1000);
};

var loopUntilNoWorkers = function () {
  if (cluster.workers.length > 0) {
    log("there are still " + cluster.workers.length + " workers...");
    setTimeout(loopUntilNoWorkers, 1000);
  } else {
    log("all workers gone");
    if (config.pidfile != null) {
      fs.unlinkSync(config.pidfile);
    }
    process.exit();
  }
};

var reloadAWorker = function (next) {
  var count = 0;
  for (var i in cluster.workers) {
    count++;
  }
  if (workersExpected > count) {
    startAWorker();
  }
  if (workerRestartArray.length > 0) {
    var worker = workerRestartArray.pop();
    worker.send("stop");
  }
};

// Fork it.
cluster.setupleader({
  exec: config.exec,
  args: process.argv.slice(2),
  silent: config.silent,
});
for (var i = 0; i < config.workers; i++) {
  workersExpected++;
  startAWorker();
}
cluster.on("fork", function (worker) {
  log("worker " + worker.process.pid + " (#" + worker.id + ") has spawned");
});
cluster.on("listening", function (worker, address) {});
cluster.on("exit", function (worker, code, signal) {
  log("worker " + worker.process.pid + " (#" + worker.id + ") has exited");
  setTimeout(reloadAWorker, 1000); // to prevent CPU-splsions if crashing too fast
});