Class: Sidekiq::Routing::QueueComposition

Inherits:
Object
  • Object
show all
Defined in:
lib/sidekiq/routing/queue_composition.rb

Overview

Read-only, capped scan of a live queue grouped by displayed job class. Used during incidents to answer: "which class is flooding this SLA tier?"

Defined Under Namespace

Classes: Report

Constant Summary collapse

DEFAULT_SCAN_LIMIT = 
250_000

Instance Method Summary collapse

Constructor Details

#initialize(queue_name, scan_limit: DEFAULT_SCAN_LIMIT, now: Time.now) ⇒ QueueComposition

Returns a new instance of QueueComposition.



37
38
39
40
41
 
# File 'lib/sidekiq/routing/queue_composition.rb', line 37

def initialize(queue_name, scan_limit: DEFAULT_SCAN_LIMIT, now: Time.now)
  @queue_name = normalize_queue_name(queue_name)
  @scan_limit = normalize_scan_limit(scan_limit)
  @now = now
end

Instance Method Details

#call ⇒ Object 



43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/sidekiq/routing/queue_composition.rb', line 43

def call
  by_class = Hash.new { |hash, key| hash[key] = { count: 0, oldest_at: nil } }
  queue = Sidekiq::Queue.new(@queue_name)
  scanned = 0

  queue.each do |job|
    break if scanned >= @scan_limit

    scanned += 1
    row = by_class[job.display_class]
    row[:count] += 1

    enqueued_at = job.enqueued_at
    row[:oldest_at] = enqueued_at if older?(enqueued_at, row[:oldest_at])
  end

  Report.new(
    queue: @queue_name,
    rows: build_rows(by_class),
    scanned: scanned,
    size: queue.size,
    scan_limit: @scan_limit
  )
end