Module: Wurk::API::Queues

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

Overview

The observe plane: how deep the queues are, what is waiting to retry, what died, and the counters over the lot.

Every route reads through a canonical inspector — Stats, Queue, RetrySet, ScheduledSet, DeadSet, Batch::Status — the same objects the dashboard reads. Not to save code: a second reader over the same Redis keys is a second place for the size-descending queue order, the latency math, the paused set and the ActiveJob unwrapping to drift, and a dashboard and an API that disagree about how deep a queue is are worse than either on its own.

Pause and unpause are the only writes, and they go through Queue#pause! / Queue#unpause!, which expire this process's fetcher cache of the paused set. A bare SADD here would leave the swarm fetching from a queue the API had just reported paused.

Constant Summary collapse

TABLE =

Verb, pattern, the scope a caller must hold, handler. A table rather than nine router.get calls because the scope column is the part that has to be read at a glance: the two writes take :admin, not :read. Pausing default stops the fleet from working without enqueueing or deleting anything, so it belongs with the destructive routes rather than with the listings it sits beside.

[
  [:get, '/stats', :read, :stats],
  [:get, '/queues', :read, :index],
  [:get, '/queues/:name', :read, :show],
  [:get, '/retries', :read, :retries],
  [:get, '/scheduled', :read, :scheduled],
  [:get, '/dead', :read, :dead],
  [:get, '/batches/:bid', :read, :batch],
  [:post, '/queues/:name/pause', :admin, :pause],
  [:post, '/queues/:name/unpause', :admin, :unpause]
].freeze

Class Method Summary collapse

Class Method Details

.batch(request) ⇒ Object 



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

def batch(request)
  bid = Validation.bid!(request.path_params[:bid])
  status = ::Wurk::Batch::Status.new(bid)
  return batch_not_found(request, bid) unless status.exists?

  Response.json(200, status.data)
end

.batch_not_found(request, bid) ⇒ Object 



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

def batch_not_found(request, bid)
  Problem.render(
    Problem::BATCH_NOT_FOUND,
    status: 404,
    detail: "No batch has bid #{bid}; it was never created, or it has expired.",
    instance: request.path,
    bid: bid
  )
end

.dead(request) ⇒ Object 



85
 
# File 'lib/wurk/api/queues.rb', line 85

def dead(request) = sorted_set(request, ::Wurk::DeadSet.new)

.draw(router) ⇒ Object

The refusal rendering wraps every handler here rather than sitting in each one: they all validate something that came off the network, and a rescue repeated per route is one a new route gets written without.



50
51
52
53
54
55
56
57
58
 
# File 'lib/wurk/api/queues.rb', line 50

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

.index(_request) ⇒ Object

Stats#queue_summaries rather than Queue.all: one pipeline instead of an LLEN round trip per queue, and it comes back in the size-descending order every other Wurk surface lists queues in.



67
68
69
 
# File 'lib/wurk/api/queues.rb', line 67

def index(_request)
  Response.json(200, queues: ::Wurk::Stats.new.queue_summaries.map { |q| Serializers.queue_summary(q) })
end

.pause(request) ⇒ Object 



87
 
# File 'lib/wurk/api/queues.rb', line 87

def pause(request) = toggle(request, paused: true)

.retries(request) ⇒ Object 



83
 
# File 'lib/wurk/api/queues.rb', line 83

def retries(request) = sorted_set(request, ::Wurk::RetrySet.new)

.scheduled(request) ⇒ Object 



84
 
# File 'lib/wurk/api/queues.rb', line 84

def scheduled(request) = sorted_set(request, ::Wurk::ScheduledSet.new)

.show(request) ⇒ Object

No 404 for a name nothing was ever enqueued under. A queue is not an entity in the Sidekiq schema — it is a LIST that exists only while it has members — so "never used" and "drained" are the same state in Redis, and size: 0 is the honest reading of both. Pausing one before its first job is a legitimate pre-deploy move for the same reason.



76
77
78
79
80
81
 
# File 'lib/wurk/api/queues.rb', line 76

def show(request)
  window = Page.window!(request)
  queue = ::Wurk::Queue.new(Validation.queue_name!(request.path_params[:name]))
  jobs = Page.slice(queue, window) { |record| Serializers.job_record(record) }
  Response.json(200, Serializers.queue_gauges(queue).merge(page: window.page, count: window.count, jobs: jobs))
end

.sorted_set(request, set) ⇒ Object

total is read before the page is walked, so it describes the set the page was taken from rather than the one left after a poller drained it.



100
101
102
103
104
105
 
# File 'lib/wurk/api/queues.rb', line 100

def sorted_set(request, set)
  window = Page.window!(request)
  total = set.size
  jobs = Page.slice(set, window) { |entry| Serializers.sorted_entry(entry) }
  Response.json(200, name: set.name, total: total, page: window.page, count: window.count, jobs: jobs)
end

.stats(_request) ⇒ Object 



60
61
62
 
# File 'lib/wurk/api/queues.rb', line 60

def stats(_request)
  Response.json(200, Serializers.stats(::Wurk::Stats.new))
end

.toggle(request, paused:) ⇒ Object

Idempotent both ways (SADD/SREM), and the resulting state comes back so a client does not need a second request to confirm the toggle.



92
93
94
95
96
 
# File 'lib/wurk/api/queues.rb', line 92

def toggle(request, paused:)
  queue = ::Wurk::Queue.new(Validation.queue_name!(request.path_params[:name]))
  paused ? queue.pause! : queue.unpause!
  Response.json(200, name: queue.name, paused: paused)
end

.unpause(request) ⇒ Object 



88
 
# File 'lib/wurk/api/queues.rb', line 88

def unpause(request) = toggle(request, paused: false)