From cab5509076942975a61ea4190dae9ea0438f75b4 Mon Sep 17 00:00:00 2001 From: James Allen Date: Thu, 7 Aug 2014 11:45:19 +0100 Subject: [PATCH 1/6] Listen on queue via BLPOP rather than pub/sub --- services/document-updater/app.coffee | 9 ++- .../app/coffee/WorkersManager.coffee | 43 +++++++++++ .../coffee/helpers/DocUpdaterClient.coffee | 2 +- .../WorkersManager/WorkersManagerTests.coffee | 73 +++++++++++++++++++ 4 files changed, 123 insertions(+), 4 deletions(-) create mode 100644 services/document-updater/app/coffee/WorkersManager.coffee create mode 100644 services/document-updater/test/unit/coffee/WorkersManager/WorkersManagerTests.coffee diff --git a/services/document-updater/app.coffee b/services/document-updater/app.coffee index a52b26f500..79db012081 100644 --- a/services/document-updater/app.coffee +++ b/services/document-updater/app.coffee @@ -3,8 +3,9 @@ http = require("http") Settings = require('settings-sharelatex') logger = require('logger-sharelatex') logger.initialize("documentupdater") -RedisManager = require('./app/js/RedisManager.js') -UpdateManager = require('./app/js/UpdateManager.js') +RedisManager = require('./app/js/RedisManager') +UpdateManager = require('./app/js/UpdateManager') +WorkersManager = require('./app/js/WorkersManager') Keys = require('./app/js/RedisKeyBuilder') redis = require('redis') Errors = require "./app/js/Errors" @@ -32,7 +33,9 @@ rclient.on "message", (channel, doc_key) -> UpdateManager.processOutstandingUpdatesWithLock project_id, doc_id, (error) -> logger.error err: error, project_id: project_id, doc_id: doc_id, "error processing update" if error? else - logger.log project_id: project_id, doc_id: doc_id, "ignoring incoming update" + logger.log project_id: project_id, doc_id: doc_id, "ignoring incoming update" + +WorkersManager.createAndStartWorkers(Settings.workerCount || 10) UpdateManager.resumeProcessing() diff --git a/services/document-updater/app/coffee/WorkersManager.coffee b/services/document-updater/app/coffee/WorkersManager.coffee new file mode 100644 index 0000000000..bfc2d705e4 --- /dev/null +++ b/services/document-updater/app/coffee/WorkersManager.coffee @@ -0,0 +1,43 @@ +Settings = require('settings-sharelatex') +logger = require('logger-sharelatex') +Keys = require('./RedisKeyBuilder') +redis = require('redis') +UpdateManager = require('./UpdateManager') + +module.exports = WorkersManager = + createWorker: () -> + redisConf = Settings.redis.web + client = redis.createClient(redisConf.port, redisConf.host) + client.auth(redisConf.password) + + worker = { + client: client + waitForAndProcessUpdate: (callback = (error) ->) -> + worker.client.blpop "pending-updates-list", 0, (error, result) -> + return callback(error) if error? + return callback() if !result? + [list_name, doc_key] = result + [project_id, doc_id] = Keys.splitProjectIdAndDocId(doc_key) + UpdateManager.processOutstandingUpdatesWithLock project_id, doc_id, (error) -> + logger.error err: error, project_id: project_id, doc_id: doc_id, "error processing update" if error? + return callback(error) if error? + return callback() + + run: () -> + return if Settings.shuttingDown + worker.waitForAndProcessUpdate (error) => + if error? + logger.error err: error, "Error in worker process, waiting 1 second before continuing" + setTimeout () -> + worker.run() + , 1000 + else + worker.run() + } + + return worker + + createAndStartWorkers: (number) -> + for i in [1..number] + worker = WorkersManager.createWorker() + worker.run() \ No newline at end of file diff --git a/services/document-updater/test/acceptance/coffee/helpers/DocUpdaterClient.coffee b/services/document-updater/test/acceptance/coffee/helpers/DocUpdaterClient.coffee index 53fc274f58..e9fd0b0c34 100644 --- a/services/document-updater/test/acceptance/coffee/helpers/DocUpdaterClient.coffee +++ b/services/document-updater/test/acceptance/coffee/helpers/DocUpdaterClient.coffee @@ -14,7 +14,7 @@ module.exports = DocUpdaterClient = doc_key = "#{project_id}:#{doc_id}" rclient.sadd "DocsWithPendingUpdates", doc_key, (error) -> return callback(error) if error? - rclient.publish "pending-updates", doc_key, callback + rclient.rpush "pending-updates-list", doc_key, callback sendUpdates: (project_id, doc_id, updates, callback = (error) ->) -> DocUpdaterClient.preloadDoc project_id, doc_id, (error) -> diff --git a/services/document-updater/test/unit/coffee/WorkersManager/WorkersManagerTests.coffee b/services/document-updater/test/unit/coffee/WorkersManager/WorkersManagerTests.coffee new file mode 100644 index 0000000000..4a86d7cdd0 --- /dev/null +++ b/services/document-updater/test/unit/coffee/WorkersManager/WorkersManagerTests.coffee @@ -0,0 +1,73 @@ +sinon = require('sinon') +chai = require('chai') +should = chai.should() +modulePath = "../../../../app/js/WorkersManager.js" +SandboxedModule = require('sandboxed-module') + +describe "WorkersManager", -> + beforeEach -> + @WorkersManager = SandboxedModule.require modulePath, requires: + "./UpdateManager" : @UpdateManager = {} + "logger-sharelatex": @logger = { log: sinon.stub() } + "settings-sharelatex": @settings = + redis: + web: {} + "redis": @redis = {} + @callback = sinon.stub() + + describe "each worker", -> + beforeEach -> + @client = + auth: sinon.stub() + @redis.createClient = sinon.stub().returns @client + + @worker = @WorkersManager.createWorker() + + it "should create a new redis client", -> + @redis.createClient.called.should.equal true + + describe "waitForAndProcessUpdate", -> + beforeEach -> + @project_id = "project-id-123" + @doc_id = "doc-id-123" + @doc_key = "#{@project_id}:#{@doc_id}" + @client.blpop = sinon.stub().callsArgWith(2, null, ["pending-updates-list", @doc_key]) + @UpdateManager.processOutstandingUpdatesWithLock = sinon.stub().callsArg(2) + + @worker.waitForAndProcessUpdate @callback + + it "should call redis with BLPOP", -> + @client.blpop + .calledWith("pending-updates-list", 0) + .should.equal true + + it "should call processOutstandingUpdatesWithLock", -> + @UpdateManager.processOutstandingUpdatesWithLock + .calledWith(@project_id, @doc_id) + .should.equal true + + it "should call the callback", -> + @callback.called.should.equal true + + describe "run", -> + it "should call waitForAndProcessUpdate until shutting down", (done) -> + callCount = 0 + @worker.waitForAndProcessUpdate = (callback = (error) ->) => + callCount++ + if callCount == 3 + @settings.shuttingDown = true + setTimeout () -> + callback() + , 10 + sinon.spy @worker, "waitForAndProcessUpdate" + + + @worker.run() + + setTimeout () => + @worker.waitForAndProcessUpdate.callCount.should.equal 3 + done() + , 100 + + + \ No newline at end of file From 773e17df66f702e600ef9b57adc4808aef0c903c Mon Sep 17 00:00:00 2001 From: James Allen Date: Thu, 7 Aug 2014 13:16:11 +0100 Subject: [PATCH 2/6] Send worker idle waiting times to statsd --- services/document-updater/app/coffee/WorkersManager.coffee | 3 +++ 1 file changed, 3 insertions(+) diff --git a/services/document-updater/app/coffee/WorkersManager.coffee b/services/document-updater/app/coffee/WorkersManager.coffee index bfc2d705e4..b421422f29 100644 --- a/services/document-updater/app/coffee/WorkersManager.coffee +++ b/services/document-updater/app/coffee/WorkersManager.coffee @@ -3,6 +3,7 @@ logger = require('logger-sharelatex') Keys = require('./RedisKeyBuilder') redis = require('redis') UpdateManager = require('./UpdateManager') +Metrics = require('./Metrics') module.exports = WorkersManager = createWorker: () -> @@ -13,7 +14,9 @@ module.exports = WorkersManager = worker = { client: client waitForAndProcessUpdate: (callback = (error) ->) -> + timer = new Metrics.Timer "worker.waiting" worker.client.blpop "pending-updates-list", 0, (error, result) -> + timer.done() return callback(error) if error? return callback() if !result? [list_name, doc_key] = result From 9da118c809c08904b7469acc303aafc6caa93983 Mon Sep 17 00:00:00 2001 From: James Allen Date: Mon, 11 Aug 2014 14:16:05 +0100 Subject: [PATCH 3/6] Process updates in the background so that workers only act as dispatchers --- services/document-updater/app.coffee | 4 ++-- ...sManager.coffee => DispatchManager.coffee} | 22 +++++++++---------- .../DispatchManagerTests.coffee} | 20 ++++++++--------- 3 files changed, 22 insertions(+), 24 deletions(-) rename services/document-updater/app/coffee/{WorkersManager.coffee => DispatchManager.coffee} (70%) rename services/document-updater/test/unit/coffee/{WorkersManager/WorkersManagerTests.coffee => DispatchManager/DispatchManagerTests.coffee} (72%) diff --git a/services/document-updater/app.coffee b/services/document-updater/app.coffee index 79db012081..e246d366e8 100644 --- a/services/document-updater/app.coffee +++ b/services/document-updater/app.coffee @@ -5,7 +5,7 @@ logger = require('logger-sharelatex') logger.initialize("documentupdater") RedisManager = require('./app/js/RedisManager') UpdateManager = require('./app/js/UpdateManager') -WorkersManager = require('./app/js/WorkersManager') +DispatchManager = require('./app/js/DispatchManager') Keys = require('./app/js/RedisKeyBuilder') redis = require('redis') Errors = require "./app/js/Errors" @@ -35,7 +35,7 @@ rclient.on "message", (channel, doc_key) -> else logger.log project_id: project_id, doc_id: doc_id, "ignoring incoming update" -WorkersManager.createAndStartWorkers(Settings.workerCount || 10) +DispatchManager.createAndStartDispatchers(Settings.dispatcherCount || 10) UpdateManager.resumeProcessing() diff --git a/services/document-updater/app/coffee/WorkersManager.coffee b/services/document-updater/app/coffee/DispatchManager.coffee similarity index 70% rename from services/document-updater/app/coffee/WorkersManager.coffee rename to services/document-updater/app/coffee/DispatchManager.coffee index b421422f29..afb1a5d176 100644 --- a/services/document-updater/app/coffee/WorkersManager.coffee +++ b/services/document-updater/app/coffee/DispatchManager.coffee @@ -5,15 +5,15 @@ redis = require('redis') UpdateManager = require('./UpdateManager') Metrics = require('./Metrics') -module.exports = WorkersManager = - createWorker: () -> +module.exports = DispatchManager = + createDispatcher: () -> redisConf = Settings.redis.web client = redis.createClient(redisConf.port, redisConf.host) client.auth(redisConf.password) worker = { client: client - waitForAndProcessUpdate: (callback = (error) ->) -> + _waitForUpdateThenDispatchWorker: (callback = (error) ->) -> timer = new Metrics.Timer "worker.waiting" worker.client.blpop "pending-updates-list", 0, (error, result) -> timer.done() @@ -21,26 +21,24 @@ module.exports = WorkersManager = return callback() if !result? [list_name, doc_key] = result [project_id, doc_id] = Keys.splitProjectIdAndDocId(doc_key) + # Dispatch this in the background UpdateManager.processOutstandingUpdatesWithLock project_id, doc_id, (error) -> logger.error err: error, project_id: project_id, doc_id: doc_id, "error processing update" if error? - return callback(error) if error? - return callback() + callback() run: () -> return if Settings.shuttingDown - worker.waitForAndProcessUpdate (error) => + worker._waitForUpdateThenDispatchWorker (error) => if error? - logger.error err: error, "Error in worker process, waiting 1 second before continuing" - setTimeout () -> - worker.run() - , 1000 + logger.error err: error, "Error in worker process" + throw error else worker.run() } return worker - createAndStartWorkers: (number) -> + createAndStartDispatchers: (number) -> for i in [1..number] - worker = WorkersManager.createWorker() + worker = DispatchManager.createDispatcher() worker.run() \ No newline at end of file diff --git a/services/document-updater/test/unit/coffee/WorkersManager/WorkersManagerTests.coffee b/services/document-updater/test/unit/coffee/DispatchManager/DispatchManagerTests.coffee similarity index 72% rename from services/document-updater/test/unit/coffee/WorkersManager/WorkersManagerTests.coffee rename to services/document-updater/test/unit/coffee/DispatchManager/DispatchManagerTests.coffee index 4a86d7cdd0..3d4be1cfec 100644 --- a/services/document-updater/test/unit/coffee/WorkersManager/WorkersManagerTests.coffee +++ b/services/document-updater/test/unit/coffee/DispatchManager/DispatchManagerTests.coffee @@ -1,12 +1,12 @@ sinon = require('sinon') chai = require('chai') should = chai.should() -modulePath = "../../../../app/js/WorkersManager.js" +modulePath = "../../../../app/js/DispatchManager.js" SandboxedModule = require('sandboxed-module') -describe "WorkersManager", -> +describe "DispatchManager", -> beforeEach -> - @WorkersManager = SandboxedModule.require modulePath, requires: + @DispatchManager = SandboxedModule.require modulePath, requires: "./UpdateManager" : @UpdateManager = {} "logger-sharelatex": @logger = { log: sinon.stub() } "settings-sharelatex": @settings = @@ -21,12 +21,12 @@ describe "WorkersManager", -> auth: sinon.stub() @redis.createClient = sinon.stub().returns @client - @worker = @WorkersManager.createWorker() + @worker = @DispatchManager.createDispatcher() it "should create a new redis client", -> @redis.createClient.called.should.equal true - describe "waitForAndProcessUpdate", -> + describe "_waitForUpdateThenDispatchWorker", -> beforeEach -> @project_id = "project-id-123" @doc_id = "doc-id-123" @@ -34,7 +34,7 @@ describe "WorkersManager", -> @client.blpop = sinon.stub().callsArgWith(2, null, ["pending-updates-list", @doc_key]) @UpdateManager.processOutstandingUpdatesWithLock = sinon.stub().callsArg(2) - @worker.waitForAndProcessUpdate @callback + @worker._waitForUpdateThenDispatchWorker @callback it "should call redis with BLPOP", -> @client.blpop @@ -50,22 +50,22 @@ describe "WorkersManager", -> @callback.called.should.equal true describe "run", -> - it "should call waitForAndProcessUpdate until shutting down", (done) -> + it "should call _waitForUpdateThenDispatchWorker until shutting down", (done) -> callCount = 0 - @worker.waitForAndProcessUpdate = (callback = (error) ->) => + @worker._waitForUpdateThenDispatchWorker = (callback = (error) ->) => callCount++ if callCount == 3 @settings.shuttingDown = true setTimeout () -> callback() , 10 - sinon.spy @worker, "waitForAndProcessUpdate" + sinon.spy @worker, "_waitForUpdateThenDispatchWorker" @worker.run() setTimeout () => - @worker.waitForAndProcessUpdate.callCount.should.equal 3 + @worker._waitForUpdateThenDispatchWorker.callCount.should.equal 3 done() , 100 From 48c5f66af38c53d1065e57cf67be38ed1a7c992a Mon Sep 17 00:00:00 2001 From: James Allen Date: Fri, 15 Aug 2014 11:06:31 +0100 Subject: [PATCH 4/6] Remove acceptance test from travis --- services/document-updater/.travis.yml | 8 -------- 1 file changed, 8 deletions(-) diff --git a/services/document-updater/.travis.yml b/services/document-updater/.travis.yml index 6adc08643a..5c181b1569 100644 --- a/services/document-updater/.travis.yml +++ b/services/document-updater/.travis.yml @@ -10,13 +10,5 @@ install: - npm install - grunt install -before_script: - - grunt forever:app:start - script: - grunt test:unit - - grunt test:acceptance - -services: - - redis-server - - mongodb From 3ffe96f5486723c17ad22d58472b2a4bbf0b8b1b Mon Sep 17 00:00:00 2001 From: James Allen Date: Tue, 19 Aug 2014 14:06:56 +0100 Subject: [PATCH 5/6] Lock down module versions --- services/document-updater/package.json | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/services/document-updater/package.json b/services/document-updater/package.json index 25feab87d8..da4f09d5d8 100644 --- a/services/document-updater/package.json +++ b/services/document-updater/package.json @@ -1,6 +1,11 @@ { "name": "document-updater-sharelatex", "version": "0.0.1", + "description": "An API for applying incoming updates to documents in real-time", + "repository": { + "type": "git", + "url": "https://github.com/sharelatex/document-updater-sharelatex.git" + }, "dependencies": { "express": "3.3.4", "underscore": "1.2.2", @@ -12,9 +17,9 @@ "async": "", "lynx": "0.0.11", "coffee-script": "1.4.0", - "settings-sharelatex": "git+https://github.com/sharelatex/settings-sharelatex.git#master", - "logger-sharelatex": "git+https://github.com/sharelatex/logger-sharelatex.git#master", - "metrics-sharelatex": "git+https://github.com/sharelatex/metrics-sharelatex.git#master", + "settings-sharelatex": "git+https://github.com/sharelatex/settings-sharelatex.git#v1.0.0", + "logger-sharelatex": "git+https://github.com/sharelatex/logger-sharelatex.git#v1.0.0", + "metrics-sharelatex": "git+https://github.com/sharelatex/metrics-sharelatex.git#v1.0.0", "sinon": "~1.5.2", "mongojs": "0.9.11" }, From a4a0eef15383ff5d347269cbdb50c80fc40a7599 Mon Sep 17 00:00:00 2001 From: James Allen Date: Tue, 19 Aug 2014 14:07:14 +0100 Subject: [PATCH 6/6] Release version 0.1.0 --- services/document-updater/package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/services/document-updater/package.json b/services/document-updater/package.json index da4f09d5d8..417627bbcf 100644 --- a/services/document-updater/package.json +++ b/services/document-updater/package.json @@ -1,6 +1,6 @@ { "name": "document-updater-sharelatex", - "version": "0.0.1", + "version": "0.1.0", "description": "An API for applying incoming updates to documents in real-time", "repository": { "type": "git",