Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

add nodeAddressMap config for cluster #1827

Merged
merged 5 commits into from
Feb 14, 2022
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
23 changes: 23 additions & 0 deletions docs/clustering.md
Original file line number Diff line number Diff line change
Expand Up @@ -38,9 +38,32 @@ import { createCluster } from 'redis';
| defaults | | The default configuration values for every client in the cluster. Use this for example when specifying an ACL user to connect with |
| useReplicas | `false` | When `true`, distribute load by executing readonly commands (such as `GET`, `GEOSEARCH`, etc.) across all cluster nodes. When `false`, only use master nodes |
| maxCommandRedirections | `16` | The maximum number of times a command will be redirected due to `MOVED` or `ASK` errors |
| nodeAddressMap | | Object defining the [node address mapping](#node-address-map) |
| modules | | Object defining which [Redis Modules](../README.md#modules) to include |
| scripts | | Object defining Lua Scripts to use with this client (see [Lua Scripts](../README.md#lua-scripts)) |

## Node Address Map

Your cluster might be configured to work within an internal network that your local environment doesn't have access to. For example, your development machine could only have access to external addresses, but the cluster returns its internal addresses. In this scenario, it's useful to provide a map from those internal addresses to the external ones.

The configuration for this is a simple mapping. Just provide a `nodeAddressMap` property mapping the internal addresses and ports to the external addresses and ports. Then, any address provided to `rootNodes` or returned from the cluster will be mapped accordingly:

```javascript
createCluster({
rootNodes: [{
url: '10.0.0.1:30001'
}, {
url: '10.0.0.2:30002'
}],
nodeAddressMap: {
'10.0.0.1:30001': 'external-host-1.io:30001',
'10.0.0.2:30002': 'external-host-2.io:30002'
}
});
```

> This is a common problem when using ElastiCache. See [Accessing ElastiCache from outside AWS](https://docs.aws.amazon.com/AmazonElastiCache/latest/red-ug/accessing-elasticache.html) for more information on that.

## Command Routing

### Commands that operate on Redis Keys
Expand Down
80 changes: 51 additions & 29 deletions packages/client/lib/cluster/cluster-slots.ts
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,15 @@ export interface ClusterNode<M extends RedisModules, S extends RedisScripts> {
client: RedisClientType<M, S>;
}

interface NodeAddress {
host: string;
port: number;
}

export type NodeAddressMap = {
[address: string]: NodeAddress;
} | ((address: string) => NodeAddress | undefined);

interface SlotNodes<M extends RedisModules, S extends RedisScripts> {
master: ClusterNode<M, S>;
replicas: Array<ClusterNode<M, S>>;
Expand All @@ -26,7 +35,7 @@ export default class RedisClusterSlots<M extends RedisModules, S extends RedisSc
readonly #options: RedisClusterOptions<M, S>;
readonly #Client: InstantiableRedisClient<M, S>;
readonly #onError: OnError;
readonly #nodeByUrl = new Map<string, ClusterNode<M, S>>();
readonly #nodeByAddress = new Map<string, ClusterNode<M, S>>();
readonly #slots: Array<SlotNodes<M, S>> = [];

constructor(options: RedisClusterOptions<M, S>, onError: OnError) {
Expand All @@ -37,7 +46,7 @@ export default class RedisClusterSlots<M extends RedisModules, S extends RedisSc

async connect(): Promise<void> {
for (const rootNode of this.#options.rootNodes) {
if (await this.#discoverNodes(this.#clientOptionsDefaults(rootNode))) return;
if (await this.#discoverNodes(rootNode)) return;
}

throw new RootNodesUnavailableError();
Expand Down Expand Up @@ -75,7 +84,7 @@ export default class RedisClusterSlots<M extends RedisModules, S extends RedisSc
async #rediscover(startWith: RedisClientType<M, S>): Promise<void> {
if (await this.#discoverNodes(startWith.options)) return;

for (const { client } of this.#nodeByUrl.values()) {
for (const { client } of this.#nodeByAddress.values()) {
if (client === startWith) continue;

if (await this.#discoverNodes(client.options)) return;
Expand All @@ -85,7 +94,7 @@ export default class RedisClusterSlots<M extends RedisModules, S extends RedisSc
}

async #reset(masters: Array<RedisClusterMasterNode>): Promise<void> {
// Override this.#slots and add not existing clients to this.#nodeByUrl
// Override this.#slots and add not existing clients to this.#nodeByAddress
const promises: Array<Promise<void>> = [],
clientsInUse = new Set<string>();
for (const master of masters) {
Expand All @@ -94,7 +103,7 @@ export default class RedisClusterSlots<M extends RedisModules, S extends RedisSc
replicas: this.#options.useReplicas ?
master.replicas.map(replica => this.#initiateClientForNode(replica, true, clientsInUse, promises)) :
[],
clientIterator: undefined // will be initiated in use
clientIterator: undefined // will be initiated in use
};

for (const { from, to } of master.slots) {
Expand All @@ -104,12 +113,12 @@ export default class RedisClusterSlots<M extends RedisModules, S extends RedisSc
}
}

// Remove unused clients from this.#nodeByUrl using clientsInUse
for (const [url, { client }] of this.#nodeByUrl.entries()) {
if (clientsInUse.has(url)) continue;
// Remove unused clients from this.#nodeByAddress using clientsInUse
for (const [address, { client }] of this.#nodeByAddress.entries()) {
if (clientsInUse.has(address)) continue;

promises.push(client.disconnect());
this.#nodeByUrl.delete(url);
this.#nodeByAddress.delete(address);
}

await Promise.all(promises);
Expand All @@ -118,38 +127,49 @@ export default class RedisClusterSlots<M extends RedisModules, S extends RedisSc
#clientOptionsDefaults(options?: RedisClusterClientOptions): RedisClusterClientOptions | undefined {
if (!this.#options.defaults) return options;

const merged = Object.assign({}, this.#options.defaults, options);

if (options?.socket && this.#options.defaults.socket) {
Object.assign({}, this.#options.defaults.socket, options.socket);
}

return merged;
return {
...this.#options.defaults,
...options,
socket: this.#options.defaults.socket && options?.socket ? {
...this.#options.defaults.socket,
...options.socket
} : this.#options.defaults.socket ?? options?.socket
};
}

#initiateClient(options?: RedisClusterClientOptions): RedisClientType<M, S> {
return new this.#Client(this.#clientOptionsDefaults(options))
.on('error', this.#onError);
}

#getNodeAddress(address: string): NodeAddress | undefined {
switch (typeof this.#options.nodeAddressMap) {
case 'object':
return this.#options.nodeAddressMap[address];

case 'function':
return this.#options.nodeAddressMap(address);
}
}

#initiateClientForNode(nodeData: RedisClusterMasterNode | RedisClusterReplicaNode, readonly: boolean, clientsInUse: Set<string>, promises: Array<Promise<void>>): ClusterNode<M, S> {
const url = `${nodeData.host}:${nodeData.port}`;
clientsInUse.add(url);
const address = `${nodeData.host}:${nodeData.port}`;
clientsInUse.add(address);

let node = this.#nodeByUrl.get(url);
let node = this.#nodeByAddress.get(address);
if (!node) {
node = {
id: nodeData.id,
client: this.#initiateClient({
socket: {
socket: this.#getNodeAddress(address) ?? {
host: nodeData.host,
port: nodeData.port
},
readonly
})
};
promises.push(node.client.connect());
this.#nodeByUrl.set(url, node);
this.#nodeByAddress.set(address, node);
}

return node;
Expand Down Expand Up @@ -186,12 +206,12 @@ export default class RedisClusterSlots<M extends RedisModules, S extends RedisSc
#randomClientIterator?: IterableIterator<ClusterNode<M, S>>;

#getRandomClient(): RedisClientType<M, S> {
if (!this.#nodeByUrl.size) {
if (!this.#nodeByAddress.size) {
throw new Error('Cluster is not connected');
}

if (!this.#randomClientIterator) {
this.#randomClientIterator = this.#nodeByUrl.values();
this.#randomClientIterator = this.#nodeByAddress.values();
}

const {done, value} = this.#randomClientIterator.next();
Expand All @@ -218,8 +238,7 @@ export default class RedisClusterSlots<M extends RedisModules, S extends RedisSc

getMasters(): Array<ClusterNode<M, S>> {
const masters = [];

for (const node of this.#nodeByUrl.values()) {
for (const node of this.#nodeByAddress.values()) {
if (node.client.options?.readonly) continue;

masters.push(node);
Expand All @@ -228,8 +247,11 @@ export default class RedisClusterSlots<M extends RedisModules, S extends RedisSc
return masters;
}

getNodeByUrl(url: string): ClusterNode<M, S> | undefined {
return this.#nodeByUrl.get(url);
getNodeByAddress(address: string): ClusterNode<M, S> | undefined {
const mappedAddress = this.#getNodeAddress(address);
return this.#nodeByAddress.get(
mappedAddress ? `${mappedAddress.host}:${mappedAddress.port}` : address
);
}

quit(): Promise<void> {
Expand All @@ -242,13 +264,13 @@ export default class RedisClusterSlots<M extends RedisModules, S extends RedisSc

async #destroy(fn: (client: RedisClientType<M, S>) => Promise<unknown>): Promise<void> {
const promises = [];
for (const { client } of this.#nodeByUrl.values()) {
for (const { client } of this.#nodeByAddress.values()) {
promises.push(fn(client));
}

await Promise.all(promises);

this.#nodeByUrl.clear();
this.#nodeByAddress.clear();
this.#slots.splice(0);
}
}
11 changes: 6 additions & 5 deletions packages/client/lib/cluster/index.ts
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import COMMANDS from './commands';
import { RedisCommand, RedisCommandArgument, RedisCommandArguments, RedisCommandRawReply, RedisCommandReply, RedisModules, RedisPlugins, RedisScript, RedisScripts } from '../commands';
import { ClientCommandOptions, RedisClientCommandSignature, RedisClientOptions, RedisClientType, WithModules, WithScripts } from '../client';
import RedisClusterSlots, { ClusterNode } from './cluster-slots';
import RedisClusterSlots, { ClusterNode, NodeAddressMap } from './cluster-slots';
import { extendWithModulesAndScripts, transformCommandArguments, transformCommandReply, extendWithCommands } from '../commander';
import { EventEmitter } from 'events';
import RedisClusterMultiCommand, { RedisClusterMultiCommandType } from './multi-command';
Expand All @@ -17,6 +17,7 @@ export interface RedisClusterOptions<
defaults?: Partial<RedisClusterClientOptions>;
useReplicas?: boolean;
maxCommandRedirections?: number;
nodeAddressMap?: NodeAddressMap;
}

type WithCommands = {
Expand Down Expand Up @@ -144,16 +145,16 @@ export default class RedisCluster<M extends RedisModules, S extends RedisScripts
}

if (err.message.startsWith('ASK')) {
const url = err.message.substring(err.message.lastIndexOf(' ') + 1);
if (this.#slots.getNodeByUrl(url)?.client === client) {
const address = err.message.substring(err.message.lastIndexOf(' ') + 1);
if (this.#slots.getNodeByAddress(address)?.client === client) {
await client.asking();
continue;
}

await this.#slots.rediscover(client);
const redirectTo = this.#slots.getNodeByUrl(url);
const redirectTo = this.#slots.getNodeByAddress(address);
if (!redirectTo) {
throw new Error(`Cannot find node ${url}`);
throw new Error(`Cannot find node ${address}`);
}

await redirectTo.client.asking();
Expand Down
12 changes: 6 additions & 6 deletions packages/client/lib/commands/CLUSTER_NODES.spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ describe('CLUSTER NODES', () => {
].join('\n')),
[{
id: 'master',
url: '127.0.0.1:30001@31001',
address: '127.0.0.1:30001@31001',
host: '127.0.0.1',
port: 30001,
cport: 31001,
Expand All @@ -34,7 +34,7 @@ describe('CLUSTER NODES', () => {
}],
replicas: [{
id: 'slave',
url: '127.0.0.1:30002@31002',
address: '127.0.0.1:30002@31002',
host: '127.0.0.1',
port: 30002,
cport: 31002,
Expand All @@ -48,14 +48,14 @@ describe('CLUSTER NODES', () => {
);
});

it('should support urls without cport', () => {
it('should support addresses without cport', () => {
assert.deepEqual(
transformReply(
'id 127.0.0.1:30001 master - 0 0 0 connected 0-16384\n'
),
[{
id: 'id',
url: '127.0.0.1:30001',
address: '127.0.0.1:30001',
host: '127.0.0.1',
port: 30001,
cport: null,
Expand All @@ -80,7 +80,7 @@ describe('CLUSTER NODES', () => {
),
[{
id: 'id',
url: '127.0.0.1:30001@31001',
address: '127.0.0.1:30001@31001',
host: '127.0.0.1',
port: 30001,
cport: 31001,
Expand All @@ -102,7 +102,7 @@ describe('CLUSTER NODES', () => {
),
[{
id: 'id',
url: '127.0.0.1:30001@31001',
address: '127.0.0.1:30001@31001',
host: '127.0.0.1',
port: 30001,
cport: 31001,
Expand Down
28 changes: 14 additions & 14 deletions packages/client/lib/commands/CLUSTER_NODES.ts
Original file line number Diff line number Diff line change
Expand Up @@ -7,15 +7,15 @@ export enum RedisClusterNodeLinkStates {
DISCONNECTED = 'disconnected'
}

interface RedisClusterNodeTransformedUrl {
interface RedisClusterNodeAddress {
host: string;
port: number;
cport: number | null;
}

export interface RedisClusterReplicaNode extends RedisClusterNodeTransformedUrl {
export interface RedisClusterReplicaNode extends RedisClusterNodeAddress {
id: string;
url: string;
address: string;
flags: Array<string>;
pingSent: number;
pongRecv: number;
Expand All @@ -39,11 +39,11 @@ export function transformReply(reply: string): Array<RedisClusterMasterNode> {
replicasMap = new Map<string, Array<RedisClusterReplicaNode>>();

for (const line of lines) {
const [id, url, flags, masterId, pingSent, pongRecv, configEpoch, linkState, ...slots] = line.split(' '),
const [id, address, flags, masterId, pingSent, pongRecv, configEpoch, linkState, ...slots] = line.split(' '),
node = {
id,
url,
...transformNodeUrl(url),
address,
...transformNodeAddress(address),
flags: flags.split(','),
pingSent: Number(pingSent),
pongRecv: Number(pongRecv),
Expand Down Expand Up @@ -84,22 +84,22 @@ export function transformReply(reply: string): Array<RedisClusterMasterNode> {
return [...mastersMap.values()];
}

function transformNodeUrl(url: string): RedisClusterNodeTransformedUrl {
const indexOfColon = url.indexOf(':'),
indexOfAt = url.indexOf('@', indexOfColon),
host = url.substring(0, indexOfColon);
function transformNodeAddress(address: string): RedisClusterNodeAddress {
const indexOfColon = address.indexOf(':'),
indexOfAt = address.indexOf('@', indexOfColon),
host = address.substring(0, indexOfColon);

if (indexOfAt === -1) {
return {
host,
port: Number(url.substring(indexOfColon + 1)),
port: Number(address.substring(indexOfColon + 1)),
cport: null
};
}

return {
host: url.substring(0, indexOfColon),
port: Number(url.substring(indexOfColon + 1, indexOfAt)),
cport: Number(url.substring(indexOfAt + 1))
host: address.substring(0, indexOfColon),
port: Number(address.substring(indexOfColon + 1, indexOfAt)),
cport: Number(address.substring(indexOfAt + 1))
};
}