out_file: Avoid performing multi-worker check in configure()

[fujimotos]

Which issue(s) this PR fixes:
Fixes #3941

What this PR does / why we need it:

The check did not work well with `<worker 0-N>` directive for
the following reasons:

 * When Fluentd starts up, supervisor performs sanity check on
   configuration by creating actual instances of plugins.

 * This check precedes SeverModule, which prepares various assets
   for workers (lockdir, RPC endpoints and shared sockets etc.)

 * Fluent::Plugin::Base dynamically overrides `system_config.workers`
   during start-up, and leaves the value > 1 only when `<worker 0-N>`
   directive is used.

Fix this issue by avoiding faulty sanity check in `configure()`.

Signed-off-by: Fujimoto Seiji <fujimoto@clear-code.com>

Docs Changes:

N/A

Release Note:

out_file: Fix the multi-worker check with <worker 0-N> directive

[fujimotos]

@ashie Can you check this PR before the next release? If it looks good,
I want to included this in td-agent v4.4.2 along with #3939.

[ashie]

• When Fluentd starts up, supervisor performs sanity check on
  configuration by creating actual instances of plugins.

fluentd/lib/fluent/supervisor.rb

Lines 727 to 736 in a777183

	begin
	ServerEngine::Privilege.change(@chuser, @chgroup)
	MessagePackFactory.init(enable_time_support: @system_config.enable_msgpack_time_support)
	Fluent::Engine.init(@system_config, supervisor_mode: true)
	Fluent::Engine.run_configure(@conf, dry_run: dry_run)
	rescue Fluent::ConfigError => e
	$log.error 'config error', file: @config_path, error: e
	$log.debug_backtrace
	exit!(1)
	end

• This check precedes SeverModule, which prepares various assets
  for workers (lockdir, RPC endpoints and shared sockets etc.)

fluentd/lib/fluent/supervisor.rb

Line 742 in a777183

supervise

fluentd/lib/fluent/supervisor.rb

Lines 898 to 900 in a777183

	se = ServerEngine.create(ServerModule, WorkerModule){
	Fluent::Supervisor.load_config(@config_path, params)
	}

fluentd/lib/fluent/supervisor.rb

Lines 41 to 48 in a777183

	def before_run
	@fluentd_conf = config[:fluentd_conf]
	@rpc_endpoint = nil
	@rpc_server = nil
	@counter = nil
	@fluentd_lock_dir = Dir.mktmpdir("fluentd-lock-")
	ENV['FLUENTD_LOCK_DIR'] = @fluentd_lock_dir

• Fluent::Plugin::Base dynamically overrides system_config.workers
  during start-up, and leaves the value > 1 only when <worker 0-N>
  directive is used.

fluentd/lib/fluent/plugin/base.rb

Lines 56 to 62 in a777183

	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)

[ashie]

• Fluent::Plugin::Base dynamically overrides system_config.workers
  during start-up, and leaves the value > 1 only when <worker 0-N>
  directive is used.

fluentd/lib/fluent/plugin/base.rb

Lines 56 to 62 in a777183

	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)

conf.target_worker_ids is set only when <worker 0-N> is used:

fluentd/lib/fluent/root_agent.rb

Lines 78 to 121 in a777183

	target_worker_ids = target_worker_id_str.split("-")
	if target_worker_ids.size == 2
	first_worker_id = target_worker_ids.first.to_i
	last_worker_id = target_worker_ids.last.to_i
	if first_worker_id > last_worker_id
	raise Fluent::ConfigError, "greater first_worker_id<#{first_worker_id}> than last_worker_id<#{last_worker_id}> specified by <worker> directive is not allowed. Available multi worker assign syntax is <smaller_worker_id>-<greater_worker_id>"
	end
	target_worker_ids = []
	first_worker_id.step(last_worker_id, 1) do |worker_id|
	target_worker_id = worker_id.to_i
	target_worker_ids << target_worker_id
	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
	available_worker_ids.delete(target_worker_id) if available_worker_ids.include?(target_worker_id)
	if used_worker_ids.include?(target_worker_id)
	raise Fluent::ConfigError, "specified worker_id<#{worker_id}> collisions is detected on <worker> directive. Available worker id(s): #{available_worker_ids}"
	end
	used_worker_ids << target_worker_id
	e.elements.each do |elem|
	unless ['source', 'match', 'filter', 'label'].include?(elem.name)
	raise Fluent::ConfigError, "<worker> section cannot have <#{elem.name}> directive"
	end
	end
	unless target_worker_ids.empty?
	e.set_target_worker_ids(target_worker_ids.uniq)
	end
	end
	else
	target_worker_id = target_worker_id_str.to_i
	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
	e.elements.each do |elem|
	unless ['source', 'match', 'filter', 'label'].include?(elem.name)
	raise Fluent::ConfigError, "<worker> section cannot have <#{elem.name}> directive"
	end
	elem.set_target_worker_id(target_worker_id)
	end
	end

[ashie]

When <worker 0-N> isn't specified system_config.workers is 1 on sanity check of supervisor so that this check doesn't return error.

[ashie]

LGTM
We may need to address confusing system_config.workers issue later.

[ashie]

Thanks!

[daipom]

Thanks for the fix!