From bdda4d2b5db37a01845d1bae8642912719492b61 Mon Sep 17 00:00:00 2001 From: tarun9715m Date: Tue, 4 Aug 2026 17:45:53 +0530 Subject: [PATCH] Fix infinite loop when an error event fails in @ERROR If a record is larger than chunk_limit_size, the buffer raises BufferChunkOverflowError at emit time and the event is routed to @ERROR. When the output in @ERROR raises the same error, the event was routed to @ERROR again, and this repeated without bound. Nothing marked the event as having already been through the error path, so there was no termination condition. The routing is recursive, so the worker eventually died with SystemStackError. @ERROR is the last resort route. An error event which fails while it is being routed there is now dumped with a distinct error log and dropped, instead of being routed back to @ERROR. The behavior without an @ERROR label is unchanged, including re-raising the error and the emit_error_log_interval suppression. Fixes #5462 Signed-off-by: tarun9715m Co-Authored-By: Claude Opus 5 (1M context) --- lib/fluent/root_agent.rb | 33 ++++++++++++++++++--- test/test_root_agent.rb | 63 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 92 insertions(+), 4 deletions(-) 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'