diff --git a/lib/fluent/plugin/out_opensearch_data_stream.rb b/lib/fluent/plugin/out_opensearch_data_stream.rb index 9140d8f..11557e4 100644 --- a/lib/fluent/plugin/out_opensearch_data_stream.rb +++ b/lib/fluent/plugin/out_opensearch_data_stream.rb @@ -22,7 +22,7 @@ def configure(conf) @data_stream_template_name = "#{@data_stream_name}_template" if @data_stream_template_name.nil? # ref. https://opensearch.org/docs/latest/opensearch/data-streams/ - unless placeholder?(:data_stream_name_placeholder, @data_stream_name) + unless placeholder_substitution_needed? validate_data_stream_parameters else @use_placeholder = true @@ -67,12 +67,20 @@ def validate_data_stream_parameters end end - def create_index_template(datastream_name, template_name, host = nil) + def placeholder_substitution_needed? + need_substitution = placeholder?(:data_stream_name_placeholder, @data_stream_name) || + @customize_template&.values&.any? { |value| placeholder?(:customize_template, value.to_s) } || + placeholder?(:data_stream_template_name, @data_stream_template_name) + log.debug("Needs substitution: #{need_substitution}") + need_substitution + end + + def create_index_template(datastream_name, template_name, customize_template = nil, host = nil) # Create index template from file if !dry_run? if @template_file return if data_stream_exist?(datastream_name, host) or template_exists?(template_name, host) - template_installation_actual(template_name, @customize_template, @application_name, datastream_name, host) + template_installation_actual(template_name, customize_template, @application_name, datastream_name, host) else # Create default index template return if data_stream_exist?(datastream_name, host) or template_exists?(template_name, host) body = { @@ -162,8 +170,11 @@ def write(chunk) end data_stream_name = extract_placeholders(@data_stream_name, chunk).downcase data_stream_template_name = extract_placeholders(@data_stream_template_name, chunk).downcase + if @customize_template + customize_template = @customize_template.each_with_object({}) { |(key, value), hash| hash[key] = extract_placeholders(value, chunk) } + end begin - create_index_template(data_stream_name, data_stream_template_name, host) + create_index_template(data_stream_name, data_stream_template_name, customize_template, host) rescue => e raise Fluent::ConfigError, "Failed to create data stream: <#{data_stream_name}> #{e.message}" end diff --git a/test/plugin/test_out_opensearch_data_stream.rb b/test/plugin/test_out_opensearch_data_stream.rb index 296c528..d1731fd 100644 --- a/test/plugin/test_out_opensearch_data_stream.rb +++ b/test/plugin/test_out_opensearch_data_stream.rb @@ -743,4 +743,34 @@ def test_record_with_remove_keys assert(!index_cmds[1].has_key?('remove_me')) end + def test_custom_data_stream_template_create_with_placeholders + cwd = File.dirname(__FILE__) + conf = config_element( + 'ROOT', '', { + '@type' => OPENSEARCH_DATA_STREAM_TYPE, + 'data_stream_name' => 'foo', + 'data_stream_name_placeholder' => 'foo', + 'data_stream_template_name' => '${tag}_template', + 'template_file' => File.join(cwd, 'datastream_template.json'), + 'customize_template' => '{"foo*": "${tag}--*"}', + }) + + stub_default + stub_bulk_feed('foo', 'test_templata') + stub_nonexistent_template?('test_template') + + stub_request(:put, "http://localhost:9200/_index_template/test_template") + .to_return(:status => 200, :body => "", :headers => {}) + + driver(conf).run(default_tag: 'test') do + driver.feed(sample_record) + end + + assert_requested( + :put, + "http://localhost:9200/_index_template/test_template", + body: {"index_patterns" => ["test--*"], "data_stream" => {}} + ) + + end end