mark multi-workers-ready plugins

lib/fluent/plugin/in_dummy.rb

@@ -67,6 +67,10 @@ def configure(conf)
       @storage = storage_create(usage: 'suspend', conf: config, default_type: DEFAULT_STORAGE_TYPE)
     end
 
+    def multi_workers_ready?
+      true
+    end
+
     def start
       super
 

lib/fluent/plugin/in_exec.rb

@@ -64,6 +64,10 @@ def configure(conf)
       @parser = parser_create
     end
 
+    def multi_workers_ready?
+      true
+    end
+
     def start
       super
 

lib/fluent/plugin/in_forward.rb

@@ -139,6 +139,10 @@ def configure(conf)
       @lsock = @usock = nil
     end
 
+    def multi_workers_ready?
+      true
+    end
+
     HEARTBEAT_UDP_PAYLOAD = "\0"
 
     def start
@@ -147,7 +151,7 @@ def start
       server_create_connection(
         :in_forward_server, @port,
         bind: @bind,
-        shared: false,
+        shared: true,
         resolve_name: @resolve_hostname,
         linger_timeout: @linger_timeout,
         backlog: @backlog,

lib/fluent/plugin/in_gc_stat.rb

@@ -33,6 +33,10 @@ def configure(conf)
       super
     end
 
+    def multi_workers_ready?
+      true
+    end
+
     def start
       super
 

lib/fluent/plugin/in_http.rb

@@ -111,6 +111,10 @@ def on_timer
       end
     end
 
+    def multi_workers_ready?
+      true
+    end
+
     def start
       @_event_loop_run_timeout = @blocking_timeout
 

lib/fluent/plugin/in_monitor_agent.rb

@@ -211,6 +211,10 @@ def process(req, res)
       end
     end
 
+    def multi_workers_ready?
+      true
+    end
+
     def start
       super
 

lib/fluent/plugin/in_object_space.rb

@@ -14,7 +14,6 @@
 #    limitations under the License.
 #
 
-require 'cool.io'
 require 'yajl'
 
 require 'fluent/plugin/input'
@@ -33,6 +32,10 @@ def initialize
     config_param :tag, :string
     config_param :top, :integer, default: 15
 
+    def multi_workers_ready?
+      true
+    end
+
     def start
       super
 

lib/fluent/plugin/in_syslog.rb

@@ -110,6 +110,10 @@ def configure(conf)
       @_event_loop_run_timeout = @blocking_timeout
     end
 
+    def multi_workers_ready?
+      true
+    end
+
     def start
       super
 

lib/fluent/plugin/in_tcp.rb

@@ -48,6 +48,10 @@ def configure(conf)
       @parser = parser_create
     end
 
+    def multi_workers_ready?
+      true
+    end
+
     def start
       super
 

lib/fluent/plugin/in_udp.rb

@@ -47,6 +47,10 @@ def configure(conf)
       @parser = parser_create
     end
 
+    def multi_workers_ready?
+      true
+    end
+
     def start
       super
 

lib/fluent/plugin/out_copy.rb

@@ -25,6 +25,10 @@ class CopyOutput < MultiOutput
     desc 'If true, pass different record to each `store` plugin.'
     config_param :deep_copy, :bool, default: false
 
+    def multi_workers_ready?
+      true
+    end
+
     def process(tag, es)
       unless es.repeatable?
         m = Fluent::MultiEventStream.new

lib/fluent/plugin/out_exec.rb

@@ -51,6 +51,10 @@ def configure(conf)
       @formatter = formatter_create
     end
 
+    def multi_workers_ready?
+      true
+    end
+
     NEWLINE = "\n"
 
     def format(tag, time, record)

lib/fluent/plugin/out_exec_filter.rb

@@ -172,6 +172,10 @@ def configure(conf)
       @next_log_time = Time.now.to_i
     end
 
+    def multi_workers_ready?
+      true
+    end
+
     ExecutedProcess = Struct.new(:mutex, :pid, :respawns, :readio, :writeio)
 
     def start

lib/fluent/plugin/out_file.rb

@@ -161,6 +161,11 @@ def configure(conf)
       @file_perm = system_config.file_permission || FILE_PERMISSION
     end
 
+    def multi_workers_ready?
+      ### TODO: add hack to synchronize for multi workers
+      true
+    end
+
     def format(tag, time, record)
       r = inject_values_to_record(tag, time, record)
       @formatter.format(tag, time, r)

lib/fluent/plugin/out_forward.rb

@@ -167,6 +167,10 @@ def configure(conf)
       raise Fluent::ConfigError, "ack_response_timeout must be a positive integer" if @ack_response_timeout < 1
     end
 
+    def multi_workers_ready?
+      true
+    end
+
     def prefer_delayed_commit
       @require_ack_response
     end

lib/fluent/plugin/out_null.rb

@@ -43,6 +43,10 @@ def initialize
       @feed_proc = nil
     end
 
+    def multi_workers_ready?
+      true
+    end
+
     def process(tag, es)
       # Do nothing
     end

lib/fluent/plugin/out_relabel.rb

@@ -21,6 +21,10 @@ class RelabelOutput < Output
     Fluent::Plugin.register_output('relabel', self)
     helpers :event_emitter
 
+    def multi_workers_ready?
+      true
+    end
+
     def process(tag, es)
       router.emit_stream(tag, es)
     end

lib/fluent/plugin/out_roundrobin.rb

@@ -42,6 +42,10 @@ def configure(conf)
       @rand_seed = Random.new.seed
     end
 
+    def multi_workers_ready?
+      true
+    end
+
     def start
       super
       rebuild_weight_array

lib/fluent/plugin/out_secondary_file.rb

@@ -65,6 +65,11 @@ def configure(conf)
       @file_perm = system_config.file_permission || FILE_PERMISSION
     end
 
+    def multi_workers_ready?
+      ### TODO: add hack to synchronize for multi workers
+      true
+    end
+
     def write(chunk)
       path_without_suffix = extract_placeholders(@path_without_suffix, chunk.metadata)
       path = generate_path(path_without_suffix)

lib/fluent/plugin/out_stdout.rb

@@ -41,6 +41,10 @@ def prefer_buffered_processing
       false
     end
 
+    def multi_workers_ready?
+      true
+    end
+
     attr_accessor :formatter
 
     def configure(conf)
@@ -52,6 +56,7 @@ def configure(conf)
     end
 
     def process(tag, es)
+      es = inject_values_to_event_stream(tag, es)
       es.each {|time,record|
         $log.write(format(tag, time, record))
       }