Class: Wurk::Metrics::Prometheus

Inherits:
Object
  • Object
show all
Includes:
Exposition
Defined in:
lib/wurk/metrics/prometheus.rb

Overview

Prometheus text exposition (format 0.0.4) served by the health listener at GET /metrics. Reads only keys Sidekiq already writes — no new Redis keys.

Two families, and the difference matters when aggregating:

* cluster gauges/counters (`wurk_queue_*`, `wurk_processed_total`, …)
come straight from shared Redis, so every scrape target reports the
same value. Aggregate with `max`, never `sum`.
* `wurk_process_*` / `wurk_swarm_*` describe THIS host's process group:
heartbeat identities sharing this process's nonce (the swarm parent
stamps one nonce that every child inherits), so a scrape of one pod
covers all of its children no matter which child owns the port.
`wurk_leader` is the same: 1 on every scrape target in the group that
holds the lease.
* `wurk_fetch_errors_total`, `wurk_reaper_reclaimed_total` and the other
ProcessCounters families, the Redis pool gauges and
`wurk_job_duration_seconds` count in the memory of the process that
owns the port, so under a swarm they are a sample of one child. `rate`
them; a child restart resets them.
* `wurk_swarm_child_restarts_total` / `wurk_swarm_crash_streak` are the
swarm parent's, read through Swarm::RestartCounters, so they survive
child restarts; present only inside a swarm child.

One Redis checkout per refresh — two pipelines, because the second needs the queue and process names the first returns — and the snapshot is cached for ttl seconds, so a scrape storm costs at most one refresh per window. A failed refresh is cached for the same window: a dead Redis is not hammered once per scrape.

Defined Under Namespace

Classes: ProcessRow, QueueRow, Snapshot

Constant Summary collapse

CONTENT_TYPE = 
'text/plain; version=0.0.4; charset=utf-8'
TTL = 
1.0
PROCESS_FIELDS = 
%w[busy concurrency beat rss quiet info rtt_us].freeze
RUN_AT_RE =

The last field of a heartbeat work entry ({"queue":…,"payload":…,"run_at":…}); anchored at the end so a run_at inside the escaped payload never matches.

/"run_at":(\d+(?:\.\d+)?)\}\z/
PSS_FIELDS = 
%w[Pss].freeze
SCALARS = 
{
  processed: ['GET', Keys::STAT_PROCESSED],
  failed: ['GET', 'stat:failed'],
  scheduled: ['ZCARD', Keys::SCHEDULE],
  retries: ['ZCARD', Keys::RETRY],
  dead: ['ZCARD', Keys::DEAD]
}.freeze
SET_HEADS =

Score of each set's earliest member: past due by now - score when the scheduler is behind.

Scheduled::SETS.map { |set| ['ZRANGE', set, 0, 0, 'WITHSCORES'] }.freeze
FIRST_PASS =

Names come last: the detail pass is built from them.

[*SCALARS.values, *SET_HEADS, ['GET', Wurk::Leader::DEFAULT_KEY],
['SMEMBERS', Keys::QUEUES_SET], ['SMEMBERS', Keys::PROCESSES]].freeze
QUEUE_GAUGES = 
[
  ['wurk_queue_size', 'Jobs enqueued per queue.', :depth.to_proc],
  ['wurk_queue_latency_seconds', 'Age of the oldest job per queue.', :latency.to_proc],
  ['wurk_queue_consumers', 'Live processes subscribed to the queue, quiet ones included.', :consumers.to_proc]
].freeze
CLUSTER_GAUGES = 
[
  ['wurk_scheduled_size', 'Jobs in the schedule set.', :scheduled.to_proc],
  ['wurk_retry_size', 'Jobs in the retry set.', :retries.to_proc],
  ['wurk_dead_size', 'Jobs in the dead set.', :dead.to_proc],
  ['wurk_processes', 'Live worker processes, cluster-wide.', ->(snap) { snap.processes.size }],
  ['wurk_busy', 'Jobs executing right now, cluster-wide.', ->(snap) { snap.processes.sum(&:busy) }],
  ['wurk_concurrency', 'Worker threads, cluster-wide.', ->(snap) { snap.processes.sum(&:concurrency) }]
].freeze
PROCESS_GAUGES = 
[
  ['wurk_process_busy', 'Jobs executing in this local process.', ->(row, _now) { row.busy }],
  ['wurk_process_concurrency', 'Worker threads in this local process.', ->(row, _now) { row.concurrency }],
  ['wurk_process_rss_bytes', 'Resident set size reported by the heartbeat.', ->(row, _now) { row.rss_kb * 1024 }],
  ['wurk_process_heartbeat_age_seconds', 'Seconds since the last heartbeat.',
   ->(row, now) { row.beat ? [now - row.beat, 0.0].max : 0.0 }],
  ['wurk_process_quiet', '1 when the process is quiet (TSTP).', ->(row, _now) { row.quiet ? 1 : 0 }],
  ['wurk_process_redis_rtt_seconds', 'Round trip of the last heartbeat write.',
   ->(row, _now) { row.rtt_us / 1_000_000.0 }],
  ['wurk_process_oldest_job_seconds', 'Age of the longest-running job in the process; 0 when idle.',
   ->(row, now) { row.oldest_run_at ? [now - row.oldest_run_at, 0.0].max : 0.0 }],
  ['wurk_process_pss_bytes', 'Proportional set size from /proc/<pid>/smaps_rollup (Linux only).',
   ->(row, _now) { pss_bytes(row.pid) }]
].freeze

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config, ttl: TTL, nonce: Component::PROCESS_NONCE, hostname: Component.hostname, clock: -> { ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) }) ⇒ Prometheus

Returns a new instance of Prometheus.



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

def initialize(config, ttl: TTL, nonce: Component::PROCESS_NONCE, hostname: Component.hostname,
               clock: -> { ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) })
  @config = config
  @ttl = ttl
  @local_suffix = ":#{nonce}"
  @local_prefix = "#{hostname}:"
  @clock = clock
  @lock = ::Mutex.new
  @snapshot = nil
end

Class Method Details

.pss_bytes(pid) ⇒ Object 



90
91
92
93
94
95
 
# File 'lib/wurk/metrics/prometheus.rb', line 90

def self.pss_bytes(pid)
  kb = Heartbeat.smaps_kb(pid, PSS_FIELDS)
  kb && (kb * 1024)
rescue ::SystemCallError, ::IOError
  nil
end

Instance Method Details

#fresh_local(snap, window, now = ::Time.now.to_f) ⇒ Object

Local processes whose heartbeat landed within window seconds.



129
130
131
 
# File 'lib/wurk/metrics/prometheus.rb', line 129

def fresh_local(snap, window, now = ::Time.now.to_f)
  local_processes(snap).select { |row| row.beat && now - row.beat < window }
end

#local_processes(snap) ⇒ Object

This host's process group (see class doc), from a Snapshot.



124
125
126
 
# File 'lib/wurk/metrics/prometheus.rb', line 124

def local_processes(snap)
  snap.processes.select { |row| local?(row.identity) }
end

#render(expected_children: nil, fresh_window: 30, swarm: nil) ⇒ Object

swarm: is a Swarm::RestartCounters::Snapshot, nil outside a swarm child.



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

def render(expected_children: nil, fresh_window: 30, swarm: nil)
  snap = snapshot
  out = +''
  build_info(out)
  gauge(out, 'wurk_redis_up', 'Whether the last metrics refresh could read Redis.', [[nil, snap.ok ? 1 : 0]])
  ProcessFamilies.render(out, swarm: swarm)
  if snap.ok
    cluster(out, snap)
    local(out, snap, expected_children, fresh_window)
  end
  out
end

#snapshot ⇒ Object

Cached Snapshot; ok false when Redis could not be read.



115
116
117
118
119
120
121
 
# File 'lib/wurk/metrics/prometheus.rb', line 115

def snapshot
  @lock.synchronize do
    now = @clock.call
    @snapshot = fetch(now) if @snapshot.nil? || now - @snapshot.taken_at >= @ttl
    @snapshot
  end
end