Module: Sidekiq::Routing

Defined in:
lib/sidekiq/routing.rb,
lib/sidekiq/routing/mover.rb,
lib/sidekiq/routing/store.rb,
lib/sidekiq/routing/sweeper.rb,
lib/sidekiq/routing/version.rb,
lib/sidekiq/routing/configuration.rb,
lib/sidekiq/routing/web_extension.rb,
lib/sidekiq/routing/parked_processor.rb,
lib/sidekiq/routing/middleware/client.rb,
lib/sidekiq/routing/middleware/server.rb,
lib/sidekiq/routing/queue_composition.rb

Defined Under Namespace

Modules: Auto, Middleware, Mover, Store, WebExtension Classes: Configuration, ParkedProcessor, QueueComposition, Sweeper

Constant Summary collapse

PARKED_QUEUE_DEFAULT = 
"routing_parked"
ORIGINAL_QUEUE_KEY =

Keys stamped into the job payload.

"routing_original_queue"
NO_DIVERT_KEY = 
"routing_no_divert"
MODE_PARK = 
"park"
MODE_BLACKHOLE = 
"blackhole"
MODES = 
[MODE_PARK, MODE_BLACKHOLE].freeze
EMPTY_ROUTES = 
{}.freeze
VERSION = 
"0.1.2"

Class Method Summary collapse

Class Method Details

.blackhole(klass) ⇒ Object 



81
82
83
 
# File 'lib/sidekiq/routing.rb', line 81

def blackhole(klass)
  write(klass, MODE_BLACKHOLE)
end

.class_name(klass_or_name) ⇒ Object

Resolve the effective job-class name. Prefers the ActiveJob "wrapped" class so a wrapped job is matched by its real class, not the JobWrapper (mirrors Sidekiq's own display_class). Accepts a Class, String, or nil.



171
172
173
174
175
 
# File 'lib/sidekiq/routing.rb', line 171

def class_name(klass_or_name)
  return klass_or_name if klass_or_name.is_a?(String)

  klass_or_name&.name
end

.configuration ⇒ Object 



36
37
38
 
# File 'lib/sidekiq/routing.rb', line 36

def configuration
  @_configuration ||= Configuration.new
end

.enabled? ⇒ Boolean

Returns:

  • (Boolean) 


63
64
65
 
# File 'lib/sidekiq/routing.rb', line 63

def enabled?
  configuration.enabled
end

.install! ⇒ Object

Registers the manual routing client + server middleware on both client and server configurations. This mirrors Sidekiq::Lock.install! so host apps can add the gem first and opt in to routing per app.



47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/sidekiq/routing.rb', line 47

def install!
  require "sidekiq/routing/middleware/client"
  require "sidekiq/routing/middleware/server"

  prepend_routing = ->(chain) { chain.prepend Middleware::Client }

  Sidekiq.configure_client do |config|
    config.client_middleware(&prepend_routing)
  end

  Sidekiq.configure_server do |config|
    config.client_middleware(&prepend_routing)
    config.server_middleware { |chain| install_server_middleware(chain) }
  end
end

.logger ⇒ Object 



67
68
69
 
# File 'lib/sidekiq/routing.rb', line 67

def logger
  configuration.logger
end

.mode(klass) ⇒ Object 



101
102
103
 
# File 'lib/sidekiq/routing.rb', line 101

def mode(klass)
  Store.fetch(class_name(klass))&.fetch("mode", nil)
end

.park(klass) ⇒ Object

---- operator API: managing manual routes ----



77
78
79
 
# File 'lib/sidekiq/routing.rb', line 77

def park(klass)
  write(klass, MODE_PARK)
end

.parked?(klass) ⇒ Boolean

Returns:

  • (Boolean) 


93
94
95
 
# File 'lib/sidekiq/routing.rb', line 93

def parked?(klass)
  mode(klass) == MODE_PARK
end

.parked_breakdown(sample: configuration.breakdown_sample_size) ⇒ Object

{ "SomeJob" => { "count" => 12, "by_original_queue" => { "within_1_minute" => 12 } } }

Scans at most sample jobs (default Configuration#breakdown_sample_size), not the whole queue: the parking queue can hold millions during a flood and this is called on every Web tab load. The result is a distribution over the sampled head, not exact totals — use parked_size (O(1) LLEN) for the true total. Pass sample: nil to scan all.



143
144
145
146
147
148
149
150
151
152
153
154
155
156
 
# File 'lib/sidekiq/routing.rb', line 143

def parked_breakdown(sample: configuration.breakdown_sample_size)
  result = Hash.new { |h, k| h[k] = { "count" => 0, "by_original_queue" => Hash.new(0) } }
  scanned = 0
  Sidekiq::Queue.new(parked_queue).each do |job|
    break if sample && scanned >= sample

    klass = job.display_class
    original = job.item[ORIGINAL_QUEUE_KEY] || "unknown"
    result[klass]["count"] += 1
    result[klass]["by_original_queue"][original] += 1
    scanned += 1
  end
  result
end

.parked_queue ⇒ Object 



71
72
73
 
# File 'lib/sidekiq/routing.rb', line 71

def parked_queue
  configuration.parked_queue
end

.parked_size ⇒ Object 



133
134
135
 
# File 'lib/sidekiq/routing.rb', line 133

def parked_size
  Sidekiq::Queue.new(parked_queue).size
end

.process_parked(klass: nil, limit: nil, batch_size: nil) ⇒ Object 



164
165
166
 
# File 'lib/sidekiq/routing.rb', line 164

def process_parked(klass: nil, limit: nil, batch_size: nil)
  ParkedProcessor.new.call(klass: klass && class_name(klass), limit:, batch_size:)
end

.queue_composition(queue_name, scan_limit: QueueComposition::DEFAULT_SCAN_LIMIT) ⇒ Object

---- queue introspection ----



129
130
131
 
# File 'lib/sidekiq/routing.rb', line 129

def queue_composition(queue_name, scan_limit: QueueComposition::DEFAULT_SCAN_LIMIT)
  QueueComposition.new(queue_name, scan_limit:).call
end

.reset_cache! ⇒ Object 



120
121
122
123
124
125
 
# File 'lib/sidekiq/routing.rb', line 120

def reset_cache!
  @snapshot_mutex.synchronize do
    @snapshot = nil
    @snapshot_at = nil
  end
end

.route_for(klass_or_name) ⇒ Object

Returns the route hash ("mode"=>...) for a class, or nil. Reads from a process-local snapshot of all routes, refreshed at most once per cache_ttl_seconds.



116
117
118
 
# File 'lib/sidekiq/routing.rb', line 116

def route_for(klass_or_name)
  snapshot[class_name(klass_or_name)]
end

.routed?(klass) ⇒ Boolean

Returns:

  • (Boolean) 


97
98
99
 
# File 'lib/sidekiq/routing.rb', line 97

def routed?(klass)
  !Store.fetch(class_name(klass)).nil?
end

.routes ⇒ Object

All active manual routes, read straight from Redis (uncached) so operators and the Web tab always see the truth.



107
108
109
 
# File 'lib/sidekiq/routing.rb', line 107

def routes
  Store.all
end

.setup {|configuration| ... } ⇒ Object

Yields:



40
41
42
 
# File 'lib/sidekiq/routing.rb', line 40

def setup
  yield configuration if block_given?
end

.sweep(klass, queue: nil, limit: nil, batch_size: nil) ⇒ Object

---- recovery (thin wrappers; logic in Sweeper/ParkedProcessor) ----



160
161
162
 
# File 'lib/sidekiq/routing.rb', line 160

def sweep(klass, queue: nil, limit: nil, batch_size: nil)
  Sweeper.new.call(class_name(klass), queue:, limit:, batch_size:)
end

.unpark(klass) ⇒ Object 



85
86
87
88
89
90
91
 
# File 'lib/sidekiq/routing.rb', line 85

def unpark(klass)
  name = class_name(klass)
  Store.delete(name)
  reset_cache!
  logger.warn("[Routing] unparked #{name}")
  name
end