From 7bf6feac0205209f1f21374d3ef8040db6d5c215 Mon Sep 17 00:00:00 2001 From: luin Date: Sun, 8 Jul 2018 02:24:23 +0800 Subject: [PATCH] fix(Sentinel): unreachable errors when sentinals are healthy --- lib/connectors/SentinelConnector.ts | 267 ------------------ .../SentinelConnector/SentinelIterator.ts | 43 +++ lib/connectors/SentinelConnector/index.ts | 267 ++++++++++++++++++ lib/connectors/SentinelConnector/types.ts | 4 + lib/utils/lodash.js | 11 +- 5 files changed, 323 insertions(+), 269 deletions(-) delete mode 100644 lib/connectors/SentinelConnector.ts create mode 100644 lib/connectors/SentinelConnector/SentinelIterator.ts create mode 100644 lib/connectors/SentinelConnector/index.ts create mode 100644 lib/connectors/SentinelConnector/types.ts diff --git a/lib/connectors/SentinelConnector.ts b/lib/connectors/SentinelConnector.ts deleted file mode 100644 index 0a2c6ae1..00000000 --- a/lib/connectors/SentinelConnector.ts +++ /dev/null @@ -1,267 +0,0 @@ -import {createConnection, Socket} from 'net' -import {bind, sample} from '../utils/lodash' -import {CONNECTION_CLOSED_ERROR_MSG, packObject} from '../utils/index' -import Connector, {ITcpConnectionOptions, IIpcConnectionOptions, ErrorEmitter, isIIpcConnectionOptions} from './Connector' -import { TLSSocket } from 'tls'; -const debug = require('../utils/debug')('ioredis:SentinelConnector') - -let Redis - -interface ISentinelSlavesResponse { - port: string, - ip: string, - flags?: string -} - -interface ISentinelOptions { - role: 'master' | 'slave' - name: 'string' - sentinels: any[] - sentinelRetryStrategy?: (retryAttempts: number) => number - preferredSlaves?: - ((slaves: Array) => ISentinelSlavesResponse) | - Array<{port: string, ip: string, prio?: number}> | - {port: string, ip: string, prio?: number} - connectTimeout?: number -} - -type NodeCallback = (err: Error | null, result?: T) => void - -interface ISentinelTcpConnectionOptions extends ITcpConnectionOptions, ISentinelOptions {} -interface ISentinelIpcConnectionOptions extends IIpcConnectionOptions, ISentinelOptions {} - -export default class SentinelConnector extends Connector { - private retryAttempts: number - private currentPoint: number = -1 - private sentinels: any[] - - constructor (protected options: ISentinelTcpConnectionOptions | ISentinelIpcConnectionOptions) { - super(options) - if (this.options.sentinels.length === 0) { - throw new Error('Requires at least one sentinel to connect to.') - } - if (!this.options.name) { - throw new Error('Requires the name of master.') - } - - this.sentinels = this.options.sentinels - } - - public check (info: {role?: string}): boolean { - const roleMatches: boolean = !info.role || this.options.role === info.role - if (!roleMatches) { - debug('role invalid, expected %s, but got %s', this.options.role, info.role) - } - return roleMatches - } - - connect (callback: NodeCallback, eventEmitter: ErrorEmitter): void { - this.connecting = true - this.retryAttempts = 0 - - let lastError - const _this = this - connectToNext() - - function connectToNext() { - _this.currentPoint += 1 - if (_this.currentPoint === _this.sentinels.length) { - _this.currentPoint = -1 - - const retryDelay = typeof _this.options.sentinelRetryStrategy === 'function' - ? _this.options.sentinelRetryStrategy(++_this.retryAttempts) - : null - - let errorMsg = typeof retryDelay !== 'number' - ? 'All sentinels are unreachable and retry is disabled.' - : `All sentinels are unreachable. Retrying from scratch after ${retryDelay}ms.` - - if (lastError) { - errorMsg += ` Last error: ${lastError.message}` - } - - debug(errorMsg) - - const error = new Error(errorMsg) - if (typeof retryDelay === 'number') { - setTimeout(connectToNext, retryDelay) - eventEmitter('error', error) - } else { - callback(error) - } - return - } - - const endpoint = _this.sentinels[_this.currentPoint] - _this.resolve(endpoint, function (err, resolved) { - if (!_this.connecting) { - callback(new Error(CONNECTION_CLOSED_ERROR_MSG)) - return - } - if (resolved) { - debug('resolved: %s:%s', resolved.host, resolved.port) - _this.stream = createConnection(resolved) - callback(null, _this.stream) - } else { - var endpointAddress = endpoint.host + ':' + endpoint.port - var errorMsg = err - ? 'failed to connect to sentinel ' + endpointAddress + ' because ' + err.message - : 'connected to sentinel ' + endpointAddress + ' successfully, but got an invalid reply: ' + resolved - - debug(errorMsg) - - eventEmitter('sentinelError', new Error(errorMsg)) - - if (err) { - lastError = err - } - connectToNext() - } - }) - } - } - - updateSentinels (client, callback: NodeCallback) { - var _this = this - client.sentinel('sentinels', this.options.name, function (err, result) { - if (err) { - client.disconnect() - return callback(err) - } - if (Array.isArray(result)) { - for (var i = 0; i < result.length; ++i) { - var sentinel = packObject(result[i]) - var flags = sentinel.flags ? sentinel.flags.split(',') : [] - if (flags.indexOf('disconnected') === -1 && sentinel.ip && sentinel.port) { - var endpoint = { host: sentinel.ip, port: parseInt(sentinel.port, 10) } - var isDuplicate = _this.sentinels.some(bind(isSentinelEql, null, endpoint)) - if (!isDuplicate) { - debug('adding sentinel %s:%s', endpoint.host, endpoint.port) - _this.sentinels.push(endpoint) - } - } - } - debug('sentinels', _this.sentinels) - } - callback(null) - }) - } - - resolveMaster (client, callback: NodeCallback) { - var _this = this - client.sentinel('get-master-addr-by-name', this.options.name, function (err, result) { - if (err) { - client.disconnect() - return callback(err) - } - _this.updateSentinels(client, function (err) { - client.disconnect() - if (err) { - return callback(err) - } - callback(null, Array.isArray(result) ? { host: result[0], port: result[1] } : null) - }) - }) - } - - resolveSlave (client, callback: NodeCallback) { - client.sentinel('slaves', this.options.name, (err, result) => { - client.disconnect() - if (err) { - return callback(err) - } - let selectedSlave: ISentinelSlavesResponse - if (Array.isArray(result)) { - const availableSlaves: Array<{port: string, ip: string, flags?: string}> = [] - for (var i = 0; i < result.length; ++i) { - const slave: ISentinelSlavesResponse = packObject(result[i]) - if (slave.flags && !slave.flags.match(/(disconnected|s_down|o_down)/)) { - availableSlaves.push(slave) - } - } - // allow the options to prefer particular slave(s) - let {preferredSlaves} = this.options - if (typeof preferredSlaves === 'function') { - selectedSlave = preferredSlaves(availableSlaves) - } else if (preferredSlaves !== null && typeof preferredSlaves === 'object') { - const preferredSlavesArray = Array.isArray(preferredSlaves) - ? preferredSlaves - : [preferredSlaves] - - // sort by priority - preferredSlavesArray.sort((a, b) => { - // default the priority to 1 - if (!a.prio) { - a.prio = 1 - } - if (!b.prio) { - b.prio = 1 - } - - // lowest priority first - if (a.prio < b.prio) { - return -1 - } - if (a.prio > b.prio) { - return 1 - } - return 0 - }) - - // loop over preferred slaves and return the first match - for (let p = 0; p < preferredSlavesArray.length; p++) { - for (let a = 0; a < availableSlaves.length; a++) { - const slave = availableSlaves[a] - if (slave.ip === preferredSlavesArray[p].ip) { - if (slave.port === preferredSlavesArray[p].port) { - selectedSlave = slave - break - } - } - } - if (selectedSlave) { - break - } - } - // if none of the preferred slaves are available, a random available slave is returned - } - if (!selectedSlave) { - // get a random available slave - selectedSlave = sample(availableSlaves) - } - } - callback(null, selectedSlave ? {host: selectedSlave.ip, port: Number(selectedSlave.port)} : null) - }) - } - - resolve (endpoint, callback: NodeCallback) { - if (typeof Redis === 'undefined') { - Redis = require('../redis') - } - var client = new Redis({ - port: endpoint.port || 26379, - host: endpoint.host, - family: endpoint.family || (isIIpcConnectionOptions(this.options) ? undefined : this.options.family), - retryStrategy: null, - enableReadyCheck: false, - connectTimeout: this.options.connectTimeout, - dropBufferSupport: true - }) - - // ignore the errors since resolve* methods will handle them - client.on('error', noop) - - if (this.options.role === 'slave') { - this.resolveSlave(client, callback) - } else { - this.resolveMaster(client, callback) - } - } -} - -function noop (): void {} - -function isSentinelEql (a, b): boolean { - return ((a.host || '127.0.0.1') === (b.host || '127.0.0.1')) && - ((a.port || 26379) === (b.port || 26379)) -} \ No newline at end of file diff --git a/lib/connectors/SentinelConnector/SentinelIterator.ts b/lib/connectors/SentinelConnector/SentinelIterator.ts new file mode 100644 index 00000000..d46a22a0 --- /dev/null +++ b/lib/connectors/SentinelConnector/SentinelIterator.ts @@ -0,0 +1,43 @@ +import {ISentinelAddress} from './types' + +function isSentinelEql (a: ISentinelAddress, b: ISentinelAddress): boolean { + return ((a.host || '127.0.0.1') === (b.host || '127.0.0.1')) && + ((a.port || 26379) === (b.port || 26379)) +} + +export default class SentinelIterator { + private cursor: number = 0 + + constructor (private sentinels: ISentinelAddress[]) {} + + hasNext (): boolean { + return this.cursor < this.sentinels.length + } + + next (): ISentinelAddress | null { + return this.hasNext() ? this.sentinels[this.cursor++] : null + } + + reset (success: boolean): void { + if (success && this.sentinels.length > 1 && this.cursor !== 1) { + const remains = this.sentinels.slice(this.cursor - 1) + this.sentinels = remains.concat(this.sentinels.slice(0, this.cursor - 1)) + } + this.cursor = 0 + } + + add (sentinel: ISentinelAddress): boolean { + for (let i = 0; i < this.sentinels.length; i++) { + if (isSentinelEql(sentinel, this.sentinels[i])) { + return false + } + } + + this.sentinels.push(sentinel) + return true + } + + toString (): string { + return JSON.stringify(this.sentinels) + } +} \ No newline at end of file diff --git a/lib/connectors/SentinelConnector/index.ts b/lib/connectors/SentinelConnector/index.ts new file mode 100644 index 00000000..a01bd3ba --- /dev/null +++ b/lib/connectors/SentinelConnector/index.ts @@ -0,0 +1,267 @@ +import {createConnection, Socket} from 'net' +import {sample} from '../../utils/lodash' +import {CONNECTION_CLOSED_ERROR_MSG, packObject} from '../../utils/index' +import Connector, {ITcpConnectionOptions, ErrorEmitter, isIIpcConnectionOptions} from '../Connector' +import {TLSSocket} from 'tls' +import SentinelIterator from './SentinelIterator' +import { ISentinelAddress } from './types'; +const debug = require('../../utils/debug')('ioredis:SentinelConnector') + +let Redis + +interface IAddressFromResponse { + port: string, + ip: string, + flags?: string +} + +type NodeCallback = (err: Error | null, result?: T) => void +type PreferredSlaves = + ((slaves: Array) => IAddressFromResponse) | + Array<{port: string, ip: string, prio?: number}> | + {port: string, ip: string, prio?: number} + +interface ISentinelConnectionOptions extends ITcpConnectionOptions { + role: 'master' | 'slave' + name: 'string' + sentinels: Array + sentinelRetryStrategy?: (retryAttempts: number) => number + preferredSlaves?: PreferredSlaves + connectTimeout?: number +} + +export default class SentinelConnector extends Connector { + private retryAttempts: number + private sentinelIterator: SentinelIterator + + constructor (protected options: ISentinelConnectionOptions) { + super(options) + if (this.options.sentinels.length === 0) { + throw new Error('Requires at least one sentinel to connect to.') + } + if (!this.options.name) { + throw new Error('Requires the name of master.') + } + + this.sentinelIterator = new SentinelIterator(this.options.sentinels) + } + + public check (info: {role?: string}): boolean { + const roleMatches: boolean = !info.role || this.options.role === info.role + if (!roleMatches) { + debug('role invalid, expected %s, but got %s', this.options.role, info.role) + } + return roleMatches + } + + connect (callback: NodeCallback, eventEmitter: ErrorEmitter): void { + this.connecting = true + this.retryAttempts = 0 + + let lastError + const _this = this + connectToNext() + + function connectToNext() { + if (!_this.sentinelIterator.hasNext()) { + _this.sentinelIterator.reset(false) + const retryDelay = typeof _this.options.sentinelRetryStrategy === 'function' + ? _this.options.sentinelRetryStrategy(++_this.retryAttempts) + : null + + let errorMsg = typeof retryDelay !== 'number' + ? 'All sentinels are unreachable and retry is disabled.' + : `All sentinels are unreachable. Retrying from scratch after ${retryDelay}ms.` + + if (lastError) { + errorMsg += ` Last error: ${lastError.message}` + } + + debug(errorMsg) + + const error = new Error(errorMsg) + if (typeof retryDelay === 'number') { + setTimeout(connectToNext, retryDelay) + eventEmitter('error', error) + } else { + callback(error) + } + return + } + + const endpoint = _this.sentinelIterator.next() + _this.resolve(endpoint, function (err, resolved) { + if (!_this.connecting) { + callback(new Error(CONNECTION_CLOSED_ERROR_MSG)) + return + } + if (resolved) { + debug('resolved: %s:%s', resolved.host, resolved.port) + _this.stream = createConnection(resolved) + _this.sentinelIterator.reset(true) + callback(null, _this.stream) + } else { + const endpointAddress = endpoint.host + ':' + endpoint.port + const errorMsg = err + ? 'failed to connect to sentinel ' + endpointAddress + ' because ' + err.message + : 'connected to sentinel ' + endpointAddress + ' successfully, but got an invalid reply: ' + resolved + + debug(errorMsg) + + eventEmitter('sentinelError', new Error(errorMsg)) + + if (err) { + lastError = err + } + connectToNext() + } + }) + } + } + + updateSentinels (client, callback: NodeCallback) { + client.sentinel('sentinels', this.options.name, (err, result) => { + if (err) { + client.disconnect() + return callback(err) + } + if (!Array.isArray(result)) { + return callback(null) + } + + result.map(packObject).forEach(sentinel => { + const flags = sentinel.flags ? sentinel.flags.split(',') : [] + if (flags.indexOf('disconnected') === -1 && sentinel.ip && sentinel.port) { + const endpoint = addressResponseToAddress(sentinel) + if (this.sentinelIterator.add(endpoint)) { + debug('adding sentinel %s:%s', endpoint.host, endpoint.port) + } + } + }) + debug('Updated internal sentinels: %s', this.sentinelIterator) + callback(null) + }) + } + + resolveMaster (client, callback: NodeCallback) { + client.sentinel('get-master-addr-by-name', this.options.name, (err, result) => { + if (err) { + client.disconnect() + return callback(err) + } + this.updateSentinels(client, (err) => { + client.disconnect() + if (err) { + return callback(err) + } + callback(null, Array.isArray(result) ? { host: result[0], port: Number(result[1]) } : null) + }) + }) + } + + resolveSlave (client, callback: NodeCallback): void { + client.sentinel('slaves', this.options.name, (err, result) => { + client.disconnect() + if (err) { + return callback(err) + } + + if (!Array.isArray(result)) { + return callback(null, null) + } + + const availableSlaves = result.map(packObject).filter(slave => ( + slave.flags && !slave.flags.match(/(disconnected|s_down|o_down)/) + )) + + callback(null, selectPreferredSentinel(availableSlaves, this.options.preferredSlaves)) + }) + } + + resolve (endpoint, callback: NodeCallback) { + if (typeof Redis === 'undefined') { + Redis = require('../redis') + } + var client = new Redis({ + port: endpoint.port || 26379, + host: endpoint.host, + family: endpoint.family || (isIIpcConnectionOptions(this.options) ? undefined : this.options.family), + retryStrategy: null, + enableReadyCheck: false, + connectTimeout: this.options.connectTimeout, + dropBufferSupport: true + }) + + // ignore the errors since resolve* methods will handle them + client.on('error', noop) + + if (this.options.role === 'slave') { + this.resolveSlave(client, callback) + } else { + this.resolveMaster(client, callback) + } + } +} + +function selectPreferredSentinel (availableSlaves: IAddressFromResponse[], preferredSlaves?: PreferredSlaves): ISentinelAddress | null { + if (availableSlaves.length === 0) { + return null + } + + let selectedSlave: IAddressFromResponse + if (typeof preferredSlaves === 'function') { + selectedSlave = preferredSlaves(availableSlaves) + } else if (preferredSlaves !== null && typeof preferredSlaves === 'object') { + const preferredSlavesArray = Array.isArray(preferredSlaves) + ? preferredSlaves + : [preferredSlaves] + + // sort by priority + preferredSlavesArray.sort((a, b) => { + // default the priority to 1 + if (!a.prio) { + a.prio = 1 + } + if (!b.prio) { + b.prio = 1 + } + + // lowest priority first + if (a.prio < b.prio) { + return -1 + } + if (a.prio > b.prio) { + return 1 + } + return 0 + }) + + // loop over preferred slaves and return the first match + for (let p = 0; p < preferredSlavesArray.length; p++) { + for (let a = 0; a < availableSlaves.length; a++) { + const slave = availableSlaves[a] + if (slave.ip === preferredSlavesArray[p].ip) { + if (slave.port === preferredSlavesArray[p].port) { + selectedSlave = slave + break + } + } + } + if (selectedSlave) { + break + } + } + // if none of the preferred slaves are available, a random available slave is returned + if (!selectedSlave) { + selectedSlave = sample(availableSlaves) + } + + return addressResponseToAddress(selectedSlave) + } +} + +function addressResponseToAddress (input: IAddressFromResponse): ISentinelAddress { + return {host: input.ip, port: Number(input.port)} +} + +function noop (): void {} \ No newline at end of file diff --git a/lib/connectors/SentinelConnector/types.ts b/lib/connectors/SentinelConnector/types.ts new file mode 100644 index 00000000..2b3ec617 --- /dev/null +++ b/lib/connectors/SentinelConnector/types.ts @@ -0,0 +1,4 @@ +export interface ISentinelAddress { + port: number + host: string +} diff --git a/lib/utils/lodash.js b/lib/utils/lodash.js index abed9971..74e4f875 100644 --- a/lib/utils/lodash.js +++ b/lib/utils/lodash.js @@ -3,10 +3,8 @@ exports.forEach = require('lodash.foreach'); exports.pick = require('lodash.pick'); exports.defaults = require('lodash.defaults'); -exports.noop = function () {}; exports.difference = require('lodash.difference'); exports.clone = require('lodash.clone'); -exports.sample = require('lodash.sample'); exports.flatten = require('lodash.flatten'); exports.bind = require('lodash.bind'); exports.isEmpty = require('lodash.isempty'); @@ -14,3 +12,12 @@ exports.values = require('lodash.values'); exports.shuffle = require('lodash.shuffle'); exports.partial = require('lodash.partial'); exports.cloneDeep = require('lodash.clonedeep'); + +exports.noop = function () {} + +exports.sample = (array) => { + if (!array || array.length === 0) { + return undefined + } + return array[Math.floor(Math.random() * length)] +} \ No newline at end of file