Merge pull request #1040 from bgruszka/master

add data_stream_template_use_index_patterns_wildcard in elasticsearch_data_stream

README.md

@@ -1561,6 +1561,50 @@ Default value is `false`.
 
 **NOTE:** This parameter requests to install elasticsearch-xpack gem.
 
+### data_stream_template_use_index_patterns_wildcard
+
+Specify whether index patterns should include a wildcard (*) when creating an index template. This is particularly useful to prevent errors in scenarios where index templates are generated automatically, and multiple services with distinct suffixes are in use.
+
+Default value is `true`.
+
+Consider the following JSON error response when index patterns clash due to wildcard usage:
+```json
+{
+  "error": {
+    "root_cause": [
+      {
+        "type": "illegal_argument_exception",
+        "reason": "index template [eks-kube-apiserver] has index patterns [eks-kube-apiserver*] matching patterns from existing templates [eks-kube-apiserver-audit] with patterns (eks-kube-apiserver-audit => [eks-kube-apiserver-audit*]) that have the same priority [0], multiple index templates may not match during index creation, please use a different priority"
+      }
+    ],
+    "type": "illegal_argument_exception",
+    "reason": "index template [eks-kube-apiserver] has index patterns [eks-kube-apiserver*] matching patterns from existing templates [eks-kube-apiserver-audit] with patterns (eks-kube-apiserver-audit => [eks-kube-apiserver-audit*]) that have the same priority [0], multiple index templates may not match during index creation, please use a different priority"
+  },
+  "status": 400
+}
+```
+
+#### Usage Examples
+
+When `data_stream_template_use_index_patterns_wildcard` is set to `true` (default):
+
+```
+data_stream_name: foo
+data_stream_template_use_index_patterns_wildcard: true
+```
+
+In this case, the resulting index patterns will be: `["foo*"]`
+
+When `data_stream_template_use_index_patterns_wildcard` is set to `false`:
+
+```
+data_stream_name: foo
+data_stream_template_use_index_patterns_wildcard: false
+```
+
+The resulting index patterns will be: `["foo"]`
+
+
 ## Troubleshooting
 
 See [Troubleshooting document](README.Troubleshooting.md)

lib/fluent/plugin/out_elasticsearch_data_stream.rb

@@ -13,6 +13,7 @@ class ElasticsearchOutputDataStream < ElasticsearchOutput
     config_param :data_stream_template_name, :string, :default => nil
     config_param :data_stream_ilm_policy, :string, :default => nil
     config_param :data_stream_ilm_policy_overwrite, :bool, :default => false
+    config_param :data_stream_template_use_index_patterns_wildcard, :bool, :default => true
 
     # Elasticsearch 7.9 or later always support new style of index template.
     config_set_default :use_legacy_template, false
@@ -112,8 +113,9 @@ def create_ilm_policy(datastream_name, template_name, ilm_name, host = nil)
 
     def create_index_template(datastream_name, template_name, ilm_name, host = nil)
       return if data_stream_exist?(datastream_name, host) or template_exists?(template_name, host)
+      wildcard = @data_stream_template_use_index_patterns_wildcard ? '*' : ''
       body = {
-        "index_patterns" => ["#{datastream_name}*"],
+        "index_patterns" => ["#{datastream_name}#{wildcard}"],
         "data_stream" => {},
         "template" => {
           "settings" => {

test/plugin/test_out_elasticsearch_data_stream.rb

@@ -34,6 +34,10 @@ def ilm_endpoint
     '_ilm'.freeze
   end
 
+  def index_template_endpoint
+    '_index_template'.freeze
+  end
+
   def driver(conf='', es_version=elasticsearch_version.to_i, client_version=elasticsearch_version)
     # For request stub to detect compatibility.
     @es_version ||= es_version
@@ -124,7 +128,6 @@ def stub_nonexistent_template?(name="foo_tpl", url="http://localhost:9200")
     stub_request(:get, "#{url}/_index_template/#{name}").to_return(:status => [404, TRANSPORT_CLASS::Transport::Errors::NotFound], :headers => {'x-elastic-product' => 'Elasticsearch'})
   end
 
-
   def push_bulk_request(req_body)
     # bulk data must be pair of OP and records
     # {"create": {}}\nhttp://localhost:9200/_ilm/policy/foo_ilm_bar
@@ -486,7 +489,8 @@ def test_datastream_configure
         '@type' => ELASTIC_DATA_STREAM_TYPE,
         'data_stream_name' => 'foo',
         'data_stream_ilm_name' => "foo_ilm_policy",
-        'data_stream_template_name' => "foo_tpl"
+        'data_stream_template_name' => "foo_tpl",
+        'data_stream_template_use_index_patterns_wildcard' => false
       })
     assert_equal "foo", driver(conf).instance.data_stream_name
   end
@@ -556,6 +560,21 @@ def test_configure_compression
     assert_equal true, driver(config).instance.compression
   end
 
+  def test_configure_wildcard_usage
+    omit REQUIRED_ELASTIC_MESSAGE unless data_stream_supported?
+
+    config = %{
+      data_stream_name                                        foo
+      data_stream_template_name                               foo_tpl
+      data_stream_ilm_name                                    foo_ilm_policy
+      data_stream_template_use_index_patterns_wildcard        false
+    }
+
+    stub_default
+
+    assert_equal false, driver(config).instance.data_stream_template_use_index_patterns_wildcard
+  end
+
   def test_check_compression_strategy
     omit REQUIRED_ELASTIC_MESSAGE unless data_stream_supported?
 
@@ -722,6 +741,72 @@ def test_template_and_ilm_unset
     assert_equal "foo_policy", driver(conf).instance.data_stream_ilm_name
   end
 
+  def test_template_index_patterns_with_wildcard
+    omit REQUIRED_ELASTIC_MESSAGE unless data_stream_supported?
+
+    stub_existent_ilm?
+
+    stub_nonexistent_data_stream?
+    stub_data_stream
+
+    stub_nonexistent_template?("foo_template")
+    stub_request(:put, "http://localhost:9200/#{index_template_endpoint}/foo_template").with do |req|
+      # bulk data must be pair of OP and records
+      # {"create": {}}\nhttp://localhost:9200/_index_template//foo_template
+      # {"@timestamp": ...}
+      push_bulk_request(req.body)
+    end.to_return({:status => 200, :body => "{}", :headers => { 'Content-Type' => 'json', 'x-elastic-product' => 'Elasticsearch' } })
+
+    conf = config_element(
+      'ROOT', '', {
+      '@type' => ELASTIC_DATA_STREAM_TYPE,
+      'data_stream_name' => 'foo',
+      'data_stream_ilm_name' => 'foo_ilm_policy',
+      'data_stream_template_use_index_patterns_wildcard' => false
+    })
+
+    assert_nothing_raised {
+      driver(conf)
+    }
+
+    assert_requested(:put, "http://localhost:9200/#{index_template_endpoint}/foo_template",
+                     body: '{"index_patterns":["foo"],"data_stream":{},"template":{"settings":{"index.lifecycle.name":"foo_ilm_policy"}}}',
+                     times: 1)
+  end
+
+  def test_template_index_patterns_without_wildcard
+    omit REQUIRED_ELASTIC_MESSAGE unless data_stream_supported?
+
+    stub_existent_ilm?
+
+    stub_nonexistent_data_stream?
+    stub_data_stream
+
+    stub_nonexistent_template?("foo_template")
+    stub_request(:put, "http://localhost:9200/#{index_template_endpoint}/foo_template").with do |req|
+      # bulk data must be pair of OP and records
+      # {"create": {}}\nhttp://localhost:9200/_index_template//foo_template
+      # {"@timestamp": ...}
+      push_bulk_request(req.body)
+    end.to_return({:status => 200, :body => "{}", :headers => { 'Content-Type' => 'json', 'x-elastic-product' => 'Elasticsearch' } })
+
+    conf = config_element(
+      'ROOT', '', {
+      '@type' => ELASTIC_DATA_STREAM_TYPE,
+      'data_stream_name' => 'foo',
+      'data_stream_ilm_name' => 'foo_ilm_policy',
+      'data_stream_template_use_index_patterns_wildcard' => true
+    })
+
+    assert_nothing_raised {
+      driver(conf)
+    }
+
+    assert_requested(:put, "http://localhost:9200/#{index_template_endpoint}/foo_template",
+                     body: '{"index_patterns":["foo*"],"data_stream":{},"template":{"settings":{"index.lifecycle.name":"foo_ilm_policy"}}}',
+                     times: 1)
+  end
+
   def test_placeholder
     omit REQUIRED_ELASTIC_MESSAGE unless data_stream_supported?