Module: Wurk::API::Serializers

Defined in:
lib/wurk/api/serializers.rb

Overview

Wire shapes for the observe plane.

Every value below is read off a canonical inspector — Stats, Queue, JobRecord, SortedEntry — so the API and the dashboard cannot report different numbers for the same Redis. Only the shape differs, and it differs on purpose: the dashboard's serializers (app/controllers/wurk/api/serializers.rb) answer to the SPA and change whenever it does, and they reach into Wurk::Web for host-registered extension rows — an engine dependency the API cannot take, because it also runs standalone. Under /v1 these field names are a contract.

Constant Summary collapse

STALE_AFTER_SECONDS =

Three missed beats at Heartbeat::BEAT_PAUSE cadence, which is also the window Wurk::Health calls a heartbeat stale, and half the heartbeat key's TTL — so a process that stopped beating is reported stale for ~30s before Redis reaps the row out from under this API.

::Wurk::Heartbeat::BEAT_PAUSE * 3

Class Method Summary collapse

Class Method Details

.beat_age_seconds(beat, now) ⇒ Object

Seconds since this process last beat, measured against a now the caller took once for the whole listing so rows never disagree by a round trip. Floored at zero: beat is stamped by the beating process's clock and this is read on another host's, so a skew of a few milliseconds must not surface as a negative age.



31
32
33
 
# File 'lib/wurk/api/serializers.rb', line 31

def beat_age_seconds(beat, now)
  [now - beat.to_f, 0.0].max.round(3)
end

.cron_loop(loop_obj, now:) ⇒ Object

One registered cron loop. next_fire_at is evaluated against a now the caller took once, so every row in a listing answers "next after the same instant" rather than drifting a row at a time.



106
107
108
109
110
111
112
113
114
 
# File 'lib/wurk/api/serializers.rb', line 106

def cron_loop(loop_obj, now:)
  {
    lid: loop_obj.lid, schedule: loop_obj.schedule, class: loop_obj.klass,
    queue: loop_obj.queue, args: loop_obj.args, tz: loop_obj.tz_name,
    paused: loop_obj.paused?,
    last_fired_at: loop_obj.last_fired_at,
    next_fire_at: loop_obj.next_fire_at(now)
  }
end

.declared(process, now) ⇒ Object

The half a process wrote about itself in the info JSON: who it is and what it was configured to do. Split along the same seam the heartbeat itself uses (Heartbeat#info_hash versus #beat_hash_args), so a field that moves between the two halves upstream moves between these.



53
54
55
56
57
58
59
60
61
 
# File 'lib/wurk/api/serializers.rb', line 53

def declared(process, now)
  {
    identity: process.identity, hostname: process['hostname'], pid: process['pid'],
    tag: process.tag, version: process.version, embedded: process.embedded?,
    concurrency: process['concurrency'], queues: process.queues,
    weights: process.weights, labels: process.labels,
    started_at: process['started_at'], uptime_seconds: uptime_seconds(process['started_at'], now)
  }
end

.flow(status) ⇒ Object

One flow and its graph. succeeded is emitted alongside pending because a client rendering progress would otherwise have to know that the two are complements of total — a relation this contract should not require anyone to rediscover.

nodes is empty for an abandoned flow: the kill switch releases the node records, and inventing rows for keys that are gone would report a graph nothing can still act on.



188
189
190
191
192
193
194
195
196
197
198
 
# File 'lib/wurk/api/serializers.rb', line 188

def flow(status)
  {
    fid: status.fid, state: status.state, terminal: status.terminal?,
    total: status.total, pending: status.pending, succeeded: status.succeeded_count,
    depth: status.depth, width: status.width,
    created_at: status.created_at, finished_at: status.finished_at,
    failed_at: status.failed_at, abandoned_at: status.abandoned_at,
    dead_nodes: status.dead_indexes,
    nodes: status.nodes.map { |node| flow_node(node) }
  }
end

.flow_node(node) ⇒ Object

depends_on and dependents are node indexes, which is what a node is addressed by: name is optional, and two nodes of the same class are otherwise indistinguishable. error is set only on a broken node — one whose piped input never arrived, so no job ran to leave a failure anywhere else.



205
206
207
208
209
210
211
212
 
# File 'lib/wurk/api/serializers.rb', line 205

def flow_node(node)
  {
    index: node.index, name: node.name, class: node.klass, queue: node.queue,
    jid: node.jid, bid: node.bid, state: node.state,
    depends_on: node.dependencies, dependents: node.dependents,
    remaining: node.remaining, piped: node.piped?, error: node.error
  }
end

.job_record(record) ⇒ Object

class and args are the display view, the same one the dashboard renders: an ActiveJob wrapper is unwrapped to the job the host actually wrote, and an encrypt: true job's ciphertext argument is masked. Serving record.args here would publish the raw encrypted envelope of every such job over HTTP.



169
170
171
172
173
174
175
176
177
178
 
# File 'lib/wurk/api/serializers.rb', line 169

def job_record(record)
  {
    jid: record.jid,
    class: record.display_class,
    args: record.display_args,
    queue: record.queue,
    enqueued_at: record.enqueued_at&.to_f,
    created_at: record.created_at&.to_f
  }
end

.limiter_row(name, meta, status) ⇒ Object

available, not the available? Limiter::Base#build_status keys it with: a trailing ? is Ruby's convention for a predicate, not JSON's for a Boolean field. status is null when the limiter's metadata expired between the listing that named it and this read.



90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/wurk/api/serializers.rb', line 90

def limiter_row(name, meta, status)
  {
    name: name,
    type: meta['type'].to_s,
    fingerprint: meta['fingerprint'].to_s,
    options: parse_options(meta['options']),
    status: status && {
      used: status[:used], limit: status[:limit],
      reset_at: status[:reset_at], available: status[:available?]
    }
  }
end

.measured(process, now) ⇒ Object

The half its last beat measured, plus the two values derived from it.



64
65
66
67
68
69
70
71
 
# File 'lib/wurk/api/serializers.rb', line 64

def measured(process, now)
  age = beat_age_seconds(process['beat'], now)
  {
    busy: process['busy'], rss_kb: process['rss'], rtt_us: process['rtt_us'],
    beat: process['beat'], beat_age_seconds: age, stale: stale?(age),
    quiet: process.stopping?
  }
end

.parse_options(raw) ⇒ Object 



116
117
118
119
120
121
122
 
# File 'lib/wurk/api/serializers.rb', line 116

def parse_options(raw)
  return {} if raw.nil? || raw.to_s.empty?

  ::JSON.parse(raw)
rescue ::JSON::ParserError
  {}
end

.process_row(process, leader:, now:) ⇒ Object

One live process. beat_age_seconds and stale are derived here rather than left to the client: a client comparing beat against its own clock is comparing two clocks, which is the one comparison this roll-up exists to save it from.

leader is passed in because the two callers learn it differently — a listing compares against one memoized dear-leader read, a single process asks itself — and neither should pay the other's round trips.



45
46
47
 
# File 'lib/wurk/api/serializers.rb', line 45

def process_row(process, leader:, now:)
  declared(process, now).merge(measured(process, now)).merge(leader: leader)
end

.queue_gauges(queue) ⇒ Object

The same four gauges read off a Wurk::Queue instead of a pipelined Stats::QueueSummary — three round trips rather than a share of one, which is what asking about a single queue costs. Same field names on purpose: a client that read a queue out of the listing and then fetched it should not have to reshape anything.



160
161
162
 
# File 'lib/wurk/api/serializers.rb', line 160

def queue_gauges(queue)
  { name: queue.name, size: queue.size, latency: queue.latency, paused: queue.paused? }
end

.queue_summary(summary) ⇒ Object 



151
152
153
 
# File 'lib/wurk/api/serializers.rb', line 151

def queue_summary(summary)
  { name: summary.name, size: summary.size, latency: summary.latency, paused: summary.paused? }
end

.sorted_entry(entry) ⇒ Object

at is the member's ZSET score in epoch seconds — when it is due to retry, due to run, or when it died, depending on which set it came out of. Emitted once: the score and the timestamp are the same number, and a contract that ships both invites clients to disagree about which is authoritative.



219
220
221
222
223
224
225
226
227
228
229
 
# File 'lib/wurk/api/serializers.rb', line 219

def sorted_entry(entry)
  job_record(entry).merge(
    at: entry.at.to_f,
    retry_count: entry['retry_count'],
    error_class: entry['error_class'],
    error_message: entry['error_message'],
    failed_at: entry.failed_at&.to_f,
    retried_at: entry.retried_at&.to_f,
    error_backtrace: entry.error_backtrace
  )
end

.stale?(age) ⇒ Boolean

Returns:

  • (Boolean) 


35
 
# File 'lib/wurk/api/serializers.rb', line 35

def stale?(age) = age > STALE_AFTER_SECONDS

.stats(snapshot) ⇒ Object

default_queue_latency, not the dashboard's latency: sitting beside a queues array, a bare latency reads as the whole fleet's rather than the default queue's, which is what Stats measures.



135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
 
# File 'lib/wurk/api/serializers.rb', line 135

def stats(snapshot)
  {
    processed: snapshot.processed,
    failed: snapshot.failed,
    expired: snapshot.expired,
    enqueued: snapshot.enqueued,
    busy: snapshot.workers_size,
    scheduled: snapshot.scheduled_size,
    retries: snapshot.retry_size,
    dead: snapshot.dead_size,
    processes: snapshot.processes_size,
    default_queue_latency: snapshot.default_queue_latency,
    queues: snapshot.queue_summaries.map { |summary| queue_summary(summary) }
  }
end

.uptime_seconds(started_at, now) ⇒ Object

nil rather than 0 when the heartbeat carries no started_at: a process of unknown age is not one that just booted.



126
127
128
129
130
 
# File 'lib/wurk/api/serializers.rb', line 126

def uptime_seconds(started_at, now)
  return nil if started_at.nil?

  [now - started_at.to_f, 0.0].max.round(3)
end

.work_row(process_id, thread_id, work, now:) ⇒ Object

One in-flight job. class/args are the display view for the reason #job_record gives, and elapsed_seconds is derived for the reason beat_age_seconds is — the client's clock is not the swarm's.



76
77
78
79
80
81
82
83
84
 
# File 'lib/wurk/api/serializers.rb', line 76

def work_row(process_id, thread_id, work, now:)
  record = work.job
  run_at = work.run_at.to_f
  {
    process_id: process_id, thread_id: thread_id, queue: work.queue,
    jid: record.jid, class: record.display_class, args: record.display_args,
    run_at: run_at, elapsed_seconds: [now - run_at, 0.0].max.round(3)
  }
end