Class: Wurk::Metrics::Statsd

Inherits:
Object
  • Object
show all
Includes:
Wurk::Middleware::ServerMiddleware
Defined in:
lib/wurk/metrics/statsd.rb

Overview

Pro parity (§9): emits per-job timing + counters to a statsd / dogstatsd client. The client itself is plumbed in by the host app via:

Wurk.configure_server do |config|
config.dogstatsd = -> { Datadog::Statsd.new('metrics.example.com', 8125) }
config.server_middleware { |chain| chain.add Wurk::Metrics::Statsd }
end

The dogstatsd accessor is a callable — invoked once per process, memoized — so the client is built lazily AFTER fork. Sharing a UDP socket across forks is fine, but Datadog::Statsd keeps thread-locals that must be initialized inside the child.

Per-job tuning via Statsd.options = ->(klass, job, queue) { {tags:, sample_rate:} }. Default options: tags ["worker:<klass>", "queue:<q>"], sample_rate 1.0. The dd_rate job option, when present, overrides sample_rate.

Metric naming follows Sidekiq Pro 8+: every metric prefixed sidekiq. (the prefix is hardcoded, not configurable — third-party dashboards built for Sidekiq Pro work unchanged).

Statsd.increment(metric, tags:) is the class-level fast path used by other Wurk components (Buffered client, Expiry middleware, super_fetch recovery, Batch lifecycle). No-op when no client is configured so callers never have to guard.

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

Constant Summary collapse

METRIC_PREFIX = 
'sidekiq.'
DEFAULT_SAMPLE_RATE = 
1.0

Class Attribute Summary collapse

Attributes included from Wurk::Middleware::ServerMiddleware

#config

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Wurk::Middleware::ServerMiddleware

#logger, #redis, #redis_pool

Class Attribute Details

.options ⇒ Object

Returns the value of attribute options.



50
51
52
 
# File 'lib/wurk/metrics/statsd.rb', line 50

def options
  @options
end

Class Method Details

.client ⇒ Object

Resolves the live client: invokes the configured dogstatsd proc exactly once per process and memoizes. Returns nil when no proc is configured, so callers get a clean no-op without raising.

The unconfigured answer is memoized too, which is what makes the no-client path free: Client#emit_enqueued asks once per payload, so re-reading Wurk.configuration here would cost a config lookup per job on every bulk push. Both invalidation points call reset!.



110
111
112
113
114
115
116
 
# File 'lib/wurk/metrics/statsd.rb', line 110

def client
  memo = @client
  return memo unless UNSET.equal?(memo)

  builder = Wurk.configuration.respond_to?(:dogstatsd) ? Wurk.configuration.dogstatsd : nil
  @client = builder.respond_to?(:call) ? builder.call : builder
end

.distribution(metric, value, tags: nil, sample_rate: DEFAULT_SAMPLE_RATE) ⇒ Object

Distribution send. Some statsd clients lack distribution (vanilla statsd-ruby, for example) — fall back to histogram so the metric still lands somewhere. dogstatsd-ruby always has distribution.



84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
 
# File 'lib/wurk/metrics/statsd.rb', line 84

def distribution(metric, value, tags: nil, sample_rate: DEFAULT_SAMPLE_RATE)
  client = self.client
  return nil unless client

  opts = sample_rate_kw(sample_rate)
  opts[:tags] = tags if tags
  name = "#{METRIC_PREFIX}#{metric}"
  if client.respond_to?(:distribution)
    client.distribution(name, value, **opts)
  elsif client.respond_to?(:histogram)
    client.histogram(name, value, **opts)
  end
  nil
rescue StandardError => e
  handle_error(e)
  nil
end

.gauge(metric, value, tags: nil, sample_rate: DEFAULT_SAMPLE_RATE) ⇒ Object 



68
69
70
71
72
73
74
75
76
77
78
79
 
# File 'lib/wurk/metrics/statsd.rb', line 68

def gauge(metric, value, tags: nil, sample_rate: DEFAULT_SAMPLE_RATE)
  client = self.client
  return nil unless client

  opts = sample_rate_kw(sample_rate)
  opts[:tags] = tags if tags
  client.gauge("#{METRIC_PREFIX}#{metric}", value, **opts)
  nil
rescue StandardError => e
  handle_error(e)
  nil
end

.increment(metric, tags: nil, sample_rate: DEFAULT_SAMPLE_RATE) ⇒ Object

Counter shortcut used across the codebase. Tags are forwarded as given — caller's job to namespace them ("class:Foo", "queue:bar"). No-op when no client is wired up.



55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/wurk/metrics/statsd.rb', line 55

def increment(metric, tags: nil, sample_rate: DEFAULT_SAMPLE_RATE)
  client = self.client
  return nil unless client

  opts = sample_rate_kw(sample_rate)
  opts[:tags] = tags if tags
  client.increment("#{METRIC_PREFIX}#{metric}", **opts)
  nil
rescue StandardError => e
  handle_error(e)
  nil
end

.reset! ⇒ Object

Test/lifecycle hook. Reset between specs, after fork so the parent's socket doesn't bleed into children, and on every config.dogstatsd= — a memoized nil would otherwise hide a client configured later.



132
133
134
 
# File 'lib/wurk/metrics/statsd.rb', line 132

def reset!
  @client = UNSET
end

.safe_client ⇒ Object

client without the raise: a builder proc that blows up is reported through the error handler and treated as "no client", so misconfigured metrics can never fail the work they were measuring. Callers on a hot path use this to skip building anything the emit would drop.



122
123
124
125
126
127
 
# File 'lib/wurk/metrics/statsd.rb', line 122

def safe_client
  client
rescue StandardError => e
  handle_error(e)
  nil
end

Instance Method Details

#call(_worker, job, queue) ⇒ Object

rubocop:disable Metrics/AbcSize



147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
 
# File 'lib/wurk/metrics/statsd.rb', line 147

def call(_worker, job, queue) # rubocop:disable Metrics/AbcSize
  return yield if self.class.safe_client.nil?

  klass = job['class']
  opts  = per_job_options(klass, job, queue)
  tags  = opts[:tags]
  rate  = opts.fetch(:sample_rate, DEFAULT_SAMPLE_RATE)

  emit(:increment, 'jobs.count', tags: tags, sample_rate: rate)
  started = monotonic_ms
  success = false
  begin
    yield
    success = true
  rescue Wurk::Job::Interrupted, Wurk::Job::DeadlineExceeded
    # Same arm, same reasons as Metrics::History#call (#394): a cooperative
    # interruption passes through here before InterruptHandler turns it
    # into a JobRetry::Skip, and a job cut by its deadline passes through
    # before Middleware::Expiry books it `expired` — both sit outside this
    # middleware, and without this arm either one emits `jobs.failure`.
    # Pro's statsd emitter is closed source, so the oracle is the free
    # ExecutionTracker plus the rule that the two Wurk emitters must never
    # classify one event differently. Signed off in
    # docs/plans/2026/08/07/101-beyond-sidekiq/00-semantics-signoff.md §1.
    success = true
    raise
  ensure
    duration = monotonic_ms - started
    # Metrics are best-effort: an emit failure mid-finalize must not
    # corrupt the job result the caller already produced.
    begin
      finalize(success, duration, tags: tags, sample_rate: rate)
    rescue StandardError => e
      self.class.send(:handle_error, e)
    end
  end
end