Fix value of system_config.workers at run_configure

[abetomo]

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

What this PR does / why we need it:

Override system_config only if conf.target_worker_ids.size >= 1.

• If <worker N> or <worker 0-N> is not set, conf.target_worker_ids is empty.
• If <system> workers N </system> is not set, system_config.workers defaults to 1.

Docs Changes:

Release Note:

• Fix value of system_config.workers at run_configure
  • (Fix bug that system_config.workers value can be wrongly 1 at config check at startup.)
• Change argument type of Fluent::Plugin::Base::configure() to Fluent::Config::Element only.
  • (Note: This doesn't change the specification. It was a bug that a variable whose type is not Fluent::Config::Element could be passed to Fluent::Plugin::Base::configure().)

[daipom]

Thanks for the fix!

The status of this PR is Draft. Please let me know which part is TODO.

[abetomo]

Thanks for your comments!
I will add some tests.

[daipom]

I will add some tests.

Thanks! We need tests for this issue!

In addition, this problem is difficult. Hard to know what was wrong and how we should fix it.
I want to clarify it.

The specification history related to this is as follows.

• Add <worker n> section to set a configuration for a specific worker #1507
• Support multi workers assign syntax on <worker> section. Fix #2289 #2292

What went wrong in which process?

[abetomo]

#2651 is related.

If there is no <worker> directive, conf.for_this_worker? is false.
However, if Fluent::Engine.supervisor_mode is true, it will overwrite system_config.workers regardless of conf.for_this_worker?.

Since conf.target_worker_ids is empty when there is no <worker> directive, this fix checks for it and does not override system_config.workers.

[daipom]

Thanks so much!
It certainly seems that #2651 has something to do with this problem.

There was nothing wrong at the point of #1507 and #2292.
(The method of rewriting the value of the system config in the process seems a bit too complicated, but I guess it was inevitable since SystemConfig.workers was used to determine if the plugin works in the multi-worker or not.
It seems to me that this complicated specification could be the cause of bugs in the future..., anyway, it is not directly related to this issue.)

I will see #2651!

[daipom]

It is hard to understand the purpose of code changes in #2651...

I'm concerned about the following points.

• About Support multi workers assign syntax on <worker> section. Fix #2289 #2292
  • If conf.for_this_worker? is true, then conf.target_worker_ids appears to have at least one id. Why was the condition if conf.target_worker_ids && !conf.target_worker_ids.empty? needed?
    • https://github.com/fluent/fluentd/pull/2292/files#diff-601c8bc645b7008fa857b253a5295ef4b796c6236d0428ed8d72cecd572be6edR243-R245
    • https://github.com/fluent/fluentd/pull/2292/files#diff-a011588d74f931c0c2154891220af14f9181f28eb70b978d2e017866dfba3081R54-R59
  • It seems to me that the following code is enough at this point.

  if conf.for_this_worker?
    system_config_override(workers: conf.target_worker_ids.size)
  end

• About Fix dry run mode #2651
  • Why does Fluent::Engine::run_configure() run before launching the supervisor process even when dry_run is false?
    • Shouldn't it be run only when dry_run is true?
    • https://github.com/fluent/fluentd/pull/2651/files#diff-6042a07157b33b2d729ab3d10ed6bba89ad5f71aaed2be1aeb22a179a45ebbaeR499-R508
  • When Fluent::Engine.supervisor_mode is true, what is the expected value of workers?
    • I think we have to validate the config in one process in this case, but how many workers each plugin will work with will change depending on the setting.
      For example, plugin-A works with worker-1 and worker-2, and plugin-B works with only worker-3.
      With the current structure, would it be difficult to properly reflect and validate the number of workers?

I don't think all the points should be solved now.
I just noted points of concern as I work on this issue.
If you have noticed anything in the process of making this fix, I'd like to know.

The current fix may be enough to solve the issue #4051, but it may be better to leave some code comments for the future.

[abetomo]

Thanks for your comment!
I'll look into it.

[abetomo]

• About Support multi workers assign syntax on <worker> section. Fix #2289 #2292

As for #2292 condition, I too think @daipom's code is enough.

[daipom]

Thanks!

When Fluent::Engine.supervisor_mode is true, what is the expected value of workers?
* I think we have to validate the config in one process in this case, but how many workers each plugin will work with will change depending on the setting.
For example, plugin-A works with worker-1 and worker-2, and plugin-B works with only worker-3.
With the current structure, would it be difficult to properly reflect and validate the number of workers?

I have checked this, and I understand this fix allows us to validate the config with the correct workers value!

I used this config.

<system>
  workers 8
</system>

<worker 0>
  <source>
    @id A
    @type sample
    tag test
  </source>
</worker>

<worker 1-2>
  <source>
    @id B
    @type sample
    tag test
  </source>
</worker>

<source>
  @id C
  @type sample
  tag test
</source>

<match test.**>
  @type stdout
</match>

I added a log for debugging to in_sample.

--- a/lib/fluent/plugin/in_sample.rb
+++ b/lib/fluent/plugin/in_sample.rb
@@ -63,6 +63,7 @@ module Fluent::Plugin
 
     def configure(conf)
       super
+      log.info id: @id, workers: system_config.workers
       @sample_index = 0
       config = conf.elements.select{|e| e.name == 'storage' }.first

The result of master branch:

2023-02-24 00:20:14 +0900 [info]: init supervisor logger path=nil rotate_age=nil rotate_size=nil
2023-02-24 00:20:14 +0900 [info]: parsing config file is succeeded path="../work/fluent.conf"
...
2023-02-24 00:20:14 +0900 [info]: [C]  id="C" workers=1
2023-02-24 00:20:14 +0900 [info]: [A]  id="A" workers=1
2023-02-24 00:20:14 +0900 [info]: [B]  id="B" workers=2
...
2023-02-24 00:20:14 +0900 [info]: starting fluentd-1.15.3 pid=1286518 ruby="3.2.0"
...
2023-02-24 00:20:15 +0900 [info]: #7 [C]  id="C" workers=8
...
2023-02-24 00:20:15 +0900 [info]: #1 [B]  id="B" workers=2
...
2023-02-24 00:20:15 +0900 [info]: #0 [A]  id="A" workers=1
...

Before starting Fluentd, workers value of C is wrong.

The expected workers values before stating Fluentd are as follows.

[C]  id="C" workers=8
[A]  id="A" workers=1
[B]  id="B" workers=2

When applying this fix, these values are as expected!

[daipom]

Summarize what we've learned so far.

• About Support multi workers assign syntax on <worker> section. Fix #2289 #2292

As for #2292 condition, I too think @daipom's code is enough.

fluentd/lib/fluent/config/element.rb

Lines 256 to 258 in 6c649d1

	def for_this_worker?
	@target_worker_ids.include?(Fluent::Engine.worker_id)
	end

fluentd/lib/fluent/plugin/base.rb

Lines 56 to 61 in 6c649d1

	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

Originally, there was no need to overwrite 1.

fluentd/lib/fluent/engine.rb

Lines 213 to 216 in 6c649d1

	def worker_id
	if @supervisor_mode
	return -1
	end

fluentd/lib/fluent/plugin/base.rb

Line 56 in 6c649d1

if Fluent::Engine.supervisor_mode || (conf.respond_to?(:for_this_worker?) && conf.for_this_worker?)

We need Fluent::Engine.supervisor_mode condition to validate the config before starting Fluentd.

However, with supervisor_mode == true, the current logic overwrites the workers of all plugins that are not inside the worker directive.
This is the bug we have to fix.

After starting Fluentd, the for_this_worker? condition prevented this bug, so it was hard to notice.

This PR completely fixes this problem.

[abetomo]

conf of Fluent::TextParser.configure(conf) is Hash.
Thus, I added a conf.is_a?(Fluent::Config::Element) check.

!conf.target_worker_ids.empty? check is only necessary for Fluent::Engine.supervisor_mode == true.
So I changed to (Fluent::Engine.supervisor_mode && !conf.target_worker_ids.empty?).

[daipom]

Thanks for the fix and adding tests! I will see this soon.

[abetomo]

Thanks for the review!
I fixed the use of Hash in the test with another PR.
#4074

[daipom]

LGTM.
Thanks for fixing this difficult issue!

[ashie]

LGTM

[ashie]

Thanks!