Class: Wurk::Heartbeat

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

Overview

Single-purpose owner of the Redis-side process heartbeat. Lifted out of Wurk::Launcher so the launcher can stay focused on lifecycle and so the heartbeat schema lives in one place readers can grep for.

Each beat is two pipelined round-trips. The identity write:

SADD   processes <identity>
HSET   <identity>          info concurrency busy beat quiet rss rtt_us
EXPIRE <identity>          60
UNLINK <identity>:work
HSET   <identity>:work     <tid> <json> ...    (only if WORK_STATE non-empty)
EXPIRE <identity>:work     60                  (only if WORK_STATE non-empty)
EVALSHA refresh_slots      <held slots>        (only if QueueSlot::HELD non-empty)

then the signal drain:

RPOP   <identity>-signals  BEAT_PAUSE

Two rather than one because only the first is safe to replay — see #pipelined_beat and #drain_signals.

The work hash is UNLINK-then-rewritten on every beat — a dropped beat momentarily empties it, and ProcessSet#cleanup compensates by SREM-ing identities whose info field has expired.

Heartbeat owns only the wire. Signals queued by the dashboard are pulled out of the pipelined results and returned to the caller for dispatch — TSTP/TERM semantics live in Launcher.

:heartbeat fires once on the first successful beat and again after any network partition (rearmed when beat! rescues). :beat fires every tick.

Spec: docs/target/sidekiq-free.md §12 (Launcher#❤️).

Constant Summary collapse

BEAT_PAUSE =

Cadence in seconds. Key TTL is 60s — a process is dead after ~6 misses.

10
TTL_SECONDS = 
60
OOM_REFRESH_SLOTS = 
"#{Fetcher::OomDrain::HEADER}#{Lua::SCRIPTS.fetch(:refresh_slots)}".freeze
OOM_BEAT =

KEYS = [processes, identity, work], ARGV = [info_json, beat, ttl_seconds, tid, work_json, ...]

<<~LUA.freeze
  #{Fetcher::OomDrain::HEADER}if redis.call('EXISTS', KEYS[2]) == 0 then
    redis.call('HSET', KEYS[2], 'info', ARGV[1])
    redis.call('SADD', KEYS[1], KEYS[2])
  end
  redis.call('HSET', KEYS[2], 'beat', ARGV[2])
  redis.call('EXPIRE', KEYS[2], ARGV[3])
  redis.call('UNLINK', KEYS[3])
  for i = 4, #ARGV, 2 do
    redis.call('HSET', KEYS[3], ARGV[i], ARGV[i + 1])
  end
  redis.call('EXPIRE', KEYS[3], ARGV[3])
  return 1
LUA
FALLBACK_PAGE_SIZE = 
4096
PAGE_SIZE = 
page_size

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(identity:, config:, started_at: nil, embedded: false, quiet: nil) ⇒ Heartbeat

quiet: is a callable so Launcher can keep ownership of its @done flag without an awkward setter contract.



103
104
105
106
107
108
109
110
111
 
# File 'lib/wurk/heartbeat.rb', line 103

def initialize(identity:, config:, started_at: nil, embedded: false, quiet: nil)
  @identity = identity
  @config = config
  @started_at = started_at || Time.now.to_f
  @embedded = embedded
  @quiet_proc = quiet || -> { false }
  @first_heartbeat = true
  @rtt_us = 0
end

Instance Attribute Details

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

Returns the value of attribute config.

#identity ⇒ Object (readonly)

Returns the value of attribute identity.



99
100
101
 
# File 'lib/wurk/heartbeat.rb', line 99

def identity
  @identity
end

#last_beat_at ⇒ Object (readonly)

Returns the value of attribute last_beat_at.



99
100
101
 
# File 'lib/wurk/heartbeat.rb', line 99

def last_beat_at
  @last_beat_at
end

#rtt_us ⇒ Object (readonly)

Returns the value of attribute rtt_us.



99
100
101
 
# File 'lib/wurk/heartbeat.rb', line 99

def rtt_us
  @rtt_us
end

Class Method Details

.page_size(etc = Etc) ⇒ Object

statm counts pages, and a page is not 4 KB everywhere: arm64 kernels commonly run 16 KB or 64 KB pages, where a hard-coded ×4 under-reports RSS by 4–16×.



71
72
73
74
75
76
77
 
# File 'lib/wurk/heartbeat.rb', line 71

def self.page_size(etc = Etc)
  size = etc.sysconf(Etc::SC_PAGESIZE)
  size.is_a?(Integer) && size.positive? ? size : FALLBACK_PAGE_SIZE
# NotImplementedError (a ScriptError) is what Etc raises without sysconf.
rescue StandardError, NotImplementedError
  FALLBACK_PAGE_SIZE
end

.smaps_kb(pid, fields, proc_root: '/proc') ⇒ Object

Sum of fields (kB) in /proc//smaps_rollup; nil when none is there. Raises SystemCallError when the file cannot be read. Shared with Swarm::MemoryWatch and Metrics::Prometheus.



90
91
92
93
94
95
96
97
 
# File 'lib/wurk/heartbeat.rb', line 90

def self.smaps_kb(pid, fields, proc_root: '/proc')
  total = nil
  ::File.foreach("#{proc_root}/#{pid}/smaps_rollup") do |line|
    name, value = line.split(':', 2)
    total = total.to_i + value.to_i if value && fields.include?(name)
  end
  total
end

.statm_rss_kb(statm) ⇒ Object

Resident KB from a /proc//statm line (field 1 is resident pages). Shared with Swarm, which reads its children's statm to recycle bloat.



83
84
85
 
# File 'lib/wurk/heartbeat.rb', line 83

def self.statm_rss_kb(statm)
  statm.split[1].to_i * PAGE_SIZE / 1024
end

Instance Method Details

#beat! ⇒ Object

Pipelined beat. Returns the drained signals as Array on success, or nil if Redis raised — the next successful beat refires :heartbeat so partition recovery is observable.



116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
 
# File 'lib/wurk/heartbeat.rb', line 116

def beat!
  sigs, @rtt_us = pipelined_beat
  @last_beat_at = Time.now.to_f
  if @first_heartbeat
    @first_heartbeat = false
    fire_event(:heartbeat)
  end
  fire_event(:beat, oneshot: false)
  emit_statsd_gauges
  sigs
rescue StandardError => e
  @first_heartbeat = true
  handle_exception(e, { context: 'heartbeat' })
  nil
end

#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

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

#stop! ⇒ Object

Best-effort teardown. SREM removes us from the live list; UNLINK drops the identity hash and its work mirror. Idempotent: if Redis is down the next process boot's ProcessSet#cleanup prunes us.



135
136
137
138
139
140
141
142
143
144
 
# File 'lib/wurk/heartbeat.rb', line 135

def stop!
  redis do |conn|
    conn.pipelined do |pipe|
      pipe.call('SREM', Keys::PROCESSES, @identity)
      pipe.call('UNLINK', @identity, "#{@identity}:work")
    end
  end
rescue StandardError
  nil
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.