Class: Telegraf::Sidekiq::Middleware

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

Overview

Telegraf::Sidekiq::Middleware

This Sidekiq middleware collects queue metrics and sends them to telegraf.

Tags:

  • ‘type`:

    One of "job" or "scheduled_job".

  • ‘queue`:

    The queue this job landed on.

  • ‘worker`:

    The name of the worker class that was executed.

  • ‘errors`:

    Whether or not this job errored.

  • ‘retry`:

    Whether or not this execution was a retry of a previously failed one.

Values:

  • ‘app_ms`:

    Total worker processing time.

  • ‘queue_ms`:

    How long did this job wait in the queue before being processed?
Only present for "normal" (async) jobs (with tag `type` of "job").

Instance Method Summary collapse

Constructor Details

#initialize(agent:, series: 'sidekiq', tags: {}) ⇒ Middleware

Returns a new instance of Middleware.



40
41
42
43
44
 
# File 'lib/telegraf/sidekiq.rb', line 40

def initialize(agent:, series: 'sidekiq', tags: {})
  @agent = agent
  @series = series.to_s.freeze
  @tags = tags.freeze
end

Instance Method Details

#call(worker, job, queue) ⇒ Object 



46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/telegraf/sidekiq.rb', line 46

def call(worker, job, queue)
  job_start = ::Time.now.utc

  tags = {
    **@tags,
    type: 'job',
    errors: true,
    retry: job.key?('retried_at'),
    queue: queue,
    worker: worker.class.name
  }

  values = {
    retry_count: job['retry_count']
  }.compact

  # The "enqueued_at" key is not present for scheduled jobs.
  # See https://github.com/mperham/sidekiq/wiki/Job-Format.
  if job.key?('enqueued_at')
    enqueued_at = ::Time.at(job['enqueued_at'].to_f).utc
    values[:queue_ms] = (job_start - enqueued_at) * 1000 # milliseconds
  end

  # The "at" key is only present for scheduled jobs.
  tags[:type] = 'scheduled_job' if job.key?('at')

  begin
    yield

    # If we get here, this was a successful execution
    tags[:errors] = false
  ensure
    job_stop = ::Time.now.utc

    values[:app_ms] = (job_stop - job_start) * 1000 # milliseconds

    @agent.write(
      @series, tags: tags, values: values
    )
  end
end