diff --git a/services/real-time/app/coffee/DocumentUpdaterController.coffee b/services/real-time/app/coffee/DocumentUpdaterController.coffee index 490902a837..b3706f47ba 100644 --- a/services/real-time/app/coffee/DocumentUpdaterController.coffee +++ b/services/real-time/app/coffee/DocumentUpdaterController.coffee @@ -1,7 +1,6 @@ logger = require "logger-sharelatex" settings = require 'settings-sharelatex' -redis = require("redis-sharelatex") -rclient = redis.createClient(settings.redis.pubsub) +RedisClientManager = require "./RedisClientManager" SafeJsonParse = require "./SafeJsonParse" EventLogger = require "./EventLogger" HealthCheckManager = require "./HealthCheckManager" @@ -12,13 +11,19 @@ MESSAGE_SIZE_LOG_LIMIT = 1024 * 1024 # 1Mb module.exports = DocumentUpdaterController = # DocumentUpdaterController is responsible for updates that come via Redis # Pub/Sub from the document updater. + rclientList: RedisClientManager.createClientList(settings.redis.pubsub, settings.redis.unusedpubsub) listenForUpdatesFromDocumentUpdater: (io) -> - rclient.subscribe "applied-ops" - rclient.on "message", (channel, message) -> - metrics.inc "rclient", 0.001 # global event rate metric - EventLogger.debugEvent(channel, message) if settings.debugEvents > 0 - DocumentUpdaterController._processMessageFromDocumentUpdater(io, channel, message) + logger.log {rclients: @rclientList.length}, "listening for applied-ops events" + for rclient, i in @rclientList + rclient.subscribe "applied-ops" + rclient.on "message", (channel, message) -> + metrics.inc "rclient", 0.001 # global event rate metric + EventLogger.debugEvent(channel, message) if settings.debugEvents > 0 + DocumentUpdaterController._processMessageFromDocumentUpdater(io, channel, message) + do (i) -> + rclient.on "message", () -> + metrics.inc "rclient-#{i}", 0.001 # per client event rate metric _processMessageFromDocumentUpdater: (io, channel, message) -> SafeJsonParse.parse message, (error, message) -> diff --git a/services/real-time/app/coffee/RedisClientManager.coffee b/services/real-time/app/coffee/RedisClientManager.coffee new file mode 100644 index 0000000000..1d573df9b8 --- /dev/null +++ b/services/real-time/app/coffee/RedisClientManager.coffee @@ -0,0 +1,18 @@ +redis = require("redis-sharelatex") +logger = require 'logger-sharelatex' + +module.exports = RedisClientManager = + createClientList: (configs...) -> + # create a dynamic list of redis clients, excluding any configurations which are not defined + clientList = for x in configs when x? + redisType = if x.cluster? + "cluster" + else if x.sentinels? + "sentinel" + else if x.host? + "single" + else + "unknown" + logger.log {redis: redisType}, "creating redis client" + redis.createClient(x) + return clientList \ No newline at end of file diff --git a/services/real-time/app/coffee/WebsocketLoadBalancer.coffee b/services/real-time/app/coffee/WebsocketLoadBalancer.coffee index e4ed673f8a..10f0c0c646 100644 --- a/services/real-time/app/coffee/WebsocketLoadBalancer.coffee +++ b/services/real-time/app/coffee/WebsocketLoadBalancer.coffee @@ -1,15 +1,13 @@ Settings = require 'settings-sharelatex' logger = require 'logger-sharelatex' -redis = require("redis-sharelatex") +RedisClientManager = require "./RedisClientManager" SafeJsonParse = require "./SafeJsonParse" -rclientPub = redis.createClient(Settings.redis.pubsub) -rclientSub = redis.createClient(Settings.redis.pubsub) EventLogger = require "./EventLogger" HealthCheckManager = require "./HealthCheckManager" module.exports = WebsocketLoadBalancer = - rclientPub: rclientPub - rclientSub: rclientSub + rclientPubList: RedisClientManager.createClientList(Settings.redis.pubsub) + rclientSubList: RedisClientManager.createClientList(Settings.redis.pubsub, Settings.redis.unusedpubsub) emitToRoom: (room_id, message, payload...) -> if !room_id? @@ -20,16 +18,20 @@ module.exports = WebsocketLoadBalancer = message: message payload: payload logger.log {room_id, message, payload, length: data.length}, "emitting to room" - @rclientPub.publish "editor-events", data + for rclientPub in @rclientPubList + rclientPub.publish "editor-events", data emitToAll: (message, payload...) -> @emitToRoom "all", message, payload... listenForEditorEvents: (io) -> - @rclientSub.subscribe "editor-events" - @rclientSub.on "message", (channel, message) -> - EventLogger.debugEvent(channel, message) if Settings.debugEvents > 0 - WebsocketLoadBalancer._processEditorEvent io, channel, message + logger.log {rclients: @rclientPubList.length}, "publishing editor events" + logger.log {rclients: @rclientSubList.length}, "listening for editor events" + for rclientSub in @rclientSubList + rclientSub.subscribe "editor-events" + rclientSub.on "message", (channel, message) -> + EventLogger.debugEvent(channel, message) if Settings.debugEvents > 0 + WebsocketLoadBalancer._processEditorEvent io, channel, message _processEditorEvent: (io, channel, message) -> SafeJsonParse.parse message, (error, message) -> diff --git a/services/real-time/test/unit/coffee/DocumentUpdaterControllerTests.coffee b/services/real-time/test/unit/coffee/DocumentUpdaterControllerTests.coffee index 5ed274b36c..24551396d9 100644 --- a/services/real-time/test/unit/coffee/DocumentUpdaterControllerTests.coffee +++ b/services/real-time/test/unit/coffee/DocumentUpdaterControllerTests.coffee @@ -10,6 +10,7 @@ describe "DocumentUpdaterController", -> @doc_id = "doc-id-123" @callback = sinon.stub() @io = { "mock": "socket.io" } + @rclient = [] @EditorUpdatesController = SandboxedModule.require modulePath, requires: "logger-sharelatex": @logger = { error: sinon.stub(), log: sinon.stub(), warn: sinon.stub() } "settings-sharelatex": @settings = @@ -17,9 +18,11 @@ describe "DocumentUpdaterController", -> documentupdater: key_schema: pendingUpdates: ({doc_id}) -> "PendingUpdates:#{doc_id}" - "redis-sharelatex" : - createClient: () => - @rclient = {} + pubsub: null + "redis-sharelatex" : @redis = + createClient: (name) => + @rclient.push(rclientStub = {name:name}) + return rclientStub "./SafeJsonParse": @SafeJsonParse = parse: (data, cb) => cb null, JSON.parse(data) "./EventLogger": @EventLogger = {checkEventOrder: sinon.stub()} @@ -28,15 +31,23 @@ describe "DocumentUpdaterController", -> describe "listenForUpdatesFromDocumentUpdater", -> beforeEach -> - @rclient.subscribe = sinon.stub() - @rclient.on = sinon.stub() + @rclient.length = 0 # clear any existing clients + @EditorUpdatesController.rclientList = [@redis.createClient("first"), @redis.createClient("second")] + @rclient[0].subscribe = sinon.stub() + @rclient[0].on = sinon.stub() + @rclient[1].subscribe = sinon.stub() + @rclient[1].on = sinon.stub() @EditorUpdatesController.listenForUpdatesFromDocumentUpdater() it "should subscribe to the doc-updater stream", -> - @rclient.subscribe.calledWith("applied-ops").should.equal true + @rclient[0].subscribe.calledWith("applied-ops").should.equal true it "should register a callback to handle updates", -> - @rclient.on.calledWith("message").should.equal true + @rclient[0].on.calledWith("message").should.equal true + + it "should subscribe to any additional doc-updater stream", -> + @rclient[1].subscribe.calledWith("applied-ops").should.equal true + @rclient[1].on.calledWith("message").should.equal true describe "_processMessageFromDocumentUpdater", -> describe "with bad JSON", -> diff --git a/services/real-time/test/unit/coffee/WebsocketLoadBalancerTests.coffee b/services/real-time/test/unit/coffee/WebsocketLoadBalancerTests.coffee index e786038ab7..14df2df851 100644 --- a/services/real-time/test/unit/coffee/WebsocketLoadBalancerTests.coffee +++ b/services/real-time/test/unit/coffee/WebsocketLoadBalancerTests.coffee @@ -7,19 +7,20 @@ describe "WebsocketLoadBalancer", -> beforeEach -> @rclient = {} @WebsocketLoadBalancer = SandboxedModule.require modulePath, requires: - "redis-sharelatex": - createClient: () => @rclient + "./RedisClientManager": + createClientList: () => [] "logger-sharelatex": @logger = { log: sinon.stub(), error: sinon.stub() } "./SafeJsonParse": @SafeJsonParse = parse: (data, cb) => cb null, JSON.parse(data) "./EventLogger": {checkEventOrder: sinon.stub()} "./HealthCheckManager": {check: sinon.stub()} @io = {} - @WebsocketLoadBalancer.rclientPub = publish: sinon.stub() - @WebsocketLoadBalancer.rclientSub = + @WebsocketLoadBalancer.rclientPubList = [{publish: sinon.stub()}] + @WebsocketLoadBalancer.rclientSubList = [{ subscribe: sinon.stub() on: sinon.stub() - + }] + @room_id = "room-id" @message = "message-to-editor" @payload = ["argument one", 42] @@ -29,7 +30,7 @@ describe "WebsocketLoadBalancer", -> @WebsocketLoadBalancer.emitToRoom(@room_id, @message, @payload...) it "should publish the message to redis", -> - @WebsocketLoadBalancer.rclientPub.publish + @WebsocketLoadBalancer.rclientPubList[0].publish .calledWith("editor-events", JSON.stringify( room_id: @room_id, message: @message @@ -53,12 +54,12 @@ describe "WebsocketLoadBalancer", -> @WebsocketLoadBalancer.listenForEditorEvents() it "should subscribe to the editor-events channel", -> - @WebsocketLoadBalancer.rclientSub.subscribe + @WebsocketLoadBalancer.rclientSubList[0].subscribe .calledWith("editor-events") .should.equal true it "should process the events with _processEditorEvent", -> - @WebsocketLoadBalancer.rclientSub.on + @WebsocketLoadBalancer.rclientSubList[0].on .calledWith("message", sinon.match.func) .should.equal true