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.

Constant Summary

Constants included from Component

Component::DEFAULT_THREAD_PRIORITY, Component::LEADER_CACHE_TTL_MS, Component::PROCESS_NONCE

Instance Attribute Summary

Attributes included from Component

#config

Instance Method Summary collapse

Methods included from Component

#default_tag, #fire_event, #handle_exception, #hostname, #identity, #leader?, #logger, #mono_ms, #process_nonce, #real_ms, #redis, #safe_thread, tid, #tid, #watchdog

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 Method Details

#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