Merge pull request #1869 from eagletmt/multi-worker-debug

Support multi worker environment in in_debug_agent

lib/fluent/plugin/in_debug_agent.rb

@@ -34,13 +34,20 @@ def initialize
 
     def configure(conf)
       super
+      if system_config.workers > 1
+        @port += fluentd_worker_id
+      end
       if @unix_path
         unless ::Fluent::FileUtil.writable?(@unix_path)
           raise Fluent::ConfigError, "in_debug_agent: `#{@unix_path}` is not writable"
         end
       end
     end
 
+    def multi_workers_ready?
+      @unix_path.nil?
+    end
+
     def start
       super
 
@@ -50,7 +57,7 @@ def start
       else
         uri = "druby://#{@bind}:#{@port}"
       end
-      log.info "listening dRuby", uri: uri, object: @object
+      log.info "listening dRuby", uri: uri, object: @object, worker: fluentd_worker_id
       obj = eval(@object)
       @server = DRb::DRbServer.new(uri, obj)
     end

test/plugin/test_in_debug_agent.rb

@@ -25,4 +25,25 @@ def test_unix_path_writable
       create_driver %[unix_path #{TMP_DIR}/does_not_exist/test_path]
     end
   end
+
+  def test_multi_worker_environment_with_port
+    Fluent::SystemConfig.overwrite_system_config('workers' => 4) do
+      d = Fluent::Test::Driver::Input.new(Fluent::Plugin::DebugAgentInput)
+      d.instance.instance_eval { @_fluentd_worker_id = 2 }
+      d.configure('port 24230')
+
+      assert_true d.instance.multi_workers_ready?
+      assert_equal(24232, d.instance.instance_variable_get(:@port))
+    end
+  end
+
+  def test_multi_worker_environment_with_unix_path
+    Fluent::SystemConfig.overwrite_system_config('workers' => 4) do
+      d = Fluent::Test::Driver::Input.new(Fluent::Plugin::DebugAgentInput)
+      d.instance.instance_eval { @_fluentd_worker_id = 2 }
+      d.configure("unix_path #{TMP_DIR}/test_path")
+
+      assert_false d.instance.multi_workers_ready?
+    end
+  end
 end