Fix infinite loop when an error event fails in @ERROR

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

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 }
+      # <match> in @ERROR raises an error at emit time, just like
+      # BufferChunkOverflowError raised for a record larger than chunk_limit_size
+      conf = <<-EOC
+<match **>
+  @type test_out
+</match>
+<label @ERROR>
+  <match **>
+    @type test_out_error
+  </match>
+</label>
+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'