Module: Wurk::Api::Serializers

Defined in:
app/controllers/wurk/api/serializers.rb

Overview

Pure mapping from inspector objects → JSON-shaped Hashes for the dashboard SPA. Keeping the serializers out of the controller lets the action methods stay tiny and lets future endpoints share the same field shapes without re-implementing them.

Class Method Summary collapse

Class Method Details

.cron_row(loop_obj, now_epoch) ⇒ Object 



128
129
130
131
132
133
134
135
136
137
138
139
140
 
# File 'app/controllers/wurk/api/serializers.rb', line 128

def cron_row(loop_obj, now_epoch)
  {
    lid: loop_obj.lid,
    schedule: loop_obj.schedule,
    klass: loop_obj.klass,
    queue: loop_obj.queue,
    tz: loop_obj.tz_name,
    paused: loop_obj.paused?,
    args: loop_obj.args,
    last_fire_at: loop_obj.last_fired_at,
    next_fire_at: loop_obj.next_fire_at(now_epoch)
  }
end

.flow_detail(status) ⇒ Object

The graph. Nodes ride inside the same document rather than behind a second request: the SPA lays the DAG out from depends_on, and a header fetched separately from its edges can describe a different revision of the flow than the one being drawn.



165
166
167
168
169
170
 
# File 'app/controllers/wurk/api/serializers.rb', line 165

def flow_detail(status)
  flow_row(status).merge(
    dead_nodes: status.dead_indexes,
    nodes: status.nodes.map { |node| flow_node(node) }
  )
end

.flow_node(node) ⇒ Object

error carries a broken pipe's reason (slice 11 decision 2) — the one node state whose cause is not discoverable from the job itself, because no job ever ran.



175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
 
# File 'app/controllers/wurk/api/serializers.rb', line 175

def flow_node(node)
  {
    index: node.index,
    name: node.name,
    klass: 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

.flow_row(status) ⇒ Object

One row of the flow listing. Header fields only: Flow::Status reads its node records lazily, and a page of 25 flows that each fetched a thousand of them would be 25,000 HMGETs to render a progress bar.



145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
 
# File 'app/controllers/wurk/api/serializers.rb', line 145

def flow_row(status)
  {
    fid: status.fid,
    state: status.state,
    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
  }
end

.history_point(row) ⇒ Object

One point in a cluster-total time-series (Wurk::Metrics::Query.history). at is epoch seconds at the bucket start.



198
199
200
 
# File 'app/controllers/wurk/api/serializers.rb', line 198

def history_point(row)
  { at: row[:at], processed: row[:p], failed: row[:f], runtime_ms: row[:ms] }
end

.job_record(record) ⇒ Object

Host-registered custom job-info rows (spec §25.2) ride along as custom_rows for the SPA's job-detail modal (see Config#job_info_pairs, which gates on registration so the common no-extension case is free). Divergence: wurk evaluates these during job-list serialization — the SPA renders job detail client-side — not in a dedicated server detail view.



42
43
44
45
46
47
48
49
50
51
52
53
 
# File 'app/controllers/wurk/api/serializers.rb', line 42

def job_record(record)
  base = {
    jid: record.jid,
    klass: record.display_class,
    args: record.display_args,
    queue: record.queue,
    enqueued_at: record.enqueued_at&.to_f,
    created_at: record.created_at&.to_f
  }
  rows = ::Wurk::Web.config.job_info_pairs(record)
  rows.empty? ? base : base.merge(custom_rows: rows)
end

.limiter_row(name, meta) ⇒ Object 



107
108
109
110
111
112
113
114
115
 
# File 'app/controllers/wurk/api/serializers.rb', line 107

def limiter_row(name, meta)
  {
    name: name,
    type: meta['type'].to_s,
    fingerprint: meta['fingerprint'].to_s,
    options: parse_options(meta['options']),
    status: limiter_status(name, meta)
  }
end

.limiter_status(name, meta) ⇒ Object

Reconstruct the limiter (read-only, register: false) just to read its uniform { used, limit, reset_at, available? } status for the Limits tab. Best-effort: a malformed meta hash yields nil rather than 500-ing the whole list.



121
122
123
124
125
126
 
# File 'app/controllers/wurk/api/serializers.rb', line 121

def limiter_status(name, meta)
  limiter = ::Wurk::Limiter.build(name, meta['type'], parse_options(meta['options']))
  limiter&.status
rescue StandardError
  nil
end

.metric_row(klass, totals) ⇒ Object 



192
193
194
 
# File 'app/controllers/wurk/api/serializers.rb', line 192

def metric_row(klass, totals)
  { klass: klass, processed: totals[:p], failed: totals[:f], runtime_ms: totals[:ms] }
end

.parse_options(raw) ⇒ Object 



209
210
211
212
213
214
215
 
# File 'app/controllers/wurk/api/serializers.rb', line 209

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

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

.process_row(process, leader_identity: nil) ⇒ Object

leader_identity is the cluster's dear-leader value (ProcessSet#leader, memoized to one Redis GET per request) — comparing here avoids the N+1 Process#leader? GET-per-row that a per-process lookup would cost.



69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
 
# File 'app/controllers/wurk/api/serializers.rb', line 69

def process_row(process, leader_identity: nil) # rubocop:disable Metrics/AbcSize
  {
    identity: process.identity,
    hostname: process['hostname'],
    pid: process['pid'],
    tag: process.tag,
    concurrency: process['concurrency'],
    busy: process['busy'],
    beat: process['beat'],
    quiet: process.stopping?,
    leader: !leader_identity.to_s.empty? && process.identity == leader_identity,
    rss: process['rss'],
    rtt_us: process['rtt_us'],
    started_at: process['started_at'],
    cpu_model: process['cpu_model'],
    cores: process['cores'],
    memory_total_kb: process['memory_total_kb'],
    labels: process.labels,
    queues: process.queues,
    version: process.version,
    embedded: process.embedded?
  }
end

.profile_record(rec) ⇒ Object

One profile row for the Profiles pane. key drives the view/data links.



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

def profile_record(rec)
  {
    key: rec.key,
    jid: rec.jid,
    token: rec.token,
    type: rec.type,
    size: rec.size,
    elapsed: rec.elapsed,
    started_at: rec.started_at&.to_i
  }
end

.queue_history_series(row) ⇒ Object

One queue's size/latency gauge series (Wurk::Metrics::Query.queue_history). points are oldest→newest; at is epoch seconds at the bucket start, size is queue depth, latency is head-of-line wait in seconds.



205
206
207
 
# File 'app/controllers/wurk/api/serializers.rb', line 205

def queue_history_series(row)
  { name: row[:name], points: row[:points].map { |p| { at: p[:at], size: p[:size], latency: p[:latency] } } }
end

.queue_summary(summary) ⇒ Object 



33
34
35
 
# File 'app/controllers/wurk/api/serializers.rb', line 33

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

.sorted_entry(entry) ⇒ Object 



55
56
57
58
59
60
61
62
63
64
 
# File 'app/controllers/wurk/api/serializers.rb', line 55

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

.stats_payload(stats) ⇒ Object

Wire-shape consumed by the SolidJS dashboard's landing page + SSE feed. Field names match the SPA's StatsSnapshot interface in frontend/src/hooks/useSSE.ts — keep them in sync. The canonical Sidekiq-compatible accessors on Wurk::Stats use _size suffixes; this serializer renames them for the dashboard's wire shape only.



17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
 
# File 'app/controllers/wurk/api/serializers.rb', line 17

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

.work_row(process_id, thread_id, work) ⇒ Object

One in-flight job (WorkSet row) for the Busy page's process detail.



94
95
96
97
98
99
100
101
102
103
104
105
 
# File 'app/controllers/wurk/api/serializers.rb', line 94

def work_row(process_id, thread_id, work)
  record = work.job
  {
    process_id: process_id,
    thread_id: thread_id,
    queue: work.queue,
    klass: record.display_class,
    args: record.display_args,
    jid: record.jid,
    run_at: work.run_at.to_f
  }
end