From 876a48a1abaa38b85495f55bb59a1b08679f931b Mon Sep 17 00:00:00 2001 From: Brusk Awat Date: Mon, 5 Aug 2024 00:54:02 +0300 Subject: [PATCH 1/4] Improves error handling and shut down. Moves code for reusability --- lib/rabbit_carrots.rb | 1 + lib/rabbit_carrots/core.rb | 113 ++++++++++++++++++++++++++++++ lib/rabbit_carrots/tasks/rmq.rake | 74 ++----------------- 3 files changed, 119 insertions(+), 69 deletions(-) create mode 100644 lib/rabbit_carrots/core.rb diff --git a/lib/rabbit_carrots.rb b/lib/rabbit_carrots.rb index 953d5dd..2f0a2c4 100644 --- a/lib/rabbit_carrots.rb +++ b/lib/rabbit_carrots.rb @@ -2,6 +2,7 @@ require_relative 'rabbit_carrots/version' require 'rabbit_carrots/connection' +require 'rabbit_carrots/core' require 'rabbit_carrots/configuration' require 'rabbit_carrots/railtie' if defined?(Rails) diff --git a/lib/rabbit_carrots/core.rb b/lib/rabbit_carrots/core.rb new file mode 100644 index 0000000..7846d45 --- /dev/null +++ b/lib/rabbit_carrots/core.rb @@ -0,0 +1,113 @@ +module RabbitCarrots + class Core + attr_reader :logger + + DatabaseAgonsticNotNullViolation = defined?(ActiveRecord) ? ActiveRecord::NotNullViolation : RabbitCarrots::EventHandlers::Errors::PlaceholderError + DatabaseAgonsticConnectionNotEstablished = defined?(ActiveRecord) ? ActiveRecord::ConnectionNotEstablished : Mongo::Error::SocketError + DatabaseAgnosticRecordInvalid = defined?(ActiveRecord) ? ActiveRecord::RecordInvalid : Mongoid::Errors::Validations + + def initialize(logger: nil) + @logger = logger || Logger.new(Rails.env.production? ? '/proc/self/fd/1' : $stdout) + @threads = [] + @running = true + @shutdown_requested = false + end + + def start(kill_to_restart_on_standard_error: false) + channels = RabbitCarrots.configuration.routing_key_mappings.map do |mapping| + { **mapping, handler: mapping[:handler].constantize } + end + + channels.each do |channel| + handler_class = channel[:handler] + raise "#{handler_class.name} must respond to `handle!`" unless handler_class.respond_to?(:handle!) + + @threads << Thread.new do + run_task( + queue_name: channel[:queue], + handler_class:, + routing_keys: channel[:routing_keys], + queue_arguments: channel[:arguments], + kill_to_restart_on_standard_error: + ) + end + end + + Signal.trap('INT') { request_shutdown } + Signal.trap('TERM') { request_shutdown } + + while @running + if @shutdown_requested + request_shutdown + sleep 1 + break + end + sleep 1 + end + + @threads.each(&:join) + rescue StandardError => e + logger.error "Error starting Rabbit Carrots: #{e.message}" + end + + def request_shutdown + # Workaround to a known issue with Signal Traps and logs + Thread.start do + logger.info 'Shutting down Rabbit Carrots service...' + end + @shutdown_requested = true + @threads.each(&:kill) + stop + end + + def stop + # Workaround to a known issue with Signal Traps and logs + Thread.start do + logger.info 'Stoppig the Rabbit Carrots service...' + end + @running = false + end + + def run_task(queue_name:, handler_class:, routing_keys:, queue_arguments: {}, kill_to_restart_on_standard_error: false) + RabbitCarrots::Connection.instance.channel.with do |channel| + exchange = channel.topic(RabbitCarrots.configuration.rabbitmq_exchange_name, durable: true) + + logger.info "Listening on QUEUE: #{queue_name} for ROUTING KEYS: #{routing_keys}" + queue = channel.queue(queue_name, durable: true, arguments: queue_arguments) + + routing_keys.map(&:strip).each { |k| queue.bind(exchange, routing_key: k) } + + queue.subscribe(block: false, manual_ack: true, prefetch: 10) do |delivery_info, properties, payload| + break if @shutdown_requested + + logger.info "Received from queue: #{queue_name}, Routing Keys: #{routing_keys}" + handler_class.handle!(channel, delivery_info, properties, payload) + channel.ack(delivery_info.delivery_tag, false) + rescue RabbitCarrots::EventHandlers::Errors::NackMessage, JSON::ParserError => _e + logger.info "Nacked message: #{payload}" + channel.nack(delivery_info.delivery_tag, false, false) + rescue RabbitCarrots::EventHandlers::Errors::NackAndRequeueMessage => _e + logger.info "Nacked and Requeued message: #{payload}" + channel.nack(delivery_info.delivery_tag, false, true) + rescue DatabaseAgonsticNotNullViolation, DatabaseAgnosticRecordInvalid => e + logger.info "Null constraint or Invalid violation: #{payload}. Error: #{e.message}" + channel.ack(delivery_info.delivery_tag, false) + rescue DatabaseAgonsticConnectionNotEstablished => e + logger.info "Error connection not established to the database: #{payload}. Error: #{e.message}" + sleep 3 + channel.nack(delivery_info.delivery_tag, false, true) + rescue StandardError => e + logger.info "Error handling message: #{payload}. Error: #{e.message}" + sleep 3 + channel.nack(delivery_info.delivery_tag, false, true) + Process.kill('SIGTERM', Process.pid) if kill_to_restart_on_standard_error + end + + logger.info "Ending task for queue: #{queue_name}" + end + rescue StandardError => e + logger.error "Bunny session error: #{e.message}" + request_shutdown + end + end +end diff --git a/lib/rabbit_carrots/tasks/rmq.rake b/lib/rabbit_carrots/tasks/rmq.rake index 541f45b..33efb5f 100644 --- a/lib/rabbit_carrots/tasks/rmq.rake +++ b/lib/rabbit_carrots/tasks/rmq.rake @@ -1,77 +1,13 @@ -require 'bunny' - namespace :rabbit_carrots do - desc 'Listener for Queue' + desc 'Rake task for standalone RabbitCarrots mode' task eat: :environment do Rails.application.eager_load! - # rubocop:disable Lint/ConstantDefinitionInBlock - DatabaseAgonsticNotNullViolation = defined?(ActiveRecord) ? ActiveRecord::NotNullViolation : RabbitCarrots::EventHandlers::Errors::PlaceholderError - DatabaseAgonsticConnectionNotEstablished = defined?(ActiveRecord) ? ActiveRecord::ConnectionNotEstablished : Mongo::Error::SocketError - DatabaseAgnosticRecordInvalid = defined?(ActiveRecord) ? ActiveRecord::RecordInvalid : Mongoid::Errors::Validations - # rubocop:enable Lint/ConstantDefinitionInBlock + logger = Logger.new(Rails.env.production? ? '/proc/self/fd/1' : $stdout) + logger.level = Logger::INFO - channels = RabbitCarrots.configuration.routing_key_mappings.map do |mapping| - # This will be supplied in initializer. At that time, the Handler will not be available to be loaded and will throw Uninitialized Constant - { **mapping, handler: mapping[:handler].constantize } - end + core_service = RabbitCarrots::Core.new(logger:) - Rails.logger = Logger.new(Rails.env.production? ? '/proc/self/fd/1' : $stdout) - - # Run RMQ Subscriber for each channel - channels.each do |channel| - handler_class = channel[:handler] - - raise "#{handler_class.name} must respond to `handle!`" unless handler_class.respond_to?(:handle!) - - run_task(queue_name: channel[:queue], handler_class:, routing_keys: channel[:routing_keys], queue_arguments: channel[:arguments]) - end - - # Infinite loop to keep the process running - loop do - sleep 1 - end - end -end - -def run_task(queue_name:, handler_class:, routing_keys:, queue_arguments: {}) - RabbitCarrots::Connection.instance.channel.with do |channel| - exchange = channel.topic(RabbitCarrots.configuration.rabbitmq_exchange_name, durable: true) - - Rails.logger.info "Listening on QUEUE: #{queue_name} for ROUTING KEYS: #{routing_keys}" - queue = channel.queue(queue_name, durable: true, arguments: queue_arguments) - - routing_keys.map(&:strip).each { |k| queue.bind(exchange, routing_key: k) } - - queue.subscribe(block: false, manual_ack: true, prefetch: 10) do |delivery_info, properties, payload| - Rails.logger.info "Received from queue: #{queue_name}, Routing Keys: #{routing_keys}" - handler_class.handle!(channel, delivery_info, properties, payload) - channel.ack(delivery_info.delivery_tag, false) - rescue RabbitCarrots::EventHandlers::Errors::NackMessage, JSON::ParserError => _e - Rails.logger.info "Nacked message: #{payload}" - channel.nack(delivery_info.delivery_tag, false, false) - rescue RabbitCarrots::EventHandlers::Errors::NackAndRequeueMessage => _e - Rails.logger.info "Nacked and Requeued message: #{payload}" - channel.nack(delivery_info.delivery_tag, false, true) - rescue DatabaseAgonsticNotNullViolation, DatabaseAgnosticRecordInvalid => e - # on null constraint violation, we want to ack the message - Rails.logger.error "Null constraint or Invalid violation: #{payload}. Error: #{e.message}" - channel.ack(delivery_info.delivery_tag, false) - rescue DatabaseAgonsticConnectionNotEstablished => e - # on connection not established, we want to requeue the message and sleep for 3 seconds - Rails.logger.error "Error connection not established to the database: #{payload}. Error: #{e.message}" - # delay for 3 seconds before requeuing - sleep 3 - channel.nack(delivery_info.delivery_tag, false, true) - rescue StandardError => e - Rails.logger.error "Error handling message: #{payload}. Error: #{e.message}" - # requeue the message then kill the container - sleep 3 - channel.nack(delivery_info.delivery_tag, false, true) - # kill the container with sigterm - Process.kill('SIGTERM', Process.pid) - end - - Rails.logger.info 'RUN TASK ENDED' + core_service.start(kill_to_restart_on_standard_error: true) end end From 28b620b7b485633e7eef8ad02c6214ada96165d0 Mon Sep 17 00:00:00 2001 From: Brusk Awat Date: Mon, 5 Aug 2024 00:54:19 +0300 Subject: [PATCH 2/4] Adds automatic recovery and exit strategy --- lib/rabbit_carrots/configuration.rb | 11 ++++++++++- lib/rabbit_carrots/connection.rb | 6 +++++- 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/lib/rabbit_carrots/configuration.rb b/lib/rabbit_carrots/configuration.rb index 410561a..f1e4d92 100644 --- a/lib/rabbit_carrots/configuration.rb +++ b/lib/rabbit_carrots/configuration.rb @@ -9,6 +9,15 @@ module RabbitCarrots end class Configuration - attr_accessor :rabbitmq_host, :rabbitmq_port, :rabbitmq_user, :rabbitmq_password, :rabbitmq_vhost, :routing_key_mappings, :rabbitmq_exchange_name + attr_accessor :rabbitmq_host, + :rabbitmq_port, + :rabbitmq_user, + :rabbitmq_password, + :rabbitmq_vhost, + :routing_key_mappings, + :rabbitmq_exchange_name, + :automatically_recover, + :network_recovery_interval, + :recovery_attempts end end diff --git a/lib/rabbit_carrots/connection.rb b/lib/rabbit_carrots/connection.rb index 8a1b246..81dfc95 100644 --- a/lib/rabbit_carrots/connection.rb +++ b/lib/rabbit_carrots/connection.rb @@ -11,7 +11,11 @@ module RabbitCarrots port: RabbitCarrots.configuration.rabbitmq_port, user: RabbitCarrots.configuration.rabbitmq_user, password: RabbitCarrots.configuration.rabbitmq_password, - vhost: RabbitCarrots.configuration.rabbitmq_vhost + vhost: RabbitCarrots.configuration.rabbitmq_vhost, + automatically_recover: RabbitCarrots.configuration.automatically_recover || true, + network_recovery_interval: RabbitCarrots.configuration.network_recovery_interval || 5, + recovery_attempts: RabbitCarrots.configuration.recovery_attempts || 5, + recovery_attempts_exhausted: -> { Process.kill('TERM', Process.pid) } ) @connection.start From a51eed8101e38b64fbccd93be3ca360d7dac530a Mon Sep 17 00:00:00 2001 From: Brusk Awat Date: Mon, 5 Aug 2024 01:03:10 +0300 Subject: [PATCH 3/4] Adds support for puma --- lib/puma/plugin/rabbit_carrots.rb | 84 +++++++++++++++++++++++++++++++ lib/rabbit_carrots/core.rb | 20 ++++---- 2 files changed, 94 insertions(+), 10 deletions(-) create mode 100644 lib/puma/plugin/rabbit_carrots.rb diff --git a/lib/puma/plugin/rabbit_carrots.rb b/lib/puma/plugin/rabbit_carrots.rb new file mode 100644 index 0000000..d49ce86 --- /dev/null +++ b/lib/puma/plugin/rabbit_carrots.rb @@ -0,0 +1,84 @@ +# rabbit_carrots.rb + +require 'puma/plugin' + +Puma::Plugin.create do + attr_reader :puma_pid, :rabbit_carrots_pid, :log_writer, :core_service + + def start(launcher) + @log_writer = launcher.log_writer + @puma_pid = $$ + + @core_service = RabbitCarrots::Core.new(logger: log_writer) + + in_background do + monitor_rabbit_carrots + end + + launcher.events.on_booted do + @rabbit_carrots_pid = fork do + Thread.new { monitor_puma } + start_rabbit_carrots_consumer + end + end + + launcher.events.on_stopped { stop_rabbit_carrots } + launcher.events.on_restart { stop_rabbit_carrots } + end + + private + + def start_rabbit_carrots_consumer + core_service.start(kill_to_restart_on_standard_error: true) + rescue StandardError => e + Rails.logger.error "Error starting Rabbit Carrots: #{e.message}" + end + + def stop_rabbit_carrots + return unless rabbit_carrots_pid + + log 'Stopping Rabbit Carrots...' + core_service.request_shutdown + Process.kill('TERM', rabbit_carrots_pid) + Process.wait(rabbit_carrots_pid) + rescue Errno::ECHILD, Errno::ESRCH + end + + def monitor_puma + monitor(:puma_dead?, 'Detected Puma has gone away, stopping Rabbit Carrots...') + end + + def monitor_rabbit_carrots + monitor(:rabbit_carrots_dead?, 'Rabbits Carrot is dead, stopping Puma...') + end + + def monitor(process_dead, message) + loop do + if send(process_dead) + log message + Process.kill('TERM', $$) + break + end + sleep 2 + end + end + + def rabbit_carrots_dead? + Process.waitpid(rabbit_carrots_pid, Process::WNOHANG) if rabbit_carrots_started? + false + rescue Errno::ECHILD, Errno::ESRCH + true + end + + def rabbit_carrots_started? + rabbit_carrots_pid.present? + end + + def puma_dead? + Process.ppid != puma_pid + end + + def log(...) + log_writer.log(...) + end +end diff --git a/lib/rabbit_carrots/core.rb b/lib/rabbit_carrots/core.rb index 7846d45..dba740d 100644 --- a/lib/rabbit_carrots/core.rb +++ b/lib/rabbit_carrots/core.rb @@ -53,7 +53,7 @@ module RabbitCarrots def request_shutdown # Workaround to a known issue with Signal Traps and logs Thread.start do - logger.info 'Shutting down Rabbit Carrots service...' + logger.log 'Shutting down Rabbit Carrots service...' end @shutdown_requested = true @threads.each(&:kill) @@ -63,7 +63,7 @@ module RabbitCarrots def stop # Workaround to a known issue with Signal Traps and logs Thread.start do - logger.info 'Stoppig the Rabbit Carrots service...' + logger.log 'Stoppig the Rabbit Carrots service...' end @running = false end @@ -72,7 +72,7 @@ module RabbitCarrots RabbitCarrots::Connection.instance.channel.with do |channel| exchange = channel.topic(RabbitCarrots.configuration.rabbitmq_exchange_name, durable: true) - logger.info "Listening on QUEUE: #{queue_name} for ROUTING KEYS: #{routing_keys}" + logger.log "Listening on QUEUE: #{queue_name} for ROUTING KEYS: #{routing_keys}" queue = channel.queue(queue_name, durable: true, arguments: queue_arguments) routing_keys.map(&:strip).each { |k| queue.bind(exchange, routing_key: k) } @@ -80,30 +80,30 @@ module RabbitCarrots queue.subscribe(block: false, manual_ack: true, prefetch: 10) do |delivery_info, properties, payload| break if @shutdown_requested - logger.info "Received from queue: #{queue_name}, Routing Keys: #{routing_keys}" + logger.log "Received from queue: #{queue_name}, Routing Keys: #{routing_keys}" handler_class.handle!(channel, delivery_info, properties, payload) channel.ack(delivery_info.delivery_tag, false) rescue RabbitCarrots::EventHandlers::Errors::NackMessage, JSON::ParserError => _e - logger.info "Nacked message: #{payload}" + logger.log "Nacked message: #{payload}" channel.nack(delivery_info.delivery_tag, false, false) rescue RabbitCarrots::EventHandlers::Errors::NackAndRequeueMessage => _e - logger.info "Nacked and Requeued message: #{payload}" + logger.log "Nacked and Requeued message: #{payload}" channel.nack(delivery_info.delivery_tag, false, true) rescue DatabaseAgonsticNotNullViolation, DatabaseAgnosticRecordInvalid => e - logger.info "Null constraint or Invalid violation: #{payload}. Error: #{e.message}" + logger.log "Null constraint or Invalid violation: #{payload}. Error: #{e.message}" channel.ack(delivery_info.delivery_tag, false) rescue DatabaseAgonsticConnectionNotEstablished => e - logger.info "Error connection not established to the database: #{payload}. Error: #{e.message}" + logger.log "Error connection not established to the database: #{payload}. Error: #{e.message}" sleep 3 channel.nack(delivery_info.delivery_tag, false, true) rescue StandardError => e - logger.info "Error handling message: #{payload}. Error: #{e.message}" + logger.log "Error handling message: #{payload}. Error: #{e.message}" sleep 3 channel.nack(delivery_info.delivery_tag, false, true) Process.kill('SIGTERM', Process.pid) if kill_to_restart_on_standard_error end - logger.info "Ending task for queue: #{queue_name}" + logger.log "Ending task for queue: #{queue_name}" end rescue StandardError => e logger.error "Bunny session error: #{e.message}" From ef60e8342db7a2953f71e40fd872366ff70bd6cb Mon Sep 17 00:00:00 2001 From: Brusk Awat Date: Mon, 5 Aug 2024 01:07:15 +0300 Subject: [PATCH 4/4] Updates README --- README.md | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 9c685f6..cd683f2 100644 --- a/README.md +++ b/README.md @@ -33,6 +33,9 @@ RabbitCarrots.configure do |c| c.rabbitmq_password = ENV.fetch('RABBITMQ__PASSWORD', nil) c.rabbitmq_vhost = ENV.fetch('RABBITMQ__VHOST', nil) c.rabbitmq_exchange_name = ENV.fetch('RABBITMQ__EXCHANGE_NAME', nil) + c.automatically_recover = true + c.network_recovery_interval = 5 + c.recovery_attempts = 5 c.routing_key_mappings = [ { routing_keys: ['RK1', 'RK2'], queue: 'QUEUE_NAME', handler: 'CLASS HANDLER IN STRING' }, { routing_keys: ['RK1', 'RK2'], queue: 'QUEUE_NAME', handler: 'CLASS HANDLER IN STRING' } @@ -41,8 +44,6 @@ end ``` - - Note that handler is a class that must implement a method named ```handle!``` that takes 4 parameters as follow: ```ruby @@ -53,8 +54,6 @@ class DummyEventHandler end ``` - - Inside the handle message, you can NACK the message without re-queuing by raising ```RabbitCarrots::EventHandlers::Errors::NackMessage``` exception. To NACK and re-queue, raise ```RabbitCarrots::EventHandlers::Errors::NackAndRequeueMessage``` exception. @@ -65,8 +64,16 @@ Note: Any other unrescued exception raised inside ```handle!``` the that is a su ### Running -Then run ```bundle exec rake rabbit_carrots:eat```. +For better scalability and improved performance, you can run rabbit_carrots in standalone mode by invoking the following command: +```bundle exec rake rabbit_carrots:eat```. +#### Puma + +For small and medium sized projects, you can delegate the management of the rabbit_carrots to the Puma web server. To achieve that, add the following line to your puma.rb + +```plugin :rabbit_carrots``` + +This will make sure that Puma will manage rabbit carrots as a background service and will gracefully terminate if rabbit_carrots eventually loses connection after multiple automatic recovery. ## Development After checking out the repo, run `bin/setup` to install dependencies. Then, run `rake spec` to run the tests. You can also run `bin/console` for an interactive prompt that will allow you to experiment.