Class: Wurk::Metrics::QueueRollup

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

Overview

Leader-only background thread that snapshots each queue's current depth (LLEN) and head-of-line latency into compact per-queue gauge buckets the dashboard's "queue size / latency over time" charts read directly:

qm|1m|<epoch>   HASH {<queue>|sz, <queue>|lt}   TTL 24h
qm|5m|<epoch>   HASH {<queue>|sz, <queue>|lt}   TTL 7d
qm|1h|<epoch>   HASH {<queue>|sz, <queue>|lt}   TTL 30d

Unlike Metrics::Rollup (which SUMS counters rolled up from a source), size and latency are GAUGES — point-in-time values — so each tick samples "now" and writes it to the current bucket at every resolution. Within a coarse (5m/1h) bucket the per-minute ticks overwrite, so the bucket holds the latest sample in its window (a "last value" downsample, which is the right summary for a gauge). Leader-gated so N workers don't each sample the same queues every minute. <epoch> is the UTC start-of-bucket.

Spec: docs/target/sidekiq-ent.md §5.2 (sidekiq.queue.size / sidekiq.queue.latency gauges), §7 Historical tab.

Constant Summary collapse

PREFIX = 
'qm'
SIZE_KIND = 
'sz'
LAT_KIND = 
'lt'
BUCKETS =

Mirror Metrics::Rollup retention so the dashboard's range selector (24h·1m / 7d·5m / 30d·1h) maps 1:1 to both the throughput and the queue-gauge series.

Wurk::Metrics::Rollup::BUCKETS
DEFAULT_TICK_SECONDS = 
60

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config) ⇒ QueueRollup

Returns a new instance of QueueRollup.



46
47
48
49
50
 
# File 'lib/wurk/metrics/queue_rollup.rb', line 46

def initialize(config)
  @config = config
  @timer = TimerLoop.new(config[:metrics_rollup_interval] || DEFAULT_TICK_SECONDS)
  @thread = nil
end

Instance Attribute Details

#config ⇒ Object (readonly) Originally defined in module Component

Returns the value of attribute config.

Class Method Details

.bucket_key(bucket, epoch) ⇒ Object 



42
43
44
 
# File 'lib/wurk/metrics/queue_rollup.rb', line 42

def self.bucket_key(bucket, epoch)
  "#{PREFIX}|#{bucket}|#{epoch}"
end

Instance Method Details

#default_tag(dir = Dir.pwd) ⇒ Object Originally defined in module Component

#fire_event(event, oneshot: true, reverse: false, reraise: false) ⇒ Object Originally defined in module Component

Invokes lifecycle hooks for event. Hooks run in registration order (or LIFO when reverse: true, used for teardown). A raise in one hook is reported via handle_exception and does NOT stop the next hook unless reraise: true (used in tests / fail-fast boot). oneshot: true clears the bucket after dispatch so the event can't fire twice.

#handle_exception(ex, ctx = {}) ⇒ Object Originally defined in module Component

#hostname ⇒ Object Originally defined in module Component

#identity ⇒ Object Originally defined in module Component

#leader? ⇒ Boolean Originally defined in module Component

True iff this process currently holds the cluster dear-leader lock. Cached per Component instance for LEADER_CACHE_TTL_MS (~5s): cron and the metrics rollups call this every tick, and an uncached GET would double their Redis traffic at short intervals for no benefit — the lock's own renewal cadence (60s+, spec §6.1) easily tolerates a few-second-stale read. Returns false unconditionally when WURK_LEADER=false (or SIDEKIQ_LEADER=false) is set on the process (opt-out hot-standby). Any Redis error is swallowed → false, so a transient partition can't propagate as an exception into user code.

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

Returns:

  • (Boolean)

#logger ⇒ Object Originally defined in module Component

--- delegated to config -------------------------------------------

#mono_ms ⇒ Object Originally defined in module Component

#process_nonce ⇒ Object Originally defined in module Component

#real_ms ⇒ Object Originally defined in module Component

--- clocks ---------------------------------------------------------

#redis(idempotent: false) ⇒ Object Originally defined in module Component

#safe_thread(name, priority: nil, &block) ⇒ Object Originally defined in module Component

Spawns a named thread that runs block under watchdog(name). The parent must retain the returned Thread; otherwise GC may not, but report_on_exception is disabled so we don't double-log on death.

Priority resolution matches Sidekiq (component.rb:44-48): explicit argument, then config.thread_priority, then -1. Ruby's default of 0 buys a 100ms timeslice; each negative step halves it, so -1 keeps a CPU-heavy capsule from starving its siblings for a whole tick.

#sample(now = ::Time.now) ⇒ Object

One sampling pass, bypassing the leader gate and the sleep loop. Public so deterministic specs and a manual "sample now" can drive it directly.

A zero gauge is not stored: readers default a missing field to 0, and an idle fleet's queues are mostly empty, so storing them cost two fields per queue in every bucket for nothing. It is HDEL'd instead of skipped, so a queue that drained inside a coarse bucket reads 0 there rather than keeping the earlier sample. All three buckets go in one pipeline.



91
92
93
94
95
96
97
98
99
 
# File 'lib/wurk/metrics/queue_rollup.rb', line 91

def sample(now = ::Time.now)
  gauges = queue_gauges
  return if gauges.empty?

  set, zero = gauge_fields(gauges)
  keys = bucket_keys(now)
  redis { |c| c.pipelined { |p| keys.each { |key, ttl| write_bucket(p, key, ttl, set, zero) } } }
  nil
end

#start ⇒ Object 



52
53
54
55
56
57
 
# File 'lib/wurk/metrics/queue_rollup.rb', line 52

def start
  return @thread if @thread

  @timer.reset
  @thread = safe_thread('queue-metrics') { @timer.run { tick } }
end

#terminate ⇒ Object

Blocks until the thread is really gone: the launcher releases the cluster lock immediately after this returns, and a sample still in flight would write the same buckets as the next leader's first one.

Cleared only on a confirmed join (Thread#join returns nil on timeout): a wedged thread must stay tracked so #start's guard returns it instead of calling @timer.reset, which would un-terminate the loop it is still inside and leave two threads HSETting the same buckets.



67
68
69
70
 
# File 'lib/wurk/metrics/queue_rollup.rb', line 67

def terminate
  @timer.terminate
  @thread = nil if @thread&.join(TimerLoop::JOIN_TIMEOUT)
end

#tick(now: ::Time.now) ⇒ Object

Leader-gated: only the elected leader samples, so N workers don't each HSET the same buckets every minute.



74
75
76
77
78
79
80
 
# File 'lib/wurk/metrics/queue_rollup.rb', line 74

def tick(now: ::Time.now)
  return unless leader?

  sample(now)
rescue StandardError => e
  handle_exception(e, { context: 'queue-metrics' })
end

#tid ⇒ Object Originally defined in module Component

#watchdog(last_words) ⇒ Object Originally defined in module Component

Wraps a block at a thread boundary: any unhandled exception is reported via handle_exception (so it lands in error_handlers / the log) and then re-raised. last_words is the component label included in the context.