Merge pull request #4066 from abetomo/fix-value-of-system_config_work…

…ers-at-run_configure

Fix value of `system_config.workers` at `run_configure`

lib/fluent/plugin/base.rb

@@ -53,14 +53,12 @@ def fluentd_worker_id
       end
 
       def configure(conf)
-        if Fluent::Engine.supervisor_mode || (conf.respond_to?(:for_this_worker?) && conf.for_this_worker?)
-          workers = if conf.target_worker_ids && !conf.target_worker_ids.empty?
-                      conf.target_worker_ids.size
-                    else
-                      1
-                    end
-          system_config_override(workers: workers)
+        raise ArgumentError, "BUG: type of conf must be Fluent::Config::Element, but #{conf.class} is passed." unless conf.is_a?(Fluent::Config::Element)
+
+        if conf.for_this_worker? || (Fluent::Engine.supervisor_mode && !conf.for_every_workers?)
+          system_config_override(workers: conf.target_worker_ids.size)
         end
+
         super(conf, system_config.strict_config_value)
         @_state ||= State.new(false, false, false, false, false, false, false, false, false)
         @_state.configure = true

test/plugin/test_base.rb

@@ -146,4 +146,102 @@ class FluentPluginBaseTest::DummyPlugin2 < Fluent::Plugin::TestBase
       end
     end
   end
+
+  test '`ArgumentError` when `conf` is not `Fluent::Config::Element`' do
+    assert_raise ArgumentError.new('BUG: type of conf must be Fluent::Config::Element, but Hash is passed.') do
+      @p.configure({})
+    end
+  end
+
+  sub_test_case 'system_config.workers value after configure' do
+    def assert_system_config_workers_value(data)
+      conf = config_element()
+      conf.set_target_worker_ids(data[:target_worker_ids])
+      @p.configure(conf)
+      assert{ @p.system_config.workers == data[:expected] }
+    end
+
+    def stub_supervisor_mode
+      stub(Fluent::Engine).supervisor_mode { true }
+      stub(Fluent::Engine).worker_id { -1 }
+    end
+
+    sub_test_case 'with <system> workers 3 </system>' do
+      setup do
+        system_config = Fluent::SystemConfig.new
+        system_config.workers = 3
+        stub(Fluent::Engine).system_config { system_config }
+      end
+
+      data(
+        'without <worker> directive',
+        {
+          target_worker_ids: [],
+          expected: 3
+        },
+        keep: true
+      )
+      data(
+        'with <worker 0>',
+        {
+          target_worker_ids: [0],
+          expected: 1
+        },
+        keep: true
+      )
+      data(
+        'with <worker 0-1>',
+        {
+          target_worker_ids: [0, 1],
+          expected: 2
+        },
+        keep: true
+      )
+      data(
+        'with <worker 0-2>',
+        {
+          target_worker_ids: [0, 1, 2],
+          expected: 3
+        },
+        keep: true
+      )
+
+      test 'system_config.workers value after configure' do
+        assert_system_config_workers_value(data)
+      end
+
+      test 'system_config.workers value after configure  with supervisor_mode' do
+        stub_supervisor_mode
+        assert_system_config_workers_value(data)
+      end
+    end
+
+    sub_test_case 'without <system> directive' do
+      data(
+        'without <worker> directive',
+        {
+          target_worker_ids: [],
+          expected: 1
+        },
+        keep: true
+      )
+      data(
+        'with <worker 0>',
+        {
+          target_worker_ids: [0],
+          expected: 1
+        },
+        keep: true
+      )
+
+      test 'system_config.workers value after configure' do
+        assert_system_config_workers_value(data)
+      end
+
+      test 'system_config.workers value after configure  with supervisor_mode' do
+        stub_supervisor_mode
+        assert_system_config_workers_value(data)
+      end
+    end
+  end
 end