Class: Wurk::Metrics::History

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

Overview

Ent feature parity (§5): server middleware that records per-job-class execution metrics into Redis time-buckets. The on-the-wire schema is wire-compat with Sidekiq's history pane — Sidekiq keys the per-minute HASH as j|<YYYYMMDD>|<H>:<M>, so dashboards (and Sidekiq data migrated in place) keep resolving against the same key after a drop-in swap.

Bucket layout (spec: docs/target/sidekiq-free.md §1.6):

j|YYYYMMDD|H:M    HASH   per-minute bucket, TTL = MID_TERM (3 days)
<klass>|p       INT    processed count
<klass>|f       INT    failed count
<klass>|ms      INT    total ms spent

<klass>-YYYYMMDD-H  HASH per-class hourly histogram, TTL = MID_TERM

We deliberately do NOT write a H:m0 10-minute rollup. Its key format collides with the real minute-0 bucket, so rolling x1..x9 into it turns that minute's value into a decade total — and the read side (Query) then sums the minute-0 bucket alongside x1..x9 and double-counts. Sidekiq itself doesn't keep that rollup (the daily/hourly rollups are commented out in its ExecutionTracker); Query reads the last N per-minute keys.

Every bucket TTL is set on every write (not NX): as long as a class keeps running we keep its bucket around for the retention window measured from last write, not from first write. So we EXPIRE unconditionally.

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, in the same 6-commands-per-(class, minute) shape the per-job pipeline sent.

Constant Summary collapse

MID_TERM =

Per spec §1.6 — naming mirrors the upstream constants so anyone grepping the Sidekiq source for MID_TERM lands here.

3 * 24 * 60 * 60
MINUTE_KEY_PREFIX =

3 days, in seconds

'j|'
DATE_FORMAT =

YYYYMMDD — matches Sidekiq's j| key

'%Y%m%d'
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
HOUR_FIELDS =

Hourly buckets are already class-scoped, so their fields are bare.

%w[p f ms].freeze

Instance Attribute Summary

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 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.



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

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

  nil
end

.hour_key(klass, time) ⇒ Object 



171
172
173
174
 
# File 'lib/wurk/metrics/history.rb', line 171

def hour_key(klass, time)
  t = time.utc
  "#{klass}-#{t.strftime(DATE_FORMAT)}-#{t.hour}"
end

.minute_key(time) ⇒ Object

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



166
167
168
169
 
# File 'lib/wurk/metrics/history.rb', line 166

def minute_key(time)
  t = time.utc
  format("#{MINUTE_KEY_PREFIX}%s|%d:%d", t.strftime(DATE_FORMAT), t.hour, t.min)
end

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

No Redis. success: true → <klass>|p; success: false → <klass>|f. <klass>|ms accumulates total runtime in milliseconds for both outcomes (so an operator can ask "how much wall-clock time has FooJob consumed?" without branching on outcome).

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.



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

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

Instance Method Details

#call(_worker, job, _queue) ⇒ Object 



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
 
# File 'lib/wurk/metrics/history.rb', line 66

def call(_worker, job, _queue)
  klass = 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, 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, 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