diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
index 01517e07245..67cac68940c 100644
--- a/CHANGELOG.next.asciidoc
+++ b/CHANGELOG.next.asciidoc
@@ -437,6 +437,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Added experimental dataset `juniper/netscreen`. {pull}20820[20820]
 - Added experimental dataset `sophos/utm`. {pull}20820[20820]
 - Add Cloud Foundry tags in related events. {pull}21177[21177]
+- Add option to select the type of index template to load: legacy, component, index. {pull}21212[21212]
 
 *Auditbeat*
 
diff --git a/auditbeat/auditbeat.reference.yml b/auditbeat/auditbeat.reference.yml
index 62f29364c83..f0dafa75662 100644
--- a/auditbeat/auditbeat.reference.yml
+++ b/auditbeat/auditbeat.reference.yml
@@ -1144,6 +1144,11 @@ output.elasticsearch:
 # Set to false to disable template loading.
 #setup.template.enabled: true
 
+# Select the kind of index template. From Elasticsearch 7.8, it is possible to
+# use component templates. Available options: legacy, component, index.
+# By default auditbeat uses the legacy index templates.
+#setup.template.type: legacy
+
 # Template name. By default the template name is "auditbeat-%{[agent.version]}"
 # The template name and pattern has to be set in case the Elasticsearch index pattern is modified.
 #setup.template.name: "auditbeat-%{[agent.version]}"
diff --git a/filebeat/filebeat.reference.yml b/filebeat/filebeat.reference.yml
index bf66cfb98b9..bf29e0715ed 100644
--- a/filebeat/filebeat.reference.yml
+++ b/filebeat/filebeat.reference.yml
@@ -1870,6 +1870,11 @@ output.elasticsearch:
 # Set to false to disable template loading.
 #setup.template.enabled: true
 
+# Select the kind of index template. From Elasticsearch 7.8, it is possible to
+# use component templates. Available options: legacy, component, index.
+# By default filebeat uses the legacy index templates.
+#setup.template.type: legacy
+
 # Template name. By default the template name is "filebeat-%{[agent.version]}"
 # The template name and pattern has to be set in case the Elasticsearch index pattern is modified.
 #setup.template.name: "filebeat-%{[agent.version]}"
diff --git a/heartbeat/heartbeat.reference.yml b/heartbeat/heartbeat.reference.yml
index f3f1fea3b22..687552ef33b 100644
--- a/heartbeat/heartbeat.reference.yml
+++ b/heartbeat/heartbeat.reference.yml
@@ -1321,6 +1321,11 @@ output.elasticsearch:
 # Set to false to disable template loading.
 #setup.template.enabled: true
 
+# Select the kind of index template. From Elasticsearch 7.8, it is possible to
+# use component templates. Available options: legacy, component, index.
+# By default heartbeat uses the legacy index templates.
+#setup.template.type: legacy
+
 # Template name. By default the template name is "heartbeat-%{[agent.version]}"
 # The template name and pattern has to be set in case the Elasticsearch index pattern is modified.
 #setup.template.name: "heartbeat-%{[agent.version]}"
diff --git a/journalbeat/journalbeat.reference.yml b/journalbeat/journalbeat.reference.yml
index 08f4a04e008..4fe83567574 100644
--- a/journalbeat/journalbeat.reference.yml
+++ b/journalbeat/journalbeat.reference.yml
@@ -1086,6 +1086,11 @@ output.elasticsearch:
 # Set to false to disable template loading.
 #setup.template.enabled: true
 
+# Select the kind of index template. From Elasticsearch 7.8, it is possible to
+# use component templates. Available options: legacy, component, index.
+# By default journalbeat uses the legacy index templates.
+#setup.template.type: legacy
+
 # Template name. By default the template name is "journalbeat-%{[agent.version]}"
 # The template name and pattern has to be set in case the Elasticsearch index pattern is modified.
 #setup.template.name: "journalbeat-%{[agent.version]}"
diff --git a/libbeat/_meta/config/setup.template.reference.yml.tmpl b/libbeat/_meta/config/setup.template.reference.yml.tmpl
index 48d23d9d0c9..3f8dc077446 100644
--- a/libbeat/_meta/config/setup.template.reference.yml.tmpl
+++ b/libbeat/_meta/config/setup.template.reference.yml.tmpl
@@ -7,6 +7,11 @@
 # Set to false to disable template loading.
 #setup.template.enabled: true
 
+# Select the kind of index template. From Elasticsearch 7.8, it is possible to
+# use component templates. Available options: legacy, component, index.
+# By default {{.BeatName}} uses the legacy index templates.
+#setup.template.type: legacy
+
 # Template name. By default the template name is "{{.BeatIndexPrefix}}-%{[agent.version]}"
 # The template name and pattern has to be set in case the Elasticsearch index pattern is modified.
 #setup.template.name: "{{.BeatIndexPrefix}}-%{[agent.version]}"
diff --git a/libbeat/cmd/instance/metrics.go b/libbeat/cmd/instance/metrics.go
index 54cd3ab55d8..fa0d42bbeaf 100644
--- a/libbeat/cmd/instance/metrics.go
+++ b/libbeat/cmd/instance/metrics.go
@@ -28,6 +28,7 @@ import (
 	"github.com/elastic/beats/v7/libbeat/metric/system/cpu"
 	"github.com/elastic/beats/v7/libbeat/metric/system/process"
 	"github.com/elastic/beats/v7/libbeat/monitoring"
+	"github.com/elastic/gosigar/cgroup"
 )
 
 var (
@@ -65,10 +66,15 @@ func setupMetrics(name string) error {
 }
 
 func setupPlatformSpecificMetrics() {
+	switch runtime.GOOS {
+	case "linux":
+		monitoring.NewFunc(beatMetrics, "cgroup", reportBeatCgroups, monitoring.Report)
+	case "windows":
+		setupWindowsHandlesMetrics()
+	}
+
 	if runtime.GOOS != "windows" {
 		monitoring.NewFunc(systemMetrics, "load", reportSystemLoadAverage, monitoring.Report)
-	} else {
-		setupWindowsHandlesMetrics()
 	}
 
 	setupLinuxBSDFDMetrics()
@@ -254,3 +260,79 @@ func reportRuntime(_ monitoring.Mode, V monitoring.Visitor) {
 
 	monitoring.ReportInt(V, "goroutines", int64(runtime.NumGoroutine()))
 }
+
+func reportBeatCgroups(_ monitoring.Mode, V monitoring.Visitor) {
+	V.OnRegistryStart()
+	defer V.OnRegistryFinished()
+
+	pid, err := process.GetSelfPid()
+	if err != nil {
+		logp.Err("error getting PID for self process: %v", err)
+		return
+	}
+
+	cgroups, err := cgroup.NewReader("", true)
+	if err != nil {
+		if err == cgroup.ErrCgroupsMissing {
+			logp.Warn("cgroup data collection disabled: %v", err)
+		} else {
+			logp.Err("cgroup data collection disabled: %v", err)
+		}
+		return
+	}
+	selfStats, err := cgroups.GetStatsForProcess(pid)
+	if err != nil {
+		logp.Err("error getting group status: %v", err)
+		return
+	}
+
+	if cpu := selfStats.CPU; cpu != nil {
+		monitoring.ReportNamespace(V, "cpu", func() {
+			if cpu.ID != "" {
+				monitoring.ReportString(V, "id", cpu.ID)
+			}
+			monitoring.ReportNamespace(V, "cfs", func() {
+				monitoring.ReportNamespace(V, "period", func() {
+					monitoring.ReportInt(V, "us", int64(cpu.CFS.PeriodMicros))
+				})
+				monitoring.ReportNamespace(V, "quota", func() {
+					monitoring.ReportInt(V, "us", int64(cpu.CFS.QuotaMicros))
+				})
+			})
+			monitoring.ReportNamespace(V, "stats", func() {
+				monitoring.ReportInt(V, "periods", int64(cpu.Stats.Periods))
+				monitoring.ReportNamespace(V, "throttled", func() {
+					monitoring.ReportInt(V, "periods", int64(cpu.Stats.ThrottledPeriods))
+					monitoring.ReportInt(V, "ns", int64(cpu.Stats.ThrottledTimeNanos))
+				})
+			})
+		})
+	}
+
+	if cpuacct := selfStats.CPUAccounting; cpuacct != nil {
+		monitoring.ReportNamespace(V, "cpuacct", func() {
+			if cpuacct.ID != "" {
+				monitoring.ReportString(V, "id", cpuacct.ID)
+			}
+			monitoring.ReportNamespace(V, "total", func() {
+				monitoring.ReportInt(V, "ns", int64(cpuacct.TotalNanos))
+			})
+		})
+	}
+
+	if memory := selfStats.Memory; memory != nil {
+		monitoring.ReportNamespace(V, "memory", func() {
+			if memory.ID != "" {
+				monitoring.ReportString(V, "id", memory.ID)
+			}
+			monitoring.ReportNamespace(V, "mem", func() {
+				monitoring.ReportNamespace(V, "limit", func() {
+					monitoring.ReportInt(V, "bytes", int64(memory.Mem.Limit))
+				})
+				monitoring.ReportNamespace(V, "usage", func() {
+					monitoring.ReportInt(V, "bytes", int64(memory.Mem.Usage))
+				})
+			})
+		})
+	}
+}
diff --git a/libbeat/docs/template-config.asciidoc b/libbeat/docs/template-config.asciidoc
index 3271d567c2a..9ac888503d1 100644
--- a/libbeat/docs/template-config.asciidoc
+++ b/libbeat/docs/template-config.asciidoc
@@ -26,6 +26,11 @@ existing one.
 *`setup.template.enabled`*:: Set to false to disable template loading. If set this to false,
 you must <<load-template-manually,load the template manually>>.
 
+*`setup.template.type`*:: The type of template to use. Available options: `legacy` (default), index templates
+before Elasticsearch v7.8. Use this to avoid breaking existing deployments. New options are `composite`
+and `index`. Selecting `component` loads a component template which can be included in new index templates.
+The option `index` loads the new index template.
+
 *`setup.template.name`*:: The name of the template. The default is
 +{beatname_lc}+. The {beatname_uc} version is always appended to the given
 name, so the final name is +{beatname_lc}-%{[{beat_version_key}]}+.
diff --git a/libbeat/template/config.go b/libbeat/template/config.go
index c9e963d6135..7eb6ff522f4 100644
--- a/libbeat/template/config.go
+++ b/libbeat/template/config.go
@@ -17,7 +17,27 @@
 
 package template
 
-import "github.com/elastic/beats/v7/libbeat/mapping"
+import (
+	"fmt"
+
+	"github.com/elastic/beats/v7/libbeat/mapping"
+)
+
+const (
+	IndexTemplateLegacy IndexTemplateType = iota
+	IndexTemplateComponent
+	IndexTemplateIndex
+)
+
+var (
+	templateTypes = map[string]IndexTemplateType{
+		"legacy":    IndexTemplateLegacy,
+		"component": IndexTemplateComponent,
+		"index":     IndexTemplateIndex,
+	}
+)
+
+type IndexTemplateType uint8
 
 // TemplateConfig holds config information about the Elasticsearch template
 type TemplateConfig struct {
@@ -30,10 +50,12 @@ type TemplateConfig struct {
 		Path    string `config:"path"`
 		Name    string `config:"name"`
 	} `config:"json"`
-	AppendFields mapping.Fields   `config:"append_fields"`
-	Overwrite    bool             `config:"overwrite"`
-	Settings     TemplateSettings `config:"settings"`
-	Order        int              `config:"order"`
+	AppendFields mapping.Fields    `config:"append_fields"`
+	Overwrite    bool              `config:"overwrite"`
+	Settings     TemplateSettings  `config:"settings"`
+	Order        int               `config:"order"`
+	Priority     int               `config:"priority"`
+	Type         IndexTemplateType `config:"type"`
 }
 
 // TemplateSettings are part of the Elasticsearch template and hold index and source specific information.
@@ -45,8 +67,26 @@ type TemplateSettings struct {
 // DefaultConfig for index template
 func DefaultConfig() TemplateConfig {
 	return TemplateConfig{
-		Enabled: true,
-		Fields:  "",
-		Order:   1,
+		Enabled:  true,
+		Fields:   "",
+		Type:     IndexTemplateLegacy,
+		Order:    1,
+		Priority: 150,
 	}
 }
+
+func (t *IndexTemplateType) Unpack(v string) error {
+	if v == "" {
+		*t = IndexTemplateLegacy
+		return nil
+	}
+
+	var tt IndexTemplateType
+	var ok bool
+	if tt, ok = templateTypes[v]; !ok {
+		return fmt.Errorf("unknown index template type: %s", v)
+	}
+	*t = tt
+
+	return nil
+}
diff --git a/libbeat/template/load.go b/libbeat/template/load.go
index b018fda9788..5b0e0f58440 100644
--- a/libbeat/template/load.go
+++ b/libbeat/template/load.go
@@ -31,6 +31,14 @@ import (
 	"github.com/elastic/beats/v7/libbeat/paths"
 )
 
+var (
+	templateLoaderPath = map[IndexTemplateType]string{
+		IndexTemplateLegacy:    "/_template/",
+		IndexTemplateComponent: "/_component_template/",
+		IndexTemplateIndex:     "/_index_template/",
+	}
+)
+
 //Loader interface for loading templates
 type Loader interface {
 	Load(config TemplateConfig, info beat.Info, fields []byte, migration bool) error
@@ -97,7 +105,7 @@ func (l *ESLoader) Load(config TemplateConfig, info beat.Info, fields []byte, mi
 		templateName = config.JSON.Name
 	}
 
-	if l.templateExists(templateName) && !config.Overwrite {
+	if l.templateExists(templateName, config.Type) && !config.Overwrite {
 		l.log.Infof("Template %s already exists and will not be overwritten.", templateName)
 		return nil
 	}
@@ -107,7 +115,7 @@ func (l *ESLoader) Load(config TemplateConfig, info beat.Info, fields []byte, mi
 	if err != nil {
 		return err
 	}
-	if err := l.loadTemplate(templateName, body); err != nil {
+	if err := l.loadTemplate(templateName, config.Type, body); err != nil {
 		return fmt.Errorf("could not load template. Elasticsearch returned: %v. Template is: %s", err, body.StringToPrint())
 	}
 	l.log.Infof("template with name '%s' loaded.", templateName)
@@ -117,10 +125,11 @@ func (l *ESLoader) Load(config TemplateConfig, info beat.Info, fields []byte, mi
 // loadTemplate loads a template into Elasticsearch overwriting the existing
 // template if it exists. If you wish to not overwrite an existing template
 // then use CheckTemplate prior to calling this method.
-func (l *ESLoader) loadTemplate(templateName string, template map[string]interface{}) error {
+func (l *ESLoader) loadTemplate(templateName string, templateType IndexTemplateType, template map[string]interface{}) error {
 	l.log.Infof("Try loading template %s to Elasticsearch", templateName)
-	path := "/_template/" + templateName
-	params := esVersionParams(l.client.GetVersion())
+	clientVersion := l.client.GetVersion()
+	path := templateLoaderPath[templateType] + templateName
+	params := esVersionParams(clientVersion)
 	status, body, err := l.client.Request("PUT", path, "", params, template)
 	if err != nil {
 		return fmt.Errorf("couldn't load template: %v. Response body: %s", err, body)
@@ -133,11 +142,16 @@ func (l *ESLoader) loadTemplate(templateName string, template map[string]interfa
 
 // templateExists checks if a given template already exist. It returns true if
 // and only if Elasticsearch returns with HTTP status code 200.
-func (l *ESLoader) templateExists(templateName string) bool {
+func (l *ESLoader) templateExists(templateName string, templateType IndexTemplateType) bool {
 	if l.client == nil {
 		return false
 	}
 
+	if templateType == IndexTemplateComponent {
+		status, _, _ := l.client.Request("GET", "/_component_template/"+templateName, "", nil, nil)
+		return status == http.StatusOK
+	}
+
 	status, body, _ := l.client.Request("GET", "/_cat/templates/"+templateName, "", nil, nil)
 
 	return status == http.StatusOK && strings.Contains(string(body), templateName)
diff --git a/libbeat/template/load_integration_test.go b/libbeat/template/load_integration_test.go
index fc3d66cab4c..7a75b5d4f46 100644
--- a/libbeat/template/load_integration_test.go
+++ b/libbeat/template/load_integration_test.go
@@ -65,8 +65,8 @@ func newTestSetup(t *testing.T, cfg TemplateConfig) *testSetup {
 		t.Fatal(err)
 	}
 	s := testSetup{t: t, client: client, loader: NewESLoader(client), config: cfg}
-	client.Request("DELETE", "/_template/"+cfg.Name, "", nil, nil)
-	require.False(t, s.loader.templateExists(cfg.Name))
+	client.Request("DELETE", templateLoaderPath[cfg.Type]+cfg.Name, "", nil, nil)
+	require.False(t, s.loader.templateExists(cfg.Name, cfg.Type))
 	return &s
 }
 func (ts *testSetup) loadFromFile(fileElems []string) error {
@@ -82,7 +82,7 @@ func (ts *testSetup) load(fields []byte) error {
 
 func (ts *testSetup) mustLoad(fields []byte) {
 	require.NoError(ts.t, ts.load(fields))
-	require.True(ts.t, ts.loader.templateExists(ts.config.Name))
+	require.True(ts.t, ts.loader.templateExists(ts.config.Name, ts.config.Type))
 }
 
 func TestESLoader_Load(t *testing.T) {
@@ -91,7 +91,7 @@ func TestESLoader_Load(t *testing.T) {
 			setup := newTestSetup(t, TemplateConfig{Enabled: false})
 
 			setup.load(nil)
-			assert.False(t, setup.loader.templateExists(setup.config.Name))
+			assert.False(t, setup.loader.templateExists(setup.config.Name, setup.config.Type))
 		})
 
 		t.Run("invalid version", func(t *testing.T) {
@@ -115,14 +115,14 @@ func TestESLoader_Load(t *testing.T) {
 
 		t.Run("disabled", func(t *testing.T) {
 			setup.load(nil)
-			tmpl := getTemplate(t, setup.client, setup.config.Name)
+			tmpl := getTemplate(t, setup.client, setup.config.Name, setup.config.Type)
 			assert.Equal(t, true, tmpl.SourceEnabled())
 		})
 
 		t.Run("enabled", func(t *testing.T) {
 			setup.config.Overwrite = true
 			setup.load(nil)
-			tmpl := getTemplate(t, setup.client, setup.config.Name)
+			tmpl := getTemplate(t, setup.client, setup.config.Name, setup.config.Type)
 			assert.Equal(t, false, tmpl.SourceEnabled())
 		})
 	})
@@ -140,7 +140,7 @@ func TestESLoader_Load(t *testing.T) {
 			Name    string `config:"name"`
 		}{Enabled: true, Path: path(t, []string{"testdata", "fields.json"}), Name: nameJSON}
 		setup.load(nil)
-		assert.True(t, setup.loader.templateExists(nameJSON))
+		assert.True(t, setup.loader.templateExists(nameJSON, setup.config.Type))
 	})
 
 	t.Run("load template successful", func(t *testing.T) {
@@ -157,10 +157,19 @@ func TestESLoader_Load(t *testing.T) {
 				fields:     fields,
 				properties: []string{"foo", "bar"},
 			},
+			"default config with fields and component": {
+				cfg:        TemplateConfig{Enabled: true, Type: IndexTemplateComponent},
+				fields:     fields,
+				properties: []string{"foo", "bar"},
+			},
 			"minimal template": {
 				cfg:    TemplateConfig{Enabled: true},
 				fields: nil,
 			},
+			"minimal template component": {
+				cfg:    TemplateConfig{Enabled: true, Type: IndexTemplateComponent},
+				fields: nil,
+			},
 			"fields from file": {
 				cfg:        TemplateConfig{Enabled: true, Fields: path(t, []string{"testdata", "fields.yml"})},
 				fields:     fields,
@@ -181,7 +190,7 @@ func TestESLoader_Load(t *testing.T) {
 				setup.mustLoad(data.fields)
 
 				// Fetch properties
-				tmpl := getTemplate(t, setup.client, setup.config.Name)
+				tmpl := getTemplate(t, setup.client, setup.config.Name, setup.config.Type)
 				val, err := tmpl.GetValue("mappings.properties")
 				if data.properties == nil {
 					assert.Error(t, err)
@@ -203,7 +212,7 @@ func TestESLoader_Load(t *testing.T) {
 func TestTemplate_LoadFile(t *testing.T) {
 	setup := newTestSetup(t, TemplateConfig{Enabled: true})
 	assert.NoError(t, setup.loadFromFile([]string{"..", "fields.yml"}))
-	assert.True(t, setup.loader.templateExists(setup.config.Name))
+	assert.True(t, setup.loader.templateExists(setup.config.Name, setup.config.Type))
 }
 
 func TestLoadInvalidTemplate(t *testing.T) {
@@ -211,9 +220,9 @@ func TestLoadInvalidTemplate(t *testing.T) {
 
 	// Try to load invalid template
 	template := map[string]interface{}{"json": "invalid"}
-	err := setup.loader.loadTemplate(setup.config.Name, template)
+	err := setup.loader.loadTemplate(setup.config.Name, setup.config.Type, template)
 	assert.Error(t, err)
-	assert.False(t, setup.loader.templateExists(setup.config.Name))
+	assert.False(t, setup.loader.templateExists(setup.config.Name, setup.config.Type))
 }
 
 // Tests loading the templates for each beat
@@ -225,7 +234,7 @@ func TestLoadBeatsTemplate_fromFile(t *testing.T) {
 	for _, beat := range beats {
 		setup := newTestSetup(t, TemplateConfig{Name: beat, Enabled: true})
 		assert.NoError(t, setup.loadFromFile([]string{"..", "..", beat, "fields.yml"}))
-		assert.True(t, setup.loader.templateExists(setup.config.Name))
+		assert.True(t, setup.loader.templateExists(setup.config.Name, setup.config.Type))
 	}
 }
 
@@ -238,7 +247,7 @@ func TestTemplateSettings(t *testing.T) {
 	require.NoError(t, setup.loadFromFile([]string{"..", "fields.yml"}))
 
 	// Check that it contains the mapping
-	templateJSON := getTemplate(t, setup.client, setup.config.Name)
+	templateJSON := getTemplate(t, setup.client, setup.config.Name, setup.config.Type)
 	assert.Equal(t, 1, templateJSON.NumberOfShards())
 	assert.Equal(t, false, templateJSON.SourceEnabled())
 }
@@ -289,7 +298,7 @@ var dataTests = []struct {
 func TestTemplateWithData(t *testing.T) {
 	setup := newTestSetup(t, TemplateConfig{Enabled: true})
 	require.NoError(t, setup.loadFromFile([]string{"testdata", "fields.yml"}))
-	require.True(t, setup.loader.templateExists(setup.config.Name))
+	require.True(t, setup.loader.templateExists(setup.config.Name, setup.config.Type))
 	esClient := setup.client.(*eslegclient.Connection)
 	for _, test := range dataTests {
 		_, _, err := esClient.Index(setup.config.Name, "_doc", "", nil, test.data)
@@ -302,14 +311,29 @@ func TestTemplateWithData(t *testing.T) {
 	}
 }
 
-func getTemplate(t *testing.T, client ESClient, templateName string) testTemplate {
-	status, body, err := client.Request("GET", "/_template/"+templateName, "", nil, nil)
+func getTemplate(t *testing.T, client ESClient, templateName string, templateType IndexTemplateType) testTemplate {
+	status, body, err := client.Request("GET", templateLoaderPath[templateType]+templateName, "", nil, nil)
 	require.NoError(t, err)
 	require.Equal(t, status, 200)
 
 	var response common.MapStr
 	err = json.Unmarshal(body, &response)
 	require.NoError(t, err)
+	require.NotNil(t, response)
+
+	if templateType == IndexTemplateComponent {
+		var tmpl map[string]interface{}
+		components := response["component_templates"].([]interface{})
+		for _, ct := range components {
+			componentTemplate := ct.(map[string]interface{})["component_template"].(map[string]interface{})
+			tmpl = componentTemplate["template"].(map[string]interface{})
+		}
+		return testTemplate{
+			t:      t,
+			client: client,
+			MapStr: common.MapStr(tmpl),
+		}
+	}
 
 	return testTemplate{
 		t:      t,
diff --git a/libbeat/template/template.go b/libbeat/template/template.go
index dac3a920196..2aaa7712d02 100644
--- a/libbeat/template/template.go
+++ b/libbeat/template/template.go
@@ -46,14 +46,16 @@ var (
 // Template holds information for the ES template.
 type Template struct {
 	sync.Mutex
-	name        string
-	pattern     string
-	beatVersion common.Version
-	beatName    string
-	esVersion   common.Version
-	config      TemplateConfig
-	migration   bool
-	order       int
+	name         string
+	pattern      string
+	beatVersion  common.Version
+	beatName     string
+	esVersion    common.Version
+	config       TemplateConfig
+	migration    bool
+	templateType IndexTemplateType
+	order        int
+	priority     int
 }
 
 // New creates a new template instance
@@ -123,14 +125,16 @@ func New(
 	}
 
 	return &Template{
-		pattern:     pattern,
-		name:        name,
-		beatVersion: *bV,
-		esVersion:   esVersion,
-		beatName:    beatName,
-		config:      config,
-		migration:   migration,
-		order:       config.Order,
+		pattern:      pattern,
+		name:         name,
+		beatVersion:  *bV,
+		esVersion:    esVersion,
+		beatName:     beatName,
+		config:       config,
+		migration:    migration,
+		templateType: config.Type,
+		order:        config.Order,
+		priority:     config.Priority,
 	}, nil
 }
 
@@ -184,23 +188,56 @@ func (t *Template) LoadBytes(data []byte) (common.MapStr, error) {
 
 // LoadMinimal loads the template only with the given configuration
 func (t *Template) LoadMinimal() (common.MapStr, error) {
-	keyPattern, patterns := buildPatternSettings(t.esVersion, t.GetPattern())
-	m := common.MapStr{
-		keyPattern: patterns,
-		"order":    t.order,
-		"settings": common.MapStr{
-			"index": t.config.Settings.Index,
-		},
+	m := common.MapStr{}
+	switch t.templateType {
+	case IndexTemplateLegacy:
+		m = t.loadMinimalLegacy()
+	case IndexTemplateComponent:
+		m = t.loadMinimalComponent()
+	case IndexTemplateIndex:
+		m = t.loadMinimalIndex()
+	default:
+		return nil, fmt.Errorf("unknown template type %v", t.templateType)
 	}
+
 	if t.config.Settings.Source != nil {
 		m["mappings"] = buildMappings(
 			t.beatVersion, t.esVersion, t.beatName,
 			nil, nil,
 			common.MapStr(t.config.Settings.Source))
 	}
+
 	return m, nil
 }
 
+func (t *Template) loadMinimalLegacy() common.MapStr {
+	keyPattern, patterns := buildPatternSettings(t.esVersion, t.GetPattern())
+	return common.MapStr{
+		keyPattern: patterns,
+		"order":    t.order,
+		"settings": common.MapStr{
+			"index": t.config.Settings.Index,
+		},
+	}
+}
+
+func (t *Template) loadMinimalComponent() common.MapStr {
+	return common.MapStr{
+		"template": common.MapStr{
+			"settings": common.MapStr{
+				"index": t.config.Settings.Index,
+			},
+		},
+	}
+}
+
+func (t *Template) loadMinimalIndex() common.MapStr {
+	m := t.loadMinimalLegacy()
+	m["priority"] = t.priority
+	delete(m, "order")
+	return m
+}
+
 // GetName returns the name of the template
 func (t *Template) GetName() string {
 	return t.name
@@ -214,6 +251,19 @@ func (t *Template) GetPattern() string {
 // Generate generates the full template
 // The default values are taken from the default variable.
 func (t *Template) Generate(properties common.MapStr, dynamicTemplates []common.MapStr) common.MapStr {
+	switch t.templateType {
+	case IndexTemplateLegacy:
+		return t.generateLegacy(properties)
+	case IndexTemplateComponent:
+		return t.generateComponent(properties)
+	case IndexTemplateIndex:
+		return t.generateIndex(properties)
+	default:
+	}
+	return nil
+}
+
+func (t *Template) generateLegacy(properties common.MapStr) common.MapStr {
 	keyPattern, patterns := buildPatternSettings(t.esVersion, t.GetPattern())
 	return common.MapStr{
 		keyPattern: patterns,
@@ -232,6 +282,31 @@ func (t *Template) Generate(properties common.MapStr, dynamicTemplates []common.
 	}
 }
 
+func (t *Template) generateComponent(properties common.MapStr) common.MapStr {
+	return common.MapStr{
+		"template": common.MapStr{
+			"mappings": buildMappings(
+				t.beatVersion, t.esVersion, t.beatName,
+				properties,
+				append(dynamicTemplates, buildDynTmpl(t.esVersion)),
+				common.MapStr(t.config.Settings.Source)),
+			"settings": common.MapStr{
+				"index": buildIdxSettings(
+					t.esVersion,
+					t.config.Settings.Index,
+				),
+			},
+		},
+	}
+}
+
+func (t *Template) generateIndex(properties common.MapStr) common.MapStr {
+	tmpl := t.generateLegacy(properties)
+	tmpl["priority"] = t.priority
+	delete(tmpl, "order")
+	return tmpl
+}
+
 func buildPatternSettings(ver common.Version, pattern string) (string, interface{}) {
 	if ver.Major < 6 {
 		return "template", pattern
diff --git a/metricbeat/metricbeat.reference.yml b/metricbeat/metricbeat.reference.yml
index f40a7d572ea..9b6f37eb447 100644
--- a/metricbeat/metricbeat.reference.yml
+++ b/metricbeat/metricbeat.reference.yml
@@ -1910,6 +1910,11 @@ output.elasticsearch:
 # Set to false to disable template loading.
 #setup.template.enabled: true
 
+# Select the kind of index template. From Elasticsearch 7.8, it is possible to
+# use component templates. Available options: legacy, component, index.
+# By default metricbeat uses the legacy index templates.
+#setup.template.type: legacy
+
 # Template name. By default the template name is "metricbeat-%{[agent.version]}"
 # The template name and pattern has to be set in case the Elasticsearch index pattern is modified.
 #setup.template.name: "metricbeat-%{[agent.version]}"
diff --git a/packetbeat/packetbeat.reference.yml b/packetbeat/packetbeat.reference.yml
index 6dc0f3a01d6..0dc551698e9 100644
--- a/packetbeat/packetbeat.reference.yml
+++ b/packetbeat/packetbeat.reference.yml
@@ -1570,6 +1570,11 @@ output.elasticsearch:
 # Set to false to disable template loading.
 #setup.template.enabled: true
 
+# Select the kind of index template. From Elasticsearch 7.8, it is possible to
+# use component templates. Available options: legacy, component, index.
+# By default packetbeat uses the legacy index templates.
+#setup.template.type: legacy
+
 # Template name. By default the template name is "packetbeat-%{[agent.version]}"
 # The template name and pattern has to be set in case the Elasticsearch index pattern is modified.
 #setup.template.name: "packetbeat-%{[agent.version]}"
diff --git a/winlogbeat/winlogbeat.reference.yml b/winlogbeat/winlogbeat.reference.yml
index c76f369eb35..71ec0007631 100644
--- a/winlogbeat/winlogbeat.reference.yml
+++ b/winlogbeat/winlogbeat.reference.yml
@@ -1066,6 +1066,11 @@ output.elasticsearch:
 # Set to false to disable template loading.
 #setup.template.enabled: true
 
+# Select the kind of index template. From Elasticsearch 7.8, it is possible to
+# use component templates. Available options: legacy, component, index.
+# By default winlogbeat uses the legacy index templates.
+#setup.template.type: legacy
+
 # Template name. By default the template name is "winlogbeat-%{[agent.version]}"
 # The template name and pattern has to be set in case the Elasticsearch index pattern is modified.
 #setup.template.name: "winlogbeat-%{[agent.version]}"
diff --git a/x-pack/auditbeat/auditbeat.reference.yml b/x-pack/auditbeat/auditbeat.reference.yml
index 041f460ddd1..f2167a6293e 100644
--- a/x-pack/auditbeat/auditbeat.reference.yml
+++ b/x-pack/auditbeat/auditbeat.reference.yml
@@ -1200,6 +1200,11 @@ output.elasticsearch:
 # Set to false to disable template loading.
 #setup.template.enabled: true
 
+# Select the kind of index template. From Elasticsearch 7.8, it is possible to
+# use component templates. Available options: legacy, component, index.
+# By default auditbeat uses the legacy index templates.
+#setup.template.type: legacy
+
 # Template name. By default the template name is "auditbeat-%{[agent.version]}"
 # The template name and pattern has to be set in case the Elasticsearch index pattern is modified.
 #setup.template.name: "auditbeat-%{[agent.version]}"
diff --git a/x-pack/elastic-agent/CHANGELOG.next.asciidoc b/x-pack/elastic-agent/CHANGELOG.next.asciidoc
index c466d0c656d..3963dda9a12 100644
--- a/x-pack/elastic-agent/CHANGELOG.next.asciidoc
+++ b/x-pack/elastic-agent/CHANGELOG.next.asciidoc
@@ -10,6 +10,7 @@
 - Docker container is not run as root by default. {pull}21213[21213]
 
 ==== Bugfixes
+- Thread safe sorted set {pull}21290[21290]
 - Copy Action store on upgrade {pull}21298[21298]
 - Include inputs in action store actions {pull}21298[21298]
 
diff --git a/x-pack/elastic-agent/pkg/sorted/set.go b/x-pack/elastic-agent/pkg/sorted/set.go
index 739e525aac5..38099963fcf 100644
--- a/x-pack/elastic-agent/pkg/sorted/set.go
+++ b/x-pack/elastic-agent/pkg/sorted/set.go
@@ -6,6 +6,7 @@ package sorted
 
 import (
 	"sort"
+	"sync"
 )
 
 // Set is a sorted set that allow to iterate on they keys in an ordered manner, when
@@ -13,6 +14,7 @@ import (
 type Set struct {
 	mapped map[string]interface{}
 	keys   []string
+	rwlock sync.RWMutex
 }
 
 // NewSet returns an ordered set.
@@ -24,6 +26,9 @@ func NewSet() *Set {
 
 // Add adds an items to the set.
 func (s *Set) Add(k string, v interface{}) {
+	s.rwlock.Lock()
+	defer s.rwlock.Unlock()
+
 	_, ok := s.mapped[k]
 	if !ok {
 		s.keys = append(s.keys, k)
@@ -35,6 +40,9 @@ func (s *Set) Add(k string, v interface{}) {
 
 // Remove removes an items from the Set.
 func (s *Set) Remove(k string) {
+	s.rwlock.Lock()
+	defer s.rwlock.Unlock()
+
 	_, ok := s.mapped[k]
 	if !ok {
 		return
@@ -50,11 +58,17 @@ func (s *Set) Remove(k string) {
 
 // Get retrieves a specific values from the map and will return false if the key is not found.
 func (s *Set) Get(k string) (interface{}, bool) {
+	s.rwlock.RLock()
+	defer s.rwlock.RUnlock()
+
 	v, ok := s.mapped[k]
 	return v, ok
 }
 
 // Keys returns slice of keys where the keys are ordered alphabetically.
 func (s *Set) Keys() []string {
+	s.rwlock.RLock()
+	defer s.rwlock.RUnlock()
+
 	return append(s.keys[:0:0], s.keys...)
 }
diff --git a/x-pack/filebeat/filebeat.reference.yml b/x-pack/filebeat/filebeat.reference.yml
index 89beee34729..9797291bdf4 100644
--- a/x-pack/filebeat/filebeat.reference.yml
+++ b/x-pack/filebeat/filebeat.reference.yml
@@ -3305,6 +3305,11 @@ output.elasticsearch:
 # Set to false to disable template loading.
 #setup.template.enabled: true
 
+# Select the kind of index template. From Elasticsearch 7.8, it is possible to
+# use component templates. Available options: legacy, component, index.
+# By default filebeat uses the legacy index templates.
+#setup.template.type: legacy
+
 # Template name. By default the template name is "filebeat-%{[agent.version]}"
 # The template name and pattern has to be set in case the Elasticsearch index pattern is modified.
 #setup.template.name: "filebeat-%{[agent.version]}"
diff --git a/x-pack/functionbeat/functionbeat.reference.yml b/x-pack/functionbeat/functionbeat.reference.yml
index f20f73e8bfd..a55fcc56e23 100644
--- a/x-pack/functionbeat/functionbeat.reference.yml
+++ b/x-pack/functionbeat/functionbeat.reference.yml
@@ -1051,6 +1051,11 @@ output.elasticsearch:
 # Set to false to disable template loading.
 #setup.template.enabled: true
 
+# Select the kind of index template. From Elasticsearch 7.8, it is possible to
+# use component templates. Available options: legacy, component, index.
+# By default functionbeat uses the legacy index templates.
+#setup.template.type: legacy
+
 # Template name. By default the template name is "functionbeat-%{[agent.version]}"
 # The template name and pattern has to be set in case the Elasticsearch index pattern is modified.
 #setup.template.name: "functionbeat-%{[agent.version]}"
diff --git a/x-pack/heartbeat/heartbeat.reference.yml b/x-pack/heartbeat/heartbeat.reference.yml
index f3f1fea3b22..687552ef33b 100644
--- a/x-pack/heartbeat/heartbeat.reference.yml
+++ b/x-pack/heartbeat/heartbeat.reference.yml
@@ -1321,6 +1321,11 @@ output.elasticsearch:
 # Set to false to disable template loading.
 #setup.template.enabled: true
 
+# Select the kind of index template. From Elasticsearch 7.8, it is possible to
+# use component templates. Available options: legacy, component, index.
+# By default heartbeat uses the legacy index templates.
+#setup.template.type: legacy
+
 # Template name. By default the template name is "heartbeat-%{[agent.version]}"
 # The template name and pattern has to be set in case the Elasticsearch index pattern is modified.
 #setup.template.name: "heartbeat-%{[agent.version]}"
diff --git a/x-pack/metricbeat/metricbeat.reference.yml b/x-pack/metricbeat/metricbeat.reference.yml
index 507d8492485..ff9bffda33e 100644
--- a/x-pack/metricbeat/metricbeat.reference.yml
+++ b/x-pack/metricbeat/metricbeat.reference.yml
@@ -2382,6 +2382,11 @@ output.elasticsearch:
 # Set to false to disable template loading.
 #setup.template.enabled: true
 
+# Select the kind of index template. From Elasticsearch 7.8, it is possible to
+# use component templates. Available options: legacy, component, index.
+# By default metricbeat uses the legacy index templates.
+#setup.template.type: legacy
+
 # Template name. By default the template name is "metricbeat-%{[agent.version]}"
 # The template name and pattern has to be set in case the Elasticsearch index pattern is modified.
 #setup.template.name: "metricbeat-%{[agent.version]}"
diff --git a/x-pack/winlogbeat/winlogbeat.reference.yml b/x-pack/winlogbeat/winlogbeat.reference.yml
index 773fde0fab4..62dab077c43 100644
--- a/x-pack/winlogbeat/winlogbeat.reference.yml
+++ b/x-pack/winlogbeat/winlogbeat.reference.yml
@@ -1109,6 +1109,11 @@ output.elasticsearch:
 # Set to false to disable template loading.
 #setup.template.enabled: true
 
+# Select the kind of index template. From Elasticsearch 7.8, it is possible to
+# use component templates. Available options: legacy, component, index.
+# By default winlogbeat uses the legacy index templates.
+#setup.template.type: legacy
+
 # Template name. By default the template name is "winlogbeat-%{[agent.version]}"
 # The template name and pattern has to be set in case the Elasticsearch index pattern is modified.
 #setup.template.name: "winlogbeat-%{[agent.version]}"