import { getConfig } from '@/config/configuration' ; import setupLogger from '@/config/logger' ; import { IClusterDefinition , IInstanceDefinition } from '@/typings/typings' ; import Redis from 'ioredis' ; const { DYNOMITE_REDIS_PORT } = getConfig ( ) ; const logger = setupLogger ( module ) ; export default class DynomiteCluster { private connections : Redis . Redis [ ] ; private hostConnectionMap : { [ hostname ] : Redis . Redis | null ; } ; private instances : IInstanceDefinition [ ] ; param instances param connector constructor ( readonly cluster ) { this . connections = [ ] ; this . instances = cluster . instances ; this . hostConnectionMap = { } ; this . cluster . instances . forEach ( ( instance ) => { this . hostConnectionMap [ instance . hostname ] = null ; } ) ; } returns see getConnectionToHost public async getConnection ( ) { if ( ! this . instances || this . instances . length === 0 ) { throw new Error ( 'Unable to get connection. No instances are available in this pool' ) ; } if ( this . connections . length > 0 ) { return this . connections [ 0 ] ; } return this . _connect ( this . instances [ 0 ] ) ; } public disconnect ( ) { this . connections . forEach ( ( conn ) => conn . disconnect ( ) ) ; } param hostname public async getConnectionToHost ( hostname ) { const hostConnection = this . hostConnectionMap [ hostname ] ; if ( hostConnection ) { logger . debug ( `using existing connection in the pool for: ${ hostname }` ) ; return hostConnection ; } logger . debug ( `no connection available for: ${ hostname }. attempting to connect...` ) ; const instance = this . instances . find ( ( inst ) => inst . hostname === hostname ) ; if ( ! instance ) { throw new Error ( `Could not find hostname "${ hostname }" in the list of available instances` ) ; } return this . _connect ( instance ) ; } public executeCommandInSingleZone ( callback ) { const zones = new Set ( this . instances . map ( ( instance ) => instance . az ) ) ; logger . debug ( `executing command against the following instances: ${ zones }` ) ; const firstZone = zones . values ( ) . next ( ) . value ; return this . executeCommandInZone ( firstZone , callback ) ; } public executeCommandOnAllInstances ( callback ) { return this . _executeOnGivenInstances ( this . instances , callback ) ; } public getFirstRingMembers ( ) { const members = this . getRingMembers ( ) ; const firstAz = Object . keys ( members ) [ 0 ] ; logger . debug ( `getFirstRingMembers firstAz: ${ firstAz }` ) ; return members [ firstAz ] . map ( ( instance ) => instance . hostname ) ; } param zone param callback returns private executeCommandInZone ( zone , callback ) { logger . debug ( `executing command in zone: ${ zone }` ) ; const instancesInZone = this . instances . filter ( ( instance ) => instance . az === zone ) ; if ( instancesInZone . length === 0 ) { return Promise . reject ( new Error ( `No instances found in zone ${ zone }` ) ) ; } return this . _executeOnGivenInstances ( instancesInZone , callback ) ; } param instances param callback returns private _executeOnGivenInstances ( instances , callback ) { logger . debug ( `executing command against the following instances: ${ JSON . stringify ( instances ) }` ) ; return new Promise ( ( resolve , reject ) => { const connectionPromises = new Array < Promise < any > > ( ) ; instances . forEach ( ( instance ) => { connectionPromises . push ( this . getConnectionToHost ( instance . hostname ) ) ; } ) ; Promise . all ( connectionPromises ) . then ( ( connections ) => { const callbackPromises = connections . map ( ( connection ) => callback ( connection ) ) ; Promise . all ( callbackPromises ) . then ( resolve ) . catch ( reject ) ; } ) . catch ( ( err ) => { logger . error ( err ) ; reject ( err ) ; } ) ; } ) ; } private getRingMembers ( ) { const rings = { } as { [ az ] : IInstanceDefinition [ ] ; } ; this . instances . forEach ( ( instance ) => { const az = instance . az ; let ringMembers = rings [ az ] ; if ( ! ringMembers ) { ringMembers = [ ] ; rings [ az ] = ringMembers ; } ringMembers . push ( instance ) ; } ) ; return rings ; } param instance returns private async _connect ( instance ) { try { const fqdn = `${ instance . hostname } (${ instance . ip }:8102)` ; logger . debug ( `connecting to dynomite: ${ fqdn }` ) ; const conn = new Redis ( { port : DYNOMITE_REDIS_PORT , host : instance . ip , retryStrategy : ( times ) => { if ( times > 5 ) { logger . error ( `Reached retry limit for connection: ${ fqdn }. Removing connection from pool...` ) ; const index = this . connections . indexOf ( conn ) ; if ( index >= 0 ) { this . connections . splice ( index , 1 ) ; } return undefined ; } const delayInMs = 500 ; return delayInMs * times ; } , } ) ; conn . on ( 'error' , ( err ) => { logger . error ( `Redis connection error: ${ fqdn }` ) ; logger . debug ( err ) ; logger . debug ( err . lastNodeError ) ; } ) ; this . connections . push ( conn ) ; this . instances . forEach ( ( inst ) => { if ( instance === inst ) { this . hostConnectionMap [ inst . hostname ] = conn ; } } ) ; logger . debug ( 'got connection successfully' ) ; return conn ; } catch ( err ) { logger . error ( `Could not get a connection to instance: ${ JSON . stringify ( instance ) }` ) ; throw err ; } } }