Merge pull request #1 from sharelatex/ja-redis-cluster-refactor

Include ioredis as a driver to be able to handle cluster configs

libraries/redis-wrapper/index.coffee

@@ -13,53 +13,71 @@ module.exports = RedisSharelatex =
 			delete standardOpts.endpoints
 			delete standardOpts.masterName
 			client = require("redis-sentinel").createClient opts.endpoints, opts.masterName, standardOpts
+			client.healthCheck = RedisSharelatex.singleInstanceHealthCheckBuilder(client)
+		else if opts.cluster?
+			Redis = require("ioredis")
+			client = new Redis.Cluster(opts.cluster)
+			client.healthCheck = RedisSharelatex.clusterHealthCheckBuilder(client)
+			RedisSharelatex._monkeyPatchIoredisExec(client)
 		else
 			standardOpts = _.clone(opts)
 			delete standardOpts.port
 			delete standardOpts.host
 			client = require("redis").createClient opts.port, opts.host, standardOpts
+			client.healthCheck = RedisSharelatex.singleInstanceHealthCheckBuilder(client)
 		return client
-
-
-	activeHealthCheckRedis: (connectionInfo)->
-		sub = RedisSharelatex.createClient(connectionInfo)
-		pub = RedisSharelatex.createClient(connectionInfo)
+	
+	HEARTBEAT_TIMEOUT: 2000
+	singleInstanceHealthCheckBuilder: (client) ->
+		healthCheck = (callback) ->
+			RedisSharelatex._checkClient(client, callback)
+		return healthCheck
+	
+	clusterHealthCheckBuilder: (client) ->
+		healthCheck = (callback) ->
+			jobs = client.rclient.nodes("all").map (node) =>
+				(cb) => RedisSharelatex._checkClient(node, cb)
+			async.parallel jobs, callback
 		
-		redisIsOk = true
-		lastPingMessage = ""
-		heartbeatInterval = 2000 #ms
-		isAliveTimeout = 10000 #ms
+		return healthCheck
+	
+	_checkClient: (client, callback) ->
+		callback = _.once(callback)
+		timer = setTimeout () ->
+			error = new Error("redis client ping check timed out")
+			console.error {
+				err: error,
+				key: client.options?.key # only present for cluster
+			}, "client timed out"
+			callback(error)
+		, RedisSharelatex.HEARTBEAT_TIMEOUT
+		client.ping (err) ->
+			clearTimeout timer
+			callback(err)
 		
-		id = require("crypto").pseudoRandomBytes(16).toString("hex")
-		heartbeatChannel = "heartbeat-#{id}"
-		lastHeartbeat = Date.now()
+	_monkeyPatchIoredisExec: (client) ->
+		_multi = client.multi
+		client.multi = (args...) ->
+			multi = _multi.call(client, args...)
+			_exec = multi.exec
+			multi.exec = (args..., callback) ->
+				_exec.call multi, args..., (error, result) ->
+					# ioredis exec returns an results like:
+					# [ [null, 42], [null, "foo"] ]
+					# where the first entries in each 2-tuple are
+					# presumably errors for each individual command,
+					# and the second entry is the result. We need to transform
+					# this into the same result as the old redis driver:
+					# [ 42, "foo" ]
+					filtered_result = []
+					for entry in result or []
+						if entry[0]?
+							return callback(entry[0])
+						else
+							filtered_result.push entry[1]
+					callback error, filtered_result
+			return multi
+	
 		
-		sub.subscribe heartbeatChannel, (error) ->
-			if error?
-				console.error "ERROR: failed to subscribe to #{heartbeatChannel} channel", error
-		sub.on "message", (channel, message) ->
-			if lastPingMessage == message #we got the same message twice
-				redisIsOk = false
-			lastPingMessage = message
-			if channel == heartbeatChannel
-				lastHeartbeat = Date.now()
-		
-		setInterval ->
-			message = "ping:#{Date.now()}"
-			pub.publish heartbeatChannel, message
-		, heartbeatInterval
-
-		isAlive = ->
-			timeSinceLastHeartbeat = Date.now() - lastHeartbeat
-			if !redisIsOk
-				return false
-			else if timeSinceLastHeartbeat > isAliveTimeout
-				console.error "heartbeat from redis timed out"
-				redisIsOk = false
-				return false
-			else
-				return true
+	
 
-		return {
-			isAlive:isAlive
-		}

libraries/redis-wrapper/package.json

@@ -1,9 +1,9 @@
 {
   "name": "redis-sharelatex",
-  "version": "0.0.9",
-  "description": "redis wrapper for node which will either use sentinal or normal redis",
+  "version": "1.0.0",
+  "description": "Redis wrapper for node which will either use cluster, sentinal, or single instance redis",
   "main": "index.js",
-  "author": "henry oswald @ sharelatex",
+  "author": "ShareLaTeX",
   "license": "ISC",
   "dependencies": {
     "chai": "1.9.1",
@@ -11,6 +11,7 @@
     "grunt": "0.4.5",
     "grunt-contrib-coffee": "0.11.1",
     "grunt-mocha-test": "0.12.0",
+    "ioredis": "^2.5.0",
     "mocha": "1.21.4",
     "redis": "0.12.1",
     "redis-sentinel": "0.1.1",

libraries/redis-wrapper/test.coffee

@@ -21,15 +21,17 @@ describe "index", ->
 		@redis = SandboxedModule.require modulePath, requires:
 			"redis-sentinel":@sentinel
 			"redis":@normalRedis
+			"ioredis": @ioredis =
+				Cluster: class Cluster
+					constructor: (@config) ->
 		@auth_pass = "1234 pass"
 		@endpoints = [
 				{host: '127.0.0.1', port: 26379},
 				{host: '127.0.0.1', port: 26380}
 			]
-	describe "sentinel", ->
 
+	describe "sentinel", ->
 		beforeEach ->
-			
 			@masterName = "my master"
 			@sentinelOptions =
 				endpoints:@endpoints
@@ -44,11 +46,10 @@ describe "index", ->
 
 		it "should pass the options correctly though", ->
 			client = @redis.createClient @sentinelOptions
-			@sentinel.createClient.calledWith(@endpoints, @masterName, auth_pass:@auth_pass).should.equal true
+			@sentinel.createClient.calledWith(@endpoints, @masterName, {auth_pass:@auth_pass, retry_max_delay: 5000}).should.equal true
 			client.should.equal @sentinelClient
 
 	describe "normal redis", ->
-
 		beforeEach ->
 			@standardOpts =
 				auth_pass: @auth_pass
@@ -63,9 +64,37 @@ describe "index", ->
 
 		it "should use the normal redis driver if a non array is passed", ->
 			client = @redis.createClient @standardOpts
-			@normalRedis.createClient.calledWith(@standardOpts.port, @standardOpts.host, auth_pass:@auth_pass).should.equal true
+			@normalRedis.createClient.calledWith(@standardOpts.port, @standardOpts.host, {auth_pass:@auth_pass, retry_max_delay: 5000}).should.equal true
 
+	describe "cluster", ->
+		beforeEach ->
+			@cluster = [{"mock": "cluster"}, { "mock": "cluster2"}]
 
+		it "should pass the options correctly though", ->
+			client = @redis.createClient cluster: @cluster
+			assert(client instanceof @ioredis.Cluster)
+			client.config.should.deep.equal @cluster
+
+	describe "monkey patch ioredis exec", ->
+		beforeEach ->
+			@callback = sinon.stub()
+			@results = []
+			@multiOrig = { exec: sinon.stub().yields(null, @results)}
+			@client = { multi: sinon.stub().returns(@multiOrig) }
+			@redis._monkeyPatchIoredisExec(@client)
+			@multi = @client.multi()
+
+		it "should return the old redis format for an array", ->
+			@results[0] = [null, 42]
+			@results[1] = [null, "foo"]
+			@multi.exec @callback
+			@callback.calledWith(null, [42, "foo"]).should.equal true
+
+		it "should return the old redis format when there is an error", ->
+			@results[0] = [null, 42]
+			@results[1] = ["error", "foo"]
+			@multi.exec @callback
+			@callback.calledWith("error").should.equal true
 
 	describe "setting the password", ->
 		beforeEach ->
@@ -81,10 +110,10 @@ describe "index", ->
 
 		it "should set the auth_pass from password if password exists for normal redis", ->
 			client = @redis.createClient @standardOpts
-			@normalRedis.createClient.calledWith(@standardOpts.port, @standardOpts.host, auth_pass:@auth_pass).should.equal true
+			@normalRedis.createClient.calledWith(@standardOpts.port, @standardOpts.host, {auth_pass:@auth_pass, retry_max_delay: 5000}).should.equal true
 		
 		it "should set the auth_pass from password if password exists for sentinal", ->
 			client = @redis.createClient @sentinelOptions
-			@sentinel.createClient.calledWith(@endpoints, @masterName, auth_pass:@auth_pass).should.equal true
+			@sentinel.createClient.calledWith(@endpoints, @masterName, {auth_pass:@auth_pass, retry_max_delay: 5000}).should.equal true