alois/Verso
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #6 from heukirne/s3-archive

Browse Source 
Add S3 archive track changes feature
This commit is contained in:
Brian Gough
2015-09-21 11:25:06 +01:00
parent 0e627c92d8 aa66c5ee8c
commit a10dc4f898
22 changed files with 992 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/track-changes/app/coffee/DocArchiveManager.coffee
+74
View File
@@ -0,0 +1,74 @@
MongoManager = require "./MongoManager"
MongoAWS = require "./MongoAWS"
LockManager = require "./LockManager"
DocstoreHandler = require "./DocstoreHandler"
logger = require "logger-sharelatex"
_ = require "underscore"
async = require "async"
settings = require("settings-sharelatex")
# increase lock timeouts because archiving can be slow
LockManager.LOCK_TEST_INTERVAL = 500 # 500ms between each test of the lock
LockManager.MAX_LOCK_WAIT_TIME = 30000 # 30s maximum time to spend trying to get the lock
LockManager.LOCK_TTL = 30 # seconds
module.exports = DocArchiveManager =
archiveAllDocsChanges: (project_id, callback = (error, docs) ->) ->
DocstoreHandler.getAllDocs project_id, (error, docs) ->
if error?
return callback(error)
else if !docs?
return callback new Error("No docs for project #{project_id}")
jobs = _.map docs, (doc) ->
(cb)-> DocArchiveManager.archiveDocChangesWithLock project_id, doc._id, cb
async.series jobs, callback
archiveDocChangesWithLock: (project_id, doc_id, callback = (error) ->) ->
job = (releaseLock) ->
DocArchiveManager.archiveDocChanges project_id, doc_id, releaseLock
LockManager.runWithLock("HistoryLock:#{doc_id}", job, callback)
archiveDocChanges: (project_id, doc_id, callback)->
MongoManager.getDocChangesCount doc_id, (error, count) ->
return callback(error) if error?
if count == 0
logger.log {project_id, doc_id}, "document history is empty, not archiving"
return callback()
else
MongoManager.getLastCompressedUpdate doc_id, (error, update) ->
return callback(error) if error?
MongoAWS.archiveDocHistory project_id, doc_id, (error) ->
return callback(error) if error?
logger.log doc_id:doc_id, project_id:project_id, "exported document to S3"
MongoManager.markDocHistoryAsArchived doc_id, update, (error) ->
return callback(error) if error?
callback()
unArchiveAllDocsChanges: (project_id, callback = (error, docs) ->) ->
DocstoreHandler.getAllDocs project_id, (error, docs) ->
if error?
return callback(error)
else if !docs?
return callback new Error("No docs for project #{project_id}")
jobs = _.map docs, (doc) ->
(cb)-> DocArchiveManager.unArchiveDocChangesWithLock project_id, doc._id, cb
async.parallelLimit jobs, 4, callback
unArchiveDocChangesWithLock: (project_id, doc_id, callback = (error) ->) ->
job = (releaseLock) ->
DocArchiveManager.unArchiveDocChanges project_id, doc_id, releaseLock
LockManager.runWithLock("HistoryLock:#{doc_id}", job, callback)
unArchiveDocChanges: (project_id, doc_id, callback)->
MongoManager.getArchivedDocChanges doc_id, (error, count) ->
return callback(error) if error?
if count == 0
return callback()
else
MongoAWS.unArchiveDocHistory project_id, doc_id, (error) ->
return callback(error) if error?
logger.log doc_id:doc_id, project_id:project_id, "imported document from S3"
MongoManager.markDocHistoryAsUnarchived doc_id, (error) ->
return callback(error) if error?
callback()
services/track-changes/app/coffee/DocstoreHandler.coffee
+21
View File
@@ -0,0 +1,21 @@
request = require("request").defaults(jar: false)
logger = require "logger-sharelatex"
settings = require "settings-sharelatex"
module.exports = DocstoreHandler =
getAllDocs: (project_id, callback = (error) ->) ->
logger.log project_id: project_id, "getting all docs for project in docstore api"
url = "#{settings.apis.docstore.url}/project/#{project_id}/doc"
request.get {
url: url
json: true
}, (error, res, docs) ->
return callback(error) if error?
logger.log {error, res, docs: if docs?.length then docs.map (d) -> d._id else []}, "docstore response"
if 200 <= res.statusCode < 300
callback(null, docs)
else
error = new Error("docstore api responded with non-success code: #{res.statusCode}")
logger.error err: error, project_id: project_id, "error getting all docs from docstore"
callback(error)
services/track-changes/app/coffee/HttpController.coffee
+15
View File
@@ -3,6 +3,7 @@ DiffManager = require "./DiffManager"
PackManager = require "./PackManager"
RestoreManager = require "./RestoreManager"
logger = require "logger-sharelatex"
DocArchiveManager = require "./DocArchiveManager"
module.exports = HttpController =
flushDoc: (req, res, next = (error) ->) ->
@@ -66,3 +67,17 @@ module.exports = HttpController =
RestoreManager.restoreToBeforeVersion project_id, doc_id, version, user_id, (error) ->
return next(error) if error?
res.send 204
archiveProject: (req, res, next = (error) ->) ->
project_id = req.params.project_id
logger.log project_id: project_id, "archiving all track changes to s3"
DocArchiveManager.archiveAllDocsChanges project_id, (error) ->
return next(error) if error?
res.send 204
unArchiveProject: (req, res, next = (error) ->) ->
project_id = req.params.project_id
logger.log project_id: project_id, "unarchiving all track changes from s3"
DocArchiveManager.unArchiveAllDocsChanges project_id, (error) ->
return next(error) if error?
res.send 204
services/track-changes/app/coffee/MongoAWS.coffee
+105
View File
@@ -0,0 +1,105 @@
settings = require "settings-sharelatex"
logger = require "logger-sharelatex"
AWS = require 'aws-sdk'
S3S = require 's3-streams'
{db, ObjectId} = require "./mongojs"
JSONStream = require "JSONStream"
ReadlineStream = require "readline-stream"
module.exports = MongoAWS =
MAX_SIZE: 1024*1024 # almost max size
MAX_COUNT: 1024 # almost max count
archiveDocHistory: (project_id, doc_id, _callback = (error) ->) ->
callback = (args...) ->
_callback(args...)
_callback = () ->
query = {
doc_id: ObjectId(doc_id)
expiresAt: {$exists : false}
}
AWS.config.update {
accessKeyId: settings.filestore.s3.key
secretAccessKey: settings.filestore.s3.secret
}
upload = S3S.WriteStream new AWS.S3(), {
"Bucket": settings.filestore.stores.user_files,
"Key": project_id+"/changes-"+doc_id
}
db.docHistory.find(query)
.on 'error', (err) ->
callback(err)
.pipe JSONStream.stringify()
.pipe upload
.on 'error', (err) ->
callback(err)
.on 'finish', () ->
return callback(null)
unArchiveDocHistory: (project_id, doc_id, _callback = (error) ->) ->
callback = (args...) ->
_callback(args...)
_callback = () ->
AWS.config.update {
accessKeyId: settings.filestore.s3.key
secretAccessKey: settings.filestore.s3.secret
}
download = S3S.ReadStream new AWS.S3(), {
"Bucket": settings.filestore.stores.user_files,
"Key": project_id+"/changes-"+doc_id
}, {
encoding: "utf8"
}
lineStream = new ReadlineStream();
ops = []
sz = 0
download
.on 'open', (obj) ->
return 1
.on 'error', (err) ->
callback(err)
.pipe lineStream
.on 'data', (line) ->
if line.length > 2
ops.push(JSON.parse(line))
sz += line.length
if ops.length >= MongoAWS.MAX_COUNT || sz >= MongoAWS.MAX_SIZE
download.pause()
MongoAWS.handleBulk ops.slice(0), sz, () ->
download.resume()
ops.splice(0,ops.length)
sz = 0
.on 'end', () ->
MongoAWS.handleBulk ops, sz, callback
.on 'error', (err) ->
return callback(err)
handleBulk: (ops, size, cb) ->
bulk = db.docHistory.initializeUnorderedBulkOp();
for op in ops
op._id = ObjectId(op._id)
op.doc_id = ObjectId(op.doc_id)
op.project_id = ObjectId(op.project_id)
bulk.find({_id:op._id}).upsert().updateOne(op)
if ops.length > 0
bulk.execute (err, result) ->
if err?
logger.error err:err, "error bulking ReadlineStream"
else
logger.log count:ops.length, result:result, size: size, "bulked ReadlineStream"
cb(err)
else
cb()
services/track-changes/app/coffee/MongoAWSexternal.coffee
+123
View File
@@ -0,0 +1,123 @@
settings = require "settings-sharelatex"
child_process = require "child_process"
mongoUri = require "mongo-uri";
logger = require "logger-sharelatex"
AWS = require 'aws-sdk'
fs = require 'fs'
S3S = require 's3-streams'
module.exports = MongoAWSexternal =
archiveDocHistory: (project_id, doc_id, callback = (error) ->) ->
MongoAWS.mongoExportDocHistory doc_id, (error, filepath) ->
MongoAWS.s3upStream project_id, doc_id, filepath, callback
#delete temp file?
unArchiveDocHistory: (project_id, doc_id, callback = (error) ->) ->
MongoAWS.s3downStream project_id, doc_id, (error, filepath) ->
if error == null
MongoAWS.mongoImportDocHistory filepath, callback
#delete temp file?
else
callback
mongoExportDocHistory: (doc_id, callback = (error, filepath) ->) ->
uriData = mongoUri.parse(settings.mongo.url);
filepath = settings.path.dumpFolder + '/' + doc_id + '.jsonUp'
args = []
args.push '-h'
args.push uriData.hosts[0]
args.push '-d'
args.push uriData.database
args.push '-c'
args.push 'docHistory'
args.push '-q'
args.push "{doc_id: ObjectId('#{doc_id}') , expiresAt: {$exists : false} }"
args.push '-o'
args.push filepath
proc = child_process.spawn "mongoexport", args
proc.on "error", callback
stderr = ""
proc.stderr.on "data", (chunk) -> stderr += chunk.toString()
proc.on "close", (code) ->
if code == 0
return callback(null,filepath)
else
return callback(new Error("mongodump failed: #{stderr}"),null)
mongoImportDocHistory: (filepath, callback = (error) ->) ->
uriData = mongoUri.parse(settings.mongo.url);
args = []
args.push '-h'
args.push uriData.hosts[0]
args.push '-d'
args.push uriData.database
args.push '-c'
args.push 'docHistory'
args.push '--file'
args.push filepath
proc = child_process.spawn "mongoimport", args
proc.on "error", callback
stderr = ""
proc.stderr.on "data", (chunk) -> stderr += chunk.toString()
proc.on "close", (code) ->
if code == 0
return callback(null,filepath)
else
return callback(new Error("mongodump failed: #{stderr}"),null)
s3upStream: (project_id, doc_id, filepath, callback = (error) ->) ->
AWS.config.update {
accessKeyId: settings.filestore.s3.key
secretAccessKey: settings.filestore.s3.secret
}
upload = S3S.WriteStream new AWS.S3(), {
"Bucket": settings.filestore.stores.user_files,
"Key": project_id+"/changes-"+doc_id
}
fs.createReadStream(filepath)
.on 'open', (obj) ->
return 1
.pipe(upload)
.on 'finish', () ->
return callback(null)
.on 'error', (err) ->
return callback(err)
s3downStream: (project_id, doc_id, callback = (error, filepath) ->) ->
filepath = settings.path.dumpFolder + '/' + doc_id + '.jsonDown'
AWS.config.update {
accessKeyId: settings.filestore.s3.key
secretAccessKey: settings.filestore.s3.secret
}
download = S3S.ReadStream new AWS.S3(), {
"Bucket": settings.filestore.stores.user_files,
"Key": project_id+"/changes-"+doc_id
}
download
.on 'open', (obj) ->
return 1
.pipe(fs.createWriteStream(filepath))
.on 'finish', () ->
return callback(null, filepath)
.on 'error', (err) ->
return callback(err, null)
services/track-changes/app/coffee/MongoManager.coffee
+22
View File
@@ -1,6 +1,7 @@
{db, ObjectId} = require "./mongojs"
PackManager = require "./PackManager"
async = require "async"
logger = require "logger-sharelatex"
module.exports = MongoManager =
getLastCompressedUpdate: (doc_id, callback = (error, update) ->) ->
@@ -47,6 +48,7 @@ module.exports = MongoManager =
insertCompressedUpdate: (project_id, doc_id, update, temporary, callback = (error) ->) ->
inS3 = update.inS3?
update = {
doc_id: ObjectId(doc_id.toString())
project_id: ObjectId(project_id.toString())
@@ -54,6 +56,9 @@ module.exports = MongoManager =
meta: update.meta
v: update.v
}
if inS3
update.inS3 = true
if temporary
seconds = 1000
minutes = 60 * seconds
@@ -126,3 +131,20 @@ module.exports = MongoManager =
# For finding documents which need packing
db.docHistoryStats.ensureIndex { doc_id: 1 }, { background: true }
db.docHistoryStats.ensureIndex { updates: -1, doc_id: 1 }, { background: true }
getDocChangesCount: (doc_id, callback)->
db.docHistory.count { doc_id : ObjectId(doc_id.toString()), inS3 : { $exists : false }}, {}, callback
getArchivedDocChanges: (doc_id, callback)->
db.docHistory.count { doc_id: ObjectId(doc_id.toString()) , inS3: true }, {}, callback
markDocHistoryAsArchived: (doc_id, update, callback)->
db.docHistory.update { _id: update._id }, { $set : { inS3 : true } }, (error)->
return callback(error) if error?
db.docHistory.remove { doc_id : ObjectId(doc_id.toString()), inS3 : { $exists : false }, v: { $lt : update.v }, expiresAt: {$exists : false} }, (error)->
return callback(error) if error?
callback(error)
markDocHistoryAsUnarchived: (doc_id, callback)->
db.docHistory.update { doc_id: ObjectId(doc_id.toString()) }, { $unset : { inS3 : true } }, { multi: true }, (error)->
callback(error)
services/track-changes/app/coffee/UpdatesManager.coffee
+16 -1
View File
@@ -6,6 +6,8 @@ WebApiManager = require "./WebApiManager"
UpdateTrimmer = require "./UpdateTrimmer"
logger = require "logger-sharelatex"
async = require "async"
DocArchiveManager = require "./DocArchiveManager"
_ = require "underscore"
module.exports = UpdatesManager =
compressAndSaveRawUpdates: (project_id, doc_id, rawUpdates, temporary, callback = (error) ->) ->
@@ -32,6 +34,9 @@ module.exports = UpdatesManager =
return
compressedUpdates = UpdateCompressor.compressRawUpdates lastCompressedUpdate, rawUpdates
if lastCompressedUpdate?.inS3? and not _.some(compressedUpdates, (update) -> update.inS3)
compressedUpdates[compressedUpdates.length-1].inS3 = lastCompressedUpdate.inS3
MongoManager.insertCompressedUpdates project_id, doc_id, compressedUpdates, temporary,(error) ->
return callback(error) if error?
logger.log project_id: project_id, doc_id: doc_id, rawUpdatesLength: length, compressedUpdatesLength: compressedUpdates.length, "compressed doc updates"
@@ -94,7 +99,17 @@ module.exports = UpdatesManager =
getProjectUpdates: (project_id, options = {}, callback = (error, updates) ->) ->
UpdatesManager.processUncompressedUpdatesForProject project_id, (error) ->
return callback(error) if error?
MongoManager.getProjectUpdates project_id, options, callback
MongoManager.getProjectUpdates project_id, options, (error, updates) ->
jobs = []
for update in updates
if update.inS3?
do (update) ->
jobs.push (callback) -> DocArchiveManager.unArchiveDocChanges update.project_id, update.doc_id, callback
if jobs.length?
async.series jobs, (err) ->
MongoManager.getProjectUpdates project_id, options, callback
else
callback(error, updates)
getProjectUpdatesWithUserInfo: (project_id, options = {}, callback = (error, updates) ->) ->
UpdatesManager.getProjectUpdates project_id, options, (error, updates) ->