Module: Wurk::API::Swarm

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

Overview

The swarm plane: how the fleet is laid out, what each process is doing right now, and the two signals an operator sends it.

Read-only against Redis structures that already exist — the processes SET and its per-identity heartbeat HASHes, <identity>:work, dear-leader, the limiter and cron registries. No new key, no new field on the beat, and nothing here makes a process beat more often. A monitoring client polling GET /swarm every few seconds must cost the swarm nothing, so every route reads through the canonical inspectors (ProcessSet, WorkSet, Cron::LoopSet, Web::Enterprise::Limits) exactly as the dashboard does.

One deliberate divergence from the dashboard: ProcessSet.new(false) everywhere. The default constructor also runs the SREM sweep for expired identities, which is a write, rate-limited by a SET NX EX that is itself a write. Every count reported here comes from #each, which already skips identities whose info has expired, so the sweep would buy no accuracy — only a Redis write on each poll of a plane that is supposed to be free to watch. The dashboard keeps the sweep; an operator has it open for minutes, not milliseconds.

Constant Summary collapse

ALL =

Signal targets. An identity is <hostname>:<pid>:<nonce>, so nothing live can be named all and the sentinel needs no escaping.

'all'
QUIET_SIGNAL = 
'TSTP'
STOP_SIGNAL = 
'TERM'
MAX_FILTER_LENGTH = 
255
TABLE =

The two signals take :admin. Quieting the fleet stops it working without enqueueing or deleting anything — the same reasoning that puts queue pausing with the destructive routes rather than the listings.

[
  [:get, '/swarm', :read, :cluster],
  [:get, '/processes', :read, :index],
  [:get, '/processes/:identity', :read, :show],
  [:post, '/processes/:identity/quiet', :admin, :quiet],
  [:post, '/processes/:identity/stop', :admin, :stop],
  [:get, '/busy', :read, :busy],
  [:get, '/health', :read, :health],
  [:get, '/limiters', :read, :limiters],
  [:get, '/cron', :read, :cron]
].freeze

Class Method Summary collapse

Class Method Details

.broadcast(method, name) ⇒ Object

One answer shape for both ways of addressing, so a client that broadcasts and a client that names one process parse the same body. An embedded process is skipped here rather than refused: a broadcast asks about the fleet, and one member that cannot be signalled is no reason to refuse the rest.



179
180
181
182
183
184
185
186
187
188
189
 
# File 'lib/wurk/api/swarm.rb', line 179

def broadcast(method, name)
  sent = []
  skipped = []
  ::Wurk::ProcessSet.new(false).each do |process|
    next skipped << process.identity if process.embedded?

    process.public_send(method)
    sent << process.identity
  end
  signal_result(name, sent, skipped)
end

.busy(request) ⇒ Object 



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

def busy(request)
  window = Page.window!(request)
  now = ::Time.now.to_f
  rows = ::Wurk::WorkSet.new.to_a
  jobs = Page.slice(rows, window) { |row| Serializers.work_row(*row, now: now) }
  Response.json(200, total: rows.size, page: window.page, count: window.count, jobs: jobs)
end

.cluster(_request) ⇒ Object 



69
70
71
72
 
# File 'lib/wurk/api/swarm.rb', line 69

def cluster(_request)
  set = ::Wurk::ProcessSet.new(false)
  Response.json(200, RollUp.cluster(set.to_a, leader_identity: set.leader, now: ::Time.now.to_f))
end

.cron(request) ⇒ Object

Walked in full before serializing. LoopSet#each yields from inside its own Redis checkout and Loop#last_fired_at takes another, so serializing mid-iteration would hold a pool slot for the length of the listing. The registry is a deploy-sized list, not a job-sized one.



146
147
148
149
150
151
152
 
# File 'lib/wurk/api/swarm.rb', line 146

def cron(request)
  window = Page.window!(request)
  loops = ::Wurk::Cron::LoopSet.new.to_a
  now = ::Time.now.to_i
  rows = Page.slice(loops, window) { |loop_obj| Serializers.cron_loop(loop_obj, now: now) }
  Response.json(200, total: loops.size, page: window.page, count: window.count, cron: rows)
end

.draw(router) ⇒ Object 



59
60
61
62
63
64
65
66
67
 
# File 'lib/wurk/api/swarm.rb', line 59

def draw(router)
  TABLE.each do |verb, pattern, scope, handler|
    router.public_send(verb, pattern, scope: scope) do |request|
      public_send(handler, request)
    rescue Validation::Invalid => e
      Problem.from(e, instance: request.path)
    end
  end
end

.filter!(query) ⇒ Object

A repeated ?filter=a&filter=b parses to an Array, and an unbounded one is a substring scan over every limiter name run on the caller's behalf — both answered rather than coerced.

Raises:



266
267
268
269
270
271
272
 
# File 'lib/wurk/api/swarm.rb', line 266

def filter!(query)
  raw = query['filter']
  return nil if raw.nil? || raw == ''
  return raw if raw.is_a?(::String) && raw.length <= MAX_FILTER_LENGTH

  raise Validation::Invalid, "The 'filter' parameter is a string of up to #{MAX_FILTER_LENGTH} characters."
end

.health(_request) ⇒ Object

Re-derived, not proxied. Health is a raw TCPServer inside each worker process, answering about its own launcher and its own heartbeat; this runs in a process that has neither, so it answers the same two questions — is Redis reachable, is anything beating — about the cluster instead. quiet is reported but not folded into the verdict: a draining process is still finishing work, and Health does not fail readiness for it either.

Not a Kubernetes probe replacement — a probe must not carry a bearer token. config.health_check(port:) is still that surface.



124
125
126
127
128
129
130
131
132
 
# File 'lib/wurk/api/swarm.rb', line 124

def health(_request)
  redis = ping_redis
  return unhealthy(redis, nil, 'redis unreachable') unless redis[:ok]

  counts = process_health
  return unhealthy(redis, counts, 'no live processes') if counts[:live].zero?

  Response.json(200, health_body('ok', redis, counts))
end

.health_body(status, redis, processes) ⇒ Object 



224
225
226
227
 
# File 'lib/wurk/api/swarm.rb', line 224

def health_body(status, redis, processes)
  { status: status, redis: redis, processes: processes,
    stale_after_seconds: Serializers::STALE_AFTER_SECONDS }
end

.index(request) ⇒ Object

Materialized before paging so total counts live processes rather than SCARD processes, which still holds identities whose heartbeat lapsed. ProcessSet#each already costs one pipeline for the whole set, so walking it in full buys the accurate count for nothing.



78
79
80
81
82
83
84
85
86
87
88
 
# File 'lib/wurk/api/swarm.rb', line 78

def index(request)
  window = Page.window!(request)
  set = ::Wurk::ProcessSet.new(false)
  live = set.to_a
  leader = set.leader
  now = ::Time.now.to_f
  rows = Page.slice(live, window) do |process|
    Serializers.process_row(process, leader: leader?(process, leader), now: now)
  end
  Response.json(200, total: live.size, page: window.page, count: window.count, processes: rows)
end

.leader?(process, leader_identity) ⇒ Boolean

Returns:

  • (Boolean) 


195
196
197
 
# File 'lib/wurk/api/swarm.rb', line 195

def leader?(process, leader_identity)
  !leader_identity.to_s.empty? && process.identity == leader_identity
end

.limiter_row(name) ⇒ Object

Two reads per row — the metadata HASH, then the limiter's own counters — which is what the page cap bounds. Limits owns the metadata and only the limiter type knows where its state lives, so there is no single key to pipeline both out of. Rebuilt from the metadata already in hand rather than through Limits.rebuild, which would re-read it.



245
246
247
248
 
# File 'lib/wurk/api/swarm.rb', line 245

def limiter_row(name)
  meta = ::Wurk::Web::Enterprise::Limits.metadata(name)
  Serializers.limiter_row(name, meta, limiter_status(name, meta))
end

.limiter_status(name, meta) ⇒ Object

Best-effort in the two ways a persisted limiter can refuse to come back: build answers nil for a type this version no longer knows, and the constructor raises ArgumentError on options it cannot accept (a bucket whose interval is gone, a name written by a later release). A row with a null status is a better answer than a 500 for the whole page. Deliberately not rescue StandardError — a broader net would also swallow a bug in the rebuild and report it as a missing status.



257
258
259
260
261
 
# File 'lib/wurk/api/swarm.rb', line 257

def limiter_status(name, meta)
  ::Wurk::Limiter.build(name, meta['type'], Serializers.parse_options(meta['options']))&.status
rescue ::ArgumentError, ::TypeError
  nil
end

.limiters(request) ⇒ Object 



134
135
136
137
138
139
140
 
# File 'lib/wurk/api/swarm.rb', line 134

def limiters(request)
  query = Page.query!(request)
  window = Page.window(query)
  names = ::Wurk::Web::Enterprise::Limits.list(filter: filter!(query))
  rows = Page.slice(names, window) { |name| limiter_row(name) }
  Response.json(200, total: names.size, page: window.page, count: window.count, limiters: rows)
end

.ping_redis ⇒ Object

Its own PING rather than a byproduct of another read: "is Redis reachable" has to stay answerable when every other read would raise.



231
232
233
234
235
236
237
238
 
# File 'lib/wurk/api/swarm.rb', line 231

def ping_redis
  started = ::Process.clock_gettime(::Process::CLOCK_MONOTONIC, :microsecond)
  pong = ::Wurk.redis(idempotent: true) { |conn| conn.call('PING') } == 'PONG'
  elapsed = ::Process.clock_gettime(::Process::CLOCK_MONOTONIC, :microsecond) - started
  { ok: pong, rtt_ms: (elapsed / 1000.0).round(3) }
rescue ::StandardError
  { ok: false, rtt_ms: nil }
end

.process_health ⇒ Object 



210
211
212
213
214
215
 
# File 'lib/wurk/api/swarm.rb', line 210

def process_health
  now = ::Time.now.to_f
  live = ::Wurk::ProcessSet.new(false).to_a
  stale = live.count { |entry| Serializers.stale?(Serializers.beat_age_seconds(entry['beat'], now)) }
  { total: live.size, live: live.size - stale, stale: stale, quiet: live.count(&:stopping?) }
end

.process_not_found(request, identity) ⇒ Object 



274
275
276
277
278
279
280
281
282
 
# File 'lib/wurk/api/swarm.rb', line 274

def process_not_found(request, identity)
  Problem.render(
    Problem::PROCESS_NOT_FOUND,
    status: 404,
    detail: "No live process has identity #{identity}; it has exited, or its heartbeat expired.",
    instance: request.path,
    identity: identity
  )
end

.process_not_signalable(request, identity) ⇒ Object

409, not 400: the request is well-formed and the caller is entitled to make it — the target is simply in a state that cannot accept it.



286
287
288
289
290
291
292
293
294
 
# File 'lib/wurk/api/swarm.rb', line 286

def process_not_signalable(request, identity)
  Problem.render(
    Problem::PROCESS_NOT_SIGNALABLE,
    status: 409,
    detail: "Process #{identity} is embedded in its host application; signalling it would signal the host.",
    instance: request.path,
    identity: identity
  )
end

.quiet(request) ⇒ Object 



154
 
# File 'lib/wurk/api/swarm.rb', line 154

def quiet(request) = signal(request, :quiet!, QUIET_SIGNAL)

.show(request) ⇒ Object

The same row the listing emits, plus what this process is running, so a client that read a row out of /processes and then fetched it does not have to reshape anything.



93
94
95
96
97
98
99
100
101
102
103
104
 
# File 'lib/wurk/api/swarm.rb', line 93

def show(request)
  identity = Validation.identity!(request.path_params[:identity].to_s)
  process = ::Wurk::ProcessSet[identity]
  return process_not_found(request, identity) unless process

  now = ::Time.now.to_f
  Response.json(
    200,
    process: Serializers.process_row(process, leader: process.leader?, now: now),
    work: work_for(identity, now)
  )
end

.signal(request, method, name) ⇒ Object

Both signals go through Process#quiet! / #stop! — the same <identity>-signals LPUSH the Busy page sends — rather than a second writer to that list. Asynchronous by construction: the target picks the entry up on its next beat, so 200 means queued, never applied.



161
162
163
164
165
166
167
168
169
170
171
172
 
# File 'lib/wurk/api/swarm.rb', line 161

def signal(request, method, name)
  identity = request.path_params[:identity].to_s
  return broadcast(method, name) if identity == ALL

  Validation.identity!(identity)
  process = ::Wurk::ProcessSet[identity]
  return process_not_found(request, identity) unless process
  return process_not_signalable(request, identity) if process.embedded?

  process.public_send(method)
  signal_result(name, [identity], [])
end

.signal_result(name, sent, skipped) ⇒ Object 



191
192
193
 
# File 'lib/wurk/api/swarm.rb', line 191

def signal_result(name, sent, skipped)
  Response.json(200, signal: name, signalled: sent, skipped: skipped)
end

.stop(request) ⇒ Object 



155
 
# File 'lib/wurk/api/swarm.rb', line 155

def stop(request) = signal(request, :stop!, STOP_SIGNAL)

.unhealthy(redis, processes, reason) ⇒ Object

A 503 whose body is the report itself, not a problem document: an unhealthy cluster is the answer this route was asked for, and Health's own {"status":"down","reason":…} is the shape operators already read.



220
221
222
 
# File 'lib/wurk/api/swarm.rb', line 220

def unhealthy(redis, processes, reason)
  Response.json(503, health_body('down', redis, processes).merge(reason: reason))
end

.work_for(identity, now) ⇒ Object

Filtered out of the canonical WorkSet rather than read straight off <identity>:work: a second reader of that hash is a second place for the run_at ordering and the JobRecord unwrapping to drift.



202
203
204
205
206
207
208
 
# File 'lib/wurk/api/swarm.rb', line 202

def work_for(identity, now)
  ::Wurk::WorkSet.new.filter_map do |process_id, thread_id, work|
    next unless process_id == identity

    Serializers.work_row(process_id, thread_id, work, now: now)
  end
end