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

creator-node/.eslintrc.js

@@ -3,7 +3,8 @@ module.exports = {
   parserOptions: {
     tsconfigRootDir: __dirname,
     ecmaVersion: 2020,
-    project: ['./tsconfig.json']
+    project: ['./tsconfig.json'],
+    tsconfigRootDir: __dirname
   },
   extends: [
     'standard',

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 too many processes for each
+expressAppConcurrency=2
+
 WAIT_HOSTS=
 
 # Rate limiting

creator-node/package.json

@@ -48,7 +48,7 @@
     "bl": "^4.1.0",
     "body-parser": "^1.18.3",
     "buffer": "5.4.2",
-    "bull": "4.8.2",
+    "bullmq": "^1.91.1",
     "bunyan": "^1.8.15",
     "cids": "0.8.0",
     "commander": "^6.2.1",
@@ -92,12 +92,12 @@
     "web3": "1.2.8"
   },
   "devDependencies": {
-    "@types/bull": "^3.15.8",
     "@types/bunyan": "^1.8.8",
     "@types/chai": "^4.3.3",
     "@types/eth-sig-util": "^2.1.1",
     "@types/express": "4.17.12",
     "@types/fs-extra": "^9.0.13",
+    "@types/ioredis": "^4.28.10",
     "@types/lodash": "^4.14.182",
     "@types/mocha": "^9.1.1",
     "@types/node": "^18.7.9",

creator-node/src/AsyncProcessingQueue.js

@@ -1,4 +1,4 @@
-const Bull = require('bull')
+const { Queue, Worker } = require('bullmq')
 const { logger: genericLogger } = require('./logging')
 const config = require('./config')
 const redisClient = require('./redis')
@@ -39,48 +39,52 @@ const ASYNC_PROCESSING_QUEUE_HISTORY = 500
 
 class AsyncProcessingQueue {
   constructor(libs, prometheusRegistry) {
-    this.queue = new Bull('asyncProcessing', {
-      redis: {
-        host: config.get('redisHost'),
-        port: config.get('redisPort')
-      },
+    const connection = {
+      host: config.get('redisHost'),
+      port: config.get('redisPort')
+    }
+    this.queue = new Queue('async-processing', {
+      connection,
       defaultJobOptions: {
         removeOnComplete: ASYNC_PROCESSING_QUEUE_HISTORY,
         removeOnFail: ASYNC_PROCESSING_QUEUE_HISTORY
       }
     })
 
-    prometheusRegistry.startQueueMetrics(this.queue)
-
     this.libs = libs
 
     const untracedProcessTask = this.processTask
-    this.queue.process(MAX_CONCURRENCY, async (job, done) => {
-      const { logContext, parentSpanContext, task } = job.data
-      const processTask = instrumentTracing({
-        name: `AsyncProcessingQueue.process ${task}`,
-        fn: untracedProcessTask,
-        context: this,
-        options: {
-          // if a parentSpanContext is provided
-          // reference it so the async queue job can remember
-          // who enqueued it
-          links: parentSpanContext
-            ? [
-                {
-                  context: parentSpanContext
-                }
-              ]
-            : [],
-          attributes: {
-            requestID: logContext.requestID,
-            [tracing.CODE_FILEPATH]: __filename
+    const worker = new Worker(
+      'async-processing',
+      async (job) => {
+        const { logContext, parentSpanContext, task } = job.data
+        const processTask = instrumentTracing({
+          name: `AsyncProcessingQueue.process ${task}`,
+          fn: untracedProcessTask,
+          context: this,
+          options: {
+            // if a parentSpanContext is provided
+            // reference it so the async queue job can remember
+            // who enqueued it
+            links: parentSpanContext
+              ? [
+                  {
+                    context: parentSpanContext
+                  }
+                ]
+              : [],
+            attributes: {
+              requestID: logContext.requestID,
+              [tracing.CODE_FILEPATH]: __filename
+            }
           }
-        }
-      })
+        })
 
-      await processTask(job, done)
-    })
+        await processTask(job)
+      },
+      { connection, concurrency: MAX_CONCURRENCY }
+    )
+    prometheusRegistry.startQueueMetrics(this.queue, worker)
 
     this.PROCESS_NAMES = PROCESS_NAMES
     this.PROCESS_STATES = PROCESS_STATES
@@ -90,7 +94,7 @@ class AsyncProcessingQueue {
     this.constructProcessKey = this.constructAsyncProcessingKey.bind(this)
   }
 
-  async processTask(job, done) {
+  async processTask(job) {
     const { logContext, task } = job.data
 
     const func = this.getFn(task)
@@ -107,7 +111,6 @@ class AsyncProcessingQueue {
           logContext,
           req: job.data.req
         })
-        done(null, {})
       } else {
         this.logStatus(
           `Succesfully handed off transcoding and segmenting to sp=${sp}. Wrapping up remainder of track association..`
@@ -117,12 +120,12 @@ class AsyncProcessingQueue {
           logContext,
           req: { ...job.data.req, transcodeFilePath, segmentFileNames }
         })
-        done(null, { response: { transcodeFilePath, segmentFileNames } })
+        return { response: { transcodeFilePath, segmentFileNames } }
       }
     } else {
       try {
         const response = await this.monitorProgress(task, func, job.data)
-        done(null, { response })
+        return { response }
       } catch (e) {
         tracing.recordException(e)
         this.logError(
@@ -131,7 +134,7 @@ class AsyncProcessingQueue {
           }: ${e.toString()}`,
           logContext
         )
-        done(e.toString())
+        return e.toString()
       }
     }
   }
@@ -190,7 +193,7 @@ class AsyncProcessingQueue {
       logContext
     )
 
-    const job = await this.queue.add(params)
+    const job = await this.queue.add('async-processing', params)
 
     return job
   }

creator-node/src/ImageProcessingQueue.js

@@ -1,8 +1,11 @@
-const Bull = require('bull')
+const { Queue, QueueEvents, Worker } = require('bullmq')
+const path = require('path')
 const os = require('os')
 const config = require('./config')
 const { logger: genericLogger } = require('./logging')
 
+const resizeImage = require('./resizeImage')
+
 const imageProcessingMaxConcurrency = config.get(
   'imageProcessingMaxConcurrency'
 )
@@ -22,34 +25,39 @@ const IMAGE_PROCESSING_QUEUE_HISTORY = 500
 
 class ImageProcessingQueue {
   constructor(prometheusRegistry = null) {
-    this.queue = new Bull('image-processing-queue', {
-      redis: {
-        port: config.get('redisPort'),
-        host: config.get('redisHost')
-      },
+    const connection = {
+      host: config.get('redisHost'),
+      port: config.get('redisPort')
+    }
+    this.queue = new Queue('image-processing-queue', {
+      connection,
       defaultJobOptions: {
         removeOnComplete: IMAGE_PROCESSING_QUEUE_HISTORY,
         removeOnFail: IMAGE_PROCESSING_QUEUE_HISTORY
       }
     })
 
+    // Process jobs sandboxed - https://docs.bullmq.io/guide/workers/sandboxed-processors
+    // TODO: Make sandboxed again once this is fixed: https://github.com/taskforcesh/bullmq/issues/1424
+    // const processorFile = path.join(__dirname, 'resizeImage.js')
+    // const worker = new Worker('image-processing-queue', processorFile, {
+    //   connection,
+    //   concurrency: MAX_CONCURRENCY
+    // })
+    const worker = new Worker('image-processing-queue', resizeImage, {
+      connection,
+      concurrency: MAX_CONCURRENCY
+    })
     if (prometheusRegistry !== null && prometheusRegistry !== undefined) {
-      prometheusRegistry.startQueueMetrics(this.queue)
+      prometheusRegistry.startQueueMetrics(this.queue, worker)
     }
 
-    /**
-     * Queue will process tasks concurrently if provided a concurrency number and a
-     *    path to file containing job processor function
-     * https://github.com/OptimalBits/bull/tree/013c51942e559517c57a117c27a550a0fb583aa8#separate-processes
-     */
-    this.queue.process(
-      PROCESS_NAMES.resizeImage /** job processor name */,
-      MAX_CONCURRENCY /** job processor concurrency */,
-      `${__dirname}/resizeImage.js` /** path to job processor function */
-    )
-
     this.logStatus = this.logStatus.bind(this)
     this.resizeImage = this.resizeImage.bind(this)
+
+    this.queueEvents = new QueueEvents('image-processing-queue', {
+      connection
+    })
   }
 
   /**
@@ -96,7 +104,8 @@ class ImageProcessingQueue {
       square,
       logContext
     })
-    const result = await job.finished()
+
+    const result = await job.waitUntilFinished(this.queueEvents)
     return result
   }
 }