Class: SidekiqVigil::Collector

Inherits:
Object
  • Object
show all
Defined in:
lib/sidekiq_vigil/collector.rb

Defined Under Namespace

Classes: Snapshot

Instance Method Summary collapse

Constructor Details

#initialize(clock: -> { Time.now }) ⇒ Collector

Returns a new instance of Collector.



7
8
9
10
11
12
 
# File 'lib/sidekiq_vigil/collector.rb', line 7

def initialize(clock: -> { Time.now })
  @clock = clock
  @mutex = Mutex.new
  @stats = empty_stats
  @executions = empty_executions
end

Instance Method Details

#drain ⇒ Object 



23
24
25
26
27
28
29
30
 
# File 'lib/sidekiq_vigil/collector.rb', line 23

def drain
  @mutex.synchronize do
    snapshot = Snapshot.new(stats: @stats, executions: @executions)
    @stats = empty_stats
    @executions = empty_executions
    snapshot
  end
end

#merge(snapshot) ⇒ Object 



32
33
34
35
36
37
38
39
40
41
42
43
44
 
# File 'lib/sidekiq_vigil/collector.rb', line 32

def merge(snapshot)
  @mutex.synchronize do
    snapshot.stats.each do |minute, fields|
      fields.each { |field, value| @stats[minute][field] += value }
    end
    snapshot.executions.each do |queue, values|
      current = @executions[queue]
      current[:count] += values[:count]
      current[:sum] += values[:sum]
      current[:max] = [current[:max], values[:max]].max
    end
  end
end

#record(worker:, queue:, duration:, failed:) ⇒ Object 



14
15
16
17
18
19
20
21
 
# File 'lib/sidekiq_vigil/collector.rb', line 14

def record(worker:, queue:, duration:, failed:)
  minute = clock.call.utc.strftime("%Y%m%d%H%M")
  worker_name = worker.respond_to?(:name) ? worker.name : worker.class.name
  @mutex.synchronize do
    update_stats(minute, worker_name, failed)
    update_execution(queue.to_s, duration)
  end
end