Class: Karafka::Instrumentation::StdoutListener
- Inherits:
-
Object
- Object
- Karafka::Instrumentation::StdoutListener
- Defined in:
- lib/karafka/instrumentation/stdout_listener.rb
Overview
Default listener that hooks up to our instrumentation and uses its events for logging It can be removed/replaced or anything without any harm to the Karafka app flow
Constant Summary collapse
- USED_LOG_LEVELS =
Log levels that we use in this particular listener
%i[ debug info error fatal ].freeze
Instance Method Summary collapse
-
#on_app_initializing(_event) ⇒ Object
Logs info that we’re initializing Karafka app.
-
#on_app_running(_event) ⇒ Object
Logs info that we’re running Karafka app.
-
#on_app_stopping(_event) ⇒ Object
Logs info that we’re going to stop the Karafka server.
-
#on_app_stopping_error(_event) ⇒ Object
Logs an error that Karafka was unable to stop the server gracefully and it had to do a forced exit.
-
#on_backends_inline_process(event) ⇒ Object
Logs info about processing of a certain dataset with an inline backend.
-
#on_connection_batch_delegator_call(event) ⇒ Object
Logs details about incoming batches and with which consumer we will consume them.
-
#on_connection_client_fetch_loop_error(event) ⇒ Object
Logs errors that are related to the connection itself.
-
#on_connection_listener_fetch_loop_error(event) ⇒ Object
Logs errors that occurred in a listener fetch loop.
-
#on_connection_message_delegator_call(event) ⇒ Object
Logs details about incoming message and with which consumer we will consume it.
-
#on_consumers_responders_respond_with(event) ⇒ Object
Logs info about responder usage withing a controller flow.
-
#on_fetcher_call_error(event) ⇒ Object
Logs info about crashed fetcher.
-
#on_params_params_deserialize(event) ⇒ Object
Logs details about each received message value deserialization.
-
#on_params_params_deserialize_error(event) ⇒ Object
Logs unsuccessful deserialization attempts of incoming data.
-
#on_process_notice_signal(event) ⇒ Object
Logs info about system signals that Karafka received.
Instance Method Details
#on_app_initializing(_event) ⇒ Object
Logs info that we’re initializing Karafka app
106 107 108 |
# File 'lib/karafka/instrumentation/stdout_listener.rb', line 106 def on_app_initializing(_event) info "Initializing Karafka server #{::Process.pid}" end |
#on_app_running(_event) ⇒ Object
Logs info that we’re running Karafka app
112 113 114 |
# File 'lib/karafka/instrumentation/stdout_listener.rb', line 112 def on_app_running(_event) info "Running Karafka server #{::Process.pid}" end |
#on_app_stopping(_event) ⇒ Object
Logs info that we’re going to stop the Karafka server
118 119 120 121 |
# File 'lib/karafka/instrumentation/stdout_listener.rb', line 118 def on_app_stopping(_event) # We use a separate thread as logging can't be called from trap context Thread.new { info "Stopping Karafka server #{::Process.pid}" } end |
#on_app_stopping_error(_event) ⇒ Object
Logs an error that Karafka was unable to stop the server gracefully and it had to do a
forced exit
126 127 128 129 |
# File 'lib/karafka/instrumentation/stdout_listener.rb', line 126 def on_app_stopping_error(_event) # We use a separate thread as logging can't be called from trap context Thread.new { error "Forceful Karafka server #{::Process.pid} stop" } end |
#on_backends_inline_process(event) ⇒ Object
Logs info about processing of a certain dataset with an inline backend
82 83 84 85 86 87 |
# File 'lib/karafka/instrumentation/stdout_listener.rb', line 82 def on_backends_inline_process(event) count = event[:caller].send(:params_batch).to_a.size topic = event[:caller].topic.name time = event[:time] info "Inline processing of topic #{topic} with #{count} messages took #{time} ms" end |
#on_connection_batch_delegator_call(event) ⇒ Object
Logs details about incoming batches and with which consumer we will consume them
18 19 20 21 22 23 24 25 26 27 28 29 |
# File 'lib/karafka/instrumentation/stdout_listener.rb', line 18 def on_connection_batch_delegator_call(event) consumer = event[:consumer] topic = consumer.topic.name = event[:kafka_batch]. info( <<~MSG.chomp.tr("\n", ' ') #{.count} messages on #{topic} topic delegated to #{consumer.class} MSG ) end |
#on_connection_client_fetch_loop_error(event) ⇒ Object
Karafka will attempt to reconnect, so an error not a fatal
Logs errors that are related to the connection itself
68 69 70 |
# File 'lib/karafka/instrumentation/stdout_listener.rb', line 68 def on_connection_client_fetch_loop_error(event) error "Client fetch loop error: #{event[:error]}" end |
#on_connection_listener_fetch_loop_error(event) ⇒ Object
It’s an error as we can recover from it not a fatal
Logs errors that occurred in a listener fetch loop
61 62 63 |
# File 'lib/karafka/instrumentation/stdout_listener.rb', line 61 def on_connection_listener_fetch_loop_error(event) error "Listener fetch loop error: #{event[:error]}" end |
#on_connection_message_delegator_call(event) ⇒ Object
Logs details about incoming message and with which consumer we will consume it
33 34 35 36 37 |
# File 'lib/karafka/instrumentation/stdout_listener.rb', line 33 def (event) consumer = event[:consumer] topic = consumer.topic.name info "1 message on #{topic} topic delegated to #{consumer.class}" end |
#on_consumers_responders_respond_with(event) ⇒ Object
Logs info about responder usage withing a controller flow
97 98 99 100 101 102 |
# File 'lib/karafka/instrumentation/stdout_listener.rb', line 97 def on_consumers_responders_respond_with(event) calling = event[:caller] responder = calling.topic.responder data = event[:data] info "Responded from #{calling.class} using #{responder} with following data #{data}" end |
#on_fetcher_call_error(event) ⇒ Object
If this happens, Karafka will shutdown as it means a critical error in one of the threads
Logs info about crashed fetcher
76 77 78 |
# File 'lib/karafka/instrumentation/stdout_listener.rb', line 76 def on_fetcher_call_error(event) fatal "Fetcher crash due to an error: #{event[:error]}" end |
#on_params_params_deserialize(event) ⇒ Object
Logs details about each received message value deserialization
41 42 43 44 45 46 47 48 49 50 |
# File 'lib/karafka/instrumentation/stdout_listener.rb', line 41 def on_params_params_deserialize(event) # Keep in mind, that a caller here is a param object not a controller, # so it returns a topic as a string, not a routing topic debug( <<~MSG.chomp.tr("\n", ' ') Params deserialization for #{event[:caller].topic} topic successful in #{event[:time]} ms MSG ) end |
#on_params_params_deserialize_error(event) ⇒ Object
Logs unsuccessful deserialization attempts of incoming data
54 55 56 |
# File 'lib/karafka/instrumentation/stdout_listener.rb', line 54 def on_params_params_deserialize_error(event) error "Params deserialization error for #{event[:caller].topic} topic: #{event[:error]}" end |
#on_process_notice_signal(event) ⇒ Object
Logs info about system signals that Karafka received
91 92 93 |
# File 'lib/karafka/instrumentation/stdout_listener.rb', line 91 def on_process_notice_signal(event) info "Received #{event[:signal]} system signal" end |