require 'forwardable' require 'net/http' require 'honeybadger/logging' module Honeybadger # A concurrent queue to notify the backend. # @api private class EventsWorker extend Forwardable include Honeybadger::Logging::Helper # Sub-class thread so we have a named thread (useful for debugging in Thread.list). class Thread < ::Thread; end # Used to signal the worker to shutdown. SHUTDOWN = :__hb_worker_shutdown! FLUSH = :__hb_worker_flush! CHECK_TIMEOUT = :__hb_worker_check_timeout! # 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. BASE_THROTTLE = 1.05 # TODO: These could be configurable? 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 end def push(msg) return false unless start if queue.size >= config.max_queue_size warn { sprintf('Unable to send event; reached max queue size of %s.', queue.size) } return false end queue.push(msg) end def send_now(msg) handle_response(send_to_backend(msg)) end 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 # Blocks until queue is processed up to this point in time. def flush mutex.synchronize do if thread && thread.alive? queue.push(FLUSH) queue.push(marker) marker.wait(mutex) end end end 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 attr_reader :config, :queue, :pid, :mutex, :marker, :thread, :timeout_thread, :throttle, :throttle_interval, :start_at, :send_queue, :last_sent def_delegator :config, :backend def shutdown? mutex.synchronize { @shutdown } end def suspended? mutex.synchronize { start_at && Time.now.to_i < start_at } end def can_start? return false if shutdown? return false if suspended? true end def throttled? mutex.synchronize { throttle > 0 } end 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 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 def schedule_timeout_check loop do sleep(config.events_timeout / 1000.0) queue.push(CHECK_TIMEOUT) end end 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 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 def enqueue_msg(msg) mutex.synchronize do @send_queue << msg end end def send_batch send_now(mutex.synchronize { send_queue }) mutex.synchronize do @last_sent = Time.now send_queue.clear end end 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 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 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 def send_to_backend(msg) d { 'events_worker sending to backend' } response = backend.event(msg) response end def calc_throttle_interval ((BASE_THROTTLE ** throttle) - 1).round(3) end def inc_throttle mutex.synchronize do @throttle += 1 @throttle_interval = calc_throttle_interval throttle end end def dec_throttle mutex.synchronize do return nil if throttle == 0 @throttle -= 1 @throttle_interval = calc_throttle_interval throttle end end 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 info { sprintf('Success ⚡ Event sent code=%s throttle=%s interval=%s', response.code, throttle, throttle_interval) } else info { 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 # Release the marker. Important to perform during cleanup when shutting # down, otherwise it could end up waiting indefinitely. def release_marker signal_marker(marker) end def signal_marker(marker) mutex.synchronize do marker.signal end end end end