Class: Wurk::Metrics::History

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

Overview

Server middleware that records per-job-class execution metrics into the same Redis buckets Sidekiq 8.1's ExecutionTracker writes, so history survives a swap in either direction and Sidekiq's own Metrics::Query reads what Wurk wrote (spec: docs/target/sidekiq-free.md §1.6, §20).

j|YYMMDD|H:MM   HASH  per-minute bucket, TTL = SHORT_TERM (8h)
j|YYMMDD|H:M    HASH  10-minute bucket (minute's tens digit), TTL = MID_TERM (30d)
<klass>|p     INT   executions, failures included
<klass>|f     INT   failures
<klass>|ms    INT   total ms of the executions that did not fail
h|<klass>-D-H:M BITFIELD per-minute runtime histogram (Metrics::Histogram)

The two j| shapes cannot collide: the minute is always two digits (14:05), the 10-minute bucket always one (14:0). <klass> is the wrapped class for ActiveJob payloads, like upstream.

Every bucket TTL is set on every flush (not NX), once per bucket key after its HINCRBYs, as upstream's ExecutionTracker#flush does.

The middleware is hot-path — every job pays for it — so it does not touch Redis at all. Each execution folds into a process-wide Accumulator and Wurk::Metrics::Flusher writes the whole tree out every FLUSH_INTERVAL.

Constant Summary collapse

MID_TERM =

Upstream's ExecutionTracker constants, same names. MID_TERM diverges: upstream TTLs the 10-minute buckets at 3 days, but Wurk's dashboard serves 7d/30d tabs off them (Query::MAX_HOURS = 720), so they must live 30 days. Longer retention changes no key, field, or score. config.metrics_retention= overrides it (TTL only); MID_TERM stays the default.

30 * 24 * 60 * 60
SHORT_TERM = 
8 * 60 * 60
MINUTE_KEY_FORMAT = 
'j|%y%m%d|%-H:%M'
FLUSH_INTERVAL =

Ceiling on how stale an unflushed counter can be, in seconds. A constant, never a config knob: it exists to bound the dashboard's lag, not to be tuned. Sidekiq — the engine these numbers get compared against — flushes its own process stats on a 10s heartbeat and writes nothing per job, so this is the tighter of the two. Signed off in docs/plans/2026/08/06/101-faster-than-sidekiq/00-semantics-signoff.md §2.

5
ACCUMULATOR =

Process-wide, like Processor::PROCESSED — the counters belong to the process, not to a middleware instance (the chain builds one per capsule) and not to a job.

Accumulator.new

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#config ⇒ Object Originally defined in module Wurk::Middleware::ServerMiddleware

Returns the value of attribute config.

Class Method Details

.flush(accumulator = ACCUMULATOR) ⇒ Object

Drains an accumulator into Redis, one pipeline per pool. Raises the first pool's failure after every other pool has had its turn — Wurk::Metrics::Flusher owns turning that into an error-handler call.

Only the failed pool's counts are merged back. Putting the whole drained tree back would re-send the writes that already landed, and HINCRBY would count them twice.

The argument is a collaborator, not an option: a process has exactly one accumulator and nothing in wurk passes a second. Tests pass their own so a deliberately broken pool cannot leak into a parallel test's flush through the process-wide one.

A shut-down pool is the one failure that is terminal: ConnectionPool #shutdown is one-way (see Capsule#reset_redis_pools!), so its counts can never be written by anyone. Merging them back would re-raise on every tick for the life of the process — a permanent 5s error-handler loop over a condition no operator can act on. The accumulator survives the pool that fed it (a fork inherits it; reset_redis_pools! is host- callable), so drop those counts, exactly as the retry cap already drops a window Redis stayed down through.



151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
 
# File 'lib/wurk/metrics/history.rb', line 151

def flush(accumulator = ACCUMULATOR)
  error = nil
  mid_term = Wurk.configuration[:metrics_retention] || MID_TERM
  accumulator.drain.each do |pool, minutes|
    write(pool, minutes, mid_term)
  rescue ConnectionPool::PoolShuttingDownError
    next
  rescue StandardError => e
    accumulator.merge_back(pool, minutes)
    error = e
  end
  raise error if error

  nil
end

.minute_key(time) ⇒ Object

Public formatters — Wurk::Metrics::Query reuses these so the two cannot drift on bucket-naming convention.



169
170
171
 
# File 'lib/wurk/metrics/history.rb', line 169

def minute_key(time)
  time.utc.strftime(MINUTE_KEY_FORMAT)
end

.record(klass, duration_ms, success:, redis_pool: nil, at: nil) ⇒ Object

No Redis. Every call books <klass>|p; success: false also books <klass>|f; success: true adds the runtime to <klass>|ms and the histogram (Accumulator has the why).

at: defaults to nil rather than ::Time.now: this runs once per job and the bucket only needs whole UTC minutes, which #minute_bucket reads straight off the clock as an Integer.



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

def record(klass, duration_ms, success:, redis_pool: nil, at: nil)
  return if klass.nil? || klass.empty?

  ms = duration_ms.to_i
  ms = 0 if ms.negative?
  ACCUMULATOR.add(redis_pool, klass, minute_bucket(at), ms, success)
  nil
end

.shared_instance? ⇒ Boolean

One instance per chain entry, shared by every job (Middleware::Chain::Entry): #call keeps its state in locals and writes no ivar.

Returns:

  • (Boolean) 


38
 
# File 'lib/wurk/metrics/history.rb', line 38

def self.shared_instance? = true

.ten_minute_key(time) ⇒ Object

Upstream derives the 10-minute bucket by chopping the minute key's last digit: j|250214|8:43 → j|250214|8:4.



175
176
177
 
# File 'lib/wurk/metrics/history.rb', line 175

def ten_minute_key(time)
  minute_key(time).chop
end

Instance Method Details

#call(_worker, job, _queue) ⇒ Object 



64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
 
# File 'lib/wurk/metrics/history.rb', line 64

def call(_worker, job, _queue)
  klass = job['wrapped'] || job['class']
  started = monotonic_ms
  success = false
  begin
    result = yield
    success = true
    result
  rescue Wurk::Job::Interrupted, Wurk::Job::DeadlineExceeded
    # Neither is the job failing. A cooperative interruption is not a
    # failure: InterruptHandler self-prepends, so it sits *outside* this
    # middleware (registered by
    # `Wurk.configuration.server_middleware.prepend(Wurk::Middleware::InterruptHandler)`
    # in lib/wurk/middleware/interrupt_handler.rb:50); Interrupted passes
    # through here before it becomes a
    # JobRetry::Skip, and without this arm one interrupted IterableJob
    # books a spurious `<klass>|f` (#394). Upstream's
    # ExecutionTracker#track books `p` + `ms` for that Skip and reserves
    # `f` for a real exception; `p` is the bucket for "reached perform and
    # didn't error", so the resumed run booking a second `p` is the
    # oracle's behavior, not a rounding error. Signed off in
    # docs/plans/2026/08/07/101-beyond-sidekiq/00-semantics-signoff.md §1.
    #
    # A job cut by its absolute deadline lands in the same bucket for a
    # plainer reason: Middleware::Expiry, also outside this one (registered
    # by `Wurk.configuration.server_middleware.add(Wurk::Middleware::Expiry)`
    # in lib/wurk/middleware/expiry.rb:111), catches
    # that raise and books the job `expired`. Booking `f` here too would
    # put one abandoned job in two of the three counts an operator
    # subtracts (executed = processed - failed - expired), and would make
    # it the only expired job that also reads as failed — the ones dropped
    # before `perform` never reach this middleware at all.
    success = true
    raise
  ensure
    duration = (monotonic_ms - started).round
    # Best-effort: a metrics failure must never propagate into the job
    # result. The processor already finalized the ack path. Recording is
    # in-memory now, so what this catches is a bad payload (a `class` that
    # is not a String) rather than a Redis blip — the Redis half moved to
    # Flusher, which reports on its own thread.
    begin
      self.class.record(klass, duration, success: success, redis_pool: redis_pool)
    rescue StandardError => e
      handle_error(e)
    end
  end
end

#logger ⇒ Object Originally defined in module Wurk::Middleware::ServerMiddleware

#redis(idempotent: false) ⇒ Object Originally defined in module Wurk::Middleware::ServerMiddleware

#redis_pool ⇒ Object Originally defined in module Wurk::Middleware::ServerMiddleware