Skip to content

API

Jump to bottom
Mike Perham edited this page May 4, 2024 · 61 revisions

Sidekiq has a public API allowing access to real-time information about workers, queues and jobs. See sidekiq/api for RDoc. Require the API to get access to all of the functionality described below.

require 'sidekiq/api'

The Web UI uses the API exclusively - anything you can do in the UI can be scripted with the API.

Overview

See my blog post at https://www.mikeperham.com/2021/04/20/a-tour-of-the-sidekiq-api/ for a good introduction to the API and its ethos.

Queue

Get all queues

Sidekiq::Queue.all

Get a queue

Sidekiq::Queue.new # the "default" queue
Sidekiq::Queue.new("mailer")

Gets the number of jobs within a queue.

Sidekiq::Queue.new.size # => 4

Deletes all Jobs in a Queue, by removing the queue.

Sidekiq::Queue.new.clear

Deletes jobs within the queue mailer with a jid of 'abcdef1234567890'

queue = Sidekiq::Queue.new("mailer")
queue.each do |job|
  job.klass # => 'MyWorker'
  job.args # => [1, 2, 3]
  # See also `display_class` and `display_args` methods if your job is wrapped by the ActiveJob adapter some other extension.
  # to get the original job class and only the job arguments.
  job.delete if job.jid == 'abcdef1234567890'
end

NOTE: The retrieval process is prone to race conditions. For example Sidekiq::Queue.new.each { |job| puts job.id } may skip jobs or print the same job(s) multiple times if other processes are updating the queue at the same time. This is expected behavior when mutating a shared data structure (i.e. the queue within Redis) without a lock.

Calculate the latency (in seconds) of a queue (now - when the oldest job was enqueued):

> Sidekiq::Queue.new.latency
14.5

Find a job by JID (WARNING: this is very inefficient if your queue is big!)

> Sidekiq::Queue.new.find_job(somejid)

Sidekiq Pro adds a few API extensions, including Sidekiq::Queue#delete_job and Sidekiq::Queue#delete_by_class.

Named Queues

Scheduled

The scheduled sorted set holds all scheduled jobs in chronologically-sorted order. There's much more in the code, see sidekiq/api for more detail.

ss = Sidekiq::ScheduledSet.new
ss.size
ss.clear

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 scheduled queue (inefficient).

ss = Sidekiq::ScheduledSet.new
# Use `scan` to filter elements in Redis, much faster
# This will return any job payloads that match "*SomeWorker*"
jobs = ss.scan("SomeWorker").select {|retri| retri.display_class == 'SomeWorker' }
# slower, pulls everything into Ruby
jobs = ss.select {|retri| retri.display_class == 'SomeWorker' }
jobs.each(&:delete)

Deleting a job by jid and timestamp (epoch)

time = Time.zone.now.to_f
jid = '80b1e7e46381a20c0c567285'
Sidekiq::ScheduledSet.new.delete_by_jid(time, jid)

Retries

When a job raises an error, Sidekiq places it in the RetrySet for automatic retry later. Jobs are sorted based on when they will next retry.

rs = Sidekiq::RetrySet.new
rs.size
rs.clear

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 (inefficient).

query = Sidekiq::RetrySet.new
query.select do |job|
  job.klass == 'Sidekiq::Extensions::DelayedClass' &&
    # For Sidekiq::Extensions (e.g., Foo.delay.bar(*args)),
    # the context is serialized to YAML, and must
    # be deserialized to get to the original args
    ((klass, method, args) = YAML.load(job.args[0])) &&
    klass == User &&
    method == :setup_new_subscriber
end.map(&:delete)

# a better approach using 6.0's `scan` to pre-filter
query.scan("setup_new_subscriber").select do |job|
  job.klass == 'Sidekiq::Extensions::DelayedClass' &&
    # For Sidekiq::Extensions (e.g., Foo.delay.bar(*args)),
    # the context is serialized to YAML, and must
    # be deserialized to get to the original args
    ((klass, method, args) = YAML.load(job.args[0])) &&
    klass == User &&
    method == :setup_new_subscriber
end.map(&:delete)

Dead

Like RetrySet and ScheduledSet, the DeadSet holds all jobs considered dead by Sidekiq, ordered by when they died. It supports the same basic operations as the others.

ds = Sidekiq::DeadSet.new
ds.size
ds.clear

Maybe you just deployed a long-overdue bugfix? Iterate through the DeadSet, find jobs of a certain characteristic and retry them.

ds.select do |job|
  job.display_class == 'FixedWorker' && job.args[0] == 123
end.map(&:retry)

Scan

Sidekiq 6.0 adds support for scanning the named sets by glob pattern, allowing you to filter out irrelevant jobs within Redis, much faster than doing it in Ruby. Scanning operates on the raw JSON payload so you need to ensure you are using an accurate pattern:

ss = Sidekiq::ScheduledSet.new
# match jobs of the given class
ss.scan("\"class\":\"HardWorker\"") { |job| ... }

# cleaner but not quite the same: "HardWorker" might appear in arguments or other elements of the job payload
ss.scan("HardWorker") { |job| ... }

If you don't pass a block, scan returns an Enumerator.

See Redis docs for more details on glob pattern matching - full regular expressions are not supported. For convenience Sidekiq wraps your pattern with a splat (*) on either side unless you include a *, in which case you must wrap the string yourself if you need it.

Processes

Sidekiq::ProcessSet gets you access to near real-time (updated every 5 sec) info about the current set of Sidekiq processes running. You can remotely control the processes also:

ps = Sidekiq::ProcessSet.new
ps.size # => 2
ps.each do |process|
  p process['busy']     # => 3
  p process['hostname'] # => 'myhost.local'
  p process['pid']      # => 16131
end
ps.each(&:quiet!) # equivalent to the TSTP signal (USR1 for version < 5)
ps.each(&:stop!) # equivalent to the TERM signal
ps.each(&:dump_threads) # equivalent to the TTIN signal

Remote control can be useful in situations where signals are not supported: Windows, JRuby and the JVM or Heroku for instance.

Workers

Programmatic access to the current active worker set for all Sidekiq processes. A 'worker' is defined as a thread currently processing a job.

Note: This data changes asynchronously, updating every 5 seconds. It is not millisecond-precise.

worker_set = Sidekiq::WorkSet.new
worker_set.size # => 2
worker_set.each do |process_id, thread_id, work|
  # process_id is a unique identifier per Sidekiq process
  # thread_id is a unique identifier per thread
  # work is a Hash which looks like:
  # { 'queue' => name, 'run_at' => timestamp, 'payload' => msg }
  # run_at is an epoch Integer.
  # payload is a Hash which looks like:
  # { 'retry' => true,
  #   'queue' => 'default',
  #   'class' => 'Redacted',
  #   'args' => [1, 2, 'foo'],
  #   'jid' => '80b1e7e46381a20c0c567285',
  #   'enqueued_at' => 1427811033.2067106 }
end

Stats

Various stats about your Sidekiq installation.

stats = Sidekiq::Stats.new
stats.processed # => 100
stats.failed # => 3
stats.queues # => { "default" => 1001, "email" => 50 }

Gets the number of jobs enqueued in all queues (does NOT include retries and scheduled jobs).

stats.enqueued # => 5

Stats History

All dates are UTC and history stats are cleared after 5 years.

Get history of failed/processed stats:

s = Sidekiq::Stats::History.new(2) # Indicates how many days of data you want starting from today (UTC)
s.failed # => { "2012-12-05" => 120, "2012-12-04" => 234 }
s.processed # => { "2012-12-05" => 1010, "2012-12-04" => 1500 }

Start from a different date:

s = Sidekiq::Stats::History.new( 3, Date.parse("2012-12-3") )
s.failed # => { "2012-12-03" => 10, "2012-12-02" => 24, "2012-12-01" => 4 }
s.processed # => { "2012-12-03" => 124, "2012-12-02" => 345, "2012-12-01" => 355 }

Notes

  • The API exposes some operations which are not scalable and should not be used in an automated fashion or in bulk as part of your application functionality.
  • The API, at its core, is operating on shared mutable data structures within Redis without locks. This means that any iteration through the structure is best effort and results cannot be guaranteed. For example, if you call Sidekiq::Queue#find_job(jid), the call might miss a job if the queue is mutated during iteration.
  • A mostly quiet system may never notice these race conditions but a large, busy system likely will. You should not be scanning through queues/sets and deleting jobs unless something has gone wrong and you need to repair data manually.
  • Sidekiq Pro adds a few API extensions, such as Sidekiq::Queue#delete_job, Sidekiq::Queue#delete_by_class, and an ability to pause queues.

Previous: Monitoring Next: Middleware

Home | The Basics | Best Practices | Using Redis | Error Handling | Advanced Options | Problems?

Clone this wiki locally