Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 5 additions & 7 deletions lib/fluent/plugin/base.rb
Original file line number Diff line number Diff line change
Expand Up @@ -53,14 +53,12 @@ def fluentd_worker_id
end

def configure(conf)
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)
raise ArgumentError, "BUG: type of conf must be Fluent::Config::Element, but #{conf.class} is passed." unless conf.is_a?(Fluent::Config::Element)

if conf.for_this_worker? || (Fluent::Engine.supervisor_mode && !conf.for_every_workers?)
system_config_override(workers: conf.target_worker_ids.size)
end

super(conf, system_config.strict_config_value)
@_state ||= State.new(false, false, false, false, false, false, false, false, false)
@_state.configure = true
Expand Down
98 changes: 98 additions & 0 deletions test/plugin/test_base.rb
Original file line number Diff line number Diff line change
Expand Up @@ -146,4 +146,102 @@ class FluentPluginBaseTest::DummyPlugin2 < Fluent::Plugin::TestBase
end
end
end

test '`ArgumentError` when `conf` is not `Fluent::Config::Element`' do
assert_raise ArgumentError.new('BUG: type of conf must be Fluent::Config::Element, but Hash is passed.') do
@p.configure({})
end
end

sub_test_case 'system_config.workers value after configure' do
def assert_system_config_workers_value(data)
conf = config_element()
conf.set_target_worker_ids(data[:target_worker_ids])
@p.configure(conf)
assert{ @p.system_config.workers == data[:expected] }
end

def stub_supervisor_mode
stub(Fluent::Engine).supervisor_mode { true }
stub(Fluent::Engine).worker_id { -1 }
end

sub_test_case 'with <system> workers 3 </system>' do
setup do
system_config = Fluent::SystemConfig.new
system_config.workers = 3
stub(Fluent::Engine).system_config { system_config }
end

data(
'without <worker> directive',
{
target_worker_ids: [],
expected: 3
},
keep: true
)
data(
'with <worker 0>',
{
target_worker_ids: [0],
expected: 1
},
keep: true
)
data(
'with <worker 0-1>',
{
target_worker_ids: [0, 1],
expected: 2
},
keep: true
)
data(
'with <worker 0-2>',
{
target_worker_ids: [0, 1, 2],
expected: 3
},
keep: true
)

test 'system_config.workers value after configure' do
assert_system_config_workers_value(data)
end

test 'system_config.workers value after configure with supervisor_mode' do
stub_supervisor_mode
assert_system_config_workers_value(data)
end
end

sub_test_case 'without <system> directive' do
data(
'without <worker> directive',
{
target_worker_ids: [],
expected: 1
},
keep: true
)
data(
'with <worker 0>',
{
target_worker_ids: [0],
expected: 1
},
keep: true
)

test 'system_config.workers value after configure' do
assert_system_config_workers_value(data)
end

test 'system_config.workers value after configure with supervisor_mode' do
stub_supervisor_mode
assert_system_config_workers_value(data)
end
end
end
end