Class: Wurk::Metrics::Flusher

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

Overview

Drains Wurk::Metrics::History's in-process accumulator into Redis every History::FLUSH_INTERVAL seconds.

Deliberately NOT leader-gated, unlike every other periodic component here (Metrics::Rollup, Metrics::QueueRollup, Wurk::History). Those publish cluster state that one process should own; this publishes counters that only exist inside this process's memory and that no other process can write. A leader gate would strand every follower's metrics until it died.

One tick per process, not per capsule: the accumulator is process-wide and already keyed by the pool each capsule's jobs recorded through.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config, accumulator: History::ACCUMULATOR) ⇒ Flusher

accumulator is a collaborator, not an option: every real boot flushes the process-wide one. Tests inject their own so a broken pool stays inside the test that built it.



26
27
28
29
30
31
 
# File 'lib/wurk/metrics/flusher.rb', line 26

def initialize(config, accumulator: History::ACCUMULATOR)
  @config = config
  @accumulator = accumulator
  @timer = TimerLoop.new(History::FLUSH_INTERVAL)
  @thread = nil
end

Instance Attribute Details

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

Returns the value of attribute config.

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.

#start ⇒ Object 



33
34
35
36
37
38
 
# File 'lib/wurk/metrics/flusher.rb', line 33

def start
  return @thread if @thread

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

#terminate ⇒ Object

Flushes once more in an ensure: this stops the only thread that would have written the window accumulated since the last tick, so without it a graceful shutdown would drop up to FLUSH_INTERVAL of counters — the cost the sign-off accepts for a hard kill only.

Launcher#stop joins the manager drains before the teardown tail reaches us, so by the time this runs no job is still recording.

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 draining the same accumulator.



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

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

#tick ⇒ Object

Never raises: this runs on a safe_thread, whose watchdog re-raises after reporting, and a reported-then-dead flusher would silently stop every counter in the process for a blip Redis recovers from. The accumulator has already merged the failed window back for the next tick.



63
64
65
66
67
 
# File 'lib/wurk/metrics/flusher.rb', line 63

def tick
  History.flush(@accumulator)
rescue StandardError => e
  handle_exception(e, { context: 'metrics-flush' })
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.