Class: SolidObserver::QueueStats

Inherits:
Object
  • Object
show all
Defined in:
lib/solid_observer/queue_stats.rb

Overview

:reek:TooManyMethods

Constant Summary collapse

RANGES = 
{
  "15m" => 15.minutes,
  "30m" => 30.minutes,
  "1h" => 1.hour,
  "7h" => 7.hours,
  "1d" => 1.day,
  "7d" => 7.days,
  "14d" => 14.days
}.freeze
DEFAULT_RANGE = 
"15m"
POLL_DEFAULT_RANGE = 
"15m"
POLL_EMPTY_SNAPSHOT = 
{
  ready: 0,
  scheduled: 0,
  claimed: 0,
  workers: 0,
  failed: 0,
  enqueue_rate_per_min: nil
}.freeze
TICK_EMPTY_SNAPSHOT = 
{
  ready: 0,
  scheduled: 0,
  claimed: 0,
  workers: 0,
  failed: 0
}.freeze
BUCKET_RULES = 
[
  [30.minutes.to_i, 30],
  [2.hours.to_i, 60],
  [1.day.to_i, 5.minutes.to_i]
].freeze

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.chart_data(window: 15.minutes) ⇒ Object 



53
54
55
 
# File 'lib/solid_observer/queue_stats.rb', line 53

def chart_data(window: 15.minutes)
  new.chart_data(window)
end

.parse_range(value, fallback: DEFAULT_RANGE) ⇒ Object 



61
62
63
64
 
# File 'lib/solid_observer/queue_stats.rb', line 61

def parse_range(value, fallback: DEFAULT_RANGE)
  range_key = value.to_s
  RANGES.key?(range_key) ? range_key : fallback
end

.range_duration(value, fallback: DEFAULT_RANGE) ⇒ Object 



66
67
68
 
# File 'lib/solid_observer/queue_stats.rb', line 66

def range_duration(value, fallback: DEFAULT_RANGE)
  RANGES.fetch(parse_range(value, fallback: fallback))
end

.snapshot(range: DEFAULT_RANGE) ⇒ Object 



41
42
43
 
# File 'lib/solid_observer/queue_stats.rb', line 41

def snapshot(range: DEFAULT_RANGE)
  new.snapshot(range)
end

.snapshot_for_poll(range:) ⇒ Object 



45
46
47
 
# File 'lib/solid_observer/queue_stats.rb', line 45

def snapshot_for_poll(range:)
  new.snapshot_for_poll(parse_range(range, fallback: POLL_DEFAULT_RANGE))
end

.snapshot_for_tick ⇒ Object 



49
50
51
 
# File 'lib/solid_observer/queue_stats.rb', line 49

def snapshot_for_tick
  new.snapshot_for_tick
end

.solid_queue_available? ⇒ Boolean

Returns:

  • (Boolean) 


57
58
59
 
# File 'lib/solid_observer/queue_stats.rb', line 57

def solid_queue_available?
  !!(defined?(SolidQueue) && defined?(SolidQueue::Job))
end

Instance Method Details

#chart_data(window) ⇒ Object

:reek:TooManyStatements



124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
 
# File 'lib/solid_observer/queue_stats.rb', line 124

def chart_data(window)
  seconds = window.to_i
  ready = ChartBuffer.recent(seconds)
  return {performed: [], failed: [], ready: ready} unless SolidObserver.config.persistence_mode?

  bucket_seconds = derive_bucket_seconds(window)
  {
    performed: QueueEvent.count_by_time_bucket(
      event_type: "job_completed",
      window: window,
      bucket_seconds: bucket_seconds
    ),
    failed: QueueEvent.count_by_time_bucket(
      event_type: "job_failed",
      window: window,
      bucket_seconds: bucket_seconds
    ),
    ready: ready
  }
end

#snapshot(range = DEFAULT_RANGE) ⇒ Object 



71
72
73
74
75
76
77
78
 
# File 'lib/solid_observer/queue_stats.rb', line 71

def snapshot(range = DEFAULT_RANGE)
  klass = self.class
  return snapshot_for_mode(range && klass.parse_range(range)) if klass.solid_queue_available?

  error_response("SolidQueue not available")
rescue => e
  error_response(e.message)
end

#snapshot_for_poll(range) ⇒ Object

:reek:TooManyStatements



81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
 
# File 'lib/solid_observer/queue_stats.rb', line 81

def snapshot_for_poll(range)
  empty_snapshot = POLL_EMPTY_SNAPSHOT.dup
  klass = self.class
  return empty_snapshot unless klass.solid_queue_available?

  window = klass.range_duration(range, fallback: POLL_DEFAULT_RANGE)
  persistence = SolidObserver.config.persistence_mode?
  base = {
    ready: ready_count,
    scheduled: scheduled_count,
    claimed: claimed_count,
    workers: active_workers_count,
    failed: failed_count,
    enqueue_rate_per_min: persistence ? QueueEvent.enqueue_rate_per_minute(window: window) : nil
  }

  if persistence
    base.merge!(throughput_stats(range))
    base[:queues] = queue_depths
    base[:performed_by_queue] = QueueEvent.count_by_queue_and_event_type(window: window, event_type: "job_completed")
    base[:failed_by_queue] = QueueEvent.count_by_queue_and_event_type(window: window, event_type: "job_failed")
  end

  base
rescue
  empty_snapshot
end

#snapshot_for_tick ⇒ Object 



109
110
111
112
113
114
115
116
117
118
119
120
121
 
# File 'lib/solid_observer/queue_stats.rb', line 109

def snapshot_for_tick
  return TICK_EMPTY_SNAPSHOT unless self.class.solid_queue_available?

  {
    ready: ready_count,
    scheduled: scheduled_count,
    claimed: claimed_count,
    workers: active_workers_count,
    failed: failed_count
  }
rescue
  TICK_EMPTY_SNAPSHOT
end