From bd02bea8608973c72ad85ab08bf6f62cf4a1d572 Mon Sep 17 00:00:00 2001
From: Yuta Iwama <ganmacs@gmail.com>
Date: Fri, 25 Oct 2019 11:20:23 +0900
Subject: [PATCH 01/14] Supervisor print gem list

instead of worker0

Signed-off-by: Yuta Iwama <ganmacs@gmail.com>
---
 lib/fluent/engine.rb     |  5 -----
 lib/fluent/supervisor.rb | 11 +++++++++--
 2 files changed, 9 insertions(+), 7 deletions(-)

diff --git a/lib/fluent/engine.rb b/lib/fluent/engine.rb
index aa18ac7d57..b9af3e67e9 100644
--- a/lib/fluent/engine.rb
+++ b/lib/fluent/engine.rb
@@ -113,11 +113,6 @@ def run_configure(conf)
     end
 
     def configure(conf)
-      # plugins / configuration dumps
-      Gem::Specification.find_all.select{|x| x.name =~ /^fluent(d|-(plugin|mixin)-.*)$/}.each do |spec|
-        $log.info :worker0, "gem '#{spec.name}' version '#{spec.version}'"
-      end
-
       @root_agent.configure(conf)
 
       begin
diff --git a/lib/fluent/supervisor.rb b/lib/fluent/supervisor.rb
index 36f26dffcf..38f20a5511 100644
--- a/lib/fluent/supervisor.rb
+++ b/lib/fluent/supervisor.rb
@@ -586,7 +586,7 @@ def dry_run
         Fluent::Engine.dry_run_mode = true
         change_privilege
         MessagePackFactory.init
-        init_engine
+        init_engine(supervisor: true)
         run_configure
       rescue Fluent::ConfigError => e
         $log.error "config error", file: @config_path, error: e
@@ -794,7 +794,7 @@ def change_privilege
       ServerEngine::Privilege.change(@chuser, @chgroup)
     end
 
-    def init_engine
+    def init_engine(supervisor: false)
       Fluent::Engine.init(@system_config)
 
       @libs.each {|lib|
@@ -807,6 +807,13 @@ def init_engine
           Fluent::Engine.add_plugin_dir(dir)
         end
       }
+
+      if supervisor
+        # plugins / configuration dumps
+        Gem::Specification.find_all.select { |x| x.name =~ /^fluent(d|-(plugin|mixin)-.*)$/ }.each do |spec|
+          $log.info("gem '#{spec.name}' version '#{spec.version}'")
+        end
+      end
     end
 
     def run_configure

From d44ce3ba0ab73b569993685c637f1ad95e9085dd Mon Sep 17 00:00:00 2001
From: Yuta Iwama <ganmacs@gmail.com>
Date: Fri, 25 Oct 2019 12:02:46 +0900
Subject: [PATCH 02/14] if `@log_event_router` exists, log is already enabled

https://github.com/fluent/fluentd/blob/fcd467c722c491919c7044c772e9e49ead3ef787/lib/fluent/engine.rb#L151

Signed-off-by: Yuta Iwama <ganmacs@gmail.com>
---
 lib/fluent/engine.rb | 1 -
 1 file changed, 1 deletion(-)

diff --git a/lib/fluent/engine.rb b/lib/fluent/engine.rb
index b9af3e67e9..5ffa24d358 100644
--- a/lib/fluent/engine.rb
+++ b/lib/fluent/engine.rb
@@ -204,7 +204,6 @@ def run
         start
 
         if @log_event_router
-          $log.enable_event(true)
           @log_emit_thread = Thread.new(&method(:log_event_loop))
           @log_emit_thread.abort_on_exception = true
         end

From 3a3aee0227b621b61d1d7f258147f7ff661316e0 Mon Sep 17 00:00:00 2001
From: Yuta Iwama <ganmacs@gmail.com>
Date: Fri, 25 Oct 2019 12:13:05 +0900
Subject: [PATCH 03/14] Add FluentLogEventEmitter

Signed-off-by: Yuta Iwama <ganmacs@gmail.com>
---
 lib/fluent/engine.rb | 72 +++++++++++++++++++++++++++-----------------
 1 file changed, 45 insertions(+), 27 deletions(-)

diff --git a/lib/fluent/engine.rb b/lib/fluent/engine.rb
index 5ffa24d358..7c656f9692 100644
--- a/lib/fluent/engine.rb
+++ b/lib/fluent/engine.rb
@@ -115,36 +115,12 @@ def run_configure(conf)
     def configure(conf)
       @root_agent.configure(conf)
 
-      begin
-        log_event_agent = @root_agent.find_label(Fluent::Log::LOG_EVENT_LABEL)
-        log_event_router = log_event_agent.event_router
-
-        # suppress mismatched tags only for <label @FLUENT_LOG> label.
-        # it's not suppressed in default event router for non-log-event events
-        log_event_router.suppress_missing_match!
-
-        @log_event_router = log_event_router
-
-        unmatched_tags = Fluent::Log.event_tags.select{|t| !@log_event_router.match?(t) }
-        unless unmatched_tags.empty?
-          $log.warn "match for some tags of log events are not defined (to be ignored)", tags: unmatched_tags
-        end
-      rescue ArgumentError # ArgumentError "#{label_name} label not found"
-        # use default event router if <label @FLUENT_LOG> is missing in configuration
-        log_event_router = @root_agent.event_router
+      @log_event_router = FluentLogEventEmitter.build(@root_agent)
 
-        if Fluent::Log.event_tags.any?{|t| log_event_router.match?(t) }
-          @log_event_router = log_event_router
-
-          unmatched_tags = Fluent::Log.event_tags.select{|t| !@log_event_router.match?(t) }
-          unless unmatched_tags.empty?
-            $log.warn "match for some tags of log events are not defined (to be ignored)", tags: unmatched_tags
-          end
-        end
+      if @log_event_router
+        $log.enable_event(true)
       end
 
-      $log.enable_event(true) if @log_event_router
-
       unless @suppress_config_dump
         $log.info :supervisor, "using configuration file: #{conf.to_s.rstrip}"
       end
@@ -263,5 +239,47 @@ def shutdown
     end
   end
 
+  class FluentLogEventEmitter
+    def self.build(root_agent)
+      log_event_router = nil
+
+      begin
+        log_event_agent = root_agent.find_label(Fluent::Log::LOG_EVENT_LABEL)
+        log_event_router = log_event_agent.event_router
+
+        # suppress mismatched tags only for <label @FLUENT_LOG> label.
+        # it's not suppressed in default event router for non-log-event events
+        log_event_router.suppress_missing_match!
+
+        log_event_router = log_event_router
+
+        unmatched_tags = Fluent::Log.event_tags.select{|t| !log_event_router.match?(t) }
+        unless unmatched_tags.empty?
+          $log.warn "match for some tags of log events are not defined (to be ignored)", tags: unmatched_tags
+        end
+
+      rescue ArgumentError # ArgumentError "#{label_name} label not found"
+        # use default event router if <label @FLUENT_LOG> is missing in configuration
+        log_event_router = root_agent.event_router
+
+        if Fluent::Log.event_tags.any?{|t| log_event_router.match?(t) }
+          log_event_router = log_event_router
+
+          unmatched_tags = Fluent::Log.event_tags.select{|t| !log_event_router.match?(t) }
+          unless unmatched_tags.empty?
+            $log.warn "match for some tags of log events are not defined (to be ignored)", tags: unmatched_tags
+          end
+        end
+      end
+
+      if log_event_router
+        # FluentLogEventEmitter.new(log_event_router)
+        log_event_router
+      else
+        nil # no need to create fluent log event emitter
+      end
+    end
+  end
+
   Engine = EngineClass.new
 end

From 528b50f80206ff818abfe0c0c1a825ffccff24a1 Mon Sep 17 00:00:00 2001
From: Yuta Iwama <ganmacs@gmail.com>
Date: Fri, 25 Oct 2019 12:39:13 +0900
Subject: [PATCH 04/14] move start/stop logic to class

Signed-off-by: Yuta Iwama <ganmacs@gmail.com>
---
 lib/fluent/engine.rb | 109 ++++++++++++++++++++++++++-----------------
 1 file changed, 67 insertions(+), 42 deletions(-)

diff --git a/lib/fluent/engine.rb b/lib/fluent/engine.rb
index 7c656f9692..49a1134d7c 100644
--- a/lib/fluent/engine.rb
+++ b/lib/fluent/engine.rb
@@ -32,10 +32,6 @@ def initialize
       @_worker_id = nil
 
       @log_event_router = nil
-      @log_emit_thread = nil
-      @log_event_loop_stop = false
-      @log_event_loop_graceful_stop = false
-      @log_event_queue = []
       @log_event_verbose = false
 
       @suppress_config_dump = false
@@ -48,7 +44,6 @@ def initialize
     MAINLOOP_SLEEP_INTERVAL = 0.3
 
     MATCH_CACHE_SIZE = 1024
-    LOG_EMIT_INTERVAL = 0.1
 
     attr_reader :root_agent
     attr_reader :matches, :sources
@@ -151,37 +146,13 @@ def now
       Fluent::EventTime.now
     end
 
-    def log_event_loop
-      $log.disable_events(Thread.current)
-
-      while sleep(LOG_EMIT_INTERVAL)
-        break if @log_event_loop_stop
-        break if @log_event_loop_graceful_stop && @log_event_queue.empty?
-        next if @log_event_queue.empty?
-
-        # NOTE: thead-safe of slice! depends on GVL
-        events = @log_event_queue.slice!(0..-1)
-        next if events.empty?
-
-        events.each {|tag,time,record|
-          begin
-            @log_event_router.emit(tag, time, record)
-          rescue => e
-            # This $log.error doesn't emit log events, because of `$log.disable_events(Thread.current)` above
-            $log.error "failed to emit fluentd's log event", tag: tag, event: record, error: e
-          end
-        }
-      end
-    end
-
     def run
       begin
         $log.info "starting fluentd worker", pid: Process.pid, ppid: Process.ppid, worker: worker_id
         start
 
         if @log_event_router
-          @log_emit_thread = Thread.new(&method(:log_event_loop))
-          @log_emit_thread.abort_on_exception = true
+          @log_event_router.start
         end
 
         $log.info "fluentd worker is now running", worker: worker_id
@@ -196,18 +167,16 @@ def run
 
       unless @log_event_verbose
         $log.enable_event(false)
-        if @log_emit_thread
-          # to make sure to emit all log events into router, before shutting down
-          @log_event_loop_graceful_stop = true
-          @log_emit_thread.join
-          @log_emit_thread = nil
+
+        if @log_event_router
+          @log_event_router.graceful_stop
         end
       end
       $log.info "shutting down fluentd worker", worker: worker_id
       shutdown
-      if @log_emit_thread
-        @log_event_loop_stop = true
-        @log_emit_thread.join
+
+      if @log_event_router
+        @log_event_router.stop
       end
     end
 
@@ -217,8 +186,9 @@ def stop
     end
 
     def push_log_event(tag, time, record)
-      return if @log_emit_thread.nil?
-      @log_event_queue.push([tag, time, record])
+      if @log_event_router
+        @log_event_router.emit_event([tag, time, record])
+      end
     end
 
     def worker_id
@@ -240,6 +210,8 @@ def shutdown
   end
 
   class FluentLogEventEmitter
+    LOG_EMIT_INTERVAL = 0.1
+
     def self.build(root_agent)
       log_event_router = nil
 
@@ -273,12 +245,65 @@ def self.build(root_agent)
       end
 
       if log_event_router
-        # FluentLogEventEmitter.new(log_event_router)
-        log_event_router
+        FluentLogEventEmitter.new(log_event_router)
       else
         nil # no need to create fluent log event emitter
       end
     end
+
+    def initialize(event_router)
+      @event_router = event_router
+      @thread = nil
+      @graceful_stop = false
+      @stopped = false
+      @event_queue = []
+    end
+
+    def start
+      @thread = Thread.new do
+        $log.disable_events(Thread.current)
+
+        loop do
+          sleep(LOG_EMIT_INTERVAL)
+
+          break if @stop
+          break if @graceful_stop && @event_queue.empty?
+
+          next if @event_queue.empty?
+
+          # NOTE: thead-safe of slice! depends on GVL
+          events = @event_queue.slice!(0..-1)
+          next if events.empty?
+
+          events.each do |tag, time, record|
+            begin
+              @event_router.emit(tag, time, record)
+            rescue => e
+              # This $log.error doesn't emit log events, because of `$log.disable_events(Thread.current)` above
+              $log.error "failed to emit fluentd's log event", tag: tag, event: record, error: e
+            end
+          end
+        end
+      end
+
+      @thread.abort_on_exception = true
+    end
+
+    def stop
+      @stop = true
+      # there is no problem calling Thread#join multiple times.
+      @thread.join
+    end
+
+    def graceful_stop
+      # to make sure to emit all log events into router, before shutting down
+      @graceful_stop = true
+      @thread.join
+    end
+
+    def emit_event(event)
+      @event_queue.push(event)
+    end
   end
 
   Engine = EngineClass.new

From 2b47a44d4b3787175affdb10b49f2cf3be7ebeb2 Mon Sep 17 00:00:00 2001
From: Yuta Iwama <ganmacs@gmail.com>
Date: Fri, 25 Oct 2019 12:48:10 +0900
Subject: [PATCH 05/14] Move class into file

Signed-off-by: Yuta Iwama <ganmacs@gmail.com>
---
 lib/fluent/engine.rb                  | 120 +++-----------------------
 lib/fluent/fluent_log_event_router.rb | 120 ++++++++++++++++++++++++++
 2 files changed, 132 insertions(+), 108 deletions(-)
 create mode 100644 lib/fluent/fluent_log_event_router.rb

diff --git a/lib/fluent/engine.rb b/lib/fluent/engine.rb
index 49a1134d7c..7ad197824c 100644
--- a/lib/fluent/engine.rb
+++ b/lib/fluent/engine.rb
@@ -22,6 +22,7 @@
 require 'fluent/time'
 require 'fluent/system_config'
 require 'fluent/plugin'
+require 'fluent/fluent_log_event_router'
 
 module Fluent
   class EngineClass
@@ -31,7 +32,7 @@ def initialize
       @engine_stopped = false
       @_worker_id = nil
 
-      @log_event_router = nil
+      @fluent_log_event_router = nil
       @log_event_verbose = false
 
       @suppress_config_dump = false
@@ -110,9 +111,9 @@ def run_configure(conf)
     def configure(conf)
       @root_agent.configure(conf)
 
-      @log_event_router = FluentLogEventEmitter.build(@root_agent)
+      @fleunt_log_event_router = FluentLogEventRouter.build(@root_agent)
 
-      if @log_event_router
+      if @fleunt_log_event_router
         $log.enable_event(true)
       end
 
@@ -151,8 +152,8 @@ def run
         $log.info "starting fluentd worker", pid: Process.pid, ppid: Process.ppid, worker: worker_id
         start
 
-        if @log_event_router
-          @log_event_router.start
+        if @fleunt_log_event_router
+          @fleunt_log_event_router.start
         end
 
         $log.info "fluentd worker is now running", worker: worker_id
@@ -168,15 +169,15 @@ def run
       unless @log_event_verbose
         $log.enable_event(false)
 
-        if @log_event_router
-          @log_event_router.graceful_stop
+        if @fleunt_log_event_router
+          @fleunt_log_event_router.graceful_stop
         end
       end
       $log.info "shutting down fluentd worker", worker: worker_id
       shutdown
 
-      if @log_event_router
-        @log_event_router.stop
+      if @fleunt_log_event_router
+        @fleunt_log_event_router.stop
       end
     end
 
@@ -186,8 +187,8 @@ def stop
     end
 
     def push_log_event(tag, time, record)
-      if @log_event_router
-        @log_event_router.emit_event([tag, time, record])
+      if @fleunt_log_event_router
+        @fleunt_log_event_router.emit_event([tag, time, record])
       end
     end
 
@@ -209,102 +210,5 @@ def shutdown
     end
   end
 
-  class FluentLogEventEmitter
-    LOG_EMIT_INTERVAL = 0.1
-
-    def self.build(root_agent)
-      log_event_router = nil
-
-      begin
-        log_event_agent = root_agent.find_label(Fluent::Log::LOG_EVENT_LABEL)
-        log_event_router = log_event_agent.event_router
-
-        # suppress mismatched tags only for <label @FLUENT_LOG> label.
-        # it's not suppressed in default event router for non-log-event events
-        log_event_router.suppress_missing_match!
-
-        log_event_router = log_event_router
-
-        unmatched_tags = Fluent::Log.event_tags.select{|t| !log_event_router.match?(t) }
-        unless unmatched_tags.empty?
-          $log.warn "match for some tags of log events are not defined (to be ignored)", tags: unmatched_tags
-        end
-
-      rescue ArgumentError # ArgumentError "#{label_name} label not found"
-        # use default event router if <label @FLUENT_LOG> is missing in configuration
-        log_event_router = root_agent.event_router
-
-        if Fluent::Log.event_tags.any?{|t| log_event_router.match?(t) }
-          log_event_router = log_event_router
-
-          unmatched_tags = Fluent::Log.event_tags.select{|t| !log_event_router.match?(t) }
-          unless unmatched_tags.empty?
-            $log.warn "match for some tags of log events are not defined (to be ignored)", tags: unmatched_tags
-          end
-        end
-      end
-
-      if log_event_router
-        FluentLogEventEmitter.new(log_event_router)
-      else
-        nil # no need to create fluent log event emitter
-      end
-    end
-
-    def initialize(event_router)
-      @event_router = event_router
-      @thread = nil
-      @graceful_stop = false
-      @stopped = false
-      @event_queue = []
-    end
-
-    def start
-      @thread = Thread.new do
-        $log.disable_events(Thread.current)
-
-        loop do
-          sleep(LOG_EMIT_INTERVAL)
-
-          break if @stop
-          break if @graceful_stop && @event_queue.empty?
-
-          next if @event_queue.empty?
-
-          # NOTE: thead-safe of slice! depends on GVL
-          events = @event_queue.slice!(0..-1)
-          next if events.empty?
-
-          events.each do |tag, time, record|
-            begin
-              @event_router.emit(tag, time, record)
-            rescue => e
-              # This $log.error doesn't emit log events, because of `$log.disable_events(Thread.current)` above
-              $log.error "failed to emit fluentd's log event", tag: tag, event: record, error: e
-            end
-          end
-        end
-      end
-
-      @thread.abort_on_exception = true
-    end
-
-    def stop
-      @stop = true
-      # there is no problem calling Thread#join multiple times.
-      @thread.join
-    end
-
-    def graceful_stop
-      # to make sure to emit all log events into router, before shutting down
-      @graceful_stop = true
-      @thread.join
-    end
-
-    def emit_event(event)
-      @event_queue.push(event)
-    end
-  end
-
   Engine = EngineClass.new
 end
diff --git a/lib/fluent/fluent_log_event_router.rb b/lib/fluent/fluent_log_event_router.rb
new file mode 100644
index 0000000000..c99e0e96da
--- /dev/null
+++ b/lib/fluent/fluent_log_event_router.rb
@@ -0,0 +1,120 @@
+#
+# Fluentd
+#
+#    Licensed under the Apache License, Version 2.0 (the "License");
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+#
+
+require 'fluent/log'
+
+module Fluent
+  # This class is for handling fluentd's inner log
+  # e.g. <label @FLUNT_LOG> section and <match fluent.**> section
+  class FluentLogEventRouter
+    LOG_EMIT_INTERVAL = 0.1
+
+    # @param root_agent [Fluent::RootAgent]
+    def self.build(root_agent)
+      log_event_router = nil
+
+      begin
+        log_event_agent = root_agent.find_label(Fluent::Log::LOG_EVENT_LABEL)
+        log_event_router = log_event_agent.event_router
+
+        # suppress mismatched tags only for <label @FLUENT_LOG> label.
+        # it's not suppressed in default event router for non-log-event events
+        log_event_router.suppress_missing_match!
+
+        log_event_router = log_event_router
+
+        unmatched_tags = Fluent::Log.event_tags.select { |t| !log_event_router.match?(t) }
+        unless unmatched_tags.empty?
+          $log.warn "match for some tags of log events are not defined (to be ignored)", tags: unmatched_tags
+        end
+
+      rescue ArgumentError # ArgumentError "#{label_name} label not found"
+        # use default event router if <label @FLUENT_LOG> is missing in configuration
+        log_event_router = root_agent.event_router
+
+        if Fluent::Log.event_tags.any? { |t| log_event_router.match?(t) }
+          log_event_router = log_event_router
+
+          unmatched_tags = Fluent::Log.event_tags.select { |t| !log_event_router.match?(t) }
+          unless unmatched_tags.empty?
+            $log.warn "match for some tags of log events are not defined (to be ignored)", tags: unmatched_tags
+          end
+        end
+      end
+
+      if log_event_router
+        FluentLogEventRouter.new(log_event_router)
+      else
+        nil # no need to create fluent log event emitter
+      end
+    end
+
+    # @param event_router [Fluent::EventRouter]
+    def initialize(event_router)
+      @event_router = event_router
+      @thread = nil
+      @graceful_stop = false
+      @stopped = false
+      @event_queue = []
+    end
+
+    def start
+      @thread = Thread.new do
+        $log.disable_events(Thread.current)
+
+        loop do
+          sleep(LOG_EMIT_INTERVAL)
+
+          break if @stop
+          break if @graceful_stop && @event_queue.empty?
+
+          next if @event_queue.empty?
+
+          # NOTE: thead-safe of slice! depends on GVL
+          events = @event_queue.slice!(0..-1)
+          next if events.empty?
+
+          events.each do |tag, time, record|
+            begin
+              @event_router.emit(tag, time, record)
+            rescue => e
+              # This $log.error doesn't emit log events, because of `$log.disable_events(Thread.current)` above
+              $log.error "failed to emit fluentd's log event", tag: tag, event: record, error: e
+            end
+          end
+        end
+      end
+
+      @thread.abort_on_exception = true
+    end
+
+    def stop
+      @stop = true
+      # there is no problem calling Thread#join multiple times.
+      @thread.join
+    end
+
+    def graceful_stop
+      # to make sure to emit all log events into router, before shutting down
+      @graceful_stop = true
+      @thread.join
+    end
+
+    def emit_event(event)
+      @event_queue.push(event)
+    end
+  end
+end

From 9a12bc3a4875fd9bc9bfaadef0e3353ca1d42533 Mon Sep 17 00:00:00 2001
From: Yuta Iwama <ganmacs@gmail.com>
Date: Fri, 25 Oct 2019 12:52:41 +0900
Subject: [PATCH 06/14] Add nullfluentlogeventrouter for default

Signed-off-by: Yuta Iwama <ganmacs@gmail.com>
---
 lib/fluent/engine.rb                  | 17 ++++-------------
 lib/fluent/fluent_log_event_router.rb | 16 ++++++++++++++--
 2 files changed, 18 insertions(+), 15 deletions(-)

diff --git a/lib/fluent/engine.rb b/lib/fluent/engine.rb
index 7ad197824c..f57be8e5bf 100644
--- a/lib/fluent/engine.rb
+++ b/lib/fluent/engine.rb
@@ -152,9 +152,7 @@ def run
         $log.info "starting fluentd worker", pid: Process.pid, ppid: Process.ppid, worker: worker_id
         start
 
-        if @fleunt_log_event_router
-          @fleunt_log_event_router.start
-        end
+        @fleunt_log_event_router.start
 
         $log.info "fluentd worker is now running", worker: worker_id
         sleep MAINLOOP_SLEEP_INTERVAL until @engine_stopped
@@ -168,17 +166,12 @@ def run
 
       unless @log_event_verbose
         $log.enable_event(false)
-
-        if @fleunt_log_event_router
-          @fleunt_log_event_router.graceful_stop
-        end
+        @fleunt_log_event_router.graceful_stop
       end
       $log.info "shutting down fluentd worker", worker: worker_id
       shutdown
 
-      if @fleunt_log_event_router
-        @fleunt_log_event_router.stop
-      end
+      @fleunt_log_event_router.stop
     end
 
     def stop
@@ -187,9 +180,7 @@ def stop
     end
 
     def push_log_event(tag, time, record)
-      if @fleunt_log_event_router
-        @fleunt_log_event_router.emit_event([tag, time, record])
-      end
+      @fleunt_log_event_router.emit_event([tag, time, record])
     end
 
     def worker_id
diff --git a/lib/fluent/fluent_log_event_router.rb b/lib/fluent/fluent_log_event_router.rb
index c99e0e96da..5949db3232 100644
--- a/lib/fluent/fluent_log_event_router.rb
+++ b/lib/fluent/fluent_log_event_router.rb
@@ -17,9 +17,20 @@
 require 'fluent/log'
 
 module Fluent
+  # DO NOT write any logic here
+  class NullFluentLogEventRouter
+    def start; end
+
+    def stop; end
+
+    def graceful_stop; end
+
+    def emit_event; end
+  end
+
   # This class is for handling fluentd's inner log
   # e.g. <label @FLUNT_LOG> section and <match fluent.**> section
-  class FluentLogEventRouter
+  class FluentLogEventRouter < NullFluentLogEventRouter
     LOG_EMIT_INTERVAL = 0.1
 
     # @param root_agent [Fluent::RootAgent]
@@ -58,7 +69,8 @@ def self.build(root_agent)
       if log_event_router
         FluentLogEventRouter.new(log_event_router)
       else
-        nil # no need to create fluent log event emitter
+        $log.debug('No fluent logger for internal event')
+        NullFluentLogEventRouter.new
       end
     end
 

From 2c19d5a6981a0510c9bb21536e64261928ed196b Mon Sep 17 00:00:00 2001
From: Yuta Iwama <ganmacs@gmail.com>
Date: Fri, 25 Oct 2019 13:40:06 +0900
Subject: [PATCH 07/14] Delete source and matches and merge them into one line

matches isn't needed since https://github.com/fluent/fluentd/commit/6b37daa07f94534d43d7d92c5498f874d8cd48f3#diff-a3660169e91ca999d07caabc1aac67c3

Signed-off-by: Yuta Iwama <ganmacs@gmail.com>
---
 lib/fluent/engine.rb | 7 +------
 1 file changed, 1 insertion(+), 6 deletions(-)

diff --git a/lib/fluent/engine.rb b/lib/fluent/engine.rb
index f57be8e5bf..fc44d3a750 100644
--- a/lib/fluent/engine.rb
+++ b/lib/fluent/engine.rb
@@ -44,12 +44,7 @@ def initialize
 
     MAINLOOP_SLEEP_INTERVAL = 0.3
 
-    MATCH_CACHE_SIZE = 1024
-
-    attr_reader :root_agent
-    attr_reader :matches, :sources
-    attr_reader :system_config
-
+    attr_reader :root_agent, :system_config
     attr_accessor :dry_run_mode
 
     def init(system_config)

From 99ed1a2adf9ceffad2f55c19cca7ef2733e42716 Mon Sep 17 00:00:00 2001
From: Yuta Iwama <ganmacs@gmail.com>
Date: Fri, 25 Oct 2019 13:41:44 +0900
Subject: [PATCH 08/14] default_loop is unused

https://github.com/fluent/fluentd/commit/19f5b3e34610ce72bf7861d443410ea132eb4d82#diff-a3660169e91ca999d07caabc1aac67c3

Signed-off-by: Yuta Iwama <ganmacs@gmail.com>
---
 lib/fluent/engine.rb | 1 -
 1 file changed, 1 deletion(-)

diff --git a/lib/fluent/engine.rb b/lib/fluent/engine.rb
index fc44d3a750..ff7a6ef0e3 100644
--- a/lib/fluent/engine.rb
+++ b/lib/fluent/engine.rb
@@ -28,7 +28,6 @@ module Fluent
   class EngineClass
     def initialize
       @root_agent = nil
-      @default_loop = nil
       @engine_stopped = false
       @_worker_id = nil
 

From 9957d930d9020a48899a8ebdbc037de5ce66833b Mon Sep 17 00:00:00 2001
From: Yuta Iwama <ganmacs@gmail.com>
Date: Fri, 25 Oct 2019 13:43:08 +0900
Subject: [PATCH 09/14] init without_source at initialize time

Signed-off-by: Yuta Iwama <ganmacs@gmail.com>
---
 lib/fluent/engine.rb | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/lib/fluent/engine.rb b/lib/fluent/engine.rb
index ff7a6ef0e3..eae3f85d82 100644
--- a/lib/fluent/engine.rb
+++ b/lib/fluent/engine.rb
@@ -31,11 +31,11 @@ def initialize
       @engine_stopped = false
       @_worker_id = nil
 
-      @fluent_log_event_router = nil
       @log_event_verbose = false
-
       @suppress_config_dump = false
+      @without_source = false
 
+      @fluent_log_event_router = nil
       @system_config = SystemConfig.new
 
       @dry_run_mode = false

From 2cfc4cf954410a89f68f9b25467a68a4aa1961f9 Mon Sep 17 00:00:00 2001
From: Yuta Iwama <ganmacs@gmail.com>
Date: Fri, 25 Oct 2019 13:52:57 +0900
Subject: [PATCH 10/14] Use queue to be thread safe

Signed-off-by: Yuta Iwama <ganmacs@gmail.com>
---
 lib/fluent/fluent_log_event_router.rb | 37 ++++++++++++++-------------
 1 file changed, 19 insertions(+), 18 deletions(-)

diff --git a/lib/fluent/fluent_log_event_router.rb b/lib/fluent/fluent_log_event_router.rb
index 5949db3232..e9bfa18d5d 100644
--- a/lib/fluent/fluent_log_event_router.rb
+++ b/lib/fluent/fluent_log_event_router.rb
@@ -31,8 +31,6 @@ def emit_event; end
   # This class is for handling fluentd's inner log
   # e.g. <label @FLUNT_LOG> section and <match fluent.**> section
   class FluentLogEventRouter < NullFluentLogEventRouter
-    LOG_EMIT_INTERVAL = 0.1
-
     # @param root_agent [Fluent::RootAgent]
     def self.build(root_agent)
       log_event_router = nil
@@ -74,13 +72,15 @@ def self.build(root_agent)
       end
     end
 
+    STOP = :stop
+    GRACEFUL_STOP = :graceful_stop
+
     # @param event_router [Fluent::EventRouter]
     def initialize(event_router)
       @event_router = event_router
       @thread = nil
       @graceful_stop = false
-      @stopped = false
-      @event_queue = []
+      @event_queue = Queue.new
     end
 
     def start
@@ -88,25 +88,26 @@ def start
         $log.disable_events(Thread.current)
 
         loop do
-          sleep(LOG_EMIT_INTERVAL)
-
-          break if @stop
-          break if @graceful_stop && @event_queue.empty?
-
-          next if @event_queue.empty?
-
-          # NOTE: thead-safe of slice! depends on GVL
-          events = @event_queue.slice!(0..-1)
-          next if events.empty?
-
-          events.each do |tag, time, record|
+          event = @event_queue.pop
+
+          case event
+          when GRACEFUL_STOP
+            @graceful_stop = true
+          when STOP
+            break
+          else
             begin
+              tag, time, record = event
               @event_router.emit(tag, time, record)
             rescue => e
               # This $log.error doesn't emit log events, because of `$log.disable_events(Thread.current)` above
               $log.error "failed to emit fluentd's log event", tag: tag, event: record, error: e
             end
           end
+
+          if @graceful_stop && @event_queue.empty?
+            break
+          end
         end
       end
 
@@ -114,14 +115,14 @@ def start
     end
 
     def stop
-      @stop = true
+      @event_queue.push(STOP)
       # there is no problem calling Thread#join multiple times.
       @thread.join
     end
 
     def graceful_stop
       # to make sure to emit all log events into router, before shutting down
-      @graceful_stop = true
+      @event_queue.push(GRACEFUL_STOP)
       @thread.join
     end
 

From 46937aa9c7e26d17eb11fb442423f92625ae06df Mon Sep 17 00:00:00 2001
From: Yuta Iwama <ganmacs@gmail.com>
Date: Fri, 25 Oct 2019 15:13:37 +0900
Subject: [PATCH 11/14] Add test for fluent_log_event_router

Signed-off-by: Yuta Iwama <ganmacs@gmail.com>
---
 lib/fluent/fluent_log_event_router.rb | 10 +--
 test/test_fluent_log_event_router.rb  | 88 +++++++++++++++++++++++++++
 2 files changed, 93 insertions(+), 5 deletions(-)
 create mode 100644 test/test_fluent_log_event_router.rb

diff --git a/lib/fluent/fluent_log_event_router.rb b/lib/fluent/fluent_log_event_router.rb
index e9bfa18d5d..4029c95917 100644
--- a/lib/fluent/fluent_log_event_router.rb
+++ b/lib/fluent/fluent_log_event_router.rb
@@ -52,10 +52,10 @@ def self.build(root_agent)
 
       rescue ArgumentError # ArgumentError "#{label_name} label not found"
         # use default event router if <label @FLUENT_LOG> is missing in configuration
-        log_event_router = root_agent.event_router
+        root_log_event_router = root_agent.event_router
 
-        if Fluent::Log.event_tags.any? { |t| log_event_router.match?(t) }
-          log_event_router = log_event_router
+        if Fluent::Log.event_tags.any? { |t| root_log_event_router.match?(t) }
+          log_event_router = root_log_event_router
 
           unmatched_tags = Fluent::Log.event_tags.select { |t| !log_event_router.match?(t) }
           unless unmatched_tags.empty?
@@ -117,13 +117,13 @@ def start
     def stop
       @event_queue.push(STOP)
       # there is no problem calling Thread#join multiple times.
-      @thread.join
+      @thread && @thread.join
     end
 
     def graceful_stop
       # to make sure to emit all log events into router, before shutting down
       @event_queue.push(GRACEFUL_STOP)
-      @thread.join
+      @thread && @thread.join
     end
 
     def emit_event(event)
diff --git a/test/test_fluent_log_event_router.rb b/test/test_fluent_log_event_router.rb
new file mode 100644
index 0000000000..8102794d0c
--- /dev/null
+++ b/test/test_fluent_log_event_router.rb
@@ -0,0 +1,88 @@
+require_relative 'helper'
+require 'fluent/fluent_log_event_router'
+require 'fluent/root_agent'
+require 'fluent/system_config'
+
+class FluentLogEventRouterTest < ::Test::Unit::TestCase
+  # @param config [String]
+  def build_config(config)
+    Fluent::Config.parse(config, 'fluent_log_event', '', syntax: :v1)
+  end
+
+  sub_test_case '#build' do
+    test 'NullFluentLogEventRouter if root_agent have not internal logger' do
+      root_agent = Fluent::RootAgent.new(log: $log, system_config: Fluent::SystemConfig.new)
+      root_agent.configure(build_config(''))
+
+      d = Fluent::FluentLogEventRouter.build(root_agent)
+      assert_equal Fluent::NullFluentLogEventRouter, d.class
+    end
+
+    test 'FluentLogEventRouter if <match fluent.*> exists in config' do
+      root_agent = Fluent::RootAgent.new(log: $log, system_config: Fluent::SystemConfig.new)
+      root_agent.configure(build_config(<<-CONFIG))
+        <match fluent.*>
+          @type null
+        </match>
+      CONFIG
+
+      d = Fluent::FluentLogEventRouter.build(root_agent)
+      assert_equal Fluent::FluentLogEventRouter, d.class
+    end
+
+    test 'FluentLogEventRouter if <label @FLUENT_LOG> exists in config' do
+      root_agent = Fluent::RootAgent.new(log: $log, system_config: Fluent::SystemConfig.new)
+      root_agent.configure(build_config(<<-CONFIG))
+       <label @FLUENT_LOG>
+         <match *>
+           @type null
+         </match>
+        </label>
+      CONFIG
+
+      d = Fluent::FluentLogEventRouter.build(root_agent)
+      assert_equal Fluent::FluentLogEventRouter, d.class
+    end
+  end
+
+  test 'when calling graceful_stop, it flushes all events' do
+    event_router = []
+    stub(event_router).emit do |tag, time, record|
+      event_router.push([tag, time, record])
+    end
+
+    d = Fluent::FluentLogEventRouter.new(event_router)
+
+    t = Time.now
+    msg = ['tag', t, { 'key' => 'value' }]
+    d.emit_event(msg)
+    d.graceful_stop
+    d.emit_event(msg)
+    d.start
+
+    d.graceful_stop # to call join
+    assert_equal 2, event_router.size
+    assert_equal msg, event_router[0]
+    assert_equal msg, event_router[1]
+  end
+
+  test 'when calling stop, it ignores existing events' do
+    event_router = []
+    stub(event_router).emit do |tag, time, record|
+      event_router.push([tag, time, record])
+    end
+
+    d = Fluent::FluentLogEventRouter.new(event_router)
+
+    t = Time.now
+    msg = ['tag', t, { 'key' => 'value' }]
+    d.emit_event(msg)
+    d.stop
+    d.emit_event(msg)
+    d.start
+
+    d.stop # to call join
+    assert_equal 1, event_router.size
+    assert_equal msg, event_router[0]
+  end
+end

From 0970e78b9f17b5fcccc2524f529ceba573944b90 Mon Sep 17 00:00:00 2001
From: Yuta Iwama <ganmacs@gmail.com>
Date: Mon, 28 Oct 2019 12:42:24 +0900
Subject: [PATCH 12/14] Add emittable to needs of enable_event or not

Signed-off-by: Yuta Iwama <ganmacs@gmail.com>
---
 lib/fluent/engine.rb                  | 2 +-
 lib/fluent/fluent_log_event_router.rb | 4 ++++
 2 files changed, 5 insertions(+), 1 deletion(-)

diff --git a/lib/fluent/engine.rb b/lib/fluent/engine.rb
index eae3f85d82..f96a2dd1f0 100644
--- a/lib/fluent/engine.rb
+++ b/lib/fluent/engine.rb
@@ -107,7 +107,7 @@ def configure(conf)
 
       @fleunt_log_event_router = FluentLogEventRouter.build(@root_agent)
 
-      if @fleunt_log_event_router
+      if @fleunt_log_event_router.emittable?
         $log.enable_event(true)
       end
 
diff --git a/lib/fluent/fluent_log_event_router.rb b/lib/fluent/fluent_log_event_router.rb
index 4029c95917..9bcbc7be1c 100644
--- a/lib/fluent/fluent_log_event_router.rb
+++ b/lib/fluent/fluent_log_event_router.rb
@@ -26,6 +26,10 @@ def stop; end
     def graceful_stop; end
 
     def emit_event; end
+
+    def emittable?
+      self.class != NullFluentLogEventRouter
+    end
   end
 
   # This class is for handling fluentd's inner log

From 137f6b45f52bd0d04cb677425194eb7afb6b89af Mon Sep 17 00:00:00 2001
From: Yuta Iwama <ganmacs@gmail.com>
Date: Mon, 28 Oct 2019 15:16:03 +0900
Subject: [PATCH 13/14] Add argument to match with FluentLogEventRouter's one

Signed-off-by: Yuta Iwama <ganmacs@gmail.com>
---
 lib/fluent/fluent_log_event_router.rb | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/lib/fluent/fluent_log_event_router.rb b/lib/fluent/fluent_log_event_router.rb
index 9bcbc7be1c..d6bb1c1bdb 100644
--- a/lib/fluent/fluent_log_event_router.rb
+++ b/lib/fluent/fluent_log_event_router.rb
@@ -25,7 +25,7 @@ def stop; end
 
     def graceful_stop; end
 
-    def emit_event; end
+    def emit_event(_event); end
 
     def emittable?
       self.class != NullFluentLogEventRouter

From 65b7429c03cc7fc9eead11a506338d9859c44f5e Mon Sep 17 00:00:00 2001
From: Yuta Iwama <ganmacs@gmail.com>
Date: Tue, 29 Oct 2019 10:52:22 +0900
Subject: [PATCH 14/14] Add tests for NullFluentLogEventRouter

Signed-off-by: Yuta Iwama <ganmacs@gmail.com>
---
 test/test_fluent_log_event_router.rb | 11 +++++++++++
 1 file changed, 11 insertions(+)

diff --git a/test/test_fluent_log_event_router.rb b/test/test_fluent_log_event_router.rb
index 8102794d0c..5a70cf2cca 100644
--- a/test/test_fluent_log_event_router.rb
+++ b/test/test_fluent_log_event_router.rb
@@ -9,6 +9,17 @@ def build_config(config)
     Fluent::Config.parse(config, 'fluent_log_event', '', syntax: :v1)
   end
 
+  sub_test_case 'NullFluentLogEventRouter does nothing' do
+    test 'emittable? returns false but others does nothing' do
+      null_event_router = Fluent::NullFluentLogEventRouter.new
+      null_event_router.start
+      null_event_router.stop
+      null_event_router.graceful_stop
+      null_event_router.emit_event(nil)
+      assert_false null_event_router.emittable?
+    end
+  end
+
   sub_test_case '#build' do
     test 'NullFluentLogEventRouter if root_agent have not internal logger' do
       root_agent = Fluent::RootAgent.new(log: $log, system_config: Fluent::SystemConfig.new)