diff --git a/CHANGELOG.md b/CHANGELOG.md
index 91e625ddc..271d9cacc 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,6 @@
+## 10.7.3
+ - Added composable index template support for elasticsearch version 8 [#980](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/980)
+
 ## 10.7.2
  - [DOC] Fixed links to restructured Logstash-to-cloud docs [#975](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/975)
 
diff --git a/lib/logstash/outputs/elasticsearch/http_client.rb b/lib/logstash/outputs/elasticsearch/http_client.rb
index 32a37e82a..f9bdf5e49 100644
--- a/lib/logstash/outputs/elasticsearch/http_client.rb
+++ b/lib/logstash/outputs/elasticsearch/http_client.rb
@@ -343,15 +343,19 @@ def exists?(path, use_get=false)
     end
 
     def template_exists?(name)
-      exists?("/_template/#{name}")
+      exists?("/#{template_endpoint}/#{name}")
     end
 
     def template_put(name, template)
-      path = "_template/#{name}"
+      path = "#{template_endpoint}/#{name}"
       logger.info("Installing elasticsearch template to #{path}")
       @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
diff --git a/lib/logstash/outputs/elasticsearch/template_manager.rb b/lib/logstash/outputs/elasticsearch/template_manager.rb
index 20f2cbd2c..72b505e61 100644
--- a/lib/logstash/outputs/elasticsearch/template_manager.rb
+++ b/lib/logstash/outputs/elasticsearch/template_manager.rb
@@ -34,12 +34,17 @@ def self.install(client, template_name, template, template_overwrite)
     def self.add_ilm_settings_to_template(plugin, template)
       # Overwrite any index patterns, and use the rollover alias. Use 'index_patterns' rather than 'template' for pattern
       # definition - remove any existing definition of 'template'
-      template.delete('template') if template.include?('template')
+      template.delete('template') if template.include?('template') if plugin.maximum_seen_major_version < 8
       template['index_patterns'] = "#{plugin.ilm_rollover_alias}-*"
-      if template['settings'] && (template['settings']['index.lifecycle.name'] || template['settings']['index.lifecycle.rollover_alias'])
+      settings = template_settings(plugin, template)
+      if settings && (settings['index.lifecycle.name'] || settings['index.lifecycle.rollover_alias'])
         plugin.logger.info("Overwriting index lifecycle name and rollover alias as ILM is enabled.")
       end
-      template['settings'].update({ 'index.lifecycle.name' => plugin.ilm_policy, 'index.lifecycle.rollover_alias' => plugin.ilm_rollover_alias})
+      settings.update({ 'index.lifecycle.name' => plugin.ilm_policy, 'index.lifecycle.rollover_alias' => plugin.ilm_rollover_alias})
+    end
+
+    def self.template_settings(plugin, template)
+      plugin.maximum_seen_major_version < 8 ? template['settings']: template['template']['settings']
     end
 
     # Template name - if template_name set, use it
diff --git a/lib/logstash/outputs/elasticsearch/templates/ecs-disabled/elasticsearch-8x.json b/lib/logstash/outputs/elasticsearch/templates/ecs-disabled/elasticsearch-8x.json
index 14808248a..3ffba4f52 100644
--- a/lib/logstash/outputs/elasticsearch/templates/ecs-disabled/elasticsearch-8x.json
+++ b/lib/logstash/outputs/elasticsearch/templates/ecs-disabled/elasticsearch-8x.json
@@ -1,44 +1,50 @@
 {
   "index_patterns" : "logstash-*",
   "version" : 80001,
-  "settings" : {
-    "index.refresh_interval" : "5s",
-    "number_of_shards": 1
-  },
-  "mappings" : {
-    "dynamic_templates" : [ {
-      "message_field" : {
-        "path_match" : "message",
-        "match_mapping_type" : "string",
-        "mapping" : {
-          "type" : "text",
-          "norms" : false
+  "template" : {
+    "settings" : {
+      "index.refresh_interval" : "5s",
+      "number_of_shards": 1
+    },
+    "mappings" : {
+      "dynamic_templates" : [ {
+        "message_field" : {
+          "path_match" : "message",
+          "match_mapping_type" : "string",
+          "mapping" : {
+            "type" : "text",
+            "norms" : false
+          }
         }
-      }
-    }, {
-      "string_fields" : {
-        "match" : "*",
-        "match_mapping_type" : "string",
-        "mapping" : {
-          "type" : "text", "norms" : false,
-          "fields" : {
-            "keyword" : { "type": "keyword", "ignore_above": 256 }
+      }, {
+        "string_fields" : {
+          "match" : "*",
+          "match_mapping_type" : "string",
+          "mapping" : {
+            "type" : "text", "norms" : false,
+            "fields" : {
+              "keyword" : { "type": "keyword", "ignore_above": 256 }
+            }
           }
         }
-      }
-    } ],
-    "properties" : {
-      "@timestamp": { "type": "date"},
-      "@version": { "type": "keyword"},
-      "geoip"  : {
-        "dynamic": true,
-        "properties" : {
-          "ip": { "type": "ip" },
-          "location" : { "type" : "geo_point" },
-          "latitude" : { "type" : "half_float" },
-          "longitude" : { "type" : "half_float" }
+      } ],
+      "properties" : {
+        "@timestamp": { "type": "date" },
+        "@version": { "type": "keyword" },
+        "geoip"  : {
+          "dynamic": true,
+          "properties" : {
+            "ip": { "type": "ip" },
+            "location" : { "type" : "geo_point" },
+            "latitude" : { "type" : "half_float" },
+            "longitude" : { "type" : "half_float" }
+          }
         }
       }
     }
+  },
+  "priority": 200,
+  "_meta" : {
+    "description": "index template for logstash-output-elasticsearch"
   }
 }
diff --git a/logstash-output-elasticsearch.gemspec b/logstash-output-elasticsearch.gemspec
index 7b9a259c4..f8943e7dd 100644
--- a/logstash-output-elasticsearch.gemspec
+++ b/logstash-output-elasticsearch.gemspec
@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name            = 'logstash-output-elasticsearch'
-  s.version         = '10.7.2'
+  s.version         = '10.7.3'
 
   s.licenses        = ['apache-2.0']
   s.summary         = "Stores logs in Elasticsearch"
diff --git a/spec/es_spec_helper.rb b/spec/es_spec_helper.rb
index 9d8044c0c..bef795dce 100644
--- a/spec/es_spec_helper.rb
+++ b/spec/es_spec_helper.rb
@@ -49,19 +49,10 @@ def todays_date
     Time.now.strftime("%Y.%m.%d")
   end
 
-
-  def default_mapping_from_mappings(mappings)
-    if ESHelper.es_version_satisfies?(">=7")
-      mappings
-    else
-      mappings["_default_"]
-    end
-  end
-
   def field_properties_from_template(template_name, field)
-    mappings = @es.indices.get_template(name: template_name)[template_name]["mappings"]
-    mapping = default_mapping_from_mappings(mappings)
-    mapping["properties"][field]["properties"]
+    template = get_template(@es, template_name)
+    mappings = get_template_mappings(template)
+    mappings["properties"][field]["properties"]
   end
 
   def routing_field_name
@@ -105,6 +96,7 @@ def self.es_version_satisfies?(*requirement)
 
   def clean(client)
     client.indices.delete_template(:name => "*")
+    client.indices.delete_index_template(:name => "logstash*") rescue nil
     # This can fail if there are no indexes, ignore failure.
     client.indices.delete(:index => "*") rescue nil
     clean_ilm(client) if supports_ilm?(client)
@@ -182,6 +174,34 @@ def max_age_policy(max_age)
     }
   }
   end
+
+  def get_template(client, name)
+    if ESHelper.es_version_satisfies?(">=8")
+      t = client.indices.get_index_template(name: name)
+      t['index_templates'][0]['index_template']
+    else
+      t = client.indices.get_template(name: name)
+      t[name]
+    end
+  end
+
+  def get_template_settings(template)
+    if ESHelper.es_version_satisfies?(">=8")
+      template['template']['settings']
+    else
+      template['settings']
+    end
+  end
+
+  def get_template_mappings(template)
+    if ESHelper.es_version_satisfies?(">=8")
+      template['template']['mappings']
+    elsif ESHelper.es_version_satisfies?(">=7")
+      template['mappings']
+    else
+      template['mappings']["_default_"]
+    end
+  end
 end
 
 RSpec.configure do |config|
diff --git a/spec/fixtures/template-with-policy-es8x.json b/spec/fixtures/template-with-policy-es8x.json
new file mode 100644
index 000000000..ab25cfca2
--- /dev/null
+++ b/spec/fixtures/template-with-policy-es8x.json
@@ -0,0 +1,50 @@
+{
+  "index_patterns" : "overwrite-*",
+  "version" : 80001,
+  "template" : {
+    "settings" : {
+      "index.refresh_interval" : "1s",
+      "number_of_shards": 1
+    },
+    "mappings" : {
+      "dynamic_templates" : [ {
+        "message_field" : {
+          "path_match" : "message",
+          "match_mapping_type" : "string",
+          "mapping" : {
+            "type" : "text",
+            "norms" : false
+          }
+        }
+      }, {
+        "string_fields" : {
+          "match" : "*",
+          "match_mapping_type" : "string",
+          "mapping" : {
+            "type" : "text", "norms" : false,
+            "fields" : {
+              "keyword" : { "type": "keyword", "ignore_above": 256 }
+            }
+          }
+        }
+      } ],
+      "properties" : {
+        "@timestamp": { "type": "date" },
+        "@version": { "type": "keyword" },
+        "geoip"  : {
+          "dynamic": true,
+          "properties" : {
+            "ip": { "type": "ip" },
+            "location" : { "type" : "geo_point" },
+            "latitude" : { "type" : "half_float" },
+            "longitude" : { "type" : "half_float" }
+          }
+        }
+      }
+    }
+  },
+  "priority": 200,
+  "_meta" : {
+    "description": "index template for logstash-output-elasticsearch"
+  }
+}
diff --git a/spec/integration/outputs/ilm_spec.rb b/spec/integration/outputs/ilm_spec.rb
index 9ec211d94..3f99a2b2f 100644
--- a/spec/integration/outputs/ilm_spec.rb
+++ b/spec/integration/outputs/ilm_spec.rb
@@ -8,7 +8,7 @@
     let (:settings) { super.merge("ilm_policy" => ilm_policy_name)}
 
     it 'should rollover when the policy max docs is reached' do
-      put_policy(@es,ilm_policy_name, policy)
+      put_policy(@es, ilm_policy_name, policy)
       subject.register
 
       subject.multi_receive([
@@ -108,9 +108,11 @@
   it 'should not write the ILM settings into the template' do
     subject.register
     sleep(1)
-    expect(@es.indices.get_template(name: "logstash")["logstash"]).to have_index_pattern("logstash-*")
+
+    template = get_template(@es, "logstash")
+    expect(template).to have_index_pattern("logstash-*")
     if ESHelper.es_version_satisfies?(">= 2")
-      expect(@es.indices.get_template(name: "logstash")["logstash"]["settings"]['index']['lifecycle']).to be_nil
+      expect(get_template_settings(template)['index']['lifecycle']).to be_nil
     end
   end
 
@@ -152,16 +154,17 @@
   end
 
   context 'with a custom template name' do
-    let (:template_name) { "custom_template_name" }
+    let (:template_name) { "logstash_custom_template_name" }
     let (:settings) { super.merge('template_name' => template_name)}
 
     it 'should not write the ILM settings into the template' do
       subject.register
       sleep(1)
 
-      expect(@es.indices.get_template(name: template_name)[template_name]).to have_index_pattern("logstash-*")
+      template = get_template(@es, template_name)
+      expect(template).to have_index_pattern("logstash-*")
       if ESHelper.es_version_satisfies?(">= 2")
-        expect(@es.indices.get_template(name: template_name)[template_name]["settings"]['index']['lifecycle']).to be_nil
+        expect(get_template_settings(template)['index']['lifecycle']).to be_nil
       end
     end
   end
@@ -387,16 +390,20 @@
         it 'should write the ILM settings into the template' do
           subject.register
           sleep(1)
-          expect(@es.indices.get_template(name: "logstash")["logstash"]).to have_index_pattern("logstash-*")
-          expect(@es.indices.get_template(name: "logstash")["logstash"]["settings"]['index']['lifecycle']['name']).to eq("logstash-policy")
-          expect(@es.indices.get_template(name: "logstash")["logstash"]["settings"]['index']['lifecycle']['rollover_alias']).to eq("logstash")
+
+          template = get_template(@es, "logstash")
+          expect(template).to have_index_pattern("logstash-*")
+          expect(get_template_settings(template)['index']['lifecycle']['name']).to eq("logstash-policy")
+          expect(get_template_settings(template)['index']['lifecycle']['rollover_alias']).to eq("logstash")
         end
 
         it_behaves_like 'an ILM enabled Logstash'
       end
 
       context 'with a set index and a custom index pattern' do
-        if ESHelper.es_version_satisfies?(">= 7.0")
+        if ESHelper.es_version_satisfies?(">= 8.0")
+          let (:template) { "spec/fixtures/template-with-policy-es8x.json" }
+        elsif ESHelper.es_version_satisfies?(">= 7.0")
           let (:template) { "spec/fixtures/template-with-policy-es7x.json" }
         else
           let (:template) { "spec/fixtures/template-with-policy-es6x.json" }
@@ -408,13 +415,15 @@
         it 'should not overwrite the index patterns' do
           subject.register
           sleep(1)
-          expect(@es.indices.get_template(name: "logstash")["logstash"]).to have_index_pattern("overwrite-*")
+
+          template = get_template(@es, "logstash")
+          expect(template).to have_index_pattern("overwrite-*")
         end
       end
 
 
       context 'with a custom template' do
-        let (:ilm_rollover_alias) { "the_cat_in_the_hat" }
+        let (:ilm_rollover_alias) { "logstash_the_cat_in_the_hat" }
         let (:index) { ilm_rollover_alias }
         let(:expected_index) { index }
         let (:settings) { super.merge("ilm_policy" => ilm_policy_name,
@@ -422,7 +431,9 @@
                                       "ilm_rollover_alias" => ilm_rollover_alias)}
 
 
-        if ESHelper.es_version_satisfies?(">= 7.0")
+        if ESHelper.es_version_satisfies?(">= 8.0")
+          let (:template) { "spec/fixtures/template-with-policy-es8x.json" }
+        elsif ESHelper.es_version_satisfies?(">= 7.0")
           let (:template) { "spec/fixtures/template-with-policy-es7x.json" }
         else
           let (:template) { "spec/fixtures/template-with-policy-es6x.json" }
@@ -460,13 +471,15 @@
         it 'should write the ILM settings into the template' do
           subject.register
           sleep(1)
-          expect(@es.indices.get_template(name: ilm_rollover_alias)[ilm_rollover_alias]).to have_index_pattern("#{ilm_rollover_alias}-*")
-          expect(@es.indices.get_template(name: ilm_rollover_alias)[ilm_rollover_alias]["settings"]['index']['lifecycle']['name']).to eq(ilm_policy_name)
-          expect(@es.indices.get_template(name: ilm_rollover_alias)[ilm_rollover_alias]["settings"]['index']['lifecycle']['rollover_alias']).to eq(ilm_rollover_alias)
+
+          template = get_template(@es, ilm_rollover_alias)
+          expect(template).to have_index_pattern("#{ilm_rollover_alias}-*")
+          expect(get_template_settings(template)['index']['lifecycle']['name']).to eq(ilm_policy_name)
+          expect(get_template_settings(template)['index']['lifecycle']['rollover_alias']).to eq(ilm_rollover_alias)
         end
 
         context 'with a different template_name' do
-          let (:template_name) { "custom_template_name" }
+          let (:template_name) { "logstash_custom_template_name" }
           let (:settings) { super.merge('template_name' => template_name)}
 
           it_behaves_like 'an ILM enabled Logstash'
@@ -474,9 +487,10 @@
           it 'should write the ILM settings into the template' do
             subject.register
             sleep(1)
-            expect(@es.indices.get_template(name: template_name)[template_name]).to have_index_pattern("#{ilm_rollover_alias}-*")
-            expect(@es.indices.get_template(name: template_name)[template_name]["settings"]['index']['lifecycle']['name']).to eq(ilm_policy_name)
-            expect(@es.indices.get_template(name: template_name)[template_name]["settings"]['index']['lifecycle']['rollover_alias']).to eq(ilm_rollover_alias)
+            template = get_template(@es, template_name)
+            expect(template).to have_index_pattern("#{ilm_rollover_alias}-*")
+            expect(get_template_settings(template)['index']['lifecycle']['name']).to eq(ilm_policy_name)
+            expect(get_template_settings(template)['index']['lifecycle']['rollover_alias']).to eq(ilm_rollover_alias)
           end
         end
 
diff --git a/spec/integration/outputs/metrics_spec.rb b/spec/integration/outputs/metrics_spec.rb
index 58a016365..026d47612 100644
--- a/spec/integration/outputs/metrics_spec.rb
+++ b/spec/integration/outputs/metrics_spec.rb
@@ -19,11 +19,7 @@
 
     # Clean ES of data before we start.
     @es = get_client
-    @es.indices.delete_template(:name => "*")
-
-    # This can fail if there are no indexes, ignore failure.
-    @es.indices.delete(:index => "*") rescue nil
-    #@es.indices.refresh
+    clean(@es)
     subject.register
   end
 
diff --git a/spec/unit/outputs/elasticsearch/http_client_spec.rb b/spec/unit/outputs/elasticsearch/http_client_spec.rb
index efb7ca7f7..accafa720 100644
--- a/spec/unit/outputs/elasticsearch/http_client_spec.rb
+++ b/spec/unit/outputs/elasticsearch/http_client_spec.rb
@@ -1,3 +1,4 @@
+require_relative "../../../../spec/es_spec_helper"
 require "logstash/devutils/rspec/spec_helper"
 require "logstash/outputs/elasticsearch/http_client"
 require "java"
@@ -138,6 +139,27 @@
     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) }
 
diff --git a/spec/unit/outputs/elasticsearch/template_manager_spec.rb b/spec/unit/outputs/elasticsearch/template_manager_spec.rb
index af5e43863..ce38d8507 100644
--- a/spec/unit/outputs/elasticsearch/template_manager_spec.rb
+++ b/spec/unit/outputs/elasticsearch/template_manager_spec.rb
@@ -28,4 +28,35 @@
       expect(described_class.default_template_path(7, :v1)).to end_with("/templates/ecs-v1/elasticsearch-7x.json")
     end
   end
+
+  describe "index template with ilm settings" do
+    let(:plugin_settings) { {"manage_template" => true, "template_overwrite" => true} }
+    let(:plugin) { LogStash::Outputs::ElasticSearch.new(plugin_settings) }
+
+    describe "in version 8+" do
+      let(:file_path) { described_class.default_template_path(8) }
+      let(:template) { described_class.read_template_file(file_path)}
+
+      it "should update settings" do
+        expect(plugin).to receive(:maximum_seen_major_version).at_least(:once).and_return(8)
+        described_class.add_ilm_settings_to_template(plugin, template)
+        expect(template['template']['settings']['index.lifecycle.name']).not_to eq(nil)
+        expect(template['template']['settings']['index.lifecycle.rollover_alias']).not_to eq(nil)
+        expect(template.include?('settings')).to be_falsey
+      end
+    end
+
+    describe "in version < 8" do
+      let(:file_path) { described_class.default_template_path(7) }
+      let(:template) { described_class.read_template_file(file_path)}
+
+      it "should update settings" do
+        expect(plugin).to receive(:maximum_seen_major_version).at_least(:once).and_return(7)
+        described_class.add_ilm_settings_to_template(plugin, template)
+        expect(template['settings']['index.lifecycle.name']).not_to eq(nil)
+        expect(template['settings']['index.lifecycle.rollover_alias']).not_to eq(nil)
+        expect(template.include?('template')).to be_falsey
+      end
+    end
+  end
 end