Module: Resque::Metrics

Defined in:
lib/resque/metrics.rb,
lib/resque/metrics/server.rb,
lib/resque/metrics/backends/null.rb,
lib/resque/metrics/backends/redis.rb,
lib/resque/metrics/backends/statsd.rb

Defined Under Namespace

Modules: Backends, Hooks, Server

Class Method Summary collapse

Class Method Details

.after_fork ⇒ Object 



100
101
102
103
104
105
106
107
108
109
110
111
 
# File 'lib/resque/metrics.rb', line 100

def self.after_fork
  lambda do |job|
    end_time = Time.now.to_f * 1000
    key = "_metrics_:fork_start:#{job.worker.to_s}"
    start_time = ::Resque.redis.get key
    if start_time
      total = (end_time - start_time.to_f).to_i
      ::Resque::Metrics.record_job_fork(job, total)
    end
    true
  end
end

.avg_fork_time ⇒ Object 



297
298
299
 
# File 'lib/resque/metrics.rb', line 297

def self.avg_fork_time
  get_metric "avg_fork_time"
end

.avg_fork_time_by_job(job) ⇒ Object 



305
306
307
 
# File 'lib/resque/metrics.rb', line 305

def self.avg_fork_time_by_job(job)
  get_metric "avg_fork_time:job:#{job}"
end

.avg_fork_time_by_queue(queue) ⇒ Object 



301
302
303
 
# File 'lib/resque/metrics.rb', line 301

def self.avg_fork_time_by_queue(queue)
  get_metric "avg_fork_time:queue:#{queue}"
end

.avg_job_time ⇒ Object 



225
226
227
 
# File 'lib/resque/metrics.rb', line 225

def self.avg_job_time
  get_metric "avg_job_time"
end

.avg_job_time_by_job(job) ⇒ Object 



233
234
235
 
# File 'lib/resque/metrics.rb', line 233

def self.avg_job_time_by_job(job)
  get_metric "avg_job_time:job:#{job}"
end

.avg_job_time_by_queue(queue) ⇒ Object 



229
230
231
 
# File 'lib/resque/metrics.rb', line 229

def self.avg_job_time_by_queue(queue)
  get_metric "avg_job_time:queue:#{queue}"
end

.avg_payload_size ⇒ Object 



285
286
287
 
# File 'lib/resque/metrics.rb', line 285

def self.avg_payload_size
  get_metric "avg_payload_size"
end

.avg_payload_size_by_job(job) ⇒ Object 



293
294
295
 
# File 'lib/resque/metrics.rb', line 293

def self.avg_payload_size_by_job(job)
  get_metric "avg_payload_size:job:#{job}"
end

.avg_payload_size_by_queue(queue) ⇒ Object 



289
290
291
 
# File 'lib/resque/metrics.rb', line 289

def self.avg_payload_size_by_queue(queue)
  get_metric "avg_payload_size:queue:#{queue}"
end

.backends ⇒ Object 



27
28
29
30
31
 
# File 'lib/resque/metrics.rb', line 27

def self.backends
  @_backends ||= begin
                   self.backends = [Resque::Metrics::Backends::Redis.new(redis)]
                 end
end

.backends=(new_backends) ⇒ Object 



33
34
35
 
# File 'lib/resque/metrics.rb', line 33

def self.backends=(new_backends)
  @_backends = new_backends
end

.before_fork ⇒ Object 



90
91
92
93
94
95
96
97
98
 
# File 'lib/resque/metrics.rb', line 90

def self.before_fork
  lambda do |job|
    start = Time.now.to_f * 1000
    key = "_metrics_:fork_start:#{job.worker.to_s}"
    ::Resque.redis.set key, start
    ::Resque.redis.expire key, 60 * 60 * 60
    true
  end
end

.depth_by_queue(queue) ⇒ Object 



341
342
343
 
# File 'lib/resque/metrics.rb', line 341

def self.depth_by_queue(queue)
  get_metric "depth:queue:#{queue}"
end

.extended(klass) ⇒ Object 



7
8
9
 
# File 'lib/resque/metrics.rb', line 7

def self.extended(klass)
  klass.extend(Hooks)
end

.failed_depth ⇒ Object 



333
334
335
 
# File 'lib/resque/metrics.rb', line 333

def self.failed_depth
  get_metric "depth:failed"
end

.failed_job_count ⇒ Object 



261
262
263
 
# File 'lib/resque/metrics.rb', line 261

def self.failed_job_count
  get_metric "failed_job_count"
end

.failed_job_count_by_job(job) ⇒ Object 



269
270
271
 
# File 'lib/resque/metrics.rb', line 269

def self.failed_job_count_by_job(job)
  get_metric "failed_job_count:job:#{job}"
end

.failed_job_count_by_queue(queue) ⇒ Object 



265
266
267
 
# File 'lib/resque/metrics.rb', line 265

def self.failed_job_count_by_queue(queue)
  get_metric "failed_job_count:queue:#{queue}"
end

.get_metric(metric) ⇒ Object 



209
210
211
 
# File 'lib/resque/metrics.rb', line 209

def self.get_metric(metric)
  run_first_backend(:get_metric, metric)
end

.increment_metric(metric, by = 1) ⇒ Object 



193
194
195
 
# File 'lib/resque/metrics.rb', line 193

def self.increment_metric(metric, by = 1)
  run_backends(:increment_metric, metric, by)
end

.known_jobs ⇒ Object 



205
206
207
 
# File 'lib/resque/metrics.rb', line 205

def self.known_jobs
  run_first_backend(:known_jobs)
end

.multi(&block) ⇒ Object 



189
190
191
 
# File 'lib/resque/metrics.rb', line 189

def self.multi(&block)
  use_multi? ? redis.multi(&block) : yield
end

.on_job_complete(&block) ⇒ Object 



66
67
68
 
# File 'lib/resque/metrics.rb', line 66

def self.on_job_complete(&block)
  set_callback(:on_job_complete, &block)
end

.on_job_enqueue(&block) ⇒ Object 



70
71
72
 
# File 'lib/resque/metrics.rb', line 70

def self.on_job_enqueue(&block)
  set_callback(:on_job_enqueue, &block)
end

.on_job_failure(&block) ⇒ Object 



74
75
76
 
# File 'lib/resque/metrics.rb', line 74

def self.on_job_failure(&block)
  set_callback(:on_job_failure, &block)
end

.on_job_fork(&block) ⇒ Object 



62
63
64
 
# File 'lib/resque/metrics.rb', line 62

def self.on_job_fork(&block)
  set_callback(:on_job_fork, &block)
end

.pending_depth ⇒ Object 



337
338
339
 
# File 'lib/resque/metrics.rb', line 337

def self.pending_depth
  get_metric "depth:pending"
end

.record_depth ⇒ Object 



113
114
115
116
117
118
119
120
121
122
 
# File 'lib/resque/metrics.rb', line 113

def self.record_depth
  set_metric 'depth:failed', Resque::Failure.count
  set_metric 'depth:pending', Resque.info[:pending]

  Resque.queues.each do |queue|
    set_metric "depth:queue:#{queue}", Resque.size(queue)
  end

  true
end

.record_job_completion(job_class, time) ⇒ Object 



161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
 
# File 'lib/resque/metrics.rb', line 161

def self.record_job_completion(job_class, time)
  queue = Resque.queue_from_class(job_class)
  multi do
    increment_metric "job_time", time
    increment_metric "job_time:queue:#{queue}", time
    increment_metric "job_time:job:#{job_class}", time
    increment_metric "job_count"
    increment_metric "job_count:queue:#{queue}"
    increment_metric "job_count:job:#{job_class}"
  end
  set_avg "avg_job_time", total_job_time, total_job_count
  set_avg "avg_job_time:queue:#{queue}", total_job_time_by_queue(queue) , total_job_count_by_queue(queue)
  set_avg "avg_job_time:job:#{job_class}", total_job_time_by_job(job_class) , total_job_count_by_job(job_class)
  run_callback(:on_job_complete, job_class, queue, time)
end

.record_job_enqueue(job_class, *args) ⇒ Object 



141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
 
# File 'lib/resque/metrics.rb', line 141

def self.record_job_enqueue(job_class, *args)
  queue = Resque.queue_from_class(job_class)
  increment_metric "enqueue_count"
  increment_metric "enqueue_count:job:#{job_class}"
  increment_metric "enqueue_count:queue:#{queue}"
  run_first_backend(:register_job, job_class)

  size = Resque.encode(args).length
  multi do
    increment_metric "payload_size", size
    increment_metric "payload_size:queue:#{queue}", size
    increment_metric "payload_size:job:#{job_class}", size
  end
  set_avg "avg_payload_size", total_payload_size , total_enqueue_count
  set_avg "avg_payload_size:queue:#{queue}", total_payload_size_by_queue(queue) , total_enqueue_count_by_queue(queue)
  set_avg "avg_payload_size:job:#{job_class}", total_payload_size_by_job(job_class) , total_enqueue_count_by_job(job_class)
  run_callback(:on_job_enqueue, job_class, queue, size)
  true
end

.record_job_failure(job_class, e) ⇒ Object 



177
178
179
180
181
182
183
184
185
186
187
 
# File 'lib/resque/metrics.rb', line 177

def self.record_job_failure(job_class, e)
  queue = Resque.queue_from_class(job_class)

  multi do
    increment_metric "failed_job_count"
    increment_metric "failed_job_count:queue:#{queue}"
    increment_metric "failed_job_count:job:#{job_class}"
  end

  run_callback(:on_job_failure, job_class, queue)
end

.record_job_fork(job, time) ⇒ Object 



124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
 
# File 'lib/resque/metrics.rb', line 124

def self.record_job_fork(job, time)
  job_class = job.payload_class
  queue = job.queue
  multi do
    increment_metric "fork_time", time
    increment_metric "fork_time:queue:#{queue}", time
    increment_metric "fork_time:job:#{job_class}", time
    increment_metric "fork_count"
    increment_metric "fork_count:queue:#{queue}"
    increment_metric "fork_count:job:#{job_class}"
  end
  set_avg "avg_fork_time", total_fork_time , total_fork_count
  set_avg "avg_fork_time:queue:#{queue}", total_fork_time_by_queue(queue) , total_fork_count_by_queue(queue)
  set_avg "avg_fork_time:job:#{job_class}", total_fork_time_by_job(job_class) , total_fork_count_by_job(job_class)
  run_callback(:on_job_fork, job_class, queue, time)
end

.redis ⇒ Object 



11
12
13
 
# File 'lib/resque/metrics.rb', line 11

def self.redis
  @_redis ||= ::Resque.redis
end

.redis=(redis) ⇒ Object 



15
16
17
 
# File 'lib/resque/metrics.rb', line 15

def self.redis=(redis)
  @_redis = redis
end

.run_backends(method, *args) ⇒ Object 



37
38
39
40
41
42
43
44
45
46
47
48
 
# File 'lib/resque/metrics.rb', line 37

def self.run_backends(method, *args)
  ran_any = false

  backends.each do |backend|
    if backend.respond_to?(method)
      ran_any = true
      backend.send method, *args
    end
  end

  raise "No backend responded to #{method}: #{backends.inspect}" unless ran_any
end

.run_callback(callback_name, *args) ⇒ Object 



84
85
86
87
88
 
# File 'lib/resque/metrics.rb', line 84

def self.run_callback(callback_name, *args)
  if @callbacks && @callbacks[callback_name]
    @callbacks[callback_name].each {|callback| callback.call(*args) }
  end
end

.run_first_backend(method, *args) ⇒ Object 



50
51
52
53
54
55
 
# File 'lib/resque/metrics.rb', line 50

def self.run_first_backend(method, *args)
  backend = backends.detect {|backend| backend.respond_to?(method)}
  raise "No backend responds to #{method}: #{backends.inspect}" unless backend

  backend.send method, *args
end

.set_avg(metric, num, total) ⇒ Object 



201
202
203
 
# File 'lib/resque/metrics.rb', line 201

def self.set_avg(metric, num, total)
  run_backends(:set_avg, metric, num, total)
end

.set_callback(callback_name, &block) ⇒ Object 



78
79
80
81
82
 
# File 'lib/resque/metrics.rb', line 78

def self.set_callback(callback_name, &block)
  @callbacks ||= {}
  @callbacks[callback_name] ||= []
  @callbacks[callback_name] << block
end

.set_metric(metric, val) ⇒ Object 



197
198
199
 
# File 'lib/resque/metrics.rb', line 197

def self.set_metric(metric, val)
  run_backends(:set_metric, metric, val)
end

.total_enqueue_count ⇒ Object 



213
214
215
 
# File 'lib/resque/metrics.rb', line 213

def self.total_enqueue_count
  get_metric "enqueue_count"
end

.total_enqueue_count_by_job(job) ⇒ Object 



217
218
219
 
# File 'lib/resque/metrics.rb', line 217

def self.total_enqueue_count_by_job(job)
  get_metric "enqueue_count:job:#{job}"
end

.total_enqueue_count_by_queue(queue) ⇒ Object 



221
222
223
 
# File 'lib/resque/metrics.rb', line 221

def self.total_enqueue_count_by_queue(queue)
  get_metric "enqueue_count:queue:#{queue}"
end

.total_fork_count ⇒ Object 



321
322
323
 
# File 'lib/resque/metrics.rb', line 321

def self.total_fork_count
  get_metric "fork_count"
end

.total_fork_count_by_job(job) ⇒ Object 



329
330
331
 
# File 'lib/resque/metrics.rb', line 329

def self.total_fork_count_by_job(job)
  get_metric "fork_count:job:#{job}"
end

.total_fork_count_by_queue(queue) ⇒ Object 



325
326
327
 
# File 'lib/resque/metrics.rb', line 325

def self.total_fork_count_by_queue(queue)
  get_metric "fork_count:queue:#{queue}"
end

.total_fork_time ⇒ Object 



309
310
311
 
# File 'lib/resque/metrics.rb', line 309

def self.total_fork_time
  get_metric "fork_time"
end

.total_fork_time_by_job(job) ⇒ Object 



317
318
319
 
# File 'lib/resque/metrics.rb', line 317

def self.total_fork_time_by_job(job)
  get_metric "fork_time:job:#{job}"
end

.total_fork_time_by_queue(queue) ⇒ Object 



313
314
315
 
# File 'lib/resque/metrics.rb', line 313

def self.total_fork_time_by_queue(queue)
  get_metric "fork_time:queue:#{queue}"
end

.total_job_count ⇒ Object 



249
250
251
 
# File 'lib/resque/metrics.rb', line 249

def self.total_job_count
  get_metric "job_count"
end

.total_job_count_by_job(job) ⇒ Object 



257
258
259
 
# File 'lib/resque/metrics.rb', line 257

def self.total_job_count_by_job(job)
  get_metric "job_count:job:#{job}"
end

.total_job_count_by_queue(queue) ⇒ Object 



253
254
255
 
# File 'lib/resque/metrics.rb', line 253

def self.total_job_count_by_queue(queue)
  get_metric "job_count:queue:#{queue}"
end

.total_job_time ⇒ Object 



237
238
239
 
# File 'lib/resque/metrics.rb', line 237

def self.total_job_time
  get_metric "job_time"
end

.total_job_time_by_job(job) ⇒ Object 



245
246
247
 
# File 'lib/resque/metrics.rb', line 245

def self.total_job_time_by_job(job)
  get_metric "job_time:job:#{job}"
end

.total_job_time_by_queue(queue) ⇒ Object 



241
242
243
 
# File 'lib/resque/metrics.rb', line 241

def self.total_job_time_by_queue(queue)
  get_metric "job_time:queue:#{queue}"
end

.total_payload_size ⇒ Object 



273
274
275
 
# File 'lib/resque/metrics.rb', line 273

def self.total_payload_size
  get_metric "payload_size"
end

.total_payload_size_by_job(job) ⇒ Object 



281
282
283
 
# File 'lib/resque/metrics.rb', line 281

def self.total_payload_size_by_job(job)
  get_metric "payload_size:job:#{job}"
end

.total_payload_size_by_queue(queue) ⇒ Object 



277
278
279
 
# File 'lib/resque/metrics.rb', line 277

def self.total_payload_size_by_queue(queue)
  get_metric "payload_size:queue:#{queue}"
end

.use_multi=(multi) ⇒ Object 



19
20
21
 
# File 'lib/resque/metrics.rb', line 19

def self.use_multi=(multi)
  @_use_multi = multi
end

.use_multi? ⇒ Boolean

Returns:

  • (Boolean) 


23
24
25
 
# File 'lib/resque/metrics.rb', line 23

def self.use_multi?
  @_use_multi
end

.watch_fork ⇒ Object 



57
58
59
60
 
# File 'lib/resque/metrics.rb', line 57

def self.watch_fork
  ::Resque.before_fork = before_fork
  ::Resque.after_fork = after_fork
end