Class: Wurk::Metrics::Accumulator

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

Overview

The in-memory half of Wurk::Metrics::History. Every execution folds into a {pool => {minute => {class => [processed, failed, ms, histogram]}}} tree under one mutex; Wurk::Metrics::Flusher drains it into Redis on a timer. HINCRBY / BITFIELD INCRBY are additive, so N folded executions leave Redis in the same state as N individual writes.

Counting follows Sidekiq's ExecutionTracker: processed counts every execution, failed the ones that raised, and ms plus the histogram (an Array of Histogram::SIZE counters, nil until the first success) only the ones that did not — a failed job's runtime is noise.

Knows nothing about Redis — History owns the write, this owns the counts.

The minute is the bucket the job ran in, decided by the caller at record time. That is what keeps the batching a visibility lag instead of a misattribution: a job that runs at 12:03:59 and flushes at 12:04:02 still counts toward 12:03. Signed off in docs/plans/2026/08/06/101-faster-than-sidekiq/00-semantics-signoff.md §2.

Keyed by the pool the recording middleware handed us (nil = the process default), so a multi-capsule process flushes each capsule's counts through the pool that capsule's jobs ran on.

Constant Summary collapse

MAX_RETAINED_MINUTES =

Ceiling on minute buckets retained per pool. Only reachable through #merge_back: a flush that keeps failing puts its counts back every tick, and a Redis outage that outlives the process would otherwise grow a structure fed by the job hot path without bound. The newest buckets win — they are the ones the dashboard is still drawing — and the oldest are dropped, which these counters have always been allowed to do (a Redis failure during a metrics write has never changed a job's outcome).

60
MAX_LIFETIME_CLASSES =

The process-lifetime per-class histograms behind /metrics (wurk_job_duration_seconds) stop adding classes here; any further class folds into OTHER, so dynamic class names cannot grow the table.

500
OTHER = 
'other'

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Accumulator

Returns a new instance of Accumulator.



45
46
47
48
49
 
# File 'lib/wurk/metrics/accumulator.rb', line 45

def initialize
  @lock = ::Mutex.new
  @pools = {}
  @lifetime = {}
end

Instance Method Details

#add(pool, klass, minute, ms, success) ⇒ Object

Hot path. Four hash lookups under the mutex; allocates only the first time a (pool, minute, class) triple is seen.



53
54
55
56
57
58
59
 
# File 'lib/wurk/metrics/accumulator.rb', line 53

def add(pool, klass, minute, ms, success)
  bucket = Histogram.index_for(ms) if success
  @lock.synchronize do
    fold(((@pools[pool] ||= {})[minute] ||= {})[klass] ||= [0, 0, 0, nil], ms, bucket)
    record_lifetime(klass, ms, bucket) if bucket
  end
end

#drain ⇒ Object

Hands the whole tree over and starts a fresh one, so recording never blocks behind the flush's Redis round trip.



69
70
71
72
73
74
75
 
# File 'lib/wurk/metrics/accumulator.rb', line 69

def drain
  @lock.synchronize do
    drained = @pools
    @pools = {}
    drained
  end
end

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


89
90
91
 
# File 'lib/wurk/metrics/accumulator.rb', line 89

def empty?
  @lock.synchronize { @pools.empty? }
end

#lifetime ⇒ Object

{ class => [successes, total ms, Histogram::SIZE bucket counts] } since this process started; never drained, so it is a Prometheus counter.



63
64
65
 
# File 'lib/wurk/metrics/accumulator.rb', line 63

def lifetime
  @lock.synchronize { @lifetime.transform_values { |count, ms, hist| [count, ms, hist.dup] } }
end

#merge_back(pool, minutes) ⇒ Object

Puts one pool's counts back after its write failed, so the next tick retries them instead of dropping the window. Merged rather than assigned: jobs kept recording into the fresh tree while the flush was in flight, and those counts have not been written yet either.



81
82
83
84
85
86
87
 
# File 'lib/wurk/metrics/accumulator.rb', line 81

def merge_back(pool, minutes)
  @lock.synchronize do
    into = (@pools[pool] ||= {})
    minutes.each { |minute, classes| merge_minute(into, minute, classes) }
    trim(into)
  end
end