Add an option to disable shared socket forcedly

Fluentd always creates ServerEngine::SocketManager's shared socket even
if it's not required. The shared socket isn't needed if there is no
plugin that uses it such as in_forward, in_http and in_syslog.
Although it can be disabled if Fluentd doesn't use multiple workers,
this patch adds `disable_shared_socket` instead of checking number of
workers to keep backward compatibility for third-party plugins
(PluginHelper::Server.create_server_connection uses shared socket by
default).

Signed-off-by: Takuro Ashie <ashie@clear-code.com>

lib/fluent/command/fluentd.rb

@@ -163,6 +163,10 @@
   opts[:conf_encoding] = s
 }
 
+op.on('--disable-shared-socket', "Don't open shared socket for multiple workers") { |b|
+  opts[:disable_shared_socket] = b
+}
+
 if Fluent.windows?
   require 'windows/library'
   include Windows::Library

lib/fluent/supervisor.rb

@@ -45,6 +45,7 @@ module Fluent
   module ServerModule
     def before_run
       @fluentd_conf = config[:fluentd_conf]
+      @rpc_endpoint = nil
       @rpc_server = nil
       @counter = nil
 
@@ -64,9 +65,13 @@ def before_run
         run_counter_server(counter)
       end
 
-      socket_manager_path = ServerEngine::SocketManager::Server.generate_path
-      ServerEngine::SocketManager::Server.open(socket_manager_path)
-      ENV['SERVERENGINE_SOCKETMANAGER_PATH'] = socket_manager_path.to_s
+      if config[:disable_shared_socket]
+        $log.info "shared socket for multiple workers is disabled"
+      else
+        socket_manager_path = ServerEngine::SocketManager::Server.generate_path
+        ServerEngine::SocketManager::Server.open(socket_manager_path)
+        ENV['SERVERENGINE_SOCKETMANAGER_PATH'] = socket_manager_path.to_s
+      end
     end
 
     def after_run
@@ -452,6 +457,7 @@ def self.load_config(path, params = {})
         config_path: path,
         main_cmd: params['main_cmd'],
         signame: params['signame'],
+        disable_shared_socket: params['disable_shared_socket']
       }
       if daemonize
         se_config[:pid_path] = pid_path
@@ -567,7 +573,8 @@ def self.default_options
         supervise: true,
         standalone_worker: false,
         signame: nil,
-        conf_encoding: 'utf-8'
+        conf_encoding: 'utf-8',
+        disable_shared_socket: nil
       }
     end
 
@@ -795,6 +802,7 @@ def supervise
         'counter_server' => @system_config.counter_server,
         'log_format' => @system_config.log.format,
         'log_time_format' => @system_config.log.time_format,
+        'disable_shared_socket' => @system_config.disable_shared_socket
       }
 
       se = ServerEngine.create(ServerModule, WorkerModule){

lib/fluent/system_config.rb

@@ -27,7 +27,7 @@ class SystemConfig
       :log_event_verbose, :ignore_repeated_log_interval, :ignore_same_log_interval,
       :without_source, :rpc_endpoint, :enable_get_dump, :process_name,
       :file_permission, :dir_permission, :counter_server, :counter_client,
-      :strict_config_value, :enable_msgpack_time_support
+      :strict_config_value, :enable_msgpack_time_support, :disable_shared_socket
     ]
 
     config_param :workers,   :integer, default: 1
@@ -45,6 +45,7 @@ class SystemConfig
     config_param :process_name,    :string, default: nil
     config_param :strict_config_value, :bool, default: nil
     config_param :enable_msgpack_time_support, :bool, default: nil
+    config_param :disable_shared_socket, :bool, default: nil
     config_param :file_permission, default: nil do |v|
       v.to_i(8)
     end

test/command/test_fluentd.rb

@@ -1065,4 +1065,34 @@ def multi_workers_ready?; true; end
                          "secret xxxxxx", patterns_not_match: ["secret secret0", "secret secret1"])
     end
   end
+
+  sub_test_case 'sahred socket options' do
+    test 'enable shared socket by default' do
+      conf = ""
+      conf_path = create_conf_file('empty.conf', conf)
+      assert File.exist?(conf_path)
+      assert_log_matches(create_cmdline(conf_path),
+                         patterns_not_match: ["shared socket for multiple workers is disabled"])
+    end
+
+    test 'disable shared socket by command line option' do
+      conf = ""
+      conf_path = create_conf_file('empty.conf', conf)
+      assert File.exist?(conf_path)
+      assert_log_matches(create_cmdline(conf_path, "--disable-shared-socket"),
+                         "shared socket for multiple workers is disabled",)
+    end
+
+    test 'disable shared socket by system config' do
+      conf = <<CONF
+<system>
+  disable_shared_socket
+</system>
+CONF
+      conf_path = create_conf_file('empty.conf', conf)
+      assert File.exist?(conf_path)
+      assert_log_matches(create_cmdline(conf_path, "--disable-shared-socket"),
+                         "shared socket for multiple workers is disabled",)
+    end
+  end
 end

test/test_supervisor.rb

@@ -517,6 +517,31 @@ def test_log_level_affects
     assert_equal Fluent::Log::LEVEL_ERROR, $log.level
   end
 
+  def test_enable_shared_socket
+    server = DummyServer.new
+    begin
+      server.before_run
+      assert_not_nil(ENV['SERVERENGINE_SOCKETMANAGER_PATH'])
+    ensure
+      server.after_run
+    end
+  end
+
+  def test_disable_shared_socket
+    server = DummyServer.new
+    def server.config
+      {
+        :disable_shared_socket => true,
+      }
+    end
+    begin
+      server.before_run
+      assert_nil(ENV['SERVERENGINE_SOCKETMANAGER_PATH'])
+    ensure
+      server.after_run
+    end
+  end
+
   def create_debug_dummy_logger
     dl_opts = {}
     dl_opts[:log_level] = ServerEngine::DaemonLogger::DEBUG