Add worker_id collision detector

Signed-off-by: Hiroshi Hatake <[email protected]>

lib/fluent/root_agent.rb

@@ -64,6 +64,7 @@ def initialize(log:, system_config: SystemConfig.new)
     attr_reader :labels
 
     def configure(conf)
+      used_workers_ids = []
       # initialize <worker> elements
       conf.elements(name: 'worker').each do |e|
         target_worker_id_str = e.arg
@@ -83,6 +84,10 @@ def configure(conf)
             if target_worker_id < 0 || target_worker_id > (Fluent::Engine.system_config.workers - 1)
               raise Fluent::ConfigError, "worker id #{target_worker_id} specified by <worker> directive is not allowed. Available worker id is between 0 and #{(Fluent::Engine.system_config.workers - 1)}"
             end
+            if used_workers_ids.include?(worker_id)
+              raise Fluent::ConfigError, "specified worker_id<#{worker_id}> collisions is detected on <worker> directive. Available worker id is between 0 and #{(Fluent::Engine.system_config.workers - 1)} without collisions"
+            end
+            used_workers_ids << worker_id
 
             e.elements.each do |elem|
               unless ['source', 'match', 'filter', 'label'].include?(elem.name)

test/test_root_agent.rb

@@ -697,6 +697,19 @@ def configure_ra(conf_str)
       end
     end
 
+    test 'raises configuration error for worker id collisions on multi workers syntax' do
+      errmsg = "specified worker_id<2> collisions is detected on <worker> directive. Available worker id is between 0 and 3 without collisions"
+      assert_raise Fluent::ConfigError.new(errmsg) do
+        conf = <<-EOC
+<worker 0-2>
+</worker>
+<worker 2-4>
+</worker>
+EOC
+        configure_ra(conf)
+      end
+    end
+
     test 'raises configuration error for too big worker id on invalid reversed multi workers syntax' do
       errmsg = "greater first_worker_id<3> than last_worker_id<0> specified by <worker> directive is not allowed. Available multi worker assign syntax is <smaller_worker_id>-<greater_worker_id>"
       assert_raise Fluent::ConfigError.new(errmsg) do