Support legacy template api in elasticsearch 8

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 11.10.0
+ - Add legacy template API support for Elasticsearch 8 [#1092](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/1092)
+
 ## 11.9.3
  - DOC: clarify that `http_compression` option only affects _requests_; compressed _responses_ have always been read independent of this setting [#1030 ](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/1030)
 

docs/index.asciidoc

@@ -362,6 +362,7 @@ This plugin supports the following configuration options plus the
 | <<plugins-{type}s-{plugin}-ssl_certificate_verification>> |<<boolean,boolean>>|No
 | <<plugins-{type}s-{plugin}-ssl_supported_protocols>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-template>> |a valid filesystem path|No
+| <<plugins-{type}s-{plugin}-template_api>> |<<string,string>>, one of `["auto", "legacy", "composable"]`|No
 | <<plugins-{type}s-{plugin}-template_name>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-template_overwrite>> |<<boolean,boolean>>|No
 | <<plugins-{type}s-{plugin}-timeout>> |<<number,number>>|No
@@ -1082,6 +1083,22 @@ the *$JDK_HOME/conf/security/java.security* configuration file. That is, `TLSv1.
 You can set the path to your own template here, if you so desire.
 If not set, the included template will be used.
 
+[id="plugins-{type}s-{plugin}-template_api"]
+===== `template_api`
+
+  * Value can be any of: `auto`, `legacy`, `composable`
+  * Default value is `auto`
+
+The default setting of `auto` will use
+{ref}/index-templates.html[index template API] to create index template, if the
+Elasticsearch cluster is running Elasticsearch version `8.0.0` or higher,
+and use {ref}/indices-templates-v1.html[legacy template API] otherwise.
+
+Setting this flag to `legacy` will use legacy template API to create index template.
+Setting this flag to `composable` will use index template API to create index template.
+
+NOTE: The format of template needs to match with the corresponding Elasticsearch version.
+
 [id="plugins-{type}s-{plugin}-template_name"]
 ===== `template_name`
 

lib/logstash/outputs/elasticsearch.rb

@@ -185,6 +185,11 @@ class LogStash::Outputs::ElasticSearch < LogStash::Outputs::Base
   # the "logstash" template (i.e. removing all customized settings)
   config :template_overwrite, :validate => :boolean, :default => false
 
+  # Flag for enabling legacy template api for Elasticsearch 8
+  # Default auto will use index template api for Elasticsearch 8 and use legacy api for 7
+  # Set to legacy to use legacy template api
+  config :template_api, :validate => ['auto', 'legacy', 'composable'], :default => 'auto'
+
   # The version to use for indexing. Use sprintf syntax like `%{my_version}` to use a field value here.
   # See https://www.elastic.co/blog/elasticsearch-versioning-support.
   config :version, :validate => :string

lib/logstash/outputs/elasticsearch/http_client.rb

@@ -77,12 +77,12 @@ def build_url_template
       }
     end
 
-    def template_install(name, template, force=false)
-      if template_exists?(name) && !force
+    def template_install(template_endpoint, name, template, force=false)
+      if template_exists?(template_endpoint, name) && !force
         @logger.debug("Found existing Elasticsearch template, skipping template management", name: name)
         return
       end
-      template_put(name, template)
+      template_put(template_endpoint, name, template)
     end
 
     def last_es_version
@@ -402,20 +402,16 @@ def exists?(path, use_get=false)
       response.code >= 200 && response.code <= 299
     end
 
-    def template_exists?(name)
+    def template_exists?(template_endpoint, name)
       exists?("/#{template_endpoint}/#{name}")
     end
 
-    def template_put(name, template)
+    def template_put(template_endpoint, name, template)
       path = "#{template_endpoint}/#{name}"
       logger.info("Installing Elasticsearch template", name: name)
       @pool.put(path, nil, LogStash::Json.dump(template))
     end
 
-    def template_endpoint
-      maximum_seen_major_version < 8 ? '_template' : '_index_template'
-    end
-
     # ILM methods
 
     # check whether rollover alias already exists

lib/logstash/outputs/elasticsearch/template_manager.rb

@@ -1,8 +1,18 @@
 module LogStash; module Outputs; class ElasticSearch
   class TemplateManager
+    LEGACY_TEMPLATE_ENDPOINT = '_template'.freeze
+    INDEX_TEMPLATE_ENDPOINT = '_index_template'.freeze
+
     # To be mixed into the elasticsearch plugin base
     def self.install_template(plugin)
       return unless plugin.manage_template
+
+      if plugin.maximum_seen_major_version < 8 && plugin.template_api == 'auto'
+        plugin.logger.warn("Elasticsearch Output configured with `template_api => auto` uses legacy template API to manage index template for Elasticsearch 7 or below. " +
+                             "Legacy template API will be removed in Elasticsearch 9. It is recommended to set `template_api => legacy` before any upgrade " +
+                             "and please consider to migrate to composable index templates.")
+      end
+
       if plugin.template
         plugin.logger.info("Using mapping template from", :path => plugin.template)
         template = read_template_file(plugin.template)
@@ -14,7 +24,7 @@ def self.install_template(plugin)
 
       add_ilm_settings_to_template(plugin, template) if plugin.ilm_in_use?
       plugin.logger.debug("Attempting to install template", template: template)
-      install(plugin.client, template_name(plugin), template, plugin.template_overwrite)
+      install(plugin.client, template_endpoint(plugin), template_name(plugin), template, plugin.template_overwrite)
     end
 
     private
@@ -25,8 +35,8 @@ def self.load_default_template(es_major_version, ecs_compatibility)
       fail "Failed to load default template for Elasticsearch v#{es_major_version} with ECS #{ecs_compatibility}; caused by: #{e.inspect}"
     end
 
-    def self.install(client, template_name, template, template_overwrite)
-      client.template_install(template_name, template, template_overwrite)
+    def self.install(client, template_endpoint, template_name, template, template_overwrite)
+      client.template_install(template_endpoint, template_name, template, template_overwrite)
     end
 
     def self.add_ilm_settings_to_template(plugin, template)
@@ -63,5 +73,16 @@ def self.read_template_file(template_path)
       template_data = ::IO.read(template_path)
       LogStash::Json.load(template_data)
     end
+
+    def self.template_endpoint(plugin)
+      if plugin.template_api == 'auto'
+        plugin.maximum_seen_major_version < 8 ? LEGACY_TEMPLATE_ENDPOINT : INDEX_TEMPLATE_ENDPOINT
+      elsif plugin.template_api.to_s == 'legacy'
+        LEGACY_TEMPLATE_ENDPOINT
+      else
+        INDEX_TEMPLATE_ENDPOINT
+      end
+    end
+
   end
 end end end

logstash-output-elasticsearch.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name            = 'logstash-output-elasticsearch'
-  s.version         = '11.9.3'
+  s.version         = '11.10.0'
 
   s.licenses        = ['apache-2.0']
   s.summary         = "Stores logs in Elasticsearch"

spec/unit/outputs/elasticsearch/http_client_spec.rb

@@ -140,27 +140,6 @@
     end
   end
 
-  describe "index template" do
-    subject { described_class.new(base_options) }
-    let(:template_name) { "logstash" }
-    let(:template) { {} }
-    let(:get_response) {
-      double("response", :body => {})
-    }
-
-    it "should call composable index template in version 8+" do
-      expect(subject).to receive(:maximum_seen_major_version).and_return(8)
-      expect(subject.pool).to receive(:put).with("_index_template/#{template_name}", nil, anything).and_return(get_response)
-      subject.template_put(template_name, template)
-    end
-
-    it "should call index template in version < 8" do
-      expect(subject).to receive(:maximum_seen_major_version).and_return(7)
-      expect(subject.pool).to receive(:put).with("_template/#{template_name}", nil, anything).and_return(get_response)
-      subject.template_put(template_name, template)
-    end
-  end
-
   describe "join_bulk_responses" do
     subject { described_class.new(base_options) }
 

spec/unit/outputs/elasticsearch/template_manager_spec.rb

@@ -63,4 +63,53 @@
       end
     end
   end
+
+  describe "template endpoint" do
+    describe "template_api => 'auto'" do
+      let(:plugin_settings) { {"manage_template" => true, "template_api" => 'auto'} }
+      let(:plugin) { LogStash::Outputs::ElasticSearch.new(plugin_settings) }
+
+      describe "in version 8+" do
+        it "should use index template API" do
+          expect(plugin).to receive(:maximum_seen_major_version).at_least(:once).and_return(8)
+          endpoint = described_class.template_endpoint(plugin)
+          expect(endpoint).to be_equal(LogStash::Outputs::ElasticSearch::TemplateManager::INDEX_TEMPLATE_ENDPOINT)
+        end
+      end
+
+      describe "in version < 8" do
+        it "should use legacy template API" do
+          expect(plugin).to receive(:maximum_seen_major_version).at_least(:once).and_return(7)
+          endpoint = described_class.template_endpoint(plugin)
+          expect(endpoint).to be_equal(LogStash::Outputs::ElasticSearch::TemplateManager::LEGACY_TEMPLATE_ENDPOINT)
+        end
+      end
+    end
+
+    describe "template_api => 'legacy'" do
+      let(:plugin_settings) { {"manage_template" => true, "template_api" => 'legacy'} }
+      let(:plugin) { LogStash::Outputs::ElasticSearch.new(plugin_settings) }
+
+      describe "in version 8+" do
+        it "should use legacy template API" do
+          expect(plugin).to receive(:maximum_seen_major_version).never
+          endpoint = described_class.template_endpoint(plugin)
+          expect(endpoint).to be_equal(LogStash::Outputs::ElasticSearch::TemplateManager::LEGACY_TEMPLATE_ENDPOINT)
+        end
+      end
+    end
+
+    describe "template_api => 'composable'" do
+      let(:plugin_settings) { {"manage_template" => true, "template_api" => 'composable'} }
+      let(:plugin) { LogStash::Outputs::ElasticSearch.new(plugin_settings) }
+
+      describe "in version 8+" do
+        it "should use legacy template API" do
+          expect(plugin).to receive(:maximum_seen_major_version).never
+          endpoint = described_class.template_endpoint(plugin)
+          expect(endpoint).to be_equal(LogStash::Outputs::ElasticSearch::TemplateManager:: INDEX_TEMPLATE_ENDPOINT)
+        end
+      end
+    end
+  end
 end