2014-03-07 00:53:16 -05:00
|
|
|
# encoding: utf-8
|
2012-10-20 17:03:43 -04:00
|
|
|
require 'sidekiq'
|
|
|
|
|
|
|
|
module Sidekiq
|
2012-12-04 08:11:25 -05:00
|
|
|
class Stats
|
2012-12-04 13:14:38 -05:00
|
|
|
def processed
|
2013-05-31 12:02:27 -04:00
|
|
|
Sidekiq.redis { |conn| conn.get("stat:processed") }.to_i
|
2012-12-04 13:14:38 -05:00
|
|
|
end
|
2012-12-04 08:11:25 -05:00
|
|
|
|
2012-12-04 13:14:38 -05:00
|
|
|
def failed
|
2013-05-31 12:02:27 -04:00
|
|
|
Sidekiq.redis { |conn| conn.get("stat:failed") }.to_i
|
2012-12-04 13:14:38 -05:00
|
|
|
end
|
2012-12-04 08:11:25 -05:00
|
|
|
|
2013-12-18 10:46:41 -05:00
|
|
|
def reset(*stats)
|
|
|
|
all = %w(failed processed)
|
|
|
|
stats = stats.empty? ? all : all & stats.flatten.compact.map(&:to_s)
|
|
|
|
|
2014-03-22 00:31:50 -04:00
|
|
|
mset_args = []
|
|
|
|
stats.each do |stat|
|
|
|
|
mset_args << "stat:#{stat}"
|
|
|
|
mset_args << 0
|
|
|
|
end
|
2013-04-30 18:16:21 -04:00
|
|
|
Sidekiq.redis do |conn|
|
2014-03-22 00:31:50 -04:00
|
|
|
conn.mset(*mset_args)
|
2013-04-30 18:16:21 -04:00
|
|
|
end
|
|
|
|
end
|
|
|
|
|
2012-12-04 13:14:38 -05:00
|
|
|
def queues
|
|
|
|
Sidekiq.redis do |conn|
|
|
|
|
queues = conn.smembers('queues')
|
2012-12-04 08:11:25 -05:00
|
|
|
|
2014-03-22 00:30:22 -04:00
|
|
|
lengths = conn.pipelined do
|
|
|
|
queues.each do |queue|
|
|
|
|
conn.llen("queue:#{queue}")
|
|
|
|
end
|
|
|
|
end
|
|
|
|
|
|
|
|
i = 0
|
2012-12-04 17:05:47 -05:00
|
|
|
array_of_arrays = queues.inject({}) do |memo, queue|
|
2014-03-22 00:30:22 -04:00
|
|
|
memo[queue] = lengths[i]
|
|
|
|
i += 1
|
2012-12-04 13:14:38 -05:00
|
|
|
memo
|
2012-12-04 17:05:47 -05:00
|
|
|
end.sort_by { |_, size| size }
|
|
|
|
|
|
|
|
Hash[array_of_arrays.reverse]
|
2012-12-04 08:11:25 -05:00
|
|
|
end
|
2012-12-04 13:14:38 -05:00
|
|
|
end
|
2012-12-04 08:11:25 -05:00
|
|
|
|
2012-12-04 13:14:38 -05:00
|
|
|
def enqueued
|
|
|
|
queues.values.inject(&:+) || 0
|
2012-12-04 08:11:25 -05:00
|
|
|
end
|
2012-12-05 20:35:49 -05:00
|
|
|
|
2012-12-29 23:10:36 -05:00
|
|
|
def scheduled_size
|
|
|
|
Sidekiq.redis {|c| c.zcard('schedule') }
|
|
|
|
end
|
|
|
|
|
|
|
|
def retry_size
|
|
|
|
Sidekiq.redis {|c| c.zcard('retry') }
|
|
|
|
end
|
|
|
|
|
2014-02-09 17:56:01 -05:00
|
|
|
def dead_size
|
|
|
|
Sidekiq.redis {|c| c.zcard('dead') }
|
|
|
|
end
|
|
|
|
|
2012-12-05 20:35:49 -05:00
|
|
|
class History
|
|
|
|
def initialize(days_previous, start_date = nil)
|
|
|
|
@days_previous = days_previous
|
|
|
|
@start_date = start_date || Time.now.utc.to_date
|
|
|
|
end
|
|
|
|
|
|
|
|
def processed
|
|
|
|
date_stat_hash("processed")
|
|
|
|
end
|
|
|
|
|
|
|
|
def failed
|
|
|
|
date_stat_hash("failed")
|
|
|
|
end
|
|
|
|
|
|
|
|
private
|
|
|
|
|
|
|
|
def date_stat_hash(stat)
|
|
|
|
i = 0
|
|
|
|
stat_hash = {}
|
2014-03-22 00:28:27 -04:00
|
|
|
keys = []
|
|
|
|
dates = []
|
2012-12-05 20:35:49 -05:00
|
|
|
|
2014-03-22 00:28:27 -04:00
|
|
|
while i < @days_previous
|
|
|
|
date = @start_date - i
|
|
|
|
keys << "stat:#{stat}:#{date}"
|
|
|
|
dates << date
|
|
|
|
i += 1
|
|
|
|
end
|
2012-12-05 20:35:49 -05:00
|
|
|
|
2014-03-22 00:28:27 -04:00
|
|
|
Sidekiq.redis do |conn|
|
|
|
|
conn.mget(keys).each_with_index do |value, i|
|
|
|
|
stat_hash[dates[i].to_s] = value ? value.to_i : 0
|
2012-12-05 20:35:49 -05:00
|
|
|
end
|
|
|
|
end
|
|
|
|
|
|
|
|
stat_hash
|
|
|
|
end
|
|
|
|
end
|
2012-12-04 08:11:25 -05:00
|
|
|
end
|
|
|
|
|
2012-10-20 17:03:43 -04:00
|
|
|
##
|
|
|
|
# Encapsulates a queue within Sidekiq.
|
|
|
|
# Allows enumeration of all jobs within the queue
|
|
|
|
# and deletion of jobs.
|
|
|
|
#
|
|
|
|
# queue = Sidekiq::Queue.new("mailer")
|
|
|
|
# queue.each do |job|
|
|
|
|
# job.klass # => 'MyWorker'
|
|
|
|
# job.args # => [1, 2, 3]
|
|
|
|
# job.delete if job.jid == 'abcdef1234567890'
|
|
|
|
# end
|
|
|
|
#
|
|
|
|
class Queue
|
|
|
|
include Enumerable
|
|
|
|
|
2013-06-01 17:54:29 -04:00
|
|
|
def self.all
|
|
|
|
Sidekiq.redis {|c| c.smembers('queues') }.map {|q| Sidekiq::Queue.new(q) }
|
|
|
|
end
|
|
|
|
|
2012-10-20 17:03:43 -04:00
|
|
|
attr_reader :name
|
|
|
|
|
|
|
|
def initialize(name="default")
|
|
|
|
@name = name
|
|
|
|
@rname = "queue:#{name}"
|
|
|
|
end
|
|
|
|
|
|
|
|
def size
|
|
|
|
Sidekiq.redis { |con| con.llen(@rname) }
|
|
|
|
end
|
|
|
|
|
2014-04-25 01:16:34 -04:00
|
|
|
# Sidekiq Pro overrides this
|
|
|
|
def paused?
|
|
|
|
false
|
|
|
|
end
|
|
|
|
|
2013-05-24 01:58:06 -04:00
|
|
|
def latency
|
|
|
|
entry = Sidekiq.redis do |conn|
|
|
|
|
conn.lrange(@rname, -1, -1)
|
|
|
|
end.first
|
|
|
|
return 0 unless entry
|
|
|
|
Time.now.to_f - Sidekiq.load_json(entry)['enqueued_at']
|
|
|
|
end
|
|
|
|
|
2012-10-20 17:03:43 -04:00
|
|
|
def each(&block)
|
2013-10-12 18:56:34 -04:00
|
|
|
initial_size = size
|
|
|
|
deleted_size = 0
|
2012-10-20 17:03:43 -04:00
|
|
|
page = 0
|
|
|
|
page_size = 50
|
|
|
|
|
|
|
|
loop do
|
2013-10-12 18:56:34 -04:00
|
|
|
range_start = page * page_size - deleted_size
|
|
|
|
range_end = page * page_size - deleted_size + (page_size - 1)
|
2012-10-20 17:03:43 -04:00
|
|
|
entries = Sidekiq.redis do |conn|
|
2013-10-12 18:56:34 -04:00
|
|
|
conn.lrange @rname, range_start, range_end
|
2012-10-20 17:03:43 -04:00
|
|
|
end
|
|
|
|
break if entries.empty?
|
|
|
|
page += 1
|
|
|
|
entries.each do |entry|
|
|
|
|
block.call Job.new(entry, @name)
|
|
|
|
end
|
2013-10-12 18:56:34 -04:00
|
|
|
deleted_size = initial_size - size
|
2012-10-20 17:03:43 -04:00
|
|
|
end
|
|
|
|
end
|
2012-11-25 21:43:48 -05:00
|
|
|
|
2013-04-17 14:11:29 -04:00
|
|
|
def find_job(jid)
|
|
|
|
self.detect { |j| j.jid == jid }
|
|
|
|
end
|
|
|
|
|
2012-11-25 21:43:48 -05:00
|
|
|
def clear
|
|
|
|
Sidekiq.redis do |conn|
|
|
|
|
conn.multi do
|
2013-06-20 10:14:51 -04:00
|
|
|
conn.del(@rname)
|
2012-11-25 21:43:48 -05:00
|
|
|
conn.srem("queues", name)
|
|
|
|
end
|
|
|
|
end
|
|
|
|
end
|
2014-03-07 00:53:16 -05:00
|
|
|
alias_method :💣, :clear
|
2012-10-20 17:03:43 -04:00
|
|
|
end
|
|
|
|
|
|
|
|
##
|
2012-10-30 13:06:20 -04:00
|
|
|
# Encapsulates a pending job within a Sidekiq queue or
|
|
|
|
# sorted set.
|
|
|
|
#
|
2012-10-20 17:03:43 -04:00
|
|
|
# The job should be considered immutable but may be
|
|
|
|
# removed from the queue via Job#delete.
|
|
|
|
#
|
|
|
|
class Job
|
|
|
|
attr_reader :item
|
|
|
|
|
|
|
|
def initialize(item, queue_name=nil)
|
|
|
|
@value = item
|
|
|
|
@item = Sidekiq.load_json(item)
|
|
|
|
@queue = queue_name || @item['queue']
|
|
|
|
end
|
|
|
|
|
|
|
|
def klass
|
|
|
|
@item['class']
|
|
|
|
end
|
|
|
|
|
|
|
|
def args
|
|
|
|
@item['args']
|
|
|
|
end
|
|
|
|
|
|
|
|
def jid
|
|
|
|
@item['jid']
|
|
|
|
end
|
|
|
|
|
2013-05-23 07:31:41 -04:00
|
|
|
def enqueued_at
|
2013-08-15 23:06:16 -04:00
|
|
|
Time.at(@item['enqueued_at'] || 0).utc
|
2013-05-23 07:31:41 -04:00
|
|
|
end
|
|
|
|
|
2012-10-20 17:03:43 -04:00
|
|
|
def queue
|
|
|
|
@queue
|
|
|
|
end
|
|
|
|
|
2013-05-24 22:59:40 -04:00
|
|
|
def latency
|
|
|
|
Time.now.to_f - @item['enqueued_at']
|
|
|
|
end
|
|
|
|
|
2012-10-20 17:03:43 -04:00
|
|
|
##
|
|
|
|
# Remove this job from the queue.
|
|
|
|
def delete
|
|
|
|
count = Sidekiq.redis do |conn|
|
2013-11-20 17:51:58 -05:00
|
|
|
conn.lrem("queue:#{@queue}", 1, @value)
|
2012-10-20 17:03:43 -04:00
|
|
|
end
|
|
|
|
count != 0
|
|
|
|
end
|
|
|
|
|
|
|
|
def [](name)
|
2014-02-14 09:43:34 -05:00
|
|
|
@item.__send__(:[], name)
|
2012-10-20 17:03:43 -04:00
|
|
|
end
|
|
|
|
end
|
|
|
|
|
2012-10-30 13:06:20 -04:00
|
|
|
class SortedEntry < Job
|
2012-10-20 17:03:43 -04:00
|
|
|
attr_reader :score
|
2014-02-10 00:17:05 -05:00
|
|
|
attr_reader :parent
|
2012-10-20 17:03:43 -04:00
|
|
|
|
2012-10-30 13:06:20 -04:00
|
|
|
def initialize(parent, score, item)
|
2012-10-20 17:03:43 -04:00
|
|
|
super(item)
|
|
|
|
@score = score
|
2012-10-30 13:06:20 -04:00
|
|
|
@parent = parent
|
2012-10-20 17:03:43 -04:00
|
|
|
end
|
|
|
|
|
2012-10-30 13:06:20 -04:00
|
|
|
def at
|
2013-08-15 23:06:16 -04:00
|
|
|
Time.at(score).utc
|
2012-10-20 17:09:27 -04:00
|
|
|
end
|
|
|
|
|
2012-10-20 17:03:43 -04:00
|
|
|
def delete
|
2012-11-26 14:53:22 -05:00
|
|
|
@parent.delete(score, jid)
|
2012-11-26 11:22:48 -05:00
|
|
|
end
|
|
|
|
|
2013-01-29 16:17:59 -05:00
|
|
|
def reschedule(at)
|
|
|
|
@parent.delete(score, jid)
|
|
|
|
@parent.schedule(at, item)
|
|
|
|
end
|
|
|
|
|
2013-06-21 22:43:06 -04:00
|
|
|
def add_to_queue
|
|
|
|
Sidekiq.redis do |conn|
|
2014-02-18 09:08:43 -05:00
|
|
|
results = conn.multi do
|
|
|
|
conn.zrangebyscore('schedule', score, score)
|
|
|
|
conn.zremrangebyscore('schedule', score, score)
|
|
|
|
end.first
|
2013-06-21 22:43:06 -04:00
|
|
|
results.map do |message|
|
|
|
|
msg = Sidekiq.load_json(message)
|
|
|
|
Sidekiq::Client.push(msg)
|
|
|
|
end
|
|
|
|
end
|
|
|
|
end
|
|
|
|
|
2012-11-26 11:22:48 -05:00
|
|
|
def retry
|
2014-02-10 00:17:05 -05:00
|
|
|
raise "Retry not available on jobs which have not failed" unless item["failed_at"]
|
2012-11-26 11:22:48 -05:00
|
|
|
Sidekiq.redis do |conn|
|
2014-02-18 09:08:43 -05:00
|
|
|
results = conn.multi do
|
2014-02-21 21:02:28 -05:00
|
|
|
conn.zrangebyscore(parent.name, score, score)
|
|
|
|
conn.zremrangebyscore(parent.name, score, score)
|
2014-02-18 09:08:43 -05:00
|
|
|
end.first
|
2012-11-26 11:22:48 -05:00
|
|
|
results.map do |message|
|
|
|
|
msg = Sidekiq.load_json(message)
|
|
|
|
msg['retry_count'] = msg['retry_count'] - 1
|
2013-05-24 08:22:24 -04:00
|
|
|
Sidekiq::Client.push(msg)
|
2012-11-26 11:22:48 -05:00
|
|
|
end
|
|
|
|
end
|
2012-10-20 17:03:43 -04:00
|
|
|
end
|
|
|
|
end
|
|
|
|
|
2012-10-30 13:06:20 -04:00
|
|
|
class SortedSet
|
2012-10-20 17:03:43 -04:00
|
|
|
include Enumerable
|
|
|
|
|
2014-02-10 00:17:05 -05:00
|
|
|
attr_reader :name
|
|
|
|
|
2012-10-30 13:06:20 -04:00
|
|
|
def initialize(name)
|
2014-02-10 00:17:05 -05:00
|
|
|
@name = name
|
2013-10-23 22:30:53 -04:00
|
|
|
@_size = size
|
2012-10-30 13:06:20 -04:00
|
|
|
end
|
|
|
|
|
2012-10-20 17:03:43 -04:00
|
|
|
def size
|
2014-02-10 00:17:05 -05:00
|
|
|
Sidekiq.redis {|c| c.zcard(name) }
|
2012-10-20 17:03:43 -04:00
|
|
|
end
|
|
|
|
|
2014-03-02 19:36:00 -05:00
|
|
|
def clear
|
|
|
|
Sidekiq.redis do |conn|
|
|
|
|
conn.del(name)
|
|
|
|
end
|
|
|
|
end
|
2014-03-07 00:56:59 -05:00
|
|
|
alias_method :💣, :clear
|
2014-03-02 19:36:00 -05:00
|
|
|
end
|
|
|
|
|
|
|
|
class JobSet < SortedSet
|
|
|
|
|
2013-01-29 16:17:59 -05:00
|
|
|
def schedule(timestamp, message)
|
|
|
|
Sidekiq.redis do |conn|
|
2014-02-10 00:17:05 -05:00
|
|
|
conn.zadd(name, timestamp.to_f.to_s, Sidekiq.dump_json(message))
|
2013-01-29 16:17:59 -05:00
|
|
|
end
|
|
|
|
end
|
|
|
|
|
2012-10-20 17:03:43 -04:00
|
|
|
def each(&block)
|
2013-10-23 22:30:53 -04:00
|
|
|
initial_size = @_size
|
2013-10-23 23:36:13 -04:00
|
|
|
offset_size = 0
|
2012-10-20 17:03:43 -04:00
|
|
|
page = -1
|
|
|
|
page_size = 50
|
|
|
|
|
|
|
|
loop do
|
2013-10-23 23:36:13 -04:00
|
|
|
range_start = page * page_size + offset_size
|
|
|
|
range_end = page * page_size + offset_size + (page_size - 1)
|
2012-10-30 13:06:20 -04:00
|
|
|
elements = Sidekiq.redis do |conn|
|
2014-02-10 00:17:05 -05:00
|
|
|
conn.zrange name, range_start, range_end, :with_scores => true
|
2012-10-20 17:03:43 -04:00
|
|
|
end
|
2012-10-30 13:06:20 -04:00
|
|
|
break if elements.empty?
|
2012-10-20 17:03:43 -04:00
|
|
|
page -= 1
|
2012-10-30 13:06:20 -04:00
|
|
|
elements.each do |element, score|
|
|
|
|
block.call SortedEntry.new(self, score, element)
|
2012-10-20 17:03:43 -04:00
|
|
|
end
|
2013-10-23 23:36:13 -04:00
|
|
|
offset_size = initial_size - @_size
|
2012-10-20 17:03:43 -04:00
|
|
|
end
|
|
|
|
end
|
2012-10-30 13:06:20 -04:00
|
|
|
|
2012-11-26 14:53:22 -05:00
|
|
|
def fetch(score, jid = nil)
|
|
|
|
elements = Sidekiq.redis do |conn|
|
2014-02-10 00:17:05 -05:00
|
|
|
conn.zrangebyscore(name, score, score)
|
2012-11-26 14:53:22 -05:00
|
|
|
end
|
|
|
|
|
|
|
|
elements.inject([]) do |result, element|
|
|
|
|
entry = SortedEntry.new(self, score, element)
|
|
|
|
if jid
|
|
|
|
result << entry if entry.jid == jid
|
|
|
|
else
|
|
|
|
result << entry
|
|
|
|
end
|
2012-12-02 23:32:16 -05:00
|
|
|
result
|
2012-11-26 14:53:22 -05:00
|
|
|
end
|
|
|
|
end
|
|
|
|
|
2013-04-16 15:43:24 -04:00
|
|
|
def find_job(jid)
|
2013-04-17 14:11:29 -04:00
|
|
|
self.detect { |j| j.jid == jid }
|
2013-04-16 15:43:24 -04:00
|
|
|
end
|
|
|
|
|
2012-11-26 14:53:22 -05:00
|
|
|
def delete(score, jid = nil)
|
|
|
|
if jid
|
|
|
|
elements = Sidekiq.redis do |conn|
|
2014-02-10 00:17:05 -05:00
|
|
|
conn.zrangebyscore(name, score, score)
|
2012-11-26 14:53:22 -05:00
|
|
|
end
|
|
|
|
|
|
|
|
elements_with_jid = elements.map do |element|
|
|
|
|
message = Sidekiq.load_json(element)
|
|
|
|
|
|
|
|
if message["jid"] == jid
|
2013-10-23 23:36:13 -04:00
|
|
|
_, @_size = Sidekiq.redis do |conn|
|
|
|
|
conn.multi do
|
2014-02-10 00:17:05 -05:00
|
|
|
conn.zrem(name, element)
|
|
|
|
conn.zcard name
|
2013-10-23 23:36:13 -04:00
|
|
|
end
|
|
|
|
end
|
2012-11-26 14:53:22 -05:00
|
|
|
end
|
|
|
|
end
|
2013-10-23 23:36:13 -04:00
|
|
|
elements_with_jid.count != 0
|
2012-11-26 14:53:22 -05:00
|
|
|
else
|
2013-10-23 23:36:13 -04:00
|
|
|
count, @_size = Sidekiq.redis do |conn|
|
|
|
|
conn.multi do
|
2014-02-10 00:17:05 -05:00
|
|
|
conn.zremrangebyscore(name, score, score)
|
|
|
|
conn.zcard name
|
2013-10-23 23:36:13 -04:00
|
|
|
end
|
2012-11-26 14:53:22 -05:00
|
|
|
end
|
2013-10-23 23:36:13 -04:00
|
|
|
count != 0
|
2012-10-30 13:06:20 -04:00
|
|
|
end
|
|
|
|
end
|
2012-11-25 21:43:48 -05:00
|
|
|
|
2012-10-30 13:06:20 -04:00
|
|
|
end
|
|
|
|
|
|
|
|
##
|
|
|
|
# Allows enumeration of scheduled jobs within Sidekiq.
|
|
|
|
# Based on this, you can search/filter for jobs. Here's an
|
|
|
|
# example where I'm selecting all jobs of a certain type
|
|
|
|
# and deleting them from the retry queue.
|
|
|
|
#
|
|
|
|
# r = Sidekiq::ScheduledSet.new
|
|
|
|
# r.select do |retri|
|
|
|
|
# retri.klass == 'Sidekiq::Extensions::DelayedClass' &&
|
|
|
|
# retri.args[0] == 'User' &&
|
|
|
|
# retri.args[1] == 'setup_new_subscriber'
|
|
|
|
# end.map(&:delete)
|
2014-03-02 19:36:00 -05:00
|
|
|
class ScheduledSet < JobSet
|
2012-10-30 13:06:20 -04:00
|
|
|
def initialize
|
|
|
|
super 'schedule'
|
|
|
|
end
|
|
|
|
end
|
|
|
|
|
|
|
|
##
|
|
|
|
# Allows enumeration of retries within Sidekiq.
|
|
|
|
# Based on this, you can search/filter for jobs. Here's an
|
|
|
|
# example where I'm selecting all jobs of a certain type
|
|
|
|
# and deleting them from the retry queue.
|
|
|
|
#
|
|
|
|
# r = Sidekiq::RetrySet.new
|
|
|
|
# r.select do |retri|
|
|
|
|
# retri.klass == 'Sidekiq::Extensions::DelayedClass' &&
|
|
|
|
# retri.args[0] == 'User' &&
|
|
|
|
# retri.args[1] == 'setup_new_subscriber'
|
|
|
|
# end.map(&:delete)
|
2014-03-02 19:36:00 -05:00
|
|
|
class RetrySet < JobSet
|
2012-10-30 13:06:20 -04:00
|
|
|
def initialize
|
|
|
|
super 'retry'
|
|
|
|
end
|
2013-02-19 23:36:59 -05:00
|
|
|
|
2014-02-09 17:56:01 -05:00
|
|
|
def retry_all
|
|
|
|
while size > 0
|
|
|
|
each(&:retry)
|
|
|
|
end
|
|
|
|
end
|
|
|
|
end
|
|
|
|
|
2014-03-02 19:36:00 -05:00
|
|
|
class DeadSet < JobSet
|
2014-02-09 17:56:01 -05:00
|
|
|
def initialize
|
|
|
|
super 'dead'
|
|
|
|
end
|
|
|
|
|
2013-02-19 23:36:59 -05:00
|
|
|
def retry_all
|
|
|
|
while size > 0
|
|
|
|
each(&:retry)
|
|
|
|
end
|
|
|
|
end
|
2012-10-20 17:03:43 -04:00
|
|
|
end
|
|
|
|
|
2014-03-03 00:18:26 -05:00
|
|
|
##
|
|
|
|
# Enumerates the set of Sidekiq processes which are actively working
|
|
|
|
# right now. Each process send a heartbeat to Redis every 5 seconds
|
|
|
|
# so this set should be relatively accurate, barring network partitions.
|
2014-03-08 01:41:10 -05:00
|
|
|
#
|
2014-05-13 23:33:20 -04:00
|
|
|
# Yields a Sidekiq::Process.
|
2014-03-08 01:41:10 -05:00
|
|
|
#
|
|
|
|
|
2014-03-03 00:18:26 -05:00
|
|
|
class ProcessSet
|
|
|
|
include Enumerable
|
2014-03-02 19:36:00 -05:00
|
|
|
|
|
|
|
def each(&block)
|
2014-03-08 01:41:10 -05:00
|
|
|
procs = Sidekiq.redis { |conn| conn.smembers('processes') }
|
2014-03-03 00:18:26 -05:00
|
|
|
|
|
|
|
to_prune = []
|
2014-03-22 00:24:19 -04:00
|
|
|
sorted = procs.sort
|
2014-03-08 01:41:10 -05:00
|
|
|
Sidekiq.redis do |conn|
|
2014-03-22 00:24:19 -04:00
|
|
|
# We're making a tradeoff here between consuming more memory instead of
|
|
|
|
# making more roundtrips to Redis, but if you have hundreds or thousands of workers,
|
|
|
|
# you'll be happier this way
|
|
|
|
result = conn.pipelined do
|
|
|
|
sorted.each do |key|
|
|
|
|
conn.hmget(key, 'info', 'busy', 'beat')
|
|
|
|
end
|
|
|
|
end
|
|
|
|
|
|
|
|
result.each_with_index do |(info, busy, at_s), i|
|
2014-03-09 17:32:27 -04:00
|
|
|
# the hash named key has an expiry of 60 seconds.
|
|
|
|
# if it's not found, that means the process has not reported
|
|
|
|
# in to Redis and probably died.
|
2014-03-22 00:24:19 -04:00
|
|
|
(to_prune << sorted[i]; next) if info.nil?
|
2014-03-08 01:41:10 -05:00
|
|
|
hash = Sidekiq.load_json(info)
|
2014-05-13 23:33:20 -04:00
|
|
|
yield Process.new(hash.merge('busy' => busy.to_i, 'beat' => at_s.to_f))
|
2014-03-02 19:36:00 -05:00
|
|
|
end
|
|
|
|
end
|
2014-03-03 00:18:26 -05:00
|
|
|
|
2014-03-09 17:32:27 -04:00
|
|
|
Sidekiq.redis {|conn| conn.srem('processes', to_prune) } unless to_prune.empty?
|
2014-03-03 00:18:26 -05:00
|
|
|
nil
|
2014-03-02 19:36:00 -05:00
|
|
|
end
|
2014-03-09 17:32:27 -04:00
|
|
|
|
|
|
|
# This method is not guaranteed accurate since it does not prune the set
|
|
|
|
# based on current heartbeat. #each does that and ensures the set only
|
|
|
|
# contains Sidekiq processes which have sent a heartbeat within the last
|
|
|
|
# 60 seconds.
|
|
|
|
def size
|
|
|
|
Sidekiq.redis { |conn| conn.scard('processes') }
|
|
|
|
end
|
2014-03-02 19:36:00 -05:00
|
|
|
end
|
2013-01-24 12:50:30 -05:00
|
|
|
|
2014-05-13 23:33:20 -04:00
|
|
|
#
|
|
|
|
# Sidekiq::Process has a set of attributes which look like this:
|
|
|
|
#
|
|
|
|
# {
|
|
|
|
# 'hostname' => 'app-1.example.com',
|
|
|
|
# 'started_at' => <process start time>,
|
|
|
|
# 'pid' => 12345,
|
|
|
|
# 'tag' => 'myapp'
|
|
|
|
# 'concurrency' => 25,
|
|
|
|
# 'queues' => ['default', 'low'],
|
|
|
|
# 'busy' => 10,
|
|
|
|
# 'beat' => <last heartbeat>,
|
|
|
|
# }
|
|
|
|
class Process
|
|
|
|
def initialize(hash)
|
|
|
|
@attribs = hash
|
|
|
|
end
|
|
|
|
|
|
|
|
def [](key)
|
|
|
|
@attribs[key]
|
|
|
|
end
|
|
|
|
end
|
|
|
|
|
2013-01-24 12:50:30 -05:00
|
|
|
##
|
|
|
|
# Programmatic access to the current active worker set.
|
|
|
|
#
|
|
|
|
# WARNING WARNING WARNING
|
|
|
|
#
|
|
|
|
# This is live data that can change every millisecond.
|
2013-05-23 00:50:22 -04:00
|
|
|
# If you call #size => 5 and then expect #each to be
|
2013-01-24 12:50:30 -05:00
|
|
|
# called 5 times, you're going to have a bad time.
|
|
|
|
#
|
|
|
|
# workers = Sidekiq::Workers.new
|
|
|
|
# workers.size => 2
|
2014-03-08 17:21:52 -05:00
|
|
|
# workers.each do |process_id, thread_id, work|
|
|
|
|
# # process_id is a unique identifier per Sidekiq process
|
|
|
|
# # thread_id is a unique identifier per thread
|
2013-01-24 12:50:30 -05:00
|
|
|
# # work is a Hash which looks like:
|
|
|
|
# # { 'queue' => name, 'run_at' => timestamp, 'payload' => msg }
|
2014-02-01 23:48:44 -05:00
|
|
|
# # run_at is an epoch Integer.
|
2013-01-24 12:50:30 -05:00
|
|
|
# end
|
2014-03-08 17:21:52 -05:00
|
|
|
#
|
2013-01-24 12:50:30 -05:00
|
|
|
class Workers
|
|
|
|
include Enumerable
|
|
|
|
|
|
|
|
def each(&block)
|
2014-03-08 01:41:10 -05:00
|
|
|
Sidekiq.redis do |conn|
|
|
|
|
procs = conn.smembers('processes')
|
|
|
|
procs.sort.each do |key|
|
2014-03-22 00:18:33 -04:00
|
|
|
valid, workers = conn.pipelined do
|
2014-03-08 01:41:10 -05:00
|
|
|
conn.exists(key)
|
|
|
|
conn.hgetall("#{key}:workers")
|
|
|
|
end
|
|
|
|
next unless valid
|
|
|
|
workers.each_pair do |tid, json|
|
2014-03-08 17:21:52 -05:00
|
|
|
yield key, tid, Sidekiq.load_json(json)
|
2014-03-08 01:41:10 -05:00
|
|
|
end
|
|
|
|
end
|
2013-01-24 12:50:30 -05:00
|
|
|
end
|
|
|
|
end
|
|
|
|
|
2014-03-08 17:21:52 -05:00
|
|
|
# Note that #size is only as accurate as Sidekiq's heartbeat,
|
|
|
|
# which happens every 5 seconds. It is NOT real-time.
|
|
|
|
#
|
2014-03-08 17:10:27 -05:00
|
|
|
# Not very efficient if you have lots of Sidekiq
|
|
|
|
# processes but the alternative is a global counter
|
|
|
|
# which can easily get out of sync with crashy processes.
|
2013-01-24 12:50:30 -05:00
|
|
|
def size
|
2014-03-08 17:10:27 -05:00
|
|
|
Sidekiq.redis do |conn|
|
|
|
|
procs = conn.smembers('processes')
|
|
|
|
return 0 if procs.empty?
|
|
|
|
|
2014-03-22 00:18:33 -04:00
|
|
|
conn.pipelined do
|
2014-03-08 17:10:27 -05:00
|
|
|
procs.each do |key|
|
|
|
|
conn.hget(key, 'busy')
|
|
|
|
end
|
|
|
|
end.map(&:to_i).inject(:+)
|
|
|
|
end
|
2013-01-24 12:50:30 -05:00
|
|
|
end
|
|
|
|
end
|
|
|
|
|
2012-10-20 17:03:43 -04:00
|
|
|
end
|