ProjectFlusher.coffee 2.7 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273
  1. request = require("request")
  2. Settings = require('settings-sharelatex')
  3. RedisManager = require("./RedisManager")
  4. rclient = RedisManager.rclient
  5. docUpdaterKeys = Settings.redis.documentupdater.key_schema
  6. async = require("async")
  7. ProjectManager = require("./ProjectManager")
  8. _ = require("lodash")
  9. logger = require("logger-sharelatex")
  10. ProjectFlusher =
  11. # iterate over keys asynchronously using redis scan (non-blocking)
  12. # handle all the cluster nodes or single redis server
  13. _getKeys: (pattern, limit, callback) ->
  14. nodes = rclient.nodes?('master') || [ rclient ];
  15. doKeyLookupForNode = (node, cb) ->
  16. ProjectFlusher._getKeysFromNode node, pattern, limit, cb
  17. async.concatSeries nodes, doKeyLookupForNode, callback
  18. _getKeysFromNode: (node, pattern, limit = 1000, callback) ->
  19. cursor = 0 # redis iterator
  20. keySet = {} # use hash to avoid duplicate results
  21. batchSize = if limit? then Math.min(limit, 1000) else 1000
  22. # scan over all keys looking for pattern
  23. doIteration = (cb) ->
  24. node.scan cursor, "MATCH", pattern, "COUNT", batchSize, (error, reply) ->
  25. return callback(error) if error?
  26. [cursor, keys] = reply
  27. for key in keys
  28. keySet[key] = true
  29. keys = Object.keys(keySet)
  30. noResults = cursor == "0" # redis returns string results not numeric
  31. limitReached = (limit? && keys.length >= limit)
  32. if noResults || limitReached
  33. return callback(null, keys)
  34. else
  35. setTimeout doIteration, 10 # avoid hitting redis too hard
  36. doIteration()
  37. # extract ids from keys like DocsWithHistoryOps:57fd0b1f53a8396d22b2c24b
  38. # or docsInProject:{57fd0b1f53a8396d22b2c24b} (for redis cluster)
  39. _extractIds: (keyList) ->
  40. ids = for key in keyList
  41. m = key.match(/:\{?([0-9a-f]{24})\}?/) # extract object id
  42. m[1]
  43. return ids
  44. flushAllProjects: (options, callback)->
  45. logger.log options:options, "flushing all projects"
  46. ProjectFlusher._getKeys docUpdaterKeys.docsInProject({project_id:"*"}), options.limit, (error, project_keys) ->
  47. if error?
  48. logger.err err:error, "error getting keys for flushing"
  49. return callback(error)
  50. project_ids = ProjectFlusher._extractIds(project_keys)
  51. if options.dryRun
  52. return callback(null, project_ids)
  53. jobs = _.map project_ids, (project_id)->
  54. return (cb)->
  55. ProjectManager.flushAndDeleteProjectWithLocks project_id, {background:true}, cb
  56. async.parallelLimit async.reflectAll(jobs), options.concurrency, (error, results)->
  57. success = []
  58. failure = []
  59. _.each results, (result, i)->
  60. if result.error?
  61. failure.push(project_ids[i])
  62. else
  63. success.push(project_ids[i])
  64. logger.log success:success, failure:failure, "finished flushing all projects"
  65. return callback(error, {success:success, failure:failure})
  66. module.exports = ProjectFlusher