Configurable index template loading (elastic#21212)

## What does this PR do?

The PR adds a new configuration option named `setup.template.type` to select the index template type. From ES v7.8 new index templates were introduced. Possible option:

* `legacy`: Loads the legacy index template. This is the default option, so it does not break existing deployments.
* `component`: This loads Beats' index template as a composite template, so it can be used in the users' index templates.
* `index`: Loads the new index template.

## Why is it important?

Index templates v2 was released in Elasticsearch 7.8. Previously Beats had used the legacy endpoint for installing index templates. Now we are moving to the newer version.

Closes elastic#17829

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*
 

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]}"

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]}"

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]}"

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]}"

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]}"

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}]}+.

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
+}

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)

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,17 +212,17 @@ 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) {
 	setup := newTestSetup(t, TemplateConfig{})
 
 	// 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,