Class: Sidekiq::Config

Inherits:
Object
  • Object
show all
Extended by:
Forwardable
Defined in:
lib/sidekiq/config.rb

Overview

Sidekiq::Config represents the global configuration for an instance of Sidekiq.

Constant Summary collapse

DEFAULTS = 
{
  labels: Set.new,
  require: ".",
  environment: nil,
  concurrency: 5,
  timeout: 25,
  poll_interval_average: nil,
  average_scheduled_poll_interval: 5,
  on_complex_arguments: :raise,
  error_handlers: [],
  death_handlers: [],
  lifecycle_events: {
    startup: [],
    quiet: [],
    shutdown: [],
    # triggers when we fire the first heartbeat on startup OR repairing a network partition
    heartbeat: [],
    # triggers on EVERY heartbeat call, every 10 seconds
    beat: []
  },
  dead_max_jobs: 10_000,
  dead_timeout_in_seconds: 180 * 24 * 60 * 60, # 6 months
  reloader: proc { |&block| block.call },
  backtrace_cleaner: ->(backtrace) { backtrace }
}
ERROR_HANDLER = 
->(ex, ctx) {
  cfg = ctx[:_config] || Sidekiq.default_configuration
  l = cfg.logger
  l.warn(Sidekiq.dump_json(ctx)) unless ctx.empty?
  l.warn("#{ex.class.name}: #{ex.message}")
  unless ex.backtrace.nil?
    backtrace = cfg[:backtrace_cleaner].call(ex.backtrace)
    l.warn(backtrace.join("\n"))
  end
}

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(options = {}) ⇒ Config

Returns a new instance of Config.



48
49
50
51
52
53
54
 
# File 'lib/sidekiq/config.rb', line 48

def initialize(options = {})
  @options = DEFAULTS.merge(options)
  @options[:error_handlers] << ERROR_HANDLER if @options[:error_handlers].empty?
  @directory = {}
  @redis_config = {}
  @capsules = {}
end

Instance Attribute Details

#capsules ⇒ Object (readonly)

Returns the value of attribute capsules.



57
58
59
 
# File 'lib/sidekiq/config.rb', line 57

def capsules
  @capsules
end

Instance Method Details

#average_scheduled_poll_interval=(interval) ⇒ Object

How frequently Redis should be checked by a random Sidekiq process for scheduled and retriable jobs. Each individual process will take turns by waiting some multiple of this value.

See sidekiq/scheduled.rb for an in-depth explanation of this value



209
210
211
 
# File 'lib/sidekiq/config.rb', line 209

def average_scheduled_poll_interval=(interval)
  @options[:average_scheduled_poll_interval] = interval
end

#capsule(name) {|cap| ... } ⇒ Object

register a new queue processing subsystem

Yields:

  • (cap) 


108
109
110
111
112
113
114
115
116
 
# File 'lib/sidekiq/config.rb', line 108

def capsule(name)
  nm = name.to_s
  cap = @capsules.fetch(nm) do
    cap = Sidekiq::Capsule.new(nm, self)
    @capsules[nm] = cap
  end
  yield cap if block_given?
  cap
end

#client_middleware {|@client_chain| ... } ⇒ Object

Yields:

  • (@client_chain) 


91
92
93
94
95
 
# File 'lib/sidekiq/config.rb', line 91

def client_middleware
  @client_chain ||= Sidekiq::Middleware::Chain.new(self)
  yield @client_chain if block_given?
  @client_chain
end

#concurrency ⇒ Object 



65
66
67
 
# File 'lib/sidekiq/config.rb', line 65

def concurrency
  default_capsule.concurrency
end

#concurrency=(val) ⇒ Object

LEGACY: edits the default capsule config.concurrency = 5



61
62
63
 
# File 'lib/sidekiq/config.rb', line 61

def concurrency=(val)
  default_capsule.concurrency = Integer(val)
end

#death_handlers ⇒ Object

Death handlers are called when all retries for a job have been exhausted and the job dies. It’s the notification to your application that this job will not succeed without manual intervention.

Sidekiq.configure_server do |config|

config.death_handlers << ->(job, ex) do
end

end



200
201
202
 
# File 'lib/sidekiq/config.rb', line 200

def death_handlers
  @options[:death_handlers]
end

#default_capsule(&block) ⇒ Object 



103
104
105
 
# File 'lib/sidekiq/config.rb', line 103

def default_capsule(&block)
  capsule("default", &block)
end

#error_handlers ⇒ Object

Register a proc to handle any error which occurs within the Sidekiq process.

Sidekiq.configure_server do |config|
  config.error_handlers << proc {|ex,ctx_hash| MyErrorService.notify(ex, ctx_hash) }
end

The default error handler logs errors to @logger.



220
221
222
 
# File 'lib/sidekiq/config.rb', line 220

def error_handlers
  @options[:error_handlers]
end

#handle_exception(ex, ctx = {}) ⇒ Object

INTERNAL USE ONLY



259
260
261
262
263
264
265
266
267
268
269
270
271
272
 
# File 'lib/sidekiq/config.rb', line 259

def handle_exception(ex, ctx = {})
  if @options[:error_handlers].size == 0
    p ["!!!!!", ex]
  end
  ctx[:_config] = self
  @options[:error_handlers].each do |handler|
    handler.call(ex, ctx)
  rescue => e
    l = logger
    l.error "!!! ERROR HANDLER THREW AN ERROR !!!"
    l.error e
    l.error e.backtrace.join("\n") unless e.backtrace.nil?
  end
end

#logger ⇒ Object 



238
239
240
241
242
243
244
245
246
247
 
# File 'lib/sidekiq/config.rb', line 238

def logger
  @logger ||= Sidekiq::Logger.new($stdout, level: :info).tap do |log|
    log.level = Logger::INFO
    log.formatter = if ENV["DYNO"]
      Sidekiq::Logger::Formatters::WithoutTimestamp.new
    else
      Sidekiq::Logger::Formatters::Pretty.new
    end
  end
end

#logger=(logger) ⇒ Object 



249
250
251
252
253
254
255
256
 
# File 'lib/sidekiq/config.rb', line 249

def logger=(logger)
  if logger.nil?
    self.logger.level = Logger::FATAL
    return
  end

  @logger = logger
end

#lookup(name, default_class = nil) ⇒ Object

find a singleton



183
184
185
186
187
188
189
 
# File 'lib/sidekiq/config.rb', line 183

def lookup(name, default_class = nil)
  # JNDI is just a fancy name for a hash lookup
  @directory.fetch(name) do |key|
    return nil unless default_class
    @directory[key] = default_class.new(self)
  end
end

#new_redis_pool(size, name = "unset") ⇒ Object 



133
134
135
136
137
 
# File 'lib/sidekiq/config.rb', line 133

def new_redis_pool(size, name = "unset")
  # connection pool is lazy, it will not create connections unless you actually need them
  # so don't be skimpy!
  RedisConnection.create({size: size, logger: logger, pool_name: name}.merge(@redis_config))
end

#on(event, &block) ⇒ Object

Register a block to run at a point in the Sidekiq lifecycle. :startup, :quiet or :shutdown are valid events.

Sidekiq.configure_server do |config|
  config.on(:shutdown) do
    puts "Goodbye cruel world!"
  end
end

Raises:

  • (ArgumentError) 


232
233
234
235
236
 
# File 'lib/sidekiq/config.rb', line 232

def on(event, &block)
  raise ArgumentError, "Symbols only please: #{event}" unless event.is_a?(Symbol)
  raise ArgumentError, "Invalid event name: #{event}" unless @options[:lifecycle_events].key?(event)
  @options[:lifecycle_events][event] << block
end

#queues ⇒ Object 



87
88
89
 
# File 'lib/sidekiq/config.rb', line 87

def queues
  default_capsule.queues
end

#queues=(val) ⇒ Object

Edit the default capsule. config.queues = %w( high default low ) # strict config.queues = %w( high,3 default,2 low,1 ) # weighted config.queues = %w( feature1,1 feature2,1 feature3,1 ) # random

With weighted priority, queue will be checked first (weight / total) of the time. high will be checked first (3/6) or 50% of the time. I’d recommend setting weights between 1-10. Weights in the hundreds or thousands are ridiculous and unnecessarily expensive. You can get random queue ordering by explicitly setting all weights to 1.



83
84
85
 
# File 'lib/sidekiq/config.rb', line 83

def queues=(val)
  default_capsule.queues = val
end

#redis ⇒ Object

Raises:

  • (ArgumentError) 


155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
 
# File 'lib/sidekiq/config.rb', line 155

def redis
  raise ArgumentError, "requires a block" unless block_given?
  redis_pool.with do |conn|
    retryable = true
    begin
      yield conn
    rescue RedisClientAdapter::BaseError => ex
      # 2550 Failover can cause the server to become a replica, need
      # to disconnect and reopen the socket to get back to the primary.
      # 4495 Use the same logic if we have a "Not enough replicas" error from the primary
      # 4985 Use the same logic when a blocking command is force-unblocked
      # The same retry logic is also used in client.rb
      if retryable && ex.message =~ /READONLY|NOREPLICAS|UNBLOCKED/
        conn.close
        retryable = false
        retry
      end
      raise
    end
  end
end

#redis=(hash) ⇒ Object

All capsules must use the same Redis configuration



119
120
121
 
# File 'lib/sidekiq/config.rb', line 119

def redis=(hash)
  @redis_config = @redis_config.merge(hash)
end

#redis_info ⇒ Object 



139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
 
# File 'lib/sidekiq/config.rb', line 139

def redis_info
  redis do |conn|
    conn.call("INFO") { |i| i.lines(chomp: true).map { |l| l.split(":", 2) }.select { |l| l.size == 2 }.to_h }
  rescue RedisClientAdapter::CommandError => ex
    # 2850 return fake version when INFO command has (probably) been renamed
    raise unless /unknown command/.match?(ex.message)
    {
      "redis_version" => "9.9.9",
      "uptime_in_days" => "9999",
      "connected_clients" => "9999",
      "used_memory_human" => "9P",
      "used_memory_peak_human" => "9P"
    }.freeze
  end
end

#redis_pool ⇒ Object 



123
124
125
 
# File 'lib/sidekiq/config.rb', line 123

def redis_pool
  Thread.current[:sidekiq_redis_pool] || Thread.current[:sidekiq_capsule]&.redis_pool || local_redis_pool
end

#register(name, instance) ⇒ Object

register global singletons which can be accessed elsewhere



178
179
180
 
# File 'lib/sidekiq/config.rb', line 178

def register(name, instance)
  @directory[name] = instance
end

#server_middleware {|@server_chain| ... } ⇒ Object

Yields:

  • (@server_chain) 


97
98
99
100
101
 
# File 'lib/sidekiq/config.rb', line 97

def server_middleware
  @server_chain ||= Sidekiq::Middleware::Chain.new(self)
  yield @server_chain if block_given?
  @server_chain
end

#total_concurrency ⇒ Object 



69
70
71
 
# File 'lib/sidekiq/config.rb', line 69

def total_concurrency
  capsules.each_value.sum(&:concurrency)
end