Run Content Node on multiple cluster processes and migrate Bull->BullMQ

creator-node/compose/env/base.env

@@ -13,6 +13,9 @@ redisPort=6379
 # Can be overriden.
 creatorNodeIsDebug=true
 
+# Locally we run 4 CNs so we don't want to use multiple processes for each
+expressAppConcurrency=1
+
 WAIT_HOSTS=
 
 # Rate limiting

creator-node/src/blacklistManager.js

@@ -1,3 +1,5 @@
+const cluster = require('cluster')
+
 const { logger } = require('./logging')
 const models = require('./models')
 const redis = require('./redis')
@@ -29,13 +31,16 @@ class BlacklistManager {
     try {
       this.log('Initializing BlacklistManager...')
 
-      const { trackIdsToBlacklist, userIdsToBlacklist, segmentsToBlacklist } =
-        await this.getDataToBlacklist()
-      await this.fetchCIDsAndAddToRedis({
-        trackIdsToBlacklist,
-        userIdsToBlacklist,
-        segmentsToBlacklist
-      })
+      // Adding to redis only needs to be done once, but multiple workers might all run the app with their own BlacklistManager
+      if (cluster.worker?.id === 1) {
+        const { trackIdsToBlacklist, userIdsToBlacklist, segmentsToBlacklist } =
+          await this.getDataToBlacklist()
+        await this.fetchCIDsAndAddToRedis({
+          trackIdsToBlacklist,
+          userIdsToBlacklist,
+          segmentsToBlacklist
+        })
+      }
 
       this.initialized = true
 

creator-node/src/config.js

@@ -264,6 +264,12 @@ const config = convict({
     env: 'printSequelizeLogs',
     default: false
   },
+  expressAppConcurrency: {
+    doc: 'Number of processes to spawn, where each process runs its own Content Node. Default 0 to run one process per core (auto-detected)',
+    format: 'nat',
+    env: 'expressAppConcurrency',
+    default: 0
+  },
 
   // Transcoding settings
   transcodingMaxConcurrency: {

creator-node/src/index.ts

@@ -1,5 +1,13 @@
 'use strict'
 
+import type { Cluster } from 'cluster'
+import type { CpuInfo } from 'os'
+import type { LoDashStatic } from 'lodash'
+const cluster: Cluster = require('cluster')
+const { cpus }: { cpus: () => CpuInfo[] } = require('os')
+const _: LoDashStatic = require('lodash')
+const Bull = require('bull')
+
 const { setupTracing } = require('./tracer')
 setupTracing('content-node')
 
@@ -15,58 +23,64 @@ const { logger } = require('./logging')
 const { serviceRegistry } = require('./serviceRegistry')
 const redisClient = require('./redis')
 
-const exitWithError = (...msg: any[]) => {
-  logger.error('ERROR: ', ...msg)
-  process.exit(1)
-}
+// Make sure some logic is only performed by 1 worker not all
+if (cluster.worker?.id !== 1) {
+  logger.info(
+    `Disabling queues from running on worker ID=${cluster.worker?.id}`
+  )
 
-const verifyDBConnection = async () => {
-  try {
-    logger.info('Verifying DB connection...')
-    await sequelize.authenticate() // runs SELECT 1+1 AS result to check db connection
-    logger.info('DB connected successfully!')
-  } catch (connectionError) {
-    exitWithError('Error connecting to DB:', connectionError)
-  }
+  // See https://github.com/OptimalBits/bull/blob/develop/lib/queue.js for queue functions to override
+
+  // Obliterating queues is one-time logic on startup
+  Bull.prototype.obliterate = async function (opts: any) {}
+  Bull.prototype.empty = function () {}
+  // Bull isn't quality enough to make workers not take the some jobs
+  Bull.prototype.process = function (
+    name: any,
+    concurrency: any,
+    handler: any
+  ) {}
+  Bull.prototype.start = function (concurrency: any, name: any) {}
+  Bull.prototype.run = function (concurrency: any, handlerName: any) {}
+  // We also don't want to see log spam from events or run onComplete callbacks
+  Bull.prototype.on = function (eventName: any) {}
+  Bull.prototype.constructor = Bull
 }
 
-const runDBMigrations = async () => {
-  try {
-    logger.info('Executing database migrations...')
-    await runMigrations()
-    logger.info('Migrations completed successfully')
-  } catch (migrationError) {
-    exitWithError('Error in migrations:', migrationError)
+// The primary process performs one-time validation and spawns worker processes that each run the Express app
+const startAppForPrimary = async () => {
+  const exitWithError = (...msg: any[]) => {
+    logger.error('ERROR: ', ...msg)
+    process.exit(1)
   }
-}
 
-const connectToDBAndRunMigrations = async () => {
-  await verifyDBConnection()
-  await clearRunningQueries()
-  await runDBMigrations()
-}
-
-/**
- * Setting a different port is necessary for OpenResty to work. If OpenResty
- * is enabled, have the app run on port 3000. Else, run on its configured port.
- * @returns the port number to configure the Content Node app
- */
-const getPort = () => {
-  const contentCacheLayerEnabled = config.get('contentCacheLayerEnabled')
-
-  if (contentCacheLayerEnabled) {
-    return 3000
+  const verifyDBConnection = async () => {
+    try {
+      logger.info('Verifying DB connection...')
+      await sequelize.authenticate() // runs SELECT 1+1 AS result to check db connection
+      logger.info('DB connected successfully!')
+    } catch (connectionError) {
+      exitWithError('Error connecting to DB:', connectionError)
+    }
   }
 
-  return config.get('port')
-}
+  const runDBMigrations = async () => {
+    try {
+      logger.info('Executing database migrations...')
+      await runMigrations()
+      logger.info('Migrations completed successfully')
+    } catch (migrationError) {
+      exitWithError('Error in migrations:', migrationError)
+    }
+  }
 
-const startApp = async () => {
-  logger.info('Configuring service...')
+  const connectToDBAndRunMigrations = async () => {
+    await verifyDBConnection()
+    await clearRunningQueries()
+    await runDBMigrations()
+  }
 
   await config.asyncConfig()
-
-  // fail if delegateOwnerWallet & delegatePrivateKey not present
   const delegateOwnerWallet = config.get('delegateOwnerWallet')
   const delegatePrivateKey = config.get('delegatePrivateKey')
   const creatorNodeEndpoint = config.get('creatorNodeEndpoint')
@@ -77,7 +91,9 @@ const startApp = async () => {
     )
   }
 
-  // fail if delegateOwnerWallet doesn't derive from delegatePrivateKey
+  logger.info(`Primary process with pid=${process.pid} is running`)
+
+  // Fail if delegateOwnerWallet doesn't derive from delegatePrivateKey
   const privateKeyBuffer = Buffer.from(
     config.get('delegatePrivateKey').replace('0x', ''),
     'hex'
@@ -88,15 +104,61 @@ const startApp = async () => {
     throw new Error('Invalid delegatePrivateKey/delegateOwnerWallet pair')
   }
 
+  // Fail if Trusted Notifier isn't configured properly
   const trustedNotifierEnabled = !!config.get('trustedNotifierID')
   const nodeOperatorEmailAddress = config.get('nodeOperatorEmailAddress')
-
   if (!trustedNotifierEnabled && !nodeOperatorEmailAddress) {
     exitWithError(
       'Cannot startup without a trustedNotifierID or nodeOperatorEmailAddress'
     )
   }
 
+  await connectToDBAndRunMigrations()
+
+  // Clear all redis locks
+  try {
+    await redisClient.WalletWriteLock.clearWriteLocks()
+  } catch (e: any) {
+    logger.warn(`Could not clear write locks. Skipping..: ${e.message}`)
+  }
+
+  // This is called `cpus()` but it actually returns the # of logical cores, which is possibly higher than # of physical cores if there's hyperthreading
+  const logicalCores = cpus().length
+  const numWorkers = config.get('expressAppConcurrency') || logicalCores
+  logger.info(`Spawning ${numWorkers} processes to run the Express app...`)
+  _.times(numWorkers, () => cluster.fork())
+
+  cluster.on('exit', (worker, code, signal) => {
+    logger.info(`Worker process with pid=${worker.process.pid} died`)
+  })
+}
+
+// Workers don't share memory, so each one is its own Express instance with its own version of objects like serviceRegistry
+const startAppForWorker = async () => {
+  /**
+   * Setting a different port is necessary for OpenResty to work. If OpenResty
+   * is enabled, have the app run on port 3000. Else, run on its configured port.
+   * @returns the port number to configure the Content Node app
+   */
+  const getPort = () => {
+    const contentCacheLayerEnabled = config.get('contentCacheLayerEnabled')
+
+    if (contentCacheLayerEnabled) {
+      return 3000
+    }
+
+    return config.get('port')
+  }
+
+  logger.info(
+    `Worker process with pid=${process.pid} and worker ID=${cluster.worker?.id} is running`
+  )
+
+  await config.asyncConfig()
+  const privateKeyBuffer = Buffer.from(
+    config.get('delegatePrivateKey').replace('0x', ''),
+    'hex'
+  )
   try {
     const solDelegateKeypair = Keypair.fromSeed(privateKeyBuffer)
     const solDelegatePrivateKey = solDelegateKeypair.secretKey
@@ -110,30 +172,21 @@ const startApp = async () => {
     )
   }
 
-  await connectToDBAndRunMigrations()
-
   const nodeMode = config.get('devMode') ? 'Dev Mode' : 'Production Mode'
+
   await serviceRegistry.initServices()
   logger.info(`Initialized services (Node running in ${nodeMode})`)
-
   const appInfo = initializeApp(getPort(), serviceRegistry)
   logger.info('Initialized app and server')
 
-  // Clear all redis locks
-  try {
-    await redisClient.WalletWriteLock.clearWriteLocks()
-  } catch (e: any) {
-    logger.warn(`Could not clear write locks. Skipping..: ${e.message}`)
-  }
-
   // Initialize services that do not require the server, but do not need to be awaited.
   serviceRegistry.initServicesAsynchronously()
 
   // Some Services cannot start until server is up. Start them now
   // No need to await on this as this process can take a while and can run in the background
   serviceRegistry.initServicesThatRequireServer(appInfo.app)
 
-  // when app terminates, close down any open DB connections gracefully
+  // When app terminates, close down any open DB connections gracefully
   ON_DEATH((signal: any, error: any) => {
     // NOTE: log messages emitted here may be swallowed up if using the bunyan CLI (used by
     // default in `npm start` command). To see messages emitted after a kill signal, do not
@@ -145,4 +198,11 @@ const startApp = async () => {
     }
   })
 }
-startApp()
+
+if (cluster.isMaster) {
+  startAppForPrimary()
+} else if (cluster.isWorker) {
+  startAppForWorker()
+} else {
+  throw new Error("Can't determine if process is primary or worker in cluster")
+}

creator-node/src/monitors/MonitoringQueue.js

@@ -42,7 +42,7 @@ class MonitoringQueue {
     this.queue.process(
       PROCESS_NAMES.monitor,
       /* concurrency */ 1,
-      async (job, done) => {
+      async (_) => {
         try {
           this.logStatus('Starting')
 
@@ -55,11 +55,8 @@ class MonitoringQueue {
               this.logStatus(`Error on ${monitor.name} ${e}`)
             }
           })
-
-          done(null, {})
         } catch (e) {
           this.logStatus(`Error ${e}`)
-          done(e)
         }
       }
     )

creator-node/src/services/stateMachineManager/stateMonitoring/index.js

@@ -1,5 +1,6 @@
 const BullQueue = require('bull')
 const _ = require('lodash')
+const cluster = require('cluster')
 
 const config = require('../../../config')
 const {
@@ -50,10 +51,12 @@ class StateMonitoringManager {
         duration: config.get('fetchCNodeEndpointToSpIdMapIntervalMs')
       }
     })
-    await this.startEndpointToSpIdMapQueue(
-      cNodeEndpointToSpIdMapQueue,
-      prometheusRegistry
-    )
+    if (cluster.worker?.id === 1) {
+      await this.startEndpointToSpIdMapQueue(
+        cNodeEndpointToSpIdMapQueue,
+        prometheusRegistry
+      )
+    }
 
     // Create queue to slice through batches of users and gather data to be passed to find-sync and find-replica-set-update jobs
     const monitorStateQueue = makeQueue({
@@ -107,7 +110,12 @@ class StateMonitoringManager {
     await findReplicaSetUpdatesQueue.obliterate({ force: true })
 
     // Enqueue first monitor-state job
-    await this.startMonitorStateQueue(monitorStateQueue, discoveryNodeEndpoint)
+    if (cluster.worker?.id === 1) {
+      await this.startMonitorStateQueue(
+        monitorStateQueue,
+        discoveryNodeEndpoint
+      )
+    }
 
     return {
       monitorStateQueue,

creator-node/src/services/stateMachineManager/stateReconciliation/index.js

@@ -1,3 +1,5 @@
+const cluster = require('cluster')
+
 const config = require('../../../config')
 const {
   QUEUE_HISTORY,
@@ -79,10 +81,12 @@ class StateReconciliationManager {
     await recoverOrphanedDataQueue.obliterate({ force: true })
 
     // Queue the first recoverOrphanedData job, which will re-enqueue itself
-    await this.startRecoverOrphanedDataQueue(
-      recoverOrphanedDataQueue,
-      discoveryNodeEndpoint
-    )
+    if (cluster.worker?.id === 1) {
+      await this.startRecoverOrphanedDataQueue(
+        recoverOrphanedDataQueue,
+        discoveryNodeEndpoint
+      )
+    }
 
     this.registerQueueEventHandlersAndJobProcessors({
       manualSyncQueue,

creator-node/test/StateMonitoringManager.test.js

@@ -144,6 +144,11 @@ describe('test StateMonitoringManager initialization, events, and re-enqueuing',
     const MockStateMonitoringManager = proxyquire(
       '../src/services/stateMachineManager/stateMonitoring/index.js',
       {
+        cluster: {
+          worker: {
+            id: 1
+          }
+        },
         '../../../config': config
       }
     )
@@ -174,6 +179,11 @@ describe('test StateMonitoringManager initialization, events, and re-enqueuing',
     const MockStateMonitoringManager = proxyquire(
       '../src/services/stateMachineManager/stateMonitoring/index.js',
       {
+        cluster: {
+          worker: {
+            id: 1
+          }
+        },
         '../../../config': config
       }
     )