feat: improve redis cluster handling

Author: Matt Willian

lib/queue-factory.ts

@@ -1,4 +1,5 @@
 import { Redis, Cluster, RedisOptions } from "ioredis";
+import { ConnectionOptions as TlsConnectionOptions } from "tls";
 
 import { QueueType, getQueueType, redisOptsFromUrl } from "./utils";
 import { Queue } from "bullmq";
@@ -16,6 +17,13 @@ const maxTime = 40000;
 // We keep a redis client that we can reuse for all the queues.
 let redisClients: Record<string, Redis | Cluster> = {} as any;
 
+type NestedRedisOptions = Partial<RedisOptions> & {
+  tls?: TlsConnectionOptions;
+};
+type RedisOptionsWithNested = RedisOptions & {
+  redisOptions?: NestedRedisOptions;
+};
+
 export interface FoundQueue {
   prefix: string;
   name: string;
@@ -167,21 +175,52 @@ export function getRedisClient(
 
   if (!redisClients[key]) {
     if (clusterNodes && clusterNodes.length) {
-      const { username, password } = redisOptsFromUrl(clusterNodes[0]);
+      const { username: nodeUsername, password: nodePassword } =
+        redisOptsFromUrl(clusterNodes[0]);
+      const {
+        username: redisOptsUsername,
+        password: redisOptsPassword,
+        tls: redisOptsTls,
+        redisOptions: suppliedRedisOptionsRaw,
+        ...clusterLevelOpts
+      } = redisOpts as RedisOptionsWithNested;
+      const suppliedRedisOptions: NestedRedisOptions =
+        suppliedRedisOptionsRaw ?? {};
+      const baseRedisOptions: NestedRedisOptions = {
+        ...suppliedRedisOptions,
+      };
+      delete baseRedisOptions.username;
+      delete baseRedisOptions.password;
+      delete baseRedisOptions.tls;
+
+      const finalUsername =
+        redisOptsUsername ??
+        suppliedRedisOptions.username ??
+        nodeUsername;
+      if (finalUsername !== undefined) {
+        baseRedisOptions.username = finalUsername;
+      }
+
+      const finalPassword =
+        redisOptsPassword ??
+        suppliedRedisOptions.password ??
+        nodePassword;
+      if (finalPassword !== undefined) {
+        baseRedisOptions.password = finalPassword;
+      }
+
+      const mergedTls = mergeTlsConfigs(
+        getClusterTlsFromEnv(),
+        suppliedRedisOptions.tls,
+        redisOptsTls
+      );
+      if (mergedTls) {
+        baseRedisOptions.tls = mergedTls;
+      }
+
       redisClients[key] = new Redis.Cluster(clusterNodes, {
-        ...redisOpts,
-        redisOptions: {
-          username,
-          password,
-          tls: process.env.REDIS_CLUSTER_TLS
-            ? {
-                cert: Buffer.from(
-                  process.env.REDIS_CLUSTER_TLS ?? "",
-                  "base64"
-                ).toString("ascii"),
-              }
-            : undefined,
-        },
+        ...clusterLevelOpts,
+        redisOptions: baseRedisOptions,
       });
     } else {
       redisClients[key] = new Redis(redisOpts);
@@ -300,3 +339,25 @@ export function createQueue(
       );
   }
 }
+
+function getClusterTlsFromEnv(): TlsConnectionOptions | undefined {
+  if (!process.env.REDIS_CLUSTER_TLS) {
+    return undefined;
+  }
+  return {
+    cert: Buffer.from(
+      process.env.REDIS_CLUSTER_TLS ?? "",
+      "base64"
+    ).toString("ascii"),
+  };
+}
+
+function mergeTlsConfigs(
+  ...configs: Array<TlsConnectionOptions | undefined>
+): TlsConnectionOptions | undefined {
+  const valid = configs.filter(Boolean) as TlsConnectionOptions[];
+  if (!valid.length) {
+    return undefined;
+  }
+  return Object.assign({}, ...valid);
+}

lib/socket.ts

@@ -1,5 +1,4 @@
 import { RedisOptions } from "ioredis";
-import { pick } from "lodash";
 import { getCache, updateQueuesCache, queueKey } from "./queues-cache";
 import { WebSocketClient } from "./ws-autoreconnect";
 import {
@@ -15,13 +14,8 @@ const { version } = require(`${__dirname}/../package.json`);
 
 const chalk = require("chalk");
 
-export interface Connection {
-  port?: number;
-  host?: string;
-  password?: string;
-  db?: number;
+export interface Connection extends Partial<RedisOptions> {
   uri?: string;
-  tls?: object;
 }
 
 export const Socket = (
@@ -139,12 +133,20 @@ export const Socket = (
                 throw new Error("Unable to update queues");
               }
             }
-            const { queue, responders } =
-              cache[
-                queueKey({ name: queueName, prefix: queuePrefix || "bull" })
-              ];
+            const cacheKeyValue = queueKey({
+              name: queueName,
+              prefix: queuePrefix || "bull",
+            });
+            const cacheEntry = cache[cacheKeyValue];
 
-            if (!queue) {
+            if (!cacheEntry || !cacheEntry.queue) {
+              console.warn(
+                `${chalk.yellow("WebSocket:")} ${chalk.red(
+                  "queue cache miss for"
+                )} ${chalk.green(queueName)} ${chalk.blueBright(
+                  "prefix:"
+                )} ${chalk.green(queuePrefix || "bull")}`
+              );
               ws.send(
                 JSON.stringify({
                   id: msg.id,
@@ -153,6 +155,7 @@ export const Socket = (
                 startTime
               );
             } else {
+              const { queue, responders } = cacheEntry;
               switch (res) {
                 case "queues":
                   await responders.respondQueueCommand(ws, queue, msg);
@@ -249,31 +252,25 @@ export const Socket = (
   }
 };
 
-function redisOptsFromConnection(connection: Connection): RedisOptions {
-  let opts: RedisOptions = {
-    ...pick(connection, [
-      "host",
-      "port",
-      "username",
-      "password",
-      "family",
-      "sentinelPassword",
-      "db",
-      "tls",
-      "sentinels",
-      "name",
-    ]),
+function redisOptsFromConnection(
+  connection: Connection = {}
+): RedisOptions {
+  let opts: (RedisOptions & { uri?: string }) = {
+    ...connection,
   };
 
   if (connection.uri) {
     opts = { ...opts, ...redisOptsFromUrl(connection.uri) };
   }
+  delete opts.uri;
 
-  opts.retryStrategy = function (times: number) {
-    times = times % 8;
-    const delay = Math.round(Math.pow(2, times + 8));
-    console.log(chalk.yellow("Redis: ") + `Reconnecting in ${delay} ms`);
-    return delay;
-  };
+  if (!opts.retryStrategy) {
+    opts.retryStrategy = function (times: number) {
+      times = times % 8;
+      const delay = Math.round(Math.pow(2, times + 8));
+      console.log(chalk.yellow("Redis: ") + `Reconnecting in ${delay} ms`);
+      return delay;
+    };
+  }
   return opts;
 }