Class: Wurk::History

Inherits:
Object
  • Object
show all
Includes:
Component
Defined in:
lib/wurk/history.rb

Overview

Sidekiq Enterprise §5 Historical Metrics snapshotter. A leader-gated background thread that, every config.retain_history seconds, emits a statsd-shaped snapshot to the configured dogstatsd client — either the default §5.2 gauge set or a user-supplied collector block.

Configured in a server block:

Sidekiq.configure_server do |config|
config.dogstatsd = -> { Datadog::Statsd.new('localhost', 8125) }
config.retain_history(30)                      # default §5.2 gauges
# …or a custom collector:
config.retain_history(30) do |s|
  Sidekiq::Queue.all.each do |q|
    s.gauge("sidekiq.queue.size", q.size, tags: ["queue:#{q.name}"])
  end
end
end

The block receives the raw dogstatsd client s (quacks like Datadog::Statsd: gauge/count/histogram/batch) and writes fully-qualified sidekiq.* metric names itself, matching Sidekiq Ent. Leader-gated via the cluster dear-leader lock so exactly one process emits per cluster.

Every snapshot is also appended to the capped Redis stream history:metrics (§5.3) — the same key a migrated Sidekiq Ent install uses — so the dashboard's Historical view has a data source independent of any external statsd, and pre-existing Ent stream data renders without rewrite. The stream write happens whenever the snapshotter runs; the dogstatsd emit is skipped only when no client is configured.

Aliased as Sidekiq::History (drop-in contract). Spec: docs/target/sidekiq-ent.md §5.1–§5.3.

Constant Summary collapse

SNAPSHOT_FIELDS =

Stream field → Stats reader. Single source for both the history:metrics stream entry and the default §5.2 statsd gauge set (which prefixes sidekiq.). Order is the display order.

{
  'processed' => :processed,
  'failures' => :failed,
  'enqueued' => :enqueued,
  'retries' => :retry_size,
  'dead' => :dead_size,
  'scheduled' => :scheduled_size,
  'busy' => :workers_size
}.freeze
STREAM_CAP =

Approximate cap on retained snapshots (XADD MAXLEN ~). At the default 30s interval this is ~3.5 days of history; older points age out. ~ lets Redis trim in whole macro-nodes, so the actual length can briefly exceed the cap — matching Ent's best-effort retention.

10_000
STREAM_DEFAULT_LIMIT = 
1000

Constants included from Component

Component::DEFAULT_THREAD_PRIORITY, Component::LEADER_CACHE_TTL_MS, Component::PROCESS_NONCE

Instance Attribute Summary

Attributes included from Component

#config

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Component

#default_tag, #fire_event, #handle_exception, #hostname, #identity, #leader?, #logger, #mono_ms, #process_nonce, #real_ms, #redis, #safe_thread, #tid, #watchdog

Constructor Details

#initialize(config) ⇒ History

Returns a new instance of History.



65
66
67
68
69
70
71
 
# File 'lib/wurk/history.rb', line 65

def initialize(config)
  @config = config
  @collector = config.history_collector
  @stream_cap = config[:history_stream_cap] || STREAM_CAP
  @timer = TimerLoop.new(config.history_interval)
  @thread = nil
end

Class Method Details

.numeric(value) ⇒ Object

Coerce a stream field to Int/Float for charting; leave non-numeric Ent fields (e.g. a label) untouched so nothing is silently dropped.



126
127
128
129
130
131
 
# File 'lib/wurk/history.rb', line 126

def self.numeric(value)
  float = Float(value)
  (float % 1).zero? ? float.to_i : float
rescue ::ArgumentError, ::TypeError
  value
end

.parse_entry(entry_id, fields) ⇒ Object 



112
113
114
115
116
117
 
# File 'lib/wurk/history.rb', line 112

def self.parse_entry(entry_id, fields)
  pairs = fields.is_a?(::Array) ? fields.each_slice(2).to_h : fields
  point = { at: stream_epoch(entry_id) }
  pairs.each { |field, value| point[field.to_sym] = numeric(value) }
  point
end

.recent(limit: STREAM_DEFAULT_LIMIT) ⇒ Object

Most-recent snapshots from the history:metrics stream, oldest→newest. Each point is { at: <epoch seconds>, <field>: <numeric>, … }. Fields are read generically, so a migrated Sidekiq Ent install's entries render without rewrite regardless of which fields they carry.



106
107
108
109
110
 
# File 'lib/wurk/history.rb', line 106

def self.recent(limit: STREAM_DEFAULT_LIMIT)
  count = limit.to_i.clamp(1, STREAM_CAP)
  entries = Wurk.redis { |c| c.call('XREVRANGE', Keys::HISTORY_METRICS, '+', '-', 'COUNT', count) }
  entries.reverse.map { |entry_id, fields| parse_entry(entry_id, fields) }
end

.stream_epoch(entry_id) ⇒ Object

Redis stream IDs are "-"; the ms half is the snapshot time.



120
121
122
 
# File 'lib/wurk/history.rb', line 120

def self.stream_epoch(entry_id)
  entry_id.to_s.split('-', 2).first.to_i / 1000.0
end

Instance Method Details

#snapshot ⇒ Object

One snapshot, bypassing the leader gate and the sleep loop. Public so deterministic specs and a manual "snapshot now" can drive it directly. Always appends to the history:metrics stream (the dashboard's source); additionally emits to dogstatsd when a client is configured.



95
96
97
98
99
100
 
# File 'lib/wurk/history.rb', line 95

def snapshot
  values = collect_values
  record_stream(values)
  emit_statsd(values)
  nil
end

#start ⇒ Object 



73
74
75
 
# File 'lib/wurk/history.rb', line 73

def start
  @thread ||= safe_thread('history-snapshot') { @timer.run { tick } } # rubocop:disable Naming/MemoizedInstanceVariableName
end

#terminate ⇒ Object 



77
78
79
 
# File 'lib/wurk/history.rb', line 77

def terminate
  @timer.terminate
end

#tick ⇒ Object

Leader-gated: only the elected leader emits, so N workers don't each publish the same cluster-wide gauges every interval.



83
84
85
86
87
88
89
 
# File 'lib/wurk/history.rb', line 83

def tick
  return unless leader?

  snapshot
rescue StandardError => e
  handle_exception(e, { context: 'history-snapshot' })
end