From a23d1ea5ed63f7b1d455a02212a64977e537ccad Mon Sep 17 00:00:00 2001
From: Yuki Ito <yuki@gnnk.net>
Date: Fri, 17 Mar 2017 16:36:25 +0900
Subject: [PATCH] Add <worker n> section to set a configuration for a specific
 worker

---
 example/worker_section.conf  |  32 ++++++
 lib/fluent/agent.rb          |   7 +-
 lib/fluent/config/element.rb |  14 +++
 lib/fluent/engine.rb         |  29 +++--
 lib/fluent/plugin/base.rb    |   3 +
 lib/fluent/root_agent.rb     |  27 ++++-
 test/command/test_fluentd.rb | 204 ++++++++++++++++++++++++++++++++++-
 test/config/test_element.rb  |  25 +++++
 test/test_plugin_classes.rb  |  15 +++
 test/test_root_agent.rb      | 204 +++++++++++++++++++++++++++++++++++
 10 files changed, 546 insertions(+), 14 deletions(-)
 create mode 100644 example/worker_section.conf

diff --git a/example/worker_section.conf b/example/worker_section.conf
new file mode 100644
index 0000000000..f93962e802
--- /dev/null
+++ b/example/worker_section.conf
@@ -0,0 +1,32 @@
+<system>
+  workers 8
+  root_dir /path/fluentd/root
+</system>
+
+<source> # top-level sections works on all workers in parallel
+  @id edge_input
+  @type forward
+  @label @traffic
+  port 24224
+</source>
+
+<label @traffic> # this section works on all workers too
+  <match **>
+    @id backend_output
+    @type forward
+  </match>
+</label>
+
+<worker 0> # this section works only on first worker process
+  <source>
+    @id system_log_reader
+    @type tail
+    tag monitoring
+    path /path/now/watching/...
+  </source>
+  <match monitoring>
+    @id output_for_monitoring
+    @type elasticsearch
+    # ...
+  </match>
+</worker>
diff --git a/lib/fluent/agent.rb b/lib/fluent/agent.rb
index 42f25b2d41..345316cb63 100644
--- a/lib/fluent/agent.rb
+++ b/lib/fluent/agent.rb
@@ -62,6 +62,7 @@ def configure(conf)
 
       # initialize <match> and <filter> elements
       conf.elements('filter', 'match').each { |e|
+        next if e.has_target? && e.target_worker_id != Fluent::Engine.worker_id
         pattern = e.arg.empty? ? '**' : e.arg
         type = e['@type']
         raise ConfigError, "Missing '@type' parameter on <#{e.name}> directive" unless type
@@ -121,7 +122,8 @@ def lifecycle(desc: false)
     end
 
     def add_match(type, pattern, conf)
-      log.info :worker0, "adding match#{@context.nil? ? '' : " in #{@context}"}", pattern: pattern, type: type
+      log_type = conf.target_worker_id == Fluent::Engine.worker_id ? :default : :worker0
+      log.info log_type, "adding match#{@context.nil? ? '' : " in #{@context}"}", pattern: pattern, type: type
 
       output = Plugin.new_output(type)
       output.context_router = @event_router
@@ -142,7 +144,8 @@ def add_match(type, pattern, conf)
     end
 
     def add_filter(type, pattern, conf)
-      log.info :worker0, "adding filter#{@context.nil? ? '' : " in #{@context}"}", pattern: pattern, type: type
+      log_type = conf.target_worker_id == Fluent::Engine.worker_id ? :default : :worker0
+      log.info log_type, "adding filter#{@context.nil? ? '' : " in #{@context}"}", pattern: pattern, type: type
 
       filter = Plugin.new_filter(type)
       filter.context_router = @event_router
diff --git a/lib/fluent/config/element.rb b/lib/fluent/config/element.rb
index 0641eef7eb..ffc34b826c 100644
--- a/lib/fluent/config/element.rb
+++ b/lib/fluent/config/element.rb
@@ -35,10 +35,13 @@ def initialize(name, arg, attrs, elements, unused = nil)
 
         # it's global logger, not plugin logger: deprecated message should be global warning, not plugin level.
         @logger = defined?($log) ? $log : nil
+
+        @target_worker_id = nil
       end
 
       attr_accessor :name, :arg, :unused, :v1_config, :corresponding_proxies, :unused_in
       attr_writer :elements
+      attr_reader :target_worker_id
 
       RESERVED_PARAMETERS_COMPAT = {
         '@type' => 'type',
@@ -213,6 +216,17 @@ def self.unescape_parameter(v)
         v.each_char { |c| result << LiteralParser.unescape_char(c) }
         result
       end
+
+      def set_target_worker_id(worker_id)
+        @target_worker_id = worker_id
+        @elements.each { |e|
+          e.set_target_worker_id(worker_id)
+        }
+      end
+
+      def has_target?
+        !!@target_worker_id
+      end
     end
   end
 end
diff --git a/lib/fluent/engine.rb b/lib/fluent/engine.rb
index ea05954ab4..590d45edd2 100644
--- a/lib/fluent/engine.rb
+++ b/lib/fluent/engine.rb
@@ -31,6 +31,7 @@ def initialize
       @root_agent = nil
       @default_loop = nil
       @engine_stopped = false
+      @_worker_id = nil
 
       @log_event_router = nil
       @log_emit_thread = nil
@@ -97,12 +98,21 @@ def run_configure(conf)
                     else
                       "section <#{e.name}> is not used in <#{parent_name}>"
                     end
-          $log.warn :worker0, message
+          if !e.has_target?
+            $log.warn :worker0, message
+          elsif e.target_worker_id == worker_id
+            $log.warn message
+          end
           next
         end
         unless e.name == 'system'
           unless @without_source && e.name == 'source'
-            $log.warn :worker0, "parameter '#{key}' in #{e.to_s.strip} is not used."
+            message = "parameter '#{key}' in #{e.to_s.strip} is not used."
+            if !e.has_target?
+              $log.warn :worker0, message
+            elsif e.target_worker_id == worker_id
+              $log.warn message
+            end
           end
         end
       }
@@ -128,7 +138,7 @@ def configure(conf)
 
         unmatched_tags = Fluent::Log.event_tags.select{|t| !@log_event_router.match?(t) }
         unless unmatched_tags.empty?
-          $log.warn :worker0, "match for some tags of log events are not defined (to be ignored)", tags: unmatched_tags
+          $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
@@ -139,7 +149,7 @@ def configure(conf)
 
           unmatched_tags = Fluent::Log.event_tags.select{|t| !@log_event_router.match?(t) }
           unless unmatched_tags.empty?
-            $log.warn :worker0, "match for some tags of log events are not defined (to be ignored)", tags: unmatched_tags
+            $log.warn "match for some tags of log events are not defined (to be ignored)", tags: unmatched_tags
           end
         end
       end
@@ -200,10 +210,6 @@ def log_event_loop
     end
 
     def run
-      # if ENV doesn't have SERVERENGINE_WORKER_ID, it is a worker under --no-supervisor or in tests
-      # so it's (almost) a single worker, worker_id=0
-      worker_id = (ENV['SERVERENGINE_WORKER_ID'] || 0).to_i
-
       begin
         $log.info "starting fluentd worker", pid: Process.pid, ppid: Process.ppid, worker: worker_id
         start
@@ -251,8 +257,13 @@ def push_log_event(tag, time, record)
       @log_event_queue.push([tag, time, record])
     end
 
-    private
+    def worker_id
+      return @_worker_id if @_worker_id
+      @_worker_id = (ENV['SERVERENGINE_WORKER_ID'] || 0).to_i
+      @_worker_id
+    end
 
+    private
     def start
       @root_agent.start
     end
diff --git a/lib/fluent/plugin/base.rb b/lib/fluent/plugin/base.rb
index 4d8940364f..7f3332f859 100644
--- a/lib/fluent/plugin/base.rb
+++ b/lib/fluent/plugin/base.rb
@@ -51,6 +51,9 @@ def fluentd_worker_id
       end
 
       def configure(conf)
+        if conf.respond_to?(:target_worker_id) && conf.target_worker_id == Fluent::Engine.worker_id
+          system_config_override(workers: 1)
+        end
         super
         @_state ||= State.new(false, false, false, false, false, false, false, false, false)
         @_state.configure = true
diff --git a/lib/fluent/root_agent.rb b/lib/fluent/root_agent.rb
index c83f1df976..e6f5d20253 100644
--- a/lib/fluent/root_agent.rb
+++ b/lib/fluent/root_agent.rb
@@ -64,11 +64,34 @@ def initialize(log:, system_config: SystemConfig.new)
     attr_reader :labels
 
     def configure(conf)
+      # initialize <worker> elements
+      conf.elements(name: 'worker').each do |e|
+        target_worker_id_str = e.arg
+        if target_worker_id_str.empty?
+          raise ConfigError, "Missing worker id on <worker> directive"
+        end
+
+        target_worker_id = target_worker_id_str.to_i
+        if target_worker_id < 0 || target_worker_id > (Fluent::Engine.system_config.workers - 1)
+          raise ConfigError, "worker#{target_worker_id} specified by <worker> directive doesn't exist. Specify id between 0 and #{(Fluent::Engine.system_config.workers - 1)}"
+        end
+
+        e.elements.each do |elem|
+          unless ['source', 'match', 'filter', 'label'].include?(elem.name)
+            raise ConfigError, "<worker> section cannot have <#{elem.name}> directive"
+          end
+          elem.set_target_worker_id(target_worker_id)
+        end
+        conf += e
+      end
+      conf.elements.delete_if{|e| e.name == 'worker'}
+
       error_label_config = nil
 
       # initialize <label> elements before configuring all plugins to avoid 'label not found' in input, filter and output.
       label_configs = {}
       conf.elements(name: 'label').each { |e|
+        next if e.has_target? && e.target_worker_id != Fluent::Engine.worker_id
         name = e.arg
         raise ConfigError, "Missing symbol argument on <label> directive" if name.empty?
 
@@ -90,6 +113,7 @@ def configure(conf)
         log.info :worker0, "'--without-source' is applied. Ignore <source> sections"
       else
         conf.elements(name: 'source').each { |e|
+          next if e.has_target? && e.target_worker_id != Fluent::Engine.worker_id
           type = e['@type']
           raise ConfigError, "Missing '@type' parameter on <source> directive" unless type
           add_source(type, e)
@@ -235,7 +259,8 @@ def suppress_interval(interval_time)
     end
 
     def add_source(type, conf)
-      log.info :worker0, "adding source", type: type
+      log_type = conf.target_worker_id == Fluent::Engine.worker_id ? :default : :worker0
+      log.info log_type, "adding source", type: type
 
       input = Plugin.new_input(type)
       # <source> emits events to the top-level event router (RootAgent#event_router).
diff --git a/test/command/test_fluentd.rb b/test/command/test_fluentd.rb
index b3349f7eb1..8972fe9104 100644
--- a/test/command/test_fluentd.rb
+++ b/test/command/test_fluentd.rb
@@ -297,7 +297,7 @@ def assert_fluentd_fails_to_start(cmdline, *pattern_list, timeout: 10)
       assert_log_matches(
         create_cmdline(conf_path),
         "fluentd worker is now running",
-        '[warn]: match for some tags of log events are not defined (to be ignored) tags=["fluent.trace", "fluent.debug", "fluent.info"]',
+        '[warn]: #0 match for some tags of log events are not defined (to be ignored) tags=["fluent.trace", "fluent.debug", "fluent.info"]',
         '[warn]: #0 no patterns matched tag="fluent.info"',
       )
     end
@@ -334,7 +334,7 @@ def assert_fluentd_fails_to_start(cmdline, *pattern_list, timeout: 10)
       assert_log_matches(
         create_cmdline(conf_path),
         "fluentd worker is now running",
-        '[warn]: match for some tags of log events are not defined (to be ignored) tags=["fluent.info", "fluent.fatal"]',
+        '[warn]: #0 match for some tags of log events are not defined (to be ignored) tags=["fluent.info", "fluent.fatal"]',
         patterns_not_match: ['[warn]: no patterns matched tag="fluent.info"'],
       )
     end
@@ -614,5 +614,205 @@ def assert_fluentd_fails_to_start(cmdline, *pattern_list, timeout: 10)
         "config error file=\"#{conf_path}\" error_class=Fluent::ConfigError error=\"Plugin 'file' does not support multi workers configuration (Fluent::Plugin::FileBuffer)\"",
       )
     end
+
+    test 'failed to start workers when configured plugins as chidren of MultiOutput do not support multi worker configuration' do
+      script =  "require 'fluent/plugin/output'\n"
+      script << "module Fluent::Plugin\n"
+      script << "  class SingleOutput < Output\n"
+      script << "    Fluent::Plugin.register_output('single', self)\n"
+      script << "    def multi_workers_ready?\n"
+      script << "      false\n"
+      script << "    end\n"
+      script << "    def write(chunk)\n"
+      script << "    end\n"
+      script << "  end\n"
+      script << "end\n"
+      plugin_path = create_plugin_file('out_single.rb', script)
+
+      conf = <<CONF
+<system>
+  workers 2
+</system>
+<source>
+  @type single
+  @id single
+  @label @dummydata
+</source>
+<label @dummydata>
+  <match dummy>
+    @type copy
+    <store>
+      @type single
+    </store>
+    <store>
+      @type single
+    </store>
+  </match>
+</label>
+CONF
+      conf_path = create_conf_file('workers_invalid3.conf', conf)
+      assert_fluentd_fails_to_start(
+        create_cmdline(conf_path, "-p", File.dirname(plugin_path)),
+        "Plugin 'single' does not support multi workers configuration (Fluent::Plugin::SingleOutput)",
+      )
+    end
+
+    test 'success to start a worker with worker specific configuration' do
+      conf = <<CONF
+<system>
+  workers 2
+  root_dir #{@root_path}
+</system>
+<source>
+  @type dummy
+  @id dummy
+  @label @dummydata
+  tag dummy
+  dummy {"message": "yay!"}
+</source>
+<worker 1>
+  <source>
+    @type dummy
+    @id dummy_in_worker
+    @label @dummydata
+    tag dummy
+    dummy {"message": "yay!"}
+  </source>
+</worker>
+<label @dummydata>
+  <match dummy>
+    @type null
+    @id   blackhole
+  </match>
+</label>
+CONF
+      conf_path = create_conf_file('worker_section0.conf', conf)
+      assert Dir.exist?(@root_path)
+
+      assert_log_matches(
+        create_cmdline(conf_path),
+        "#0 fluentd worker is now running worker=0",
+        "#1 fluentd worker is now running worker=1",
+        /(?!#\d) adding source type="dummy"/,
+        '#1 adding source type="dummy"'
+      )
+    end
+
+    test 'success to start workers when configured plugins only for specific worker do not support multi worker configuration' do
+      script =  "require 'fluent/plugin/input'\n"
+      script << "module Fluent::Plugin\n"
+      script << "  class SingleInput < Input\n"
+      script << "    Fluent::Plugin.register_input('single', self)\n"
+      script << "    def multi_workers_ready?\n"
+      script << "      false\n"
+      script << "    end\n"
+      script << "  end\n"
+      script << "end\n"
+      plugin_path = create_plugin_file('in_single.rb', script)
+
+      conf = <<CONF
+<system>
+  workers 2
+</system>
+<worker 1>
+  <source>
+    @type single
+    @id single
+    @label @dummydata
+  </source>
+</worker>
+<label @dummydata>
+  <match dummy>
+    @type null
+    @id   blackhole
+  </match>
+</label>
+CONF
+      conf_path = create_conf_file('worker_section1.conf', conf)
+      assert Dir.exist?(@root_path)
+
+      assert_log_matches(
+        create_cmdline(conf_path, "-p", File.dirname(plugin_path)),
+        "#0 fluentd worker is now running worker=0",
+        "#1 fluentd worker is now running worker=1",
+        '#1 adding source type="single"'
+      )
+    end
+
+    test 'success to start workers when file buffer is configured in non-workers way only for specific worker' do
+      conf = <<CONF
+<system>
+  workers 2
+</system>
+<source>
+  @type dummy
+  @id dummy
+  tag dummy
+  dummy {"message": "yay!"}
+</source>
+<worker 1>
+  <match dummy>
+    @type null
+    @id   blackhole
+    <buffer>
+      @type file
+      path #{File.join(@root_path, "buf", "file.*.log")}
+    </buffer>
+  </match>
+</worker>
+CONF
+      conf_path = create_conf_file('worker_section2.conf', conf)
+      assert_log_matches(
+        create_cmdline(conf_path),
+        "#0 fluentd worker is now running worker=0",
+        "#1 fluentd worker is now running worker=1",
+        '#1 adding match pattern="dummy" type="null"'
+      )
+    end
+
+    test 'success to start workers when configured plugins as a chidren of MultiOutput only for specific worker do not support multi worker configuration' do
+      script =  "require 'fluent/plugin/output'\n"
+      script << "module Fluent::Plugin\n"
+      script << "  class SingleOutput < Output\n"
+      script << "    Fluent::Plugin.register_output('single', self)\n"
+      script << "    def multi_workers_ready?\n"
+      script << "      false\n"
+      script << "    end\n"
+      script << "    def write(chunk)\n"
+      script << "    end\n"
+      script << "  end\n"
+      script << "end\n"
+      plugin_path = create_plugin_file('out_single.rb', script)
+
+      conf = <<CONF
+<system>
+  workers 2
+</system>
+<source>
+  @type dummy
+  @id dummy
+  tag dummy
+  dummy {"message": "yay!"}
+</source>
+<worker 1>
+  <match dummy>
+    @type copy
+    <store>
+      @type single
+    </store>
+    <store>
+      @type single
+    </store>
+  </match>
+</worker>
+CONF
+      conf_path = create_conf_file('worker_section3.conf', conf)
+      assert_log_matches(
+        create_cmdline(conf_path, "-p", File.dirname(plugin_path)),
+        "#0 fluentd worker is now running worker=0",
+        "#1 fluentd worker is now running worker=1",
+        '#1 adding match pattern="dummy" type="copy"'
+      )
+    end
   end
 end
diff --git a/test/config/test_element.rb b/test/config/test_element.rb
index c4765089cf..9287f2c5c3 100644
--- a/test/config/test_element.rb
+++ b/test/config/test_element.rb
@@ -400,4 +400,29 @@ def element(name = 'ROOT', arg = '', attrs = {}, elements = [], unused = nil)
       end
     end
   end
+
+  sub_test_case '#set_target_worker' do
+    test 'set target_worker_id recursively' do
+      e = element('label', '@mytest', {}, [ element('filter', '**'), element('match', '**', {}, [ element('store'), element('store') ]) ])
+      e.set_target_worker_id(1)
+      assert_equal 1, e.target_worker_id
+      assert_equal 1, e.elements[0].target_worker_id
+      assert_equal 1, e.elements[1].target_worker_id
+      assert_equal 1, e.elements[1].elements[0].target_worker_id
+      assert_equal 1, e.elements[1].elements[1].target_worker_id
+    end
+  end
+
+  sub_test_case '#has_target?' do
+    test 'has target_worker_id' do
+      e = element()
+      e.set_target_worker_id(1)
+      assert e.has_target?
+    end
+
+    test "don't have target_worker_id" do
+      e = element()
+      refute e.has_target?
+    end
+  end
 end
diff --git a/test/test_plugin_classes.rb b/test/test_plugin_classes.rb
index b2e484c306..12879881c8 100644
--- a/test/test_plugin_classes.rb
+++ b/test/test_plugin_classes.rb
@@ -235,6 +235,21 @@ def filter(tag, time, record)
     end
   end
 
+  class FluentTestBuffer < Fluent::Plugin::Buffer
+    ::Fluent::Plugin.register_buffer('test_buffer', self)
+
+    def resume
+      return {}, []
+    end
+
+    def generate_chunk(metadata)
+    end
+
+    def multi_workers_ready?
+      false
+    end
+  end
+
   class TestEmitErrorHandler
     def initialize
       @events = Hash.new { |h, k| h[k] = [] }
diff --git a/test/test_root_agent.rb b/test/test_root_agent.rb
index e467bd81b4..c1bd4f4882 100644
--- a/test/test_root_agent.rb
+++ b/test/test_root_agent.rb
@@ -608,4 +608,208 @@ def setup_root_agent(conf)
       assert_equal [true, true], dyn_out.child.outputs.map{|i| i.terminated? }
     end
   end
+
+  sub_test_case 'configured at worker2 with 4 workers environment' do
+    setup do
+      ENV['SERVERENGINE_WORKER_ID'] = '2'
+      @ra = RootAgent.new(log: $log)
+      system_config = SystemConfig.new
+      system_config.workers = 4
+      stub(Engine).worker_id { 2 }
+      stub(Engine).root_agent { @ra }
+      stub(Engine).system_config { system_config }
+      @ra
+    end
+
+    teardown '' do
+      ENV.delete('SERVERENGINE_WORKER_ID')
+    end
+
+    def configure_ra(conf_str)
+      conf = Config.parse(conf_str, "(test)", "(test_dir)", true)
+      @ra.configure(conf)
+      @ra
+    end
+
+    test 'raises configuration error for missing worker id' do
+      errmsg = 'Missing worker id on <worker> directive'
+      assert_raise Fluent::ConfigError.new(errmsg) do
+        conf = <<-EOC
+<worker>
+</worker>
+EOC
+        configure_ra(conf)
+      end
+    end
+
+    test 'raises configuration error for too big worker id' do
+      errmsg = "worker4 specified by <worker> directive doesn't exist. Specify id between 0 and 3"
+      assert_raise Fluent::ConfigError.new(errmsg) do
+        conf = <<-EOC
+<worker 4>
+</worker>
+EOC
+        configure_ra(conf)
+      end
+    end
+
+    test 'raises configuration error for invalid elements as a child of worker section' do
+      errmsg = '<worker> section cannot have <system> directive'
+      assert_raise Fluent::ConfigError.new(errmsg) do
+        conf = <<-EOC
+<worker 2>
+<system>
+</system>
+</worker>
+EOC
+        configure_ra(conf)
+      end
+    end
+
+    test 'raises configuration error when configured plugins do not have support multi worker configuration' do
+      errmsg = "Plugin 'test_out' does not support multi workers configuration (FluentTest::FluentTestOutput)"
+      assert_raise Fluent::ConfigError.new(errmsg) do
+        conf = <<-EOC
+<match **>
+@type test_out
+</match>
+EOC
+        configure_ra(conf)
+      end
+    end
+
+    test 'does not raise configuration error when configured plugins in worker section do not have support multi worker configuration' do
+      assert_nothing_raised do
+        conf = <<-EOC
+<worker 2>
+<match **>
+  @type test_out
+</match>
+</worker>
+EOC
+        configure_ra(conf)
+      end
+    end
+
+    test 'does not raise configuration error when configured plugins as a children of MultiOutput in worker section do not have support multi worker configuration' do
+      assert_nothing_raised do
+        conf = <<-EOC
+<worker 2>
+<match **>
+  @type copy
+  <store>
+    @type test_out
+  </store>
+  <store>
+    @type test_out
+  </store>
+</match>
+</worker>
+EOC
+        configure_ra(conf)
+      end
+    end
+
+    test 'does not raise configuration error when configured plugins owned by plugin do not have support multi worker configuration' do
+      assert_nothing_raised do
+        conf = <<-EOC
+<worker 2>
+<match **>
+  @type test_out_buffered
+  <buffer>
+    @type test_buffer
+  </buffer>
+</match>
+</worker>
+EOC
+        configure_ra(conf)
+      end
+    end
+
+    test 'with plugins' do
+      conf = <<-EOC
+<worker 2>
+<source>
+  @type test_in
+  @id test_in
+</source>
+<filter>
+  type test_filter
+  id test_filter
+</filter>
+<match **>
+  @type relabel
+  @id test_relabel
+  @label @test
+</match>
+<label @test>
+  <match **>
+    type test_out
+    id test_out
+  </match>
+</label>
+<label @ERROR>
+  <match>
+    @type null
+  </match>
+</label>
+</worker>
+EOC
+      ra = configure_ra(conf)
+      assert_kind_of FluentTestInput, ra.inputs.first
+      assert_kind_of Plugin::RelabelOutput, ra.outputs.first
+      assert_kind_of FluentTestFilter, ra.filters.first
+      assert ra.error_collector
+
+      %W(@test @ERROR).each { |label_symbol|
+        assert_include ra.labels, label_symbol
+        assert_kind_of Label, ra.labels[label_symbol]
+      }
+
+      test_label = ra.labels['@test']
+      assert_kind_of FluentTestOutput, test_label.outputs.first
+      assert_equal ra, test_label.root_agent
+
+      error_label = ra.labels['@ERROR']
+      assert_kind_of Fluent::Plugin::NullOutput, error_label.outputs.first
+      assert_kind_of RootAgent::RootAgentProxyWithoutErrorCollector, error_label.root_agent
+    end
+
+    test 'with plugins but for another worker' do
+      conf = <<-EOC
+<worker 0>
+<source>
+  @type test_in
+  @id test_in
+</source>
+<filter>
+  type test_filter
+  id test_filter
+</filter>
+<match **>
+  @type relabel
+  @id test_relabel
+  @label @test
+</match>
+<label @test>
+  <match **>
+    type test_out
+    id test_out
+  </match>
+</label>
+<label @ERROR>
+  <match>
+    @type null
+  </match>
+</label>
+</worker>
+EOC
+      ra = configure_ra(conf)
+      assert_equal 0, ra.inputs.size
+      assert_equal 0, ra.outputs.size
+      assert_equal 0, ra.filters.size
+      assert_equal 0, ra.labels.size
+      refute ra.error_collector
+    end
+  end
 end