redis/node-redis
1
0
Fork 0
mirror of https://github.com/redis/node-redis.git synced 2025-08-04 15:02:09 +03:00
Code Activity

enhance cluster reshard handling

Browse Source
This commit is contained in:
leibale
2021-11-22 17:42:41 -05:00
parent 6946e36ba0
commit 42e36dfbb1
4 changed files with 76 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
packages/client/lib/cluster/cluster-slots.ts
View File

@ -42,20 +42,8 @@ export default class RedisClusterSlots<M extends RedisModules, S extends RedisSc
throw new Error('None of the root nodes is available');
}
async discover(startWith: RedisClientType<M, S>): Promise<void> {
if (await this.#discoverNodes(startWith.options)) return;
for (const { client } of this.#nodeByUrl.values()) {
if (client === startWith) continue;
if (await this.#discoverNodes(client.options)) return;
}
throw new Error('None of the cluster nodes is available');
}
async #discoverNodes(clientOptions?: RedisClusterClientOptions): Promise<boolean> {
const client = new this.#Client(clientOptions);
const client = this.#initiateClient(clientOptions);
await client.connect();
@ -72,6 +60,29 @@ export default class RedisClusterSlots<M extends RedisModules, S extends RedisSc
}
}
#runningRediscoverPromise?: Promise<void>;
async rediscover(startWith: RedisClientType<M, S>): Promise<void> {
if (!this.#runningRediscoverPromise) {
this.#runningRediscoverPromise = this.#rediscover(startWith)
.finally(() => this.#runningRediscoverPromise = undefined);
}
return this.#runningRediscoverPromise;
}
async #rediscover(startWith: RedisClientType<M, S>): Promise<void> {
if (await this.#discoverNodes(startWith.options)) return;
for (const { client } of this.#nodeByUrl.values()) {
if (client === startWith) continue;
if (await this.#discoverNodes(client.options)) return;
}
throw new Error('None of the cluster nodes is available');
}
async #reset(masters: Array<RedisClusterMasterNode>): Promise<void> {
// Override this.#slots and add not existing clients to this.#nodeByUrl
const promises: Array<Promise<void>> = [],
@ -103,18 +114,23 @@ export default class RedisClusterSlots<M extends RedisModules, S extends RedisSc
await Promise.all(promises);
}
#clientOptionsDefaults(options: RedisClusterClientOptions): RedisClusterClientOptions {
#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) {
if (options?.socket && this.#options.defaults.socket) {
Object.assign({}, this.#options.defaults.socket, options.socket);
}
return merged;
}
#initiateClient(options?: RedisClusterClientOptions): RedisClientType<M, S> {
return new this.#Client(this.#clientOptionsDefaults(options))
.on('error', this.#onError);
}
#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);
@ -123,15 +139,13 @@ export default class RedisClusterSlots<M extends RedisModules, S extends RedisSc
if (!node) {
node = {
id: nodeData.id,
client: new this.#Client(
this.#clientOptionsDefaults({
client: this.#initiateClient({
socket: {
host: nodeData.host,
port: nodeData.port
},
readonly
})
)
};
promises.push(node.client.connect());
this.#nodeByUrl.set(url, node);

4
packages/client/lib/cluster/index.ts
View File

@ -157,7 +157,7 @@ export default class RedisCluster<M extends RedisModules = Record<string, never>
const url = err.message.substring(err.message.lastIndexOf(' ') + 1);
let node = this.#slots.getNodeByUrl(url);
if (!node) {
await this.#slots.discover(client);
await this.#slots.rediscover(client);
node = this.#slots.getNodeByUrl(url);
if (!node) {
@ -168,7 +168,7 @@ export default class RedisCluster<M extends RedisModules = Record<string, never>
await node.client.asking();
return node.client;
} else if (err.message.startsWith('MOVED')) {
await this.#slots.discover(client);
await this.#slots.rediscover(client);
return true;
}

25
packages/client/lib/commands/CLUSTER_NODES.spec.ts
View File

@ -48,6 +48,31 @@ describe('CLUSTER NODES', () => {
);
});
it('should support urls 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',
host: '127.0.0.1',
port: 30001,
cport: null,
flags: ['master'],
pingSent: 0,
pongRecv: 0,
configEpoch: 0,
linkState: RedisClusterNodeLinkStates.CONNECTED,
slots: [{
from: 0,
to: 16384
}],
replicas: []
}]
);
});
it.skip('with importing slots', () => {
assert.deepEqual(
transformReply(

13
packages/client/lib/commands/CLUSTER_NODES.ts
View File

@ -10,7 +10,7 @@ export enum RedisClusterNodeLinkStates {
interface RedisClusterNodeTransformedUrl {
host: string;
port: number;
cport: number;
cport: number | null;
}
export interface RedisClusterReplicaNode extends RedisClusterNodeTransformedUrl {
@ -86,7 +86,16 @@ export function transformReply(reply: string): Array<RedisClusterMasterNode> {
function transformNodeUrl(url: string): RedisClusterNodeTransformedUrl {
const indexOfColon = url.indexOf(':'),
indexOfAt = url.indexOf('@', indexOfColon);
indexOfAt = url.indexOf('@', indexOfColon),
host = url.substring(0, indexOfColon);
if (indexOfAt === -1) {
return {
host,
port: Number(url.substring(indexOfColon + 1)),
cport: null
};
}
return {
host: url.substring(0, indexOfColon),