support redis clusters (#27)

* support redis clusters

* support redis cluster 2

* support redis cluster 3

* make example-cap-server deployable

* dedicated integration mode for redis cluster

* load credentials top-level so we can identify cluster mode

* add remote url to example-cap-server http

* make redis-service name generic

* polish for manifest

* update tests

* remove remote http-client.env

* less top-level invocation

* less top-level invocation 2

Author: rlindner81

.eslintrc.yml

@@ -3,9 +3,9 @@ root: true
 #  https://eslint.org/docs/rules/
 env:
   node: true
-  es2020: true
+  es2022: true
 parserOptions:
-  ecmaVersion: 2020
+  ecmaVersion: 2022
 plugins:
   - jest
 extends:

example-cap-server/manifest.yml

@@ -0,0 +1,11 @@
+# https://docs.cloudfoundry.org/devguide/deploy-apps/manifest-attributes.html
+---
+applications:
+  - name: example-cap-server
+    path: ./gen/srv
+    random-route: true
+    memory: 256M
+    buildpacks:
+      - nodejs_buildpack
+    services:
+      - redis-service

example-cap-server/package.json

@@ -3,19 +3,29 @@
   "name": "example-cap-server",
   "version": "1.0.0",
   "scripts": {
-    "start": "npm run copy-library && npm run serve",
-    "copy-library": "npx shx rm -rf node_modules/@cap-js-community/feature-toggle-library && npx shx mkdir -p node_modules/@cap-js-community/feature-toggle-library && npx shx cp -R ../package.json ../index.cds ../cds-plugin.js ../src node_modules/@cap-js-community/feature-toggle-library",
+    "start": "npm run serve",
     "serve": "cds-serve",
-    "build": "cds build --production",
+    "copy-library": "npx shx rm -rf node_modules/@cap-js-community/feature-toggle-library && npx shx mkdir -p node_modules/@cap-js-community/feature-toggle-library && npx shx cp -R ../package.json ../index.cds ../cds-plugin.js ../src node_modules/@cap-js-community/feature-toggle-library",
+    "build": "npm run build:cds && npm run build:deps && npm run build:copy-library",
+    "build:cds": "cds build --production",
+    "build:deps": "npm i --omit=dev --prefix=gen/srv",
+    "build:copy-library": "npx shx rm -rf gen/srv/node_modules/@cap-js-community/feature-toggle-library && npx shx mkdir -p gen/srv/node_modules/@cap-js-community/feature-toggle-library && npx shx cp -R ../package.json ../index.cds ../cds-plugin.js ../src gen/srv/node_modules/@cap-js-community/feature-toggle-library",
+    "deploy": "npm run build && cf push",
     "cloc": "npx cloc --vcs=git --exclude-ext=def --read-lang-def=cloc.def .",
     "upgrade": "npm up --save && npx shx rm -rf node_modules && npm i"
   },
+  "engines": {
+    "node": "^18.0.0",
+    "npm": "^9.0.0"
+  },
   "dependencies": {
     "@cap-js-community/feature-toggle-library": "*",
     "@sap/cds": "^7.2.0",
-    "@sap/cds-dk": "^7.2.0",
     "express": "^4.18.2"
   },
+  "devDependencies": {
+    "@sap/cds-dk": "^7.2.0"
+  },
   "cds": {
     "featureToggles": {
       "configFile": "./srv/feature/features.yaml",

src/env.js

@@ -2,9 +2,14 @@
 
 const { ENV } = require("./shared/static");
 
-const isOnCF = process.env[ENV.USER] === "vcap";
-
 class CfEnv {
+  isOnCf;
+  cfApp;
+  cfServices;
+  cfInstanceGuid;
+  cfInstanceIp;
+  cfInstanceIndex;
+
   static parseEnvVar(env, envVar) {
     try {
       if (Object.prototype.hasOwnProperty.call(env, envVar)) {
@@ -14,6 +19,7 @@ class CfEnv {
   }
 
   constructor(env = process.env) {
+    this.isOnCf = env[ENV.USER] === "vcap";
     this.cfApp = CfEnv.parseEnvVar(env, ENV.CF_APP) || {};
     this.cfServices = CfEnv.parseEnvVar(env, ENV.CF_SERVICES) || {};
     this.cfInstanceGuid = env[ENV.CF_INSTANCE_GUID];
@@ -31,7 +37,7 @@ class CfEnv {
   /**
    * @return CfEnv
    */
-  static getInstance() {
+  static get instance() {
     if (!CfEnv.__instance) {
       CfEnv.__instance = new CfEnv();
     }
@@ -53,7 +59,5 @@ class CfEnv {
 
 module.exports = {
   CfEnv,
-
-  isOnCF,
-  cfEnv: CfEnv.getInstance(),
+  cfEnv: CfEnv.instance,
 };

src/featureToggles.js

@@ -21,7 +21,7 @@ const yaml = require("yaml");
 const redis = require("./redisWrapper");
 const { REDIS_INTEGRATION_MODE } = redis;
 const { Logger } = require("./logger");
-const { isOnCF, cfEnv } = require("./env");
+const { cfEnv } = require("./env");
 const { HandlerCollection } = require("./shared/handlerCollection");
 const { ENV, isObject, tryRequire } = require("./shared/static");
 const { promiseAllDone } = require("./shared/promiseAllDone");
@@ -743,7 +743,7 @@ class FeatureToggles {
         await redis.subscribe(this.__redisChannel);
       } catch (err) {
         logger.warning(
-          isOnCF
+          cfEnv.isOnCf
             ? new VError(
                 {
                   name: VERROR_CLUSTER_NAME,

src/redisWrapper.js

@@ -6,43 +6,43 @@
 const redis = require("redis");
 const VError = require("verror");
 const { Logger } = require("./logger");
-const { isOnCF, cfEnv } = require("./env");
+const { cfEnv } = require("./env");
 const { HandlerCollection } = require("./shared/handlerCollection");
 const { Semaphore } = require("./shared/semaphore");
 
 const COMPONENT_NAME = "/RedisWrapper";
 const VERROR_CLUSTER_NAME = "RedisWrapperError";
 
 const INTEGRATION_MODE = Object.freeze({
+  CF_REDIS_CLUSTER: "CF_REDIS_CLUSTER",
   CF_REDIS: "CF_REDIS",
   LOCAL_REDIS: "LOCAL_REDIS",
   NO_REDIS: "NO_REDIS",
 });
+const CF_REDIS_SERVICE_LABEL = "redis-cache";
 
 const logger = new Logger(COMPONENT_NAME);
+const watchedGetSetSemaphore = new Semaphore();
 
 const MODE = Object.freeze({
   RAW: "raw",
   OBJECT: "object",
 });
 
-let redisIsOnCF = isOnCF;
-let mainClient = null;
-let subscriberClient = null;
-let messageHandlers = new HandlerCollection();
-let integrationMode = null;
-
-const watchedGetSetSemaphore = new Semaphore();
-
+let messageHandlers;
+let mainClient;
+let subscriberClient;
+let integrationMode;
 const _reset = () => {
-  redisIsOnCF = isOnCF;
+  messageHandlers = new HandlerCollection();
   mainClient = null;
   subscriberClient = null;
-  messageHandlers = new HandlerCollection();
+  integrationMode = null;
 };
+_reset();
 
 const _logErrorOnEvent = (err) =>
-  redisIsOnCF ? logger.error(err) : logger.warning("%s | %O", err.message, VError.info(err));
+  cfEnv.isOnCf ? logger.error(err) : logger.warning("%s | %O", err.message, VError.info(err));
 
 const _subscribedMessageHandler = async (message, channel) => {
   const handlers = messageHandlers.getHandlers(channel);
@@ -78,22 +78,30 @@ const _localReconnectStrategy = () =>
  * Lazily create a new redis client. Client creation transparently handles both the Cloud Foundry "redis-cache" service
  * (hyperscaler option) and a local redis-server.
  *
- * @returns {RedisClient}
+ * @returns {RedisClient|RedisCluster}
  * @private
  */
 const _createClientBase = () => {
-  if (redisIsOnCF) {
+  if (cfEnv.isOnCf) {
     try {
-      const credentials = cfEnv.cfServiceCredentialsForLabel("redis-cache");
       // NOTE: settings the user explicitly to empty resolves auth problems, see
       // https://github.com/go-redis/redis/issues/1343
-      const url = credentials.uri.replace(/(?<=rediss:\/\/)[\w-]+?(?=:)/, "");
+      const redisCredentials = cfEnv.cfServiceCredentialsForLabel(CF_REDIS_SERVICE_LABEL);
+      const redisIsCluster = redisCredentials.cluster_mode;
+      const url = redisCredentials.uri.replace(/(?<=rediss:\/\/)[\w-]+?(?=:)/, "");
+      if (redisIsCluster) {
+        return redis.createCluster({
+          rootNodes: [{ url }],
+          // https://github.com/redis/node-redis/issues/1782
+          defaults: {
+            password: redisCredentials.password,
+            socket: { tls: redisCredentials.tls },
+          },
+        });
+      }
       return redis.createClient({ url });
     } catch (err) {
-      throw new VError(
-        { name: VERROR_CLUSTER_NAME, cause: err },
-        "error during create client with redis-cache service"
-      );
+      throw new VError({ name: VERROR_CLUSTER_NAME, cause: err }, "error during create client with redis service");
     }
   } else {
     // NOTE: documentation is buried here https://github.com/redis/node-redis/blob/master/docs/client-configuration.md
@@ -144,7 +152,7 @@ const _clientErrorHandlerBase = async (client, err, clientName) => {
  *
  * Only one publisher is necessary for any number of channels.
  *
- * @returns {RedisClient}
+ * @returns {RedisClient|RedisCluster}
  * @private
  */
 const getMainClient = async () => {
@@ -163,7 +171,7 @@ const getMainClient = async () => {
  *
  * Only one subscriber is necessary for any number of channels.
  *
- * @returns {RedisClient}
+ * @returns {RedisClient|RedisCluster}
  * @private
  */
 const getSubscriberClient = async () => {
@@ -205,6 +213,12 @@ const sendCommand = async (command) => {
   }
 
   try {
+    const redisIsCluster = cfEnv.cfServiceCredentialsForLabel(CF_REDIS_SERVICE_LABEL).cluster_mode;
+    if (redisIsCluster) {
+      // NOTE: the cluster sendCommand API has a different signature, where it takes two optional args: firstKey and
+      //   isReadonly before the command
+      return await mainClient.sendCommand(undefined, undefined, command);
+    }
     return await mainClient.sendCommand(command);
   } catch (err) {
     throw new VError(
@@ -476,8 +490,9 @@ const removeMessageHandler = (channel, handler) => messageHandlers.removeHandler
 const removeAllMessageHandlers = (channel) => messageHandlers.removeAllHandlers(channel);
 
 const _getIntegrationMode = async () => {
-  if (redisIsOnCF) {
-    return INTEGRATION_MODE.CF_REDIS;
+  if (cfEnv.isOnCf) {
+    const redisIsCluster = cfEnv.cfServiceCredentialsForLabel(CF_REDIS_SERVICE_LABEL).cluster_mode;
+    return redisIsCluster ? INTEGRATION_MODE.CF_REDIS_CLUSTER : INTEGRATION_MODE.CF_REDIS;
   }
   try {
     await getMainClient();
@@ -521,7 +536,6 @@ module.exports = {
     _reset,
     _getMessageHandlers: () => messageHandlers,
     _getLogger: () => logger,
-    _setRedisIsOnCF: (value) => (redisIsOnCF = value),
     _getMainClient: () => mainClient,
     _setMainClient: (value) => (mainClient = value),
     _getSubscriberClient: () => subscriberClient,

test/__mocks__/env.js

@@ -1,19 +1,15 @@
 "use strict";
 
-let cfEnv;
-let isOnCF;
-
+let cfEnv = {};
 const _reset = () => {
-  isOnCF = false;
-  cfEnv = {
-    cfApp: {},
-    cfServices: {},
-  };
+  cfEnv.isOnCf = false;
+  cfEnv.cfApp = {};
+  cfEnv.cfServices = {};
+  cfEnv.cfServiceCredentialsForLabel = jest.fn().mockReturnValue({});
 };
 _reset();
 
 module.exports = {
   cfEnv,
-  isOnCF,
   _reset,
 };

test/redisWrapper.test.js

@@ -1,12 +1,7 @@
 "use strict";
 
-const mockCfEnv = {
-  cfServiceCredentialsForLabel: jest.fn(),
-};
-jest.mock("../src/env", () => ({
-  isOnCF: false,
-  cfEnv: mockCfEnv,
-}));
+const envMock = require("../src/env");
+jest.mock("../src/env", () => require("./__mocks__/env"));
 
 const mockMessageHandler = jest.fn();
 const mockMessageHandlerTwo = jest.fn();
@@ -52,6 +47,7 @@ describe("redis wrapper test", () => {
   afterEach(() => {
     jest.clearAllMocks();
     redisWrapper._._reset();
+    envMock._reset();
   });
 
   it("_createMainClientAndConnect/_createSubscriberClientAndConnect shortcut", async () => {
@@ -69,7 +65,7 @@ describe("redis wrapper test", () => {
   });
 
   it("_createClientBase local", async () => {
-    redisWrapper._._setRedisIsOnCF(false);
+    envMock.cfEnv.isOnCf = false;
     const client = redisWrapper._._createClientBase();
 
     expect(redis.createClient).toHaveBeenCalledTimes(1);
@@ -88,15 +84,16 @@ describe("redis wrapper test", () => {
   });
 
   it("_createClientBase on CF", async () => {
-    redisWrapper._._setRedisIsOnCF(true);
     const mockUrl = "rediss://BAD_USERNAME:pwd@mockUrl";
     const mockUrlUsable = mockUrl.replace("BAD_USERNAME", "");
-    mockCfEnv.cfServiceCredentialsForLabel.mockImplementationOnce(() => ({ uri: mockUrl }));
+
+    envMock.cfEnv.isOnCf = true;
+    envMock.cfEnv.cfServiceCredentialsForLabel.mockReturnValueOnce({ uri: mockUrl });
 
     const client = redisWrapper._._createClientBase();
 
-    expect(mockCfEnv.cfServiceCredentialsForLabel).toHaveBeenCalledTimes(1);
-    expect(mockCfEnv.cfServiceCredentialsForLabel).toHaveBeenCalledWith("redis-cache");
+    expect(envMock.cfEnv.cfServiceCredentialsForLabel).toHaveBeenCalledTimes(1);
+    expect(envMock.cfEnv.cfServiceCredentialsForLabel).toHaveBeenCalledWith("redis-cache");
     expect(redis.createClient).toHaveBeenCalledTimes(1);
     expect(redis.createClient).toHaveBeenCalledWith({ url: mockUrlUsable });
     expect(client).toBe(mockClient);
@@ -424,9 +421,11 @@ describe("redis wrapper test", () => {
   });
 
   it("_subscribedMessageHandler error", async () => {
-    redisWrapper._._setRedisIsOnCF(true);
     const mockUrl = "rediss://BAD_USERNAME:pwd@mockUrl";
-    mockCfEnv.cfServiceCredentialsForLabel.mockImplementationOnce(() => ({ uri: mockUrl }));
+
+    envMock.cfEnv.isOnCf = true;
+    envMock.cfEnv.cfServiceCredentialsForLabel.mockReturnValueOnce({ uri: mockUrl });
+
     redisWrapper.registerMessageHandler(channel, mockMessageHandler);
     redisWrapper.registerMessageHandler(channel, mockMessageHandlerTwo);
     await redisWrapper.subscribe(channel);