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



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

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

  nil
end

.hour_key(klass, time) ⇒ Object 



139
140
141
142
 
# File 'lib/wurk/metrics/history.rb', line 139

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.



133
134
135
136
137
 
# File 'lib/wurk/metrics/history.rb', line 133

def minute_key(time)
  t = time.utc
  format("#{MINUTE_KEY_PREFIX}%<date>s|%<hr>d:%<min>d",
         date: t.strftime(DATE_FORMAT), hr: t.hour, min: 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.



97
98
99
100
101
102
103
104
 
# File 'lib/wurk/metrics/history.rb', line 97

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 



65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/wurk/metrics/history.rb', line 65

def call(_worker, job, _queue)
  klass = job['class']
  started = monotonic_ms
  success = false
  begin
    result = yield
    success = true
    result
  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