class Sidekiq::Stats

Retrieve runtime statistics from Redis regarding this Sidekiq cluster.

stat = Sidekiq::Stats.new
stat.processed

Public Class Methods

new() click to toggle source
# File lib/sidekiq/api.rb, line 28
def initialize
  fetch_stats_fast!
end

Public Instance Methods

dead_size() click to toggle source
# File lib/sidekiq/api.rb, line 48
def dead_size
  stat :dead_size
end
default_queue_latency() click to toggle source
# File lib/sidekiq/api.rb, line 64
def default_queue_latency
  stat :default_queue_latency
end
enqueued() click to toggle source
# File lib/sidekiq/api.rb, line 52
def enqueued
  stat :enqueued
end
failed() click to toggle source
# File lib/sidekiq/api.rb, line 36
def failed
  stat :failed
end
fetch_stats!() click to toggle source

@api private

# File lib/sidekiq/api.rb, line 151
def fetch_stats!
  fetch_stats_fast!
  fetch_stats_slow!
end
fetch_stats_fast!() click to toggle source

O(1) redis calls @api private

# File lib/sidekiq/api.rb, line 85
def fetch_stats_fast!
  pipe1_res = Sidekiq.redis { |conn|
    conn.pipelined do |pipeline|
      pipeline.get("stat:processed")
      pipeline.get("stat:failed")
      pipeline.zcard("schedule")
      pipeline.zcard("retry")
      pipeline.zcard("dead")
      pipeline.scard("processes")
      pipeline.lindex("queue:default", -1)
    end
  }

  default_queue_latency = if (entry = pipe1_res[6])
    job = begin
      Sidekiq.load_json(entry)
    rescue
      {}
    end
    now = Time.now.to_f
    thence = job["enqueued_at"] || now
    now - thence
  else
    0
  end

  @stats = {
    processed: pipe1_res[0].to_i,
    failed: pipe1_res[1].to_i,
    scheduled_size: pipe1_res[2],
    retry_size: pipe1_res[3],
    dead_size: pipe1_res[4],
    processes_size: pipe1_res[5],

    default_queue_latency: default_queue_latency
  }
end
fetch_stats_slow!() click to toggle source

O(number of processes + number of queues) redis calls @api private

# File lib/sidekiq/api.rb, line 125
def fetch_stats_slow!
  processes = Sidekiq.redis { |conn|
    conn.sscan("processes").to_a
  }

  queues = Sidekiq.redis { |conn|
    conn.sscan("queues").to_a
  }

  pipe2_res = Sidekiq.redis { |conn|
    conn.pipelined do |pipeline|
      processes.each { |key| pipeline.hget(key, "busy") }
      queues.each { |queue| pipeline.llen("queue:#{queue}") }
    end
  }

  s = processes.size
  workers_size = pipe2_res[0...s].sum(&:to_i)
  enqueued = pipe2_res[s..].sum(&:to_i)

  @stats[:workers_size] = workers_size
  @stats[:enqueued] = enqueued
  @stats
end
processed() click to toggle source
# File lib/sidekiq/api.rb, line 32
def processed
  stat :processed
end
processes_size() click to toggle source
# File lib/sidekiq/api.rb, line 56
def processes_size
  stat :processes_size
end
queues() click to toggle source
# File lib/sidekiq/api.rb, line 68
def queues
  Sidekiq.redis do |conn|
    queues = conn.sscan("queues").to_a

    lengths = conn.pipelined { |pipeline|
      queues.each do |queue|
        pipeline.llen("queue:#{queue}")
      end
    }

    array_of_arrays = queues.zip(lengths).sort_by { |_, size| -size }
    array_of_arrays.to_h
  end
end
reset(*stats) click to toggle source

@api private

# File lib/sidekiq/api.rb, line 157
def reset(*stats)
  all = %w[failed processed]
  stats = stats.empty? ? all : all & stats.flatten.compact.map(&:to_s)

  mset_args = []
  stats.each do |stat|
    mset_args << "stat:#{stat}"
    mset_args << 0
  end
  Sidekiq.redis do |conn|
    conn.mset(*mset_args)
  end
end
retry_size() click to toggle source
# File lib/sidekiq/api.rb, line 44
def retry_size
  stat :retry_size
end
scheduled_size() click to toggle source
# File lib/sidekiq/api.rb, line 40
def scheduled_size
  stat :scheduled_size
end
workers_size() click to toggle source
# File lib/sidekiq/api.rb, line 60
def workers_size
  stat :workers_size
end

Private Instance Methods

stat(s) click to toggle source
# File lib/sidekiq/api.rb, line 173
def stat(s)
  fetch_stats_slow! if @stats[s].nil?
  @stats[s] || raise(ArgumentError, "Unknown stat #{s}")
end