Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

out_file/out_secondary_file: Support ${chunk_id} placeholder. fix #1705 #1708

Merged
merged 3 commits into from
Oct 13, 2017
Merged
Show file tree
Hide file tree
Changes from 1 commit
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
Prev Previous commit
Next Next commit
Change extract_placeholders API. Accept Chunk object in 2nd argument
  • Loading branch information
repeatedly committed Oct 10, 2017
commit 1bcad9a76fcc7f25609828dfa88b6ce8a91216cd
4 changes: 2 additions & 2 deletions lib/fluent/plugin/out_file.rb
Original file line number Diff line number Diff line change
Expand Up @@ -141,8 +141,8 @@ def configure(conf)
dummy_record_keys = get_placeholders_keys(@path_template) || ['message']
dummy_record = Hash[dummy_record_keys.zip(['data'] * dummy_record_keys.size)]

test_meta1 = metadata_for_test(dummy_tag, Fluent::Engine.now, dummy_record)
test_path = extract_placeholders(@path_template.gsub(CHUNK_ID_PLACEHOLDER_PATTERN, 'test'), test_meta1)
test_chunk1 = chunk_for_test(dummy_tag, Fluent::Engine.now, dummy_record)
test_path = extract_placeholders(@path_template, test_chunk1)
unless ::Fluent::FileUtil.writable_p?(test_path)
raise Fluent::ConfigError, "out_file: `#{test_path}` is not writable"
end
Expand Down
2 changes: 1 addition & 1 deletion lib/fluent/plugin/out_secondary_file.rb
Original file line number Diff line number Diff line change
Expand Up @@ -71,7 +71,7 @@ def multi_workers_ready?
end

def write(chunk)
path_without_suffix = extract_placeholders(@path_without_suffix.gsub(CHUNK_ID_PLACEHOLDER_PATTERN, dump_unique_id_hex(chunk.unique_id)), chunk.metadata)
path_without_suffix = extract_placeholders(@path_without_suffix, chunk)
path = generate_path(path_without_suffix)
FileUtils.mkdir_p File.dirname(path), mode: @dir_perm

Expand Down
33 changes: 30 additions & 3 deletions lib/fluent/plugin/output.rb
Original file line number Diff line number Diff line change
Expand Up @@ -684,9 +684,23 @@ def get_placeholders_keys(str)
end

# TODO: optimize this code
def extract_placeholders(str, metadata)
def extract_placeholders(str, chunk)
metadata = if chunk.is_a?(Fluent::Plugin::Buffer::Chunk)
chunk_passed = true
chunk.metadata
else
chunk_passed = false
# For existing plugins. Old plugin passes Chunk.metadata instead of Chunk
chunk
end
if metadata.empty?
str
str.sub(CHUNK_ID_PLACEHOLDER_PATTERN) {
if chunk_passed
dump_unique_id_hex(chunk.unique_id)
else
log.warn "${chunk_id} is not allowed in this plugin. Pass Chunk instead of metadata in extract_placeholders's 2nd argument"
end
}
else
rvalue = str.dup
# strftime formatting
Expand Down Expand Up @@ -721,7 +735,13 @@ def extract_placeholders(str, metadata)
if rvalue =~ CHUNK_KEY_PLACEHOLDER_PATTERN
log.warn "chunk key placeholder '#{$1}' not replaced. template:#{str}"
end
rvalue
rvalue.sub(CHUNK_ID_PLACEHOLDER_PATTERN) {
if chunk_passed
dump_unique_id_hex(chunk.unique_id)
else
log.warn "${chunk_id} is not allowed in this plugin. Pass Chunk instead of metadata in extract_placeholders's 2nd argument"
end
}
end
end

Expand Down Expand Up @@ -802,6 +822,13 @@ def metadata(tag, time, record)
end
end

def chunk_for_test(tag, time, record)
require 'fluent/plugin/buffer/memory_chunk'

m = metadata_for_test(tag, time, record)
Fluent::Plugin::Buffer::MemoryChunk.new(m)
end

def metadata_for_test(tag, time, record)
raise "BUG: #metadata_for_test is available only when no actual metadata exists" unless @buffer.metadata_list.empty?
m = metadata(tag, time, record)
Expand Down
97 changes: 77 additions & 20 deletions test/plugin/test_output.rb
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,10 @@ def create_output(type=:full)
def create_metadata(timekey: nil, tag: nil, variables: nil)
Fluent::Plugin::Buffer::Metadata.new(timekey, tag, variables)
end
def create_chunk(timekey: nil, tag: nil, variables: nil)
m = Fluent::Plugin::Buffer::Metadata.new(timekey, tag, variables)
Fluent::Plugin::Buffer::MemoryChunk.new(m)
end
def waiting(seconds)
begin
Timeout.timeout(seconds) do
Expand Down Expand Up @@ -219,88 +223,141 @@ def waiting(seconds)
assert @i.terminated?
end

test '#extract_placeholders does nothing if chunk key is not specified' do
data(:new_api => :chunk,
:old_api => :metadata)
test '#extract_placeholders does nothing if chunk key is not specified' do |api|
@i.configure(config_element('ROOT', '', {}, [config_element('buffer', '')]))
assert !@i.chunk_key_time
assert !@i.chunk_key_tag
assert_equal [], @i.chunk_keys
tmpl = "/mypath/%Y/%m/%d/${tag}/${tag[1]}/${tag[2]}/${key1}/${key2}/tail"
t = event_time('2016-04-11 20:30:00 +0900')
v = {key1: "value1", key2: "value2"}
m = create_metadata(timekey: t, tag: 'fluentd.test.output', variables: v)
assert_equal tmpl, @i.extract_placeholders(tmpl, m)
c = if api == :chunk
create_chunk(timekey: t, tag: 'fluentd.test.output', variables: v)
else
create_metadata(timekey: t, tag: 'fluentd.test.output', variables: v)
end
assert_equal tmpl, @i.extract_placeholders(tmpl, c)
end

test '#extract_placeholders can extract time if time key and range are configured' do
data(:new_api => :chunk,
:old_api => :metadata)
test '#extract_placeholders can extract time if time key and range are configured' do |api|
@i.configure(config_element('ROOT', '', {}, [config_element('buffer', 'time', {'timekey' => 60*30, 'timekey_zone' => "+0900"})]))
assert @i.chunk_key_time
assert !@i.chunk_key_tag
assert_equal [], @i.chunk_keys
tmpl = "/mypath/%Y/%m/%d/%H-%M/${tag}/${tag[1]}/${tag[2]}/${key1}/${key2}/tail"
t = event_time('2016-04-11 20:30:00 +0900')
v = {key1: "value1", key2: "value2"}
m = create_metadata(timekey: t, tag: 'fluentd.test.output', variables: v)
assert_equal "/mypath/2016/04/11/20-30/${tag}/${tag[1]}/${tag[2]}/${key1}/${key2}/tail", @i.extract_placeholders(tmpl, m)
c = if api == :chunk
create_chunk(timekey: t, tag: 'fluentd.test.output', variables: v)
else
create_metadata(timekey: t, tag: 'fluentd.test.output', variables: v)
end
assert_equal "/mypath/2016/04/11/20-30/${tag}/${tag[1]}/${tag[2]}/${key1}/${key2}/tail", @i.extract_placeholders(tmpl, c)
end

test '#extract_placeholders can extract tag and parts of tag if tag is configured' do
data(:new_api => :chunk,
:old_api => :metadata)
test '#extract_placeholders can extract tag and parts of tag if tag is configured' do |api|
@i.configure(config_element('ROOT', '', {}, [config_element('buffer', 'tag', {})]))
assert !@i.chunk_key_time
assert @i.chunk_key_tag
assert_equal [], @i.chunk_keys
tmpl = "/mypath/%Y/%m/%d/%H-%M/${tag}/${tag[1]}/${tag[2]}/${key1}/${key2}/tail"
t = event_time('2016-04-11 20:30:00 +0900')
v = {key1: "value1", key2: "value2"}
m = create_metadata(timekey: t, tag: 'fluentd.test.output', variables: v)
assert_equal "/mypath/%Y/%m/%d/%H-%M/fluentd.test.output/test/output/${key1}/${key2}/tail", @i.extract_placeholders(tmpl, m)
c = if api == :chunk
create_chunk(timekey: t, tag: 'fluentd.test.output', variables: v)
else
create_metadata(timekey: t, tag: 'fluentd.test.output', variables: v)
end
assert_equal "/mypath/%Y/%m/%d/%H-%M/fluentd.test.output/test/output/${key1}/${key2}/tail", @i.extract_placeholders(tmpl, c)
end

test '#extract_placeholders can extract variables if variables are configured' do
data(:new_api => :chunk,
:old_api => :metadata)
test '#extract_placeholders can extract variables if variables are configured' do |api|
@i.configure(config_element('ROOT', '', {}, [config_element('buffer', 'key1,key2', {})]))
assert !@i.chunk_key_time
assert !@i.chunk_key_tag
assert_equal ['key1','key2'], @i.chunk_keys
tmpl = "/mypath/%Y/%m/%d/%H-%M/${tag}/${tag[1]}/${tag[2]}/${key1}/${key2}/tail"
t = event_time('2016-04-11 20:30:00 +0900')
v = {key1: "value1", key2: "value2"}
m = create_metadata(timekey: t, tag: 'fluentd.test.output', variables: v)
assert_equal "/mypath/%Y/%m/%d/%H-%M/${tag}/${tag[1]}/${tag[2]}/value1/value2/tail", @i.extract_placeholders(tmpl, m)
c = if api == :chunk
create_chunk(timekey: t, tag: 'fluentd.test.output', variables: v)
else
create_metadata(timekey: t, tag: 'fluentd.test.output', variables: v)
end
assert_equal "/mypath/%Y/%m/%d/%H-%M/${tag}/${tag[1]}/${tag[2]}/value1/value2/tail", @i.extract_placeholders(tmpl, c)
end

test '#extract_placeholders can extract nested variables if variables are configured with dot notation' do
data(:new_api => :chunk,
:old_api => :metadata)
test '#extract_placeholders can extract nested variables if variables are configured with dot notation' do |api|
@i.configure(config_element('ROOT', '', {}, [config_element('buffer', 'key,$.nest.key', {})]))
assert !@i.chunk_key_time
assert !@i.chunk_key_tag
assert_equal ['key','$.nest.key'], @i.chunk_keys
tmpl = "/mypath/%Y/%m/%d/%H-%M/${tag}/${tag[1]}/${tag[2]}/${key}/${$.nest.key}/tail"
t = event_time('2016-04-11 20:30:00 +0900')
v = {:key => "value1", :"$.nest.key" => "value2"}
m = create_metadata(timekey: t, tag: 'fluentd.test.output', variables: v)
assert_equal "/mypath/%Y/%m/%d/%H-%M/${tag}/${tag[1]}/${tag[2]}/value1/value2/tail", @i.extract_placeholders(tmpl, m)
c = if api == :chunk
create_chunk(timekey: t, tag: 'fluentd.test.output', variables: v)
else
create_metadata(timekey: t, tag: 'fluentd.test.output', variables: v)
end
assert_equal "/mypath/%Y/%m/%d/%H-%M/${tag}/${tag[1]}/${tag[2]}/value1/value2/tail", @i.extract_placeholders(tmpl, c)
end

test '#extract_placeholders can extract all chunk keys if configured' do
data(:new_api => :chunk,
:old_api => :metadata)
test '#extract_placeholders can extract all chunk keys if configured' do |api|
@i.configure(config_element('ROOT', '', {}, [config_element('buffer', 'time,tag,key1,key2', {'timekey' => 60*30, 'timekey_zone' => "+0900"})]))
assert @i.chunk_key_time
assert @i.chunk_key_tag
assert_equal ['key1','key2'], @i.chunk_keys
tmpl = "/mypath/%Y/%m/%d/%H-%M/${tag}/${tag[1]}/${tag[2]}/${key1}/${key2}/tail"
t = event_time('2016-04-11 20:30:00 +0900')
v = {key1: "value1", key2: "value2"}
m = create_metadata(timekey: t, tag: 'fluentd.test.output', variables: v)
assert_equal "/mypath/2016/04/11/20-30/fluentd.test.output/test/output/value1/value2/tail", @i.extract_placeholders(tmpl, m)
c = if api == :chunk
create_chunk(timekey: t, tag: 'fluentd.test.output', variables: v)
else
create_metadata(timekey: t, tag: 'fluentd.test.output', variables: v)
end
assert_equal "/mypath/2016/04/11/20-30/fluentd.test.output/test/output/value1/value2/tail", @i.extract_placeholders(tmpl, c)
end

test '#extract_placeholders removes out-of-range tag part and unknown variable placeholders' do
data(:new_api => :chunk,
:old_api => :metadata)
test '#extract_placeholders removes out-of-range tag part and unknown variable placeholders' do |api|
@i.configure(config_element('ROOT', '', {}, [config_element('buffer', 'time,tag,key1,key2', {'timekey' => 60*30, 'timekey_zone' => "+0900"})]))
assert @i.chunk_key_time
assert @i.chunk_key_tag
assert_equal ['key1','key2'], @i.chunk_keys
tmpl = "/mypath/%Y/%m/%d/%H-%M/${tag}/${tag[3]}/${tag[4]}/${key3}/${key4}/tail"
t = event_time('2016-04-11 20:30:00 +0900')
v = {key1: "value1", key2: "value2"}
c = if api == :chunk
create_chunk(timekey: t, tag: 'fluentd.test.output', variables: v)
else
create_metadata(timekey: t, tag: 'fluentd.test.output', variables: v)
end
assert_equal "/mypath/2016/04/11/20-30/fluentd.test.output/////tail", @i.extract_placeholders(tmpl, c)
end

test '#extract_placeholders logs warn message if metadata is passed for ${chunk_id} placeholder' do
@i.configure(config_element('ROOT', '', {}, [config_element('buffer', '')]))
tmpl = "/mypath/${chunk_id}/tail"
t = event_time('2016-04-11 20:30:00 +0900')
v = {key1: "value1", key2: "value2"}
m = create_metadata(timekey: t, tag: 'fluentd.test.output', variables: v)
assert_equal "/mypath/2016/04/11/20-30/fluentd.test.output/////tail", @i.extract_placeholders(tmpl, m)
@i.extract_placeholders(tmpl, m)
logs = @i.log.out.logs
assert { logs.any? { |log| log.include?("${chunk_id} is not allowed in this plugin") } }
end

sub_test_case '#placeholder_validators' do
Expand Down