class Honeybadger::EventsWorker
A concurrent queue to notify the backend. @api private
Constants
- BASE_THROTTLE
-
The base number for the exponential backoff formula when calculating the throttle interval. ‘1.05 ** throttle` will reach an interval of 2 minutes after around 100 429 responses from the server.
- CHECK_TIMEOUT
- FLUSH
- SHUTDOWN
-
Used to signal the worker to shutdown.
Attributes
Public Class Methods
Source
# File lib/honeybadger/events_worker.rb, line 29 def initialize(config) @config = config @throttle = 0 @throttle_interval = 0 @mutex = Mutex.new @marker = ConditionVariable.new @queue = Queue.new @send_queue = Queue.new @shutdown = false @start_at = nil @pid = Process.pid @send_queue = [] @last_sent = nil @dropped_events = 0 end
TODO: These could be configurable?
Public Instance Methods
Source
# File lib/honeybadger/events_worker.rb, line 85 def flush mutex.synchronize do if thread && thread.alive? queue.push(FLUSH) queue.push(marker) marker.wait(mutex) end end end
Blocks until queue is processed up to this point in time.
Source
# File lib/honeybadger/events_worker.rb, line 45 def push(msg) return false unless start if queue.size >= config.events_max_queue_size @dropped_events += 1 return false end queue.push(msg) end
Source
# File lib/honeybadger/events_worker.rb, line 56 def send_now(msg) handle_response(send_to_backend(msg)) end
Source
# File lib/honeybadger/events_worker.rb, line 60 def shutdown(force = false) d { 'shutting down events worker' } mutex.synchronize do @shutdown = true end return true if force return true unless thread&.alive? if throttled? warn { sprintf('Unable to send %s event(s) to Honeybadger (currently throttled)', queue.size) } unless queue.empty? return true end info { sprintf('Waiting to send %s events(s) to Honeybadger', queue.size) } unless queue.empty? queue.push(FLUSH) queue.push(SHUTDOWN) !!thread.join ensure queue.clear kill! end
Source
# File lib/honeybadger/events_worker.rb, line 95 def start return false unless can_start? mutex.synchronize do @shutdown = false @start_at = nil return true if thread&.alive? @pid = Process.pid @thread = Thread.new { run } @timeout_thread = Thread.new { schedule_timeout_check } end true end
Private Instance Methods
Source
# File lib/honeybadger/events_worker.rb, line 262 def calc_throttle_interval ((BASE_THROTTLE ** throttle) - 1).round(3) end
Source
# File lib/honeybadger/events_worker.rb, line 127 def can_start? return false if shutdown? return false if suspended? true end
Source
# File lib/honeybadger/events_worker.rb, line 220 def check_and_send return if mutex.synchronize { send_queue.empty? } if mutex.synchronize { send_queue.length } >= config.events_batch_size send_batch end end
Source
# File lib/honeybadger/events_worker.rb, line 193 def check_timeout return if mutex.synchronize { send_queue.empty? } ms_since = (Time.now.to_f - last_sent.to_f) * 1000.0 if ms_since >= config.events_timeout send_batch end end
Source
# File lib/honeybadger/events_worker.rb, line 274 def dec_throttle mutex.synchronize do return nil if throttle == 0 @throttle -= 1 @throttle_interval = calc_throttle_interval throttle end end
Source
# File lib/honeybadger/events_worker.rb, line 201 def enqueue_msg(msg) mutex.synchronize do @send_queue << msg end end
Source
# File lib/honeybadger/events_worker.rb, line 227 def flush_send_queue return if mutex.synchronize { send_queue.empty? } send_batch rescue StandardError => e error { msg = "Error in worker thread class=%s message=%s\n\t%s" sprintf(msg, e.class, e.message.dump, Array(e.backtrace).join("\n\t")) } end
Source
# File lib/honeybadger/events_worker.rb, line 283 def handle_response(response) d { sprintf('events_worker response code=%s message=%s', response.code, response.message.to_s.dump) } case response.code when 429, 503 throttle = inc_throttle warn { sprintf('Event send failed: project is sending too many events. code=%s throttle=%s interval=%s', response.code, throttle, throttle_interval) } when 402 warn { sprintf('Event send failed: payment is required. code=%s', response.code) } suspend(3600) when 403 warn { sprintf('Event send failed: API key is invalid. code=%s', response.code) } suspend(3600) when 413 warn { sprintf('Event send failed: Payload is too large. code=%s', response.code) } when 201 if throttle = dec_throttle debug { sprintf('Success ⚡ Event sent code=%s throttle=%s interval=%s', response.code, throttle, throttle_interval) } else debug { sprintf('Success ⚡ Event sent code=%s', response.code) } end when :stubbed info { sprintf('Success ⚡ Development mode is enabled; This event will be sent after app is deployed.') } when :error warn { sprintf('Event send failed: an unknown error occurred. code=%s error=%s', response.code, response.message.to_s.dump) } else warn { sprintf('Event send failed: unknown response from server. code=%s', response.code) } end end
Source
# File lib/honeybadger/events_worker.rb, line 266 def inc_throttle mutex.synchronize do @throttle += 1 @throttle_interval = calc_throttle_interval throttle end end
Source
# File lib/honeybadger/events_worker.rb, line 137 def kill! d { 'killing worker thread' } if thread Thread.kill(thread) Thread.kill(timeout_thread) thread.join # Allow ensure blocks to execute. end true end
Source
# File lib/honeybadger/events_worker.rb, line 315 def release_marker signal_marker(marker) end
Release the marker. Important to perform during cleanup when shutting down, otherwise it could end up waiting indefinitely.
Source
# File lib/honeybadger/events_worker.rb, line 166 def run begin d { 'worker started' } mutex.synchronize do @last_sent = Time.now end loop do case msg = queue.pop when SHUTDOWN then break when CHECK_TIMEOUT then check_timeout when FLUSH then flush_send_queue when ConditionVariable then signal_marker(msg) else work(msg) end end ensure d { 'stopping worker' } end rescue Exception => e error { msg = "Error in worker thread (shutting down) class=%s message=%s\n\t%s" sprintf(msg, e.class, e.message.dump, Array(e.backtrace).join("\n\t")) } ensure release_marker end
Source
# File lib/honeybadger/events_worker.rb, line 159 def schedule_timeout_check loop do sleep(config.events_timeout / 1000.0) queue.push(CHECK_TIMEOUT) end end
Source
# File lib/honeybadger/events_worker.rb, line 207 def send_batch send_now(mutex.synchronize { send_queue }) mutex.synchronize do @last_sent = Time.now debug { sprintf('Sending %s events', send_queue.length) } send_queue.clear if @dropped_events > 0 warn { sprintf('Dropped %s messages due to exceeding max queue size of %s', @dropped_events, config.events_max_queue_size) } end @dropped_events = 0 end end
Source
# File lib/honeybadger/events_worker.rb, line 256 def send_to_backend(msg) d { 'events_worker sending to backend' } response = backend.event(msg) response end
Source
# File lib/honeybadger/events_worker.rb, line 119 def shutdown? mutex.synchronize { @shutdown } end
Source
# File lib/honeybadger/events_worker.rb, line 319 def signal_marker(marker) mutex.synchronize do marker.signal end end
Source
# File lib/honeybadger/events_worker.rb, line 149 def suspend(interval) mutex.synchronize do @start_at = Time.now.to_i + interval queue.clear end # Must be performed last since this may kill the current thread. kill! end
Source
# File lib/honeybadger/events_worker.rb, line 123 def suspended? mutex.synchronize { start_at && Time.now.to_i < start_at } end
Source
# File lib/honeybadger/events_worker.rb, line 133 def throttled? mutex.synchronize { throttle > 0 } end
Source
# File lib/honeybadger/events_worker.rb, line 237 def work(msg) enqueue_msg(msg) check_and_send if shutdown? && throttled? warn { sprintf('Unable to send %s events(s) to Honeybadger (currently throttled)', queue.size) } if queue.size > 1 kill! return end sleep(throttle_interval) rescue StandardError => e error { msg = "Error in worker thread class=%s message=%s\n\t%s" sprintf(msg, e.class, e.message.dump, Array(e.backtrace).join("\n\t")) } end