Class: Wurk::Configuration

Inherits:
Object
  • Object
show all
Defined in:
lib/wurk/configuration.rb

Overview

Owns runtime knobs (concurrency, queues, timeouts, lifecycle events, error/death handlers) and the registry of Capsules. Single source of truth for everything the swarm / managers / processors need to boot.

Spec: docs/target/sidekiq-free.md §4 (Sidekiq::Config).

Constant Summary collapse

DEFAULTS =

Mirrors Sidekiq::Config::DEFAULTS. Order and keys are part of the drop-in contract — third-party gems read @options via [] / fetch / dig.

{
  labels: Set.new,
  require: '.',
  environment: nil,
  concurrency: 5,
  timeout: 25,
  poll_interval_average: nil,
  average_scheduled_poll_interval: 5,
  on_complex_arguments: :raise,
  max_iteration_runtime: nil,
  error_handlers: [],
  death_handlers: [],
  lifecycle_events: {
    startup: [],
    quiet: [],
    shutdown: [],
    exit: [],
    heartbeat: [],
    beat: [],
    leader: []
  },
  dead_max_jobs: 10_000,
  dead_timeout_in_seconds: 180 * 24 * 60 * 60,
  reloader: proc { |&b| b.call },
  backtrace_cleaner: ->(bt) { bt },
  logged_job_attributes: %w[bid tags],
  redis_idle_timeout: nil
}.freeze
LIFECYCLE_EVENTS = 
%i[startup quiet shutdown exit heartbeat beat leader].freeze
DEFAULT_THREAD_PRIORITY = 
-1
ERROR_HANDLER =

Default error handler. Wraps the report in the thread-local Wurk::Context so logger formatters/JSON layouts can pick up jid/bid/tags. ‘full_message` (with backtrace) in dev/debug, `detailed_message` in prod —mirrors the Sidekiq behavior so log scrapers built for one work for both.

Spec: docs/target/sidekiq-free.md §4.3.

lambda do |ex, ctx, cfg = Wurk.configuration|
  safe_ctx = ctx || {}
  Wurk::Context.with(safe_ctx) do
    dev = $DEBUG || ENV['WURK_DEBUG'] || cfg.logger.debug?
    msg = dev ? ex.full_message : ex.detailed_message
    cfg.logger.info { msg }
  end
end

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(options = {}) ⇒ Configuration

Returns a new instance of Configuration.



77
78
79
80
81
82
83
84
85
86
87
88
 
# File 'lib/wurk/configuration.rb', line 77

def initialize(options = {})
  @options = deep_dup_defaults.merge(options)
  @options[:error_handlers] << ERROR_HANDLER if @options[:error_handlers].empty?
  @capsules = {}
  @directory = {}
  @client_chain = Middleware::Chain.new
  @server_chain = Middleware::Chain.new
  @redis_config = { url: ENV.fetch('REDIS_URL', 'redis://localhost:6379/0') }
  @logger = nil
  @thread_priority = DEFAULT_THREAD_PRIORITY
  @frozen = false
end

Instance Attribute Details

#capsules ⇒ Object (readonly)

Returns the value of attribute capsules.



65
66
67
 
# File 'lib/wurk/configuration.rb', line 65

def capsules
  @capsules
end

#directory ⇒ Object (readonly)

Returns the value of attribute directory.



65
66
67
 
# File 'lib/wurk/configuration.rb', line 65

def directory
  @directory
end

#dogstatsd ⇒ Object

Pro parity: callable that builds the statsd / dogstatsd client. Invoked once per process AFTER fork; see Wurk::Metrics::Statsd.client. Assignable as a Proc, lambda, or any object responding to #call:

config.dogstatsd = -> { Datadog::Statsd.new('host', 8125) }

Spec: docs/target/sidekiq-pro.md §9.1.



75
76
77
 
# File 'lib/wurk/configuration.rb', line 75

def dogstatsd
  @dogstatsd
end

#logger ⇒ Object

— Logger ———————————————————–



270
271
272
 
# File 'lib/wurk/configuration.rb', line 270

def logger
  @logger ||= default_logger
end

#redis_config ⇒ Object (readonly)

Returns the value of attribute redis_config.



65
66
67
 
# File 'lib/wurk/configuration.rb', line 65

def redis_config
  @redis_config
end

#thread_priority ⇒ Object

Returns the value of attribute thread_priority.



66
67
68
 
# File 'lib/wurk/configuration.rb', line 66

def thread_priority
  @thread_priority
end

Instance Method Details

#[](key) ⇒ Object

— Hash-like options access —————————————–



92
 
# File 'lib/wurk/configuration.rb', line 92

def [](key) = @options.[](key)

#[]=(key, val) ⇒ Object 



94
95
96
97
 
# File 'lib/wurk/configuration.rb', line 94

def []=(key, val)
  guard_frozen!
  @options[key] = val
end

#average_scheduled_poll_interval=(interval) ⇒ Object 



203
204
205
 
# File 'lib/wurk/configuration.rb', line 203

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

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

Yields:

  • (cap) 


131
132
133
134
135
136
 
# File 'lib/wurk/configuration.rb', line 131

def capsule(name)
  name = name.to_s
  cap = @capsules[name] ||= Capsule.new(name, self)
  yield cap if block_given?
  cap
end

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

— Middleware ——————————————————-

Yields:

  • (@client_chain) 


140
141
142
143
 
# File 'lib/wurk/configuration.rb', line 140

def client_middleware
  yield @client_chain if block_given?
  @client_chain
end

#concurrency ⇒ Object

— Default capsule shortcuts —————————————



111
 
# File 'lib/wurk/configuration.rb', line 111

def concurrency = default_capsule.concurrency

#concurrency=(val) ⇒ Object 



113
114
115
 
# File 'lib/wurk/configuration.rb', line 113

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

#configure_client {|_self| ... } ⇒ Object

Yields:

  • (_self)

Yield Parameters:



294
295
296
 
# File 'lib/wurk/configuration.rb', line 294

def configure_client(&block)
  yield self if block && !server?
end

#configure_server {|_self| ... } ⇒ Object

— Configure blocks (Sidekiq.configure_server / _client) ———–

Yields:

  • (_self)

Yield Parameters:



290
291
292
 
# File 'lib/wurk/configuration.rb', line 290

def configure_server(&block)
  yield self if block && server?
end

#death_handlers ⇒ Object 



199
200
201
 
# File 'lib/wurk/configuration.rb', line 199

def death_handlers
  @options[:death_handlers]
end

#default_capsule ⇒ Object 



127
128
129
 
# File 'lib/wurk/configuration.rb', line 127

def default_capsule(&)
  capsule('default', &)
end

#dig(*keys) ⇒ Object 



107
 
# File 'lib/wurk/configuration.rb', line 107

def dig(*keys) = @options.dig(*keys)

#error_handlers ⇒ Object

— Handlers ———————————————————



195
196
197
 
# File 'lib/wurk/configuration.rb', line 195

def error_handlers
  @options[:error_handlers]
end

#fetch ⇒ Object 



99
 
# File 'lib/wurk/configuration.rb', line 99

def fetch(*, &) = @options.fetch(*, &)

#freeze! ⇒ Object 



315
316
317
318
319
320
321
322
323
324
 
# File 'lib/wurk/configuration.rb', line 315

def freeze!
  return self if @frozen

  @capsules.each_value(&:freeze)
  @capsules.freeze
  @options.freeze
  @directory.freeze
  @frozen = true
  self
end

#frozen? ⇒ Boolean

Returns:

  • (Boolean) 


326
327
328
 
# File 'lib/wurk/configuration.rb', line 326

def frozen?
  @frozen
end

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



276
277
278
279
280
281
282
283
284
285
286
 
# File 'lib/wurk/configuration.rb', line 276

def handle_exception(ex, ctx = {})
  if error_handlers.empty?
    logger.error("#{ctx} #{ex.class}: #{ex.message}")
  else
    error_handlers.each do |handler|
      handler.call(ex, ctx, self)
    rescue StandardError => e
      logger.error("error_handler raised: #{e.class}: #{e.message}")
    end
  end
end

#health_check(port:, bind: '0.0.0.0', ready_window: 30) ⇒ Object

Opt-in thin HTTP listener inside the worker process for k8s probes. When called, the Launcher will start a TCP server on ‘port` bound to `bind` exposing `GET /live` (200 while not stopping) and `GET /ready` (200 only when Redis is reachable AND heartbeat fired within `ready_window` seconds).

Off by default — call this in a ‘configure_server` block to enable. Spec: docs/target/sidekiq-ent.md §7.1.2.

Raises:

  • (ArgumentError) 


244
245
246
247
248
249
250
251
252
253
254
255
 
# File 'lib/wurk/configuration.rb', line 244

def health_check(port:, bind: '0.0.0.0', ready_window: 30)
  guard_frozen!
  p = Integer(port)
  rw = Integer(ready_window)
  raise ArgumentError, 'port must be between 0 and 65535' unless (0..65535).cover?(p)
  raise ArgumentError, 'ready_window must be > 0' unless rw.positive?

  b = bind.to_s
  raise ArgumentError, 'bind must be a non-empty string' if b.empty?

  @options[:health_check_options] = { port: p, bind: b, ready_window: rw }
end

#inspect ⇒ Object 



330
331
332
 
# File 'lib/wurk/configuration.rb', line 330

def inspect
  "#<#{self.class} capsules=#{@capsules.keys} concurrency=#{total_concurrency}>"
end

#key?(key) ⇒ Boolean Also known as: has_key?

Returns:

  • (Boolean) 


100
 
# File 'lib/wurk/configuration.rb', line 100

def key?(key) = @options.key?(key)

#local_redis_pool ⇒ Object 



161
162
163
 
# File 'lib/wurk/configuration.rb', line 161

def local_redis_pool
  @local_redis_pool ||= build_redis_pool(size: 10, name: 'internal')
end

#lookup(name, default_class = nil) ⇒ Object 



189
190
191
 
# File 'lib/wurk/configuration.rb', line 189

def lookup(name, default_class = nil)
  @directory[name] ||= default_class&.new
end

#merge!(other) ⇒ Object 



102
103
104
105
 
# File 'lib/wurk/configuration.rb', line 102

def merge!(other)
  guard_frozen!
  @options.merge!(other)
end

#new_redis_pool(size, name = 'custom') ⇒ Object 



174
175
176
 
# File 'lib/wurk/configuration.rb', line 174

def new_redis_pool(size, name = 'custom')
  build_redis_pool(size: size, name: name)
end

#on(event, &block) ⇒ Object

— Lifecycle hooks ————————————————–

Raises:

  • (ArgumentError) 


259
260
261
262
263
264
265
266
 
# File 'lib/wurk/configuration.rb', line 259

def on(event, &block)
  raise ArgumentError, "block required for on(#{event.inspect})" unless block
  unless LIFECYCLE_EVENTS.include?(event)
    raise ArgumentError, "invalid event #{event.inspect}, must be one of #{LIFECYCLE_EVENTS.inspect}"
  end

  @options[:lifecycle_events][event] << block
end

#periodic {|@periodic_manager| ... } ⇒ Object

Yields a Wurk::Cron::Manager so the host app can register periodic jobs at boot. Manager state is shared per-process so multiple ‘config.periodic` blocks accumulate (matches Sidekiq Ent §2.1).

Spec: docs/target/sidekiq-ent.md §2.

Yields:

  • (@periodic_manager) 


214
215
216
217
218
219
 
# File 'lib/wurk/configuration.rb', line 214

def periodic
  require_relative 'cron'
  @periodic_manager ||= Wurk::Cron::Manager.new(self)
  yield @periodic_manager if block_given?
  @periodic_manager
end

#queues ⇒ Object 



117
 
# File 'lib/wurk/configuration.rb', line 117

def queues = default_capsule.queues

#queues=(val) ⇒ Object 



119
120
121
 
# File 'lib/wurk/configuration.rb', line 119

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

#redis ⇒ Object 



178
179
180
 
# File 'lib/wurk/configuration.rb', line 178

def redis(&)
  redis_pool.with(&)
end

#redis=(hash) ⇒ Object

— Redis ————————————————————



152
153
154
155
 
# File 'lib/wurk/configuration.rb', line 152

def redis=(hash)
  guard_frozen!
  @redis_config = @redis_config.merge(hash.transform_keys(&:to_sym))
end

#redis_pool ⇒ Object 



157
158
159
 
# File 'lib/wurk/configuration.rb', line 157

def redis_pool
  default_capsule.redis_pool
end

#register(name, instance) ⇒ Object

— Service locator (extension registry) —————————-



184
185
186
187
 
# File 'lib/wurk/configuration.rb', line 184

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

#reset_redis_pools! ⇒ Object

Disconnect and drop every cached pool — the per-capsule mains plus the config-level internal pool. Used by Wurk::Swarm so the parent never leaks sockets into forks and each child can build fresh ones.



168
169
170
171
172
 
# File 'lib/wurk/configuration.rb', line 168

def reset_redis_pools!
  @capsules.each_value(&:reset_redis_pools!)
  @local_redis_pool&.disconnect!
  @local_redis_pool = nil
end

#server? ⇒ Boolean

Returns:

  • (Boolean) 


298
299
300
 
# File 'lib/wurk/configuration.rb', line 298

def server?
  @options[:server] == true
end

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

Yields:

  • (@server_chain) 


145
146
147
148
 
# File 'lib/wurk/configuration.rb', line 145

def server_middleware
  yield @server_chain if block_given?
  @server_chain
end

#topology ⇒ Object

Worker topology for the swarm. When the host hasn’t declared one (the railtie path), default to a single flat fork running the default capsule’s queues + concurrency. Assign a custom Wurk::Topology (via ‘topology=`) for specialized slots.



306
307
308
 
# File 'lib/wurk/configuration.rb', line 306

def topology
  @topology ||= default_topology
end

#topology=(value) ⇒ Object 



310
311
312
313
 
# File 'lib/wurk/configuration.rb', line 310

def topology=(value)
  guard_frozen!
  @topology = value
end

#total_concurrency ⇒ Object 



123
124
125
 
# File 'lib/wurk/configuration.rb', line 123

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

#web ⇒ Object

Web UI configuration: the authorization hook and read-only mode. Returns the process-wide ‘Wurk::Web.config` singleton so `config.web.read_only = true` and the engine middleware share one source of truth. Lazy-requires the web layer to keep standalone boot lean.

Spec: docs/target/sidekiq-ent.md §9.2.



229
230
231
232
 
# File 'lib/wurk/configuration.rb', line 229

def web
  require_relative 'web/config'
  Wurk::Web.config
end