Cherry-pick #24332 to 7.x: Do not drop errors when checking for templ…

…ates (#24451)

CHANGELOG.next.asciidoc

@@ -98,6 +98,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 
 *Affecting all Beats*
 
+- Fix templates being overwritten if there was an error when check for the template existance. {pull}24332[24332]
 - Fix a race condition with the Kafka pipeline client, it is possible that `Close()` get called before `Connect()` . {issue}11945[11945]
 - Allow users to configure only `cluster_uuid` setting under `monitoring` namespace. {pull}14338[14338]
 - Update replicaset group to apps/v1 {pull}15854[15802]

libbeat/template/load.go

@@ -19,6 +19,7 @@ package template
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io/ioutil"
 	"net/http"
@@ -67,6 +68,10 @@ type FileClient interface {
 	Write(component string, name string, body string) error
 }
 
+type StatusError struct {
+	status int
+}
+
 // NewESLoader creates a new template loader for ES
 func NewESLoader(client ESClient) *ESLoader {
 	return &ESLoader{client: client}
@@ -81,6 +86,10 @@ func NewFileLoader(c FileClient) *FileLoader {
 // In case the template is not already loaded or overwriting is enabled, the
 // template is built and written to index
 func (l *ESLoader) Load(config TemplateConfig, info beat.Info, fields []byte, migration bool) error {
+	if l.client == nil {
+		return errors.New("can not load template without active Elasticsearch client")
+	}
+
 	//build template from config
 	tmpl, err := template(config, info, l.client.GetVersion(), migration)
 	if err != nil || tmpl == nil {
@@ -93,7 +102,12 @@ func (l *ESLoader) Load(config TemplateConfig, info beat.Info, fields []byte, mi
 		templateName = config.JSON.Name
 	}
 
-	if l.templateExists(templateName, config.Type) && !config.Overwrite {
+	exists, err := l.templateExists(templateName, config.Type)
+	if err != nil {
+		return fmt.Errorf("failure while checking if template exists: %w", err)
+	}
+
+	if exists && !config.Overwrite {
 		logp.Info("Template %s already exists and will not be overwritten.", templateName)
 		return nil
 	}
@@ -128,21 +142,52 @@ func (l *ESLoader) loadTemplate(templateName string, templateType IndexTemplateT
 	return nil
 }
 
-// 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, templateType IndexTemplateType) bool {
-	if l.client == nil {
-		return false
+func (l *ESLoader) templateExists(templateName string, templateType IndexTemplateType) (bool, error) {
+	if templateType == IndexTemplateComponent {
+		return l.checkExistsComponentTemplate(templateName)
 	}
+	return l.checkExistsTemplate(templateName)
+}
 
-	if templateType == IndexTemplateComponent {
-		status, _, _ := l.client.Request("GET", "/_component_template/"+templateName, "", nil, nil)
-		return status == http.StatusOK
+// existsTemplate checks if a given template already exist, using the
+// `_cat/templates/<name>` API.
+//
+// An error is returned if the loader failed to execute the request, or a
+// status code indicating some problems is encountered.
+func (l *ESLoader) checkExistsTemplate(name string) (bool, error) {
+	status, body, err := l.client.Request("GET", "/_cat/templates/"+name, "", nil, nil)
+	if err != nil {
+		return false, err
 	}
 
-	status, body, _ := l.client.Request("GET", "/_cat/templates/"+templateName, "", nil, nil)
+	// Elasticsearch API returns 200, even if the template does not exists. We
+	// need to validate the body to be sure the template is actually known. Any
+	// status code other than 200 will be treated as error.
+	if status != http.StatusOK {
+		return false, &StatusError{status: status}
+	}
+	return strings.Contains(string(body), name), nil
+}
 
-	return status == http.StatusOK && strings.Contains(string(body), templateName)
+// existsComponentTemplate checks if a component template exists by querying
+// the `_component_template/<name>` API.
+//
+// The resource is assumed as present if a 200 OK status is returned and missing if a 404 is returned.
+// Other status codes or IO errors during the request are reported as error.
+func (l *ESLoader) checkExistsComponentTemplate(name string) (bool, error) {
+	status, _, err := l.client.Request("GET", "/_component_template/"+name, "", nil, nil)
+
+	switch status {
+	case http.StatusNotFound:
+		return false, nil
+	case http.StatusOK:
+		return true, nil
+	default:
+		if err == nil {
+			err = &StatusError{status: status}
+		}
+		return false, err
+	}
 }
 
 // Load reads the template from the config, creates the template body and prints it to the configured file.
@@ -242,6 +287,10 @@ func buildMinimalTemplate(tmpl *Template) (common.MapStr, error) {
 	return body, nil
 }
 
+func (e *StatusError) Error() string {
+	return fmt.Sprintf("request failed with http status code %v", e.status)
+}
+
 func esVersionParams(ver common.Version) map[string]string {
 	if ver.Major == 6 && ver.Minor == 7 {
 		return map[string]string{

libbeat/template/load_integration_test.go

@@ -66,23 +66,53 @@ func newTestSetup(t *testing.T, cfg TemplateConfig) *testSetup {
 	}
 	s := testSetup{t: t, client: client, loader: NewESLoader(client), config: cfg}
 	client.Request("DELETE", templateLoaderPath[cfg.Type]+cfg.Name, "", nil, nil)
-	require.False(t, s.loader.templateExists(cfg.Name, cfg.Type))
+	s.requireTemplateDoesNotExist("")
 	return &s
 }
+
+func (ts *testSetup) mustLoadTemplate(body map[string]interface{}) {
+	err := ts.loader.loadTemplate(ts.config.Name, ts.config.Type, body)
+	require.NoError(ts.t, err)
+	ts.requireTemplateExists("")
+}
+
 func (ts *testSetup) loadFromFile(fileElems []string) error {
 	ts.config.Fields = path(ts.t, fileElems)
 	beatInfo := beat.Info{Version: version.GetDefaultVersion()}
 	return ts.loader.Load(ts.config, beatInfo, nil, false)
 }
 
+func (ts *testSetup) mustLoadFromFile(fileElems []string) {
+	require.NoError(ts.t, ts.loadFromFile(fileElems))
+	ts.requireTemplateExists("")
+}
+
 func (ts *testSetup) load(fields []byte) error {
 	beatInfo := beat.Info{Version: version.GetDefaultVersion()}
 	return ts.loader.Load(ts.config, beatInfo, fields, false)
 }
 
 func (ts *testSetup) mustLoad(fields []byte) {
 	require.NoError(ts.t, ts.load(fields))
-	require.True(ts.t, ts.loader.templateExists(ts.config.Name, ts.config.Type))
+	ts.requireTemplateExists("")
+}
+
+func (ts *testSetup) requireTemplateExists(name string) {
+	if name == "" {
+		name = ts.config.Name
+	}
+	exists, err := ts.loader.templateExists(name, ts.config.Type)
+	require.NoError(ts.t, err, "failed to query template status")
+	require.True(ts.t, exists, "template must exist")
+}
+
+func (ts *testSetup) requireTemplateDoesNotExist(name string) {
+	if name == "" {
+		name = ts.config.Name
+	}
+	exists, err := ts.loader.templateExists(name, ts.config.Type)
+	require.NoError(ts.t, err, "failed to query template status")
+	require.False(ts.t, exists, "template must not exist")
 }
 
 func TestESLoader_Load(t *testing.T) {
@@ -91,17 +121,16 @@ func TestESLoader_Load(t *testing.T) {
 			setup := newTestSetup(t, TemplateConfig{Enabled: false})
 
 			setup.load(nil)
-			assert.False(t, setup.loader.templateExists(setup.config.Name, setup.config.Type))
+			setup.requireTemplateDoesNotExist("")
 		})
 
 		t.Run("invalid version", func(t *testing.T) {
 			setup := newTestSetup(t, TemplateConfig{Enabled: true})
 
 			beatInfo := beat.Info{Version: "invalid"}
 			err := setup.loader.Load(setup.config, beatInfo, nil, false)
-			if assert.Error(t, err) {
-				assert.Contains(t, err.Error(), "version is not semver")
-			}
+			require.Error(t, err)
+			require.Contains(t, err.Error(), "version is not semver")
 		})
 	})
 
@@ -140,7 +169,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, setup.config.Type))
+		setup.requireTemplateExists(nameJSON)
 	})
 
 	t.Run("load template successful", func(t *testing.T) {
@@ -211,8 +240,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, setup.config.Type))
+	setup.mustLoadFromFile([]string{"..", "fields.yml"})
 }
 
 func TestLoadInvalidTemplate(t *testing.T) {
@@ -222,7 +250,7 @@ func TestLoadInvalidTemplate(t *testing.T) {
 	template := map[string]interface{}{"json": "invalid"}
 	err := setup.loader.loadTemplate(setup.config.Name, setup.config.Type, template)
 	assert.Error(t, err)
-	assert.False(t, setup.loader.templateExists(setup.config.Name, setup.config.Type))
+	setup.requireTemplateDoesNotExist("")
 }
 
 // Tests loading the templates for each beat
@@ -233,8 +261,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, setup.config.Type))
+		setup.mustLoadFromFile([]string{"..", "..", beat, "fields.yml"})
 	}
 }
 
@@ -244,7 +271,7 @@ func TestTemplateSettings(t *testing.T) {
 		Source: common.MapStr{"enabled": false},
 	}
 	setup := newTestSetup(t, TemplateConfig{Settings: settings, Enabled: true})
-	require.NoError(t, setup.loadFromFile([]string{"..", "fields.yml"}))
+	setup.mustLoadFromFile([]string{"..", "fields.yml"})
 
 	// Check that it contains the mapping
 	templateJSON := getTemplate(t, setup.client, setup.config.Name, setup.config.Type)
@@ -297,8 +324,8 @@ var dataTests = []struct {
 // Tests if data can be loaded into elasticsearch with right types
 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, setup.config.Type))
+	setup.mustLoadFromFile([]string{"testdata", "fields.yml"})
+
 	esClient := setup.client.(*eslegclient.Connection)
 	for _, test := range dataTests {
 		_, _, err := esClient.Index(setup.config.Name, "_doc", "", nil, test.data)