fix: Add redis config option for Bull & BullMQ (#43)

packages/bull/README.md

@@ -23,6 +23,11 @@ const { Judoscale } = require('judoscale-bull')
 const judoscale = new Judoscale({
   redis_url: process.env.REDISCLOUD_URL, // defaults to process.env.REDIS_URL
 })
+
+// You can optionally pass a Redis config object or an ioredis instance
+const judoscale = new Judoscale({
+  redis: redisConfig,
+})
 ```
 
 3. If you want to scale your workers down to zero instances, you also need to install judoscale-bull in your web process so Judoscale knows when to scale them back up:

packages/bull/src/bull-metrics-collector.js

@@ -11,8 +11,8 @@ process.on('exit', async () => {
 })
 
 class BullMetricsCollector extends WorkerMetricsCollector {
-  constructor() {
-    super('Bull')
+  constructor(config = {}) {
+    super('Bull', config)
     collectors.push(this)
     this.queues = new Map()
   }
@@ -61,8 +61,14 @@ class BullMetricsCollector extends WorkerMetricsCollector {
 
   get redis() {
     if (!this._redis) {
-      const redisUrl = this.config.redis_url || process.env.REDIS_URL || 'redis://127.0.0.1:6379'
-      this._redis = new Redis(redisUrl)
+      if (this.config.redis instanceof Redis) {
+        this._redis = this.config.redis
+      } else if (this.config.redis) {
+        this._redis = new Redis(this.config.redis)
+      } else {
+        const redisUrl = this.config.redis_url || process.env.REDIS_URL || 'redis://127.0.0.1:6379'
+        this._redis = new Redis(redisUrl)
+      }
     }
 
     return this._redis

packages/bullmq/README.md

@@ -40,6 +40,11 @@ require('judoscale-bullmq')
 const judoscale = new Judoscale({
   redis_url: process.env.REDISCLOUD_URL, // defaults to process.env.REDIS_URL
 })
+
+// You can optionally pass a Redis config object or an ioredis instance
+const judoscale = new Judoscale({
+  redis: redisConfig,
+})
 ```
 
 ## Troubleshooting

packages/bullmq/src/bull-mq-metrics-collector.js

@@ -3,8 +3,8 @@ const { Queue } = require('bullmq')
 const { Metric, WorkerMetricsCollector } = require('judoscale-node-core')
 
 class BullMQMetricsCollector extends WorkerMetricsCollector {
-  constructor() {
-    super('BullMQ')
+  constructor(config = {}) {
+    super('BullMQ', config)
     this.queueNames = new Set()
   }
 
@@ -43,8 +43,14 @@ class BullMQMetricsCollector extends WorkerMetricsCollector {
 
   get redis() {
     if (!this._redis) {
-      const redisUrl = this.config.redis_url || process.env.REDIS_URL || 'redis://127.0.0.1:6379'
-      this._redis = new Redis(redisUrl)
+      if (this.config.redis instanceof Redis) {
+        this._redis = this.config.redis
+      } else if (this.config.redis) {
+        this._redis = new Redis(this.config.redis)
+      } else {
+        const redisUrl = this.config.redis_url || process.env.REDIS_URL || 'redis://127.0.0.1:6379'
+        this._redis = new Redis(redisUrl)
+      }
     }
 
     return this._redis

packages/bullmq/test/bull-mq-metrics-collector.test.js

@@ -1,3 +1,4 @@
+const Redis = require('ioredis')
 const { Queue } = require('bullmq')
 const BullMQMetricsCollector = require('../src/bull-mq-metrics-collector')
 
@@ -31,4 +32,61 @@ describe('BullMQMetricsCollector', () => {
     expect(metrics[1].queueName).toEqual('foo')
     expect(metrics[1].value).toEqual(0)
   })
+
+  test('uses the redis_url config if provided', async () => {
+    const collector1 = new BullMQMetricsCollector({ redis_url: 'redis://localhost:6379/1' })
+    const collector2 = new BullMQMetricsCollector({ redis_url: 'redis://localhost:6379/2' })
+
+    const queue = new Queue('foo', { connection: collector1.redis })
+    await queue.add('test-job')
+
+    const metrics1 = await collector1.collect()
+    const metrics2 = await collector2.collect()
+
+    expect(metrics1.length).toEqual(2)
+    expect(metrics2.length).toEqual(0)
+
+    collector1.redis.quit()
+    collector2.redis.quit()
+  })
+
+  test('uses the redis connection if provided', async () => {
+    const redis1 = new Redis('redis://localhost:6379/1')
+    const redis2 = new Redis('redis://localhost:6379/2')
+
+    const collector1 = new BullMQMetricsCollector({ redis: redis1 })
+    const collector2 = new BullMQMetricsCollector({ redis: redis2 })
+
+    const queue = new Queue('foo', { connection: redis1 })
+    await queue.add('test-job')
+
+    const metrics1 = await collector1.collect()
+    const metrics2 = await collector2.collect()
+
+    expect(metrics1.length).toEqual(2)
+    expect(metrics2.length).toEqual(0)
+
+    redis1.quit()
+    redis2.quit()
+  })
+
+  test('uses the redis connection options if provided', async () => {
+    const redis1 = { host: 'localhost', port: 6379, db: 1 }
+    const redis2 = { host: 'localhost', port: 6379, db: 2 }
+
+    const collector1 = new BullMQMetricsCollector({ redis: redis1 })
+    const collector2 = new BullMQMetricsCollector({ redis: redis2 })
+
+    const queue = new Queue('foo', { connection: collector1.redis })
+    await queue.add('test-job')
+
+    const metrics1 = await collector1.collect()
+    const metrics2 = await collector2.collect()
+
+    expect(metrics1.length).toEqual(2)
+    expect(metrics2.length).toEqual(0)
+
+    collector1.redis.quit()
+    collector2.redis.quit()
+  })
 })

packages/node-core/src/judoscale.js

@@ -9,7 +9,7 @@ class Judoscale {
 
     // Expose config to the collectors
     for (const adapter of Judoscale.adapters) {
-      adapter.collector.config = this.config
+      adapter.collector.config = { ...this.config, ...adapter.collector.config }
     }
 
     new Reporter().start(this.config, Judoscale.adapters)

packages/node-core/src/web-metrics-collector.js

@@ -1,8 +1,8 @@
 class WebMetricsCollector {
-  constructor(store, collectorName = 'Web') {
+  constructor(store, collectorName = 'Web', config = {}) {
     this.collectorName = collectorName
     this.store = store
-    this.config = {}
+    this.config = config
   }
 
   collect() {

packages/node-core/src/worker-metrics-collector.js

@@ -1,7 +1,7 @@
 class WorkerMetricsCollector {
-  constructor(collectorName) {
+  constructor(collectorName, config = {}) {
     this.collectorName = collectorName
-    this.config = {}
+    this.config = config
   }
 
   collect() {

sample_apps/express_web/src/web.js

@@ -6,18 +6,17 @@ import 'judoscale-bullmq'
 
 const app = express()
 const port = process.env.PORT || 5000
-const redisUrl = process.env.REDIS_URL || 'redis://127.0.0.1:6379'
-const redisOpts = {
+const redis = new Redis('redis://127.0.0.1:6379', {
   maxRetriesPerRequest: null, // Since bull v4
   enableReadyCheck: false, // Since bull v4
-}
-const redis = new Redis(redisUrl, redisOpts)
+})
 
 app.set('views', './views')
 app.set('view engine', 'ejs')
 
 const judoscale = new Judoscale({
   api_base_url: process.env.JUDOSCALE_URL || 'https://judoscale-node-sample.requestcatcher.com',
+  redis: redis,
 })
 
 app.use(judoscaleMiddleware(judoscale))

sample_apps/express_web/src/worker.js

@@ -2,16 +2,15 @@ import Redis from 'ioredis'
 import { Worker } from 'bullmq'
 import { Judoscale } from 'judoscale-bullmq'
 
-const redisUrl = process.env.REDIS_URL || 'redis://127.0.0.1:6379'
-const redisOpts = {
+const redis = new Redis('redis://127.0.0.1:6379', {
   maxRetriesPerRequest: null, // Since bull v4
   enableReadyCheck: false, // Since bull v4
-}
-const redis = new Redis(redisUrl, redisOpts)
+})
 const queueNames = ['default', 'urgent']
 
 new Judoscale({
   api_base_url: process.env.JUDOSCALE_URL || 'https://judoscale-node-sample.requestcatcher.com',
+  redis: redis,
 })
 
 const workers = queueNames.map((queueName) => {