index.coffee 3.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111
  1. _ = require("underscore")
  2. async = require "async"
  3. os = require('os')
  4. crypto = require('crypto')
  5. # generate unique values for health check
  6. HOST = os.hostname()
  7. PID = process.pid
  8. RND = crypto.randomBytes(4).toString('hex')
  9. COUNT = 0
  10. module.exports = RedisSharelatex =
  11. createClient: (opts = {port: 6379, host: "localhost"})->
  12. if !opts.retry_max_delay?
  13. opts.retry_max_delay = 5000 # ms
  14. if opts.endpoints?
  15. standardOpts = _.clone(opts)
  16. delete standardOpts.endpoints
  17. delete standardOpts.masterName
  18. client = require("redis-sentinel").createClient opts.endpoints, opts.masterName, standardOpts
  19. client.healthCheck = RedisSharelatex.singleInstanceHealthCheckBuilder(client)
  20. else if opts.cluster?
  21. Redis = require("ioredis")
  22. standardOpts = _.clone(opts)
  23. delete standardOpts.cluster
  24. delete standardOpts.key_schema
  25. client = new Redis.Cluster(opts.cluster, standardOpts)
  26. client.healthCheck = RedisSharelatex.clusterHealthCheckBuilder(client)
  27. RedisSharelatex._monkeyPatchIoredisExec(client)
  28. else
  29. standardOpts = _.clone(opts)
  30. ioredis = require("ioredis")
  31. client = new ioredis(standardOpts)
  32. RedisSharelatex._monkeyPatchIoredisExec(client)
  33. client.healthCheck = RedisSharelatex.singleInstanceHealthCheckBuilder(client)
  34. return client
  35. HEARTBEAT_TIMEOUT: 2000
  36. singleInstanceHealthCheckBuilder: (client) ->
  37. healthCheck = (callback) ->
  38. RedisSharelatex._checkClient(client, callback)
  39. return healthCheck
  40. clusterHealthCheckBuilder: (client) ->
  41. healthCheck = (callback) ->
  42. jobs = client.nodes("all").map (node) =>
  43. (cb) => RedisSharelatex._checkClient(node, cb)
  44. async.parallel jobs, callback
  45. return healthCheck
  46. _checkClient: (client, callback) ->
  47. callback = _.once(callback)
  48. # check the redis connection by storing and retrieving a unique key/value pair
  49. uniqueToken = "host=#{HOST}:pid=#{PID}:random=#{RND}:time=#{Date.now()}:count=#{COUNT++}"
  50. timer = setTimeout () ->
  51. error = new Error("redis client health check timed out #{client?.options?.host}")
  52. console.error {
  53. err: error,
  54. key: client.options?.key # only present for cluster
  55. clientOptions: client.options
  56. uniqueToken: uniqueToken
  57. }, "client timed out"
  58. callback(error)
  59. , RedisSharelatex.HEARTBEAT_TIMEOUT
  60. healthCheckKey = "_redis-wrapper:healthCheckKey:{#{uniqueToken}}"
  61. healthCheckValue = "_redis-wrapper:healthCheckValue:{#{uniqueToken}}"
  62. # set the unique key/value pair
  63. multi = client.multi()
  64. multi.set healthCheckKey, healthCheckValue, "EX", 60
  65. multi.exec (err, reply) ->
  66. if err?
  67. clearTimeout timer
  68. return callback(err)
  69. # check that we can retrieve the unique key/value pair
  70. multi = client.multi()
  71. multi.get healthCheckKey
  72. multi.del healthCheckKey
  73. multi.exec (err, reply) ->
  74. clearTimeout timer
  75. return callback(err) if err?
  76. return callback(new Error("bad response from redis health check")) if reply?[0] isnt healthCheckValue or reply?[1] isnt 1
  77. return callback()
  78. _monkeyPatchIoredisExec: (client) ->
  79. _multi = client.multi
  80. client.multi = (args...) ->
  81. multi = _multi.call(client, args...)
  82. _exec = multi.exec
  83. multi.exec = (callback = () ->) ->
  84. _exec.call multi, (error, result) ->
  85. # ioredis exec returns an results like:
  86. # [ [null, 42], [null, "foo"] ]
  87. # where the first entries in each 2-tuple are
  88. # presumably errors for each individual command,
  89. # and the second entry is the result. We need to transform
  90. # this into the same result as the old redis driver:
  91. # [ 42, "foo" ]
  92. filtered_result = []
  93. for entry in result or []
  94. if entry[0]?
  95. return callback(entry[0])
  96. else
  97. filtered_result.push entry[1]
  98. callback error, filtered_result
  99. return multi