Implement app clustering (#17752)

* Install throng for easy cluster management
* Extract the Express app construction into its own file
* Switch server.js to use app clustering for deployed environments
* Worker count is based on the lesser of process.env.WEB_CONCURRENCY and the count of CPUs
* Reading clustered output is difficult, let's prefix the std{out,err} streams

Co-authored-by: Jason Etcovitch <jasonetco@github.com>

app.json

@@ -3,7 +3,8 @@
   "env": {
     "NODE_ENV": "production",
     "NPM_CONFIG_PRODUCTION": "true",
-    "ENABLED_LANGUAGES": "en"
+    "ENABLED_LANGUAGES": "en",
+    "WEB_CONCURRENCY": "1"
   },
   "buildpacks": [
     { "url": "heroku/nodejs" }

lib/app.js

@@ -0,0 +1,5 @@
+const express = require('express')
+
+const app = express()
+require('../middleware')(app)
+module.exports = app

lib/check-node-version.js

@@ -5,5 +5,5 @@ const { engines } = require('../package.json')
 if (!semver.satisfies(process.version, engines.node)) {
   console.error(`\n\nYou're using Node.js ${process.version}, but ${engines.node} is required`)
   console.error('Visit nodejs.org to download an installer for the latest LTS version.\n\n')
-  process.exit()
+  process.exit(1)
 }

lib/prefix-stream-write.js

@@ -0,0 +1,25 @@
+module.exports = function prefixStreamWrite (writableStream, prefix) {
+  const oldWrite = writableStream.write
+
+  function newWrite (...args) {
+    const [chunk, encoding] = args
+
+    // Prepend the prefix if the chunk is either a string or a Buffer.
+    // Otherwise, just leave it alone to be safe.
+    if (typeof chunk === 'string') {
+      // Only prepend the prefix is the `encoding` is safe or not provided.
+      // If it's a function, it is third arg `callback` provided as optional second
+      if (!encoding || encoding === 'utf8' || typeof encoding === 'function') {
+        args[0] = prefix + chunk
+      }
+    } else if (Buffer.isBuffer(chunk)) {
+      args[0] = Buffer.concat([Buffer.from(prefix), chunk])
+    }
+
+    return oldWrite.apply(this, args)
+  }
+
+  writableStream.write = newWrite
+
+  return writableStream
+}

package-lock.json

@@ -23396,6 +23396,21 @@
       "integrity": "sha512-fcwX4mndzpLQKBS1DVYhGAcYaYt7vsHNIvQV+WXMvnow5cgjPphq5CaayLaGsjRdSCKZFNGt7/GYAuXaNOiYCA==",
       "dev": true
     },
+    "throng": {
+      "version": "5.0.0",
+      "resolved": "https://registry.npmjs.org/throng/-/throng-5.0.0.tgz",
+      "integrity": "sha512-nrq7+qQhn/DL8yW/wiwImTepfi6ynOCAe7moSwgoYN1F32yQMdBkuFII40oAkb3cDfaL6q5BIoFTDCHdMWQ8Pw==",
+      "requires": {
+        "lodash": "^4.17.20"
+      },
+      "dependencies": {
+        "lodash": {
+          "version": "4.17.20",
+          "resolved": "https://registry.npmjs.org/lodash/-/lodash-4.17.20.tgz",
+          "integrity": "sha512-PlhdFcillOINfeV7Ni6oF1TAEayyZBoZ8bcshTHqOYJYlrqzRK5hagpagky5o4HfCzzd1TRkXPMFq6cKk9rGmA=="
+        }
+      }
+    },
     "through": {
       "version": "2.3.8",
       "resolved": "http://registry.npmjs.org/through/-/through-2.3.8.tgz",

package.json

@@ -97,6 +97,7 @@
     "slash": "^3.0.0",
     "strip-html-comments": "^1.0.0",
     "style-loader": "^1.2.1",
+    "throng": "^5.0.0",
     "unified": "^8.4.2",
     "unist-util-visit": "^2.0.3",
     "uuid": "^8.3.0",

script/enterprise-server-deprecations/archive-version.js

@@ -3,7 +3,7 @@
 const fs = require('fs')
 const path = require('path')
 const { execSync } = require('child_process')
-const server = require('../../server')
+const app = require('../lib/app')
 const port = '4001'
 const host = `http://localhost:${port}`
 const scrape = require('website-scraper')
@@ -155,7 +155,7 @@ async function main () {
     plugins: [new RewriteAssetPathsPlugin(version, tempDirectory)]
   }
 
-  server.listen(port, async () => {
+  app.listen(port, async () => {
     console.log(`started server on ${host}`)
 
     await scrape(scraperOptions).catch(err => {

server.js

@@ -1,42 +1,116 @@
 require('dotenv').config()
 
+const throng = require('throng')
+const os = require('os')
+const portUsed = require('port-used')
+const prefixStreamWrite = require('./lib/prefix-stream-write')
+
+// Intentionally require these for both cluster primary and workers
 require('./lib/check-node-version')
 require('./lib/handle-exceptions')
 require('./lib/feature-flags')
 
-const express = require('express')
-const portUsed = require('port-used')
-const warmServer = require('./lib/warm-server')
-const port = Number(process.env.PORT) || 4000
-const app = express()
+const { PORT, NODE_ENV } = process.env
+const port = Number(PORT) || 4000
 
-require('./middleware')(app)
-
-// prevent the app from starting up during tests
-/* istanbul ignore next */
-if (!module.parent) {
-  // check that the development server is not already running
-  portUsed.check(port).then(async status => {
-    if (status === false) {
-      // If in a deployed environment, warm the server at the start
-      if (process.env.NODE_ENV === 'production') {
-        // If in a true production environment, wait for the cache to be fully warmed.
-        if (process.env.HEROKU_PRODUCTION_APP || process.env.GITHUB_ACTIONS) {
-          await warmServer()
-        }
-      }
-
-      // workaround for https://github.com/expressjs/express/issues/1101
-      const server = require('http').createServer(app)
-      server.listen(port, () => console.log(`app running on http://localhost:${port}`))
-        .on('error', () => server.close())
-    } else {
-      console.log(`\n\n\nPort ${port} is not available. You may already have a server running.`)
-      console.log('Try running `killall node` to shut down all your running node processes.\n\n\n')
-      console.log('\x07') // system 'beep' sound
-      process.exit(1)
-    }
+function main () {
+  // Spin up a cluster!
+  throng({
+    master: setupPrimary,
+    worker: setupWorker,
+    count: calculateWorkerCount()
   })
 }
 
-module.exports = app
+// Start the server!
+main()
+
+// This function will only be run in the primary process
+async function setupPrimary () {
+  process.on('beforeExit', () => {
+    console.log('Shutting down primary...')
+    console.log('Exiting!')
+  })
+
+  console.log('Starting up primary...')
+
+  // Check that the development server is not already running
+  const portInUse = await portUsed.check(port)
+  if (portInUse) {
+    console.log(`\n\n\nPort ${port} is not available. You may already have a server running.`)
+    console.log('Try running `killall node` to shut down all your running node processes.\n\n\n')
+    console.log('\x07') // system 'beep' sound
+    process.exit(1)
+  }
+}
+
+// IMPORTANT: This function will be run in a separate worker process!
+async function setupWorker (id, disconnect) {
+  let exited = false
+
+  // Wrap stdout and stderr to include the worker ID as a static prefix
+  // console.log('hi') => '[worker.1]: hi'
+  const prefix = `[worker.${id}]: `
+  prefixStreamWrite(process.stdout, prefix)
+  prefixStreamWrite(process.stderr, prefix)
+
+  process.on('beforeExit', () => {
+    console.log('Exiting!')
+  })
+
+  process.on('SIGTERM', shutdown)
+  process.on('SIGINT', shutdown)
+
+  console.log('Starting up worker...')
+
+  // Load the server in each worker process and share the port via sharding
+  const app = require('./lib/app')
+  const warmServer = require('./lib/warm-server')
+
+  // If in a deployed environment...
+  if (NODE_ENV === 'production') {
+    // If in a true production environment, wait for the cache to be fully warmed.
+    if (process.env.HEROKU_PRODUCTION_APP || process.env.GITHUB_ACTIONS) {
+      await warmServer()
+    }
+  }
+
+  // Workaround for https://github.com/expressjs/express/issues/1101
+  const server = require('http').createServer(app)
+  server
+    .listen(port, () => console.log(`app running on http://localhost:${port}`))
+    .on('error', () => server.close())
+
+  function shutdown () {
+    if (exited) return
+    exited = true
+
+    console.log('Shutting down worker...')
+    disconnect()
+  }
+}
+
+function calculateWorkerCount () {
+  // Heroku's recommended WEB_CONCURRENCY count based on the WEB_MEMORY config
+  const { WEB_CONCURRENCY } = process.env
+
+  const recommendedCount = parseInt(WEB_CONCURRENCY, 10) || 1
+  const cpuCount = os.cpus().length
+
+  // Ensure the recommended count is AT LEAST 1 for safety
+  let workerCount = Math.max(recommendedCount, 1)
+
+  // Let's do some math...
+  // If in a deployed environment...
+  if (NODE_ENV === 'production') {
+    // If WEB_MEMORY or WEB_CONCURRENCY values were configured in Heroku, use
+    // the smaller value between their recommendation vs. the CPU count
+    if (WEB_CONCURRENCY) {
+      workerCount = Math.min(recommendedCount, cpuCount)
+    } else {
+      workerCount = cpuCount
+    }
+  }
+
+  return workerCount
+}

tests/helpers/supertest.js

@@ -3,7 +3,7 @@
 
 const cheerio = require('cheerio')
 const supertest = require('supertest')
-const app = require('../../server')
+const app = require('../../lib/app')
 
 const helpers = {}
 

tests/rendering/events.js

@@ -1,7 +1,7 @@
 const request = require('supertest')
 const nock = require('nock')
 const cheerio = require('cheerio')
-const app = require('../../server')
+const app = require('../../lib/app')
 
 describe('POST /events', () => {
   jest.setTimeout(60 * 1000)

tests/routing/deprecated-enterprise-versions.js

@@ -1,4 +1,4 @@
-const app = require('../../server')
+const app = require('../../lib/app')
 const enterpriseServerReleases = require('../../lib/enterprise-server-releases')
 const { get, getDOM } = require('../helpers/supertest')
 const supertest = require('supertest')

tests/routing/redirects.js

@@ -1,7 +1,7 @@
 const path = require('path')
 const { isPlainObject } = require('lodash')
 const supertest = require('supertest')
-const app = require('../../server')
+const app = require('../../lib/app')
 const enterpriseServerReleases = require('../../lib/enterprise-server-releases')
 const nonEnterpriseDefaultVersion = require('../../lib/non-enterprise-default-version')
 const Page = require('../../lib/page')