diff --git a/lib/fluent/root_agent.rb b/lib/fluent/root_agent.rb index 6c19e9af59..768003f51c 100644 --- a/lib/fluent/root_agent.rb +++ b/lib/fluent/root_agent.rb @@ -48,6 +48,11 @@ module Fluent class RootAgent < Agent ERROR_LABEL = "@ERROR".freeze # @ERROR is built-in error label + # Thread local key to detect an error which is raised while an error event is + # routed to @ERROR. @ERROR is the last resort route, so such an event must be + # dumped here instead of being routed back to @ERROR again and again. + ERROR_ROUTING_KEY = :fluentd_root_agent_routing_error_event + class SourceOnlyMode DISABLED = 0 NORMAL = 1 @@ -439,10 +444,13 @@ def find_label(label_name) def emit_error_event(tag, time, record, error) error_info = {error: error, location: (error.backtrace ? error.backtrace.first : nil), tag: tag, time: time} - if @error_collector + if @error_collector && !routing_error_event? # A record is not included in the logs because <@ERROR> handles it. This warn is for the notification log.warn "send an error event to @ERROR:", error_info - @error_collector.emit(tag, time, record) + route_error_event { @error_collector.emit(tag, time, record) } + elsif @error_collector + error_info[:record] = record + log.error "dump an error event because it failed in @ERROR. It is dropped to prevent an infinite loop:", error_info else error_info[:record] = record log.warn "dump an error event:", error_info @@ -451,9 +459,13 @@ def emit_error_event(tag, time, record, error) def handle_emits_error(tag, es, error) error_info = {error: error, location: (error.backtrace ? error.backtrace.first : nil), tag: tag} - if @error_collector + if @error_collector && !routing_error_event? log.warn "send an error event stream to @ERROR:", error_info - @error_collector.emit_stream(tag, es) + route_error_event { @error_collector.emit_stream(tag, es) } + elsif @error_collector + # Records are not dumped here because an error event stream can be large. + error_info[:record_count] = es.size if es + log.error "dump an error event stream because it failed in @ERROR. It is dropped to prevent an infinite loop:", error_info else now = Time.now.to_i if @suppress_emit_error_log_interval.zero? || now > @next_emit_error_log_time @@ -464,5 +476,18 @@ def handle_emits_error(tag, es, error) raise error end end + + private + + def routing_error_event? + !!Thread.current[ERROR_ROUTING_KEY] + end + + def route_error_event + Thread.current[ERROR_ROUTING_KEY] = true + yield + ensure + Thread.current[ERROR_ROUTING_KEY] = false + end end end diff --git a/test/test_root_agent.rb b/test/test_root_agent.rb index 7e2a07f7cd..78b2ff661f 100644 --- a/test/test_root_agent.rb +++ b/test/test_root_agent.rb @@ -671,6 +671,69 @@ def setup_root_agent(conf) end end + sub_test_case 'error events which fail in @ERROR' do + setup do + @ra = RootAgent.new(log: $log) + stub(Engine).root_agent { @ra } + # in @ERROR raises an error at emit time, just like + # BufferChunkOverflowError raised for a record larger than chunk_limit_size + conf = <<-EOC + + @type test_out + + +EOC + @ra.configure(Config.parse(conf, "(test)", "(test_dir)", true)) + @ra.log.out.reset + end + + test 'an error event is dumped and dropped, not routed to @ERROR again' do + @ra.emit_error_event("tag", event_time, {"message" => "test"}, StandardError.new("original error")) + + logs = @ra.log.out.logs + assert_equal([1, 1], + [logs.count { |line| line.include?("send an error event to @ERROR") }, + logs.count { |line| line.include?("dump an error event stream because it failed in @ERROR") }]) + end + + test 'an error event emitted from inside @ERROR is dumped and dropped' do + # A plugin in @ERROR may call router.emit_error_event by itself, e.g. a filter + # which fails to handle a record. + stub(@ra.error_collector).emit do |tag, time, record| + @ra.emit_error_event(tag, time, record, StandardError.new("error in @ERROR")) + end + @ra.emit_error_event("tag", event_time, {"message" => "test"}, StandardError.new("original error")) + + logs = @ra.log.out.logs + assert_equal([1, 1], + [logs.count { |line| line.include?("send an error event to @ERROR") }, + logs.count { |line| line.include?("dump an error event because it failed in @ERROR") }]) + end + + test 'an error event stream is dumped and dropped, not routed to @ERROR again' do + es = Fluent::OneEventStream.new(event_time, {"message" => "test"}) + @ra.handle_emits_error("tag", es, StandardError.new("original error")) + + logs = @ra.log.out.logs + assert_equal([1, 1], + [logs.count { |line| line.include?("send an error event stream to @ERROR") }, + logs.count { |line| line.include?("dump an error event stream because it failed in @ERROR") }]) + end + + test 'the next error event is routed to @ERROR again' do + 2.times do + @ra.emit_error_event("tag", event_time, {"message" => "test"}, StandardError.new("original error")) + end + + logs = @ra.log.out.logs + assert_equal(2, logs.count { |line| line.include?("send an error event to @ERROR") }) + end + end + sub_test_case 'configured at worker2 with 4 workers environment' do setup do ENV['SERVERENGINE_WORKER_ID'] = '2'