Class: Instrumental::Agent

Inherits:
Object
  • Object
show all
Defined in:
lib/instrumental/agent.rb

Constant Summary collapse

BACKOFF = 
2.0
CONNECT_TIMEOUT = 
20
EXIT_FLUSH_TIMEOUT = 
5
HOSTNAME = 
Socket.gethostbyname(Socket.gethostname).first rescue Socket.gethostname
MAX_BUFFER = 
5000
MAX_AGGREGATOR_SIZE = 
5000
MAX_RECONNECT_DELAY = 
15
REPLY_TIMEOUT = 
10
RESOLUTION_FAILURES_BEFORE_WAITING = 
3
RESOLUTION_WAIT = 
30
RESOLVE_TIMEOUT = 
1
DEFAULT_FREQUENCY = 
0
VALID_FREQUENCIES = 
[0, 1, 2, 3, 4, 5, 6, 10, 12, 15, 20, 30, 60]

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(api_key, options = {}) ⇒ Agent

Sets up a connection to the collector.

Instrumental::Agent.new(API_KEY) Instrumental::Agent.new(API_KEY, :collector => 'hostname:port')



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
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
 
# File 'lib/instrumental/agent.rb', line 49

def initialize(api_key, options = {})
  # symbolize options keys
  options.replace(
    options.inject({}) { |m, (k, v)| m[(k.to_sym rescue k) || k] = v; m }
  )

  # defaults
  # host:        collector.instrumentalapp.com
  # port:        8001
  # enabled:     true
  # synchronous: false
  # frequency:   10
  # secure:      true
  # verify:      true
  @api_key         = api_key
  @host, @port     = options[:collector].to_s.split(':')
  @host          ||= 'collector.instrumentalapp.com'
  requested_secure = options[:secure] == true
  desired_secure   = options[:secure].nil? ? allows_secure? : !!options[:secure]
  if !allows_secure? && desired_secure
    logger.warn "Cannot connect to Instrumental via encrypted transport, SSL not available"
    if requested_secure
      options[:enabled] = false
      logger.error "You requested secure protocol to connect to Instrumental, but it is not available on this system (OpenSSL is not defined). Connecting to Instrumental has been disabled."
    end
    desired_secure = false
  end
  @secure          = desired_secure
  @verify_cert     = options[:verify_cert].nil? ? true : !!options[:verify_cert]
  default_port     = @secure ? 8001 : 8000
  @port            = (@port || default_port).to_i
  @enabled         = options.has_key?(:enabled) ? !!options[:enabled] : true
  @synchronous     = !!options[:synchronous]

  if options.has_key?(:frequency)
    self.frequency = options[:frequency]
  else
    self.frequency = DEFAULT_FREQUENCY
  end

  @metrician       = options[:metrician].nil? ? true : !!options[:metrician]
  @pid             = Process.pid
  @allow_reconnect = true
  @dns_resolutions = 0
  @last_connect_at = 0

  @start_worker_mutex = Mutex.new
  @aggregator_queue = Queue.new
  @sender_queue = Queue.new


  setup_cleanup_at_exit if @enabled

  if @metrician
    Metrician.activate(self)
  end
end

Instance Attribute Details

#aggregator_queue ⇒ Object

Returns the value of attribute aggregator_queue.



30
31
32
 
# File 'lib/instrumental/agent.rb', line 30

def aggregator_queue
  @aggregator_queue
end

#connection ⇒ Object (readonly)

Returns the value of attribute connection.



31
32
33
 
# File 'lib/instrumental/agent.rb', line 31

def connection
  @connection
end

#dns_resolutions ⇒ Object

Returns the value of attribute dns_resolutions.



30
31
32
 
# File 'lib/instrumental/agent.rb', line 30

def dns_resolutions
  @dns_resolutions
end

#enabled ⇒ Object (readonly)

Returns the value of attribute enabled.



31
32
33
 
# File 'lib/instrumental/agent.rb', line 31

def enabled
  @enabled
end

#frequency ⇒ Object

Returns the value of attribute frequency.



30
31
32
 
# File 'lib/instrumental/agent.rb', line 30

def frequency
  @frequency
end

#host ⇒ Object

Returns the value of attribute host.



30
31
32
 
# File 'lib/instrumental/agent.rb', line 30

def host
  @host
end

#last_connect_at ⇒ Object

Returns the value of attribute last_connect_at.



30
31
32
 
# File 'lib/instrumental/agent.rb', line 30

def last_connect_at
  @last_connect_at
end

#port ⇒ Object

Returns the value of attribute port.



30
31
32
 
# File 'lib/instrumental/agent.rb', line 30

def port
  @port
end

#secure ⇒ Object (readonly)

Returns the value of attribute secure.



31
32
33
 
# File 'lib/instrumental/agent.rb', line 31

def secure
  @secure
end

#sender_queue ⇒ Object

Returns the value of attribute sender_queue.



30
31
32
 
# File 'lib/instrumental/agent.rb', line 30

def sender_queue
  @sender_queue
end

#synchronous ⇒ Object

Returns the value of attribute synchronous.



30
31
32
 
# File 'lib/instrumental/agent.rb', line 30

def synchronous
  @synchronous
end

Class Method Details

.logger ⇒ Object 



37
38
39
40
41
42
43
 
# File 'lib/instrumental/agent.rb', line 37

def self.logger
  if !@logger
    @logger = Logger.new(STDERR)
    @logger.level = Logger::WARN
  end
  @logger
end

.logger=(l) ⇒ Object 



33
34
35
 
# File 'lib/instrumental/agent.rb', line 33

def self.logger=(l)
  @logger = l
end

Instance Method Details

#cleanup ⇒ Object

Called when a process is exiting to give it some extra time to push events to the service. An at_exit handler is automatically registered for this method, but can be called manually in cases where at_exit is bypassed like Resque workers.



263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
 
# File 'lib/instrumental/agent.rb', line 263

def cleanup
  if running?
    logger.info "Cleaning up agent, aggregator_size: #{@aggregator_queue.size}, thread_running: #{@aggregator_thread.alive?}"
    logger.info "Cleaning up agent, queue size: #{@sender_queue.size}, thread running: #{@sender_thread.alive?}"
    @allow_reconnect = false
    begin
      with_timeout(EXIT_FLUSH_TIMEOUT) do
        @aggregator_queue << ['exit']
        @aggregator_thread.join
        @sender_queue << ['exit']
        @sender_thread.join
      end
    rescue Timeout::Error
      total_size = @sender_queue&.size.to_i +
                   @aggregator_queue&.size.to_i +
                   @event_aggregator&.size.to_i

      if total_size > 0
        logger.error "Timed out working agent thread on exit, dropping #{total_size} metrics"
      else
        logger.error "Timed out Instrumental Agent, exiting"
      end
    end
  end
end

#connected? ⇒ Boolean

Returns:

  • (Boolean) 


204
205
206
 
# File 'lib/instrumental/agent.rb', line 204

def connected?
  @socket && !@socket.closed?
end

#enabled? ⇒ Boolean

Returns:

  • (Boolean) 


200
201
202
 
# File 'lib/instrumental/agent.rb', line 200

def enabled?
  @enabled
end

#flush(allow_reconnect = false) ⇒ Object

Synchronously flush all pending metrics out to the server By default will not try to reconnect to the server if a connection failure happens during the flush, though you may optionally override this behavior by passing true.

agent.flush



193
194
195
196
197
198
 
# File 'lib/instrumental/agent.rb', line 193

def flush(allow_reconnect = false)
  queue_message('flush', {
    :synchronous => true,
    :allow_reconnect => allow_reconnect
  }) if running?
end

#gauge(metric, value, time = Time.now, count = 1) ⇒ Object

Store a gauge for a metric, optionally at a specific time.

agent.gauge('load', 1.23)



110
111
112
113
114
115
116
117
118
119
120
121
122
 
# File 'lib/instrumental/agent.rb', line 110

def gauge(metric, value, time = Time.now, count = 1)
  if valid?(metric, value, time, count) &&
     send_command(Instrumental::Command.new("gauge".freeze, metric, value, time, count))
    # tempted to "gauge" this to a symbol? Don't. Frozen strings are very fast,
    # and later we're going to to_s every one of these anyway.
    value
  else
    nil
  end
rescue Exception => e
  report_exception(e)
  nil
end

#increment(metric, value = 1, time = Time.now, count = 1) ⇒ Object

Increment a metric, optionally more than one or at a specific time.

agent.increment('users')



160
161
162
163
164
165
166
167
168
169
170
 
# File 'lib/instrumental/agent.rb', line 160

def increment(metric, value = 1, time = Time.now, count = 1)
  if valid?(metric, value, time, count) &&
     send_command(Instrumental::Command.new("increment".freeze, metric, value, time, count))
    value
  else
    nil
  end
rescue Exception => e
  report_exception(e)
  nil
end

#logger ⇒ Object 



212
213
214
 
# File 'lib/instrumental/agent.rb', line 212

def logger
  @logger || self.class.logger
end

#logger=(logger) ⇒ Object 



208
209
210
 
# File 'lib/instrumental/agent.rb', line 208

def logger=(logger)
  @logger = logger
end

#notice(note, time = Time.now, duration = 0) ⇒ Object

Send a notice to the server (deploys, downtime, etc.)

agent.notice('A notice')



175
176
177
178
179
180
181
182
183
184
185
 
# File 'lib/instrumental/agent.rb', line 175

def notice(note, time = Time.now, duration = 0)
  if valid_note?(note)
    send_command(Instrumental::Notice.new(note, time, duration))
    note
  else
    nil
  end
rescue Exception => e
  report_exception(e)
  nil
end

#stop ⇒ Object

Stopping the agent will immediately stop all communication to Instrumental. If you call this and submit another metric, the agent will start again.

Calling stop will cause all metrics waiting to be sent to be discarded. Don't call it unless you are expecting this behavior.

agent.stop



241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
 
# File 'lib/instrumental/agent.rb', line 241

def stop
  disconnect
  if @sender_thread
    @sender_thread.kill
    @sender_thread = nil
  end
  if @aggregator_thread
    @aggregator_thread.kill
    @aggregator_thread = nil
  end
  if @sender_queue
    @sender_queue.clear
  end
  if @aggregator_queue
    @aggregator_queue.clear
  end
end

#time(metric, multiplier = 1) ⇒ Object

Store the duration of a block in a metric. multiplier can be used to scale the duration to desired unit or change the duration in some meaningful way.

agent.time('response_time') do

potentially slow stuff

end

agent.time('response_time_in_ms', 1000) do

potentially slow stuff

end

ids = [1, 2, 3] agent.time('find_time_per_post', 1 / ids.size.to_f) do Post.find(ids) end



140
141
142
143
144
145
146
147
148
149
150
 
# File 'lib/instrumental/agent.rb', line 140

def time(metric, multiplier = 1)
  start = Time.now
  begin
    result = yield
  ensure
    finish = Time.now
    duration = finish - start
    gauge(metric, duration * multiplier, start)
  end
  result
end

#time_ms(metric, &block) ⇒ Object

Calls time and changes durations into milliseconds.



153
154
155
 
# File 'lib/instrumental/agent.rb', line 153

def time_ms(metric, &block)
  time(metric, 1000, &block)
end