[Elastic Agent] Add params prefer_v2_templates=true on the ES output

To make sure we use the v2 templates we must send on bulk request the
params `prefer_v2_templates=true` if not it will default to v1 and will
not try to use the v2 template at all.

Reference: elastic#17809

x-pack/elastic-agent/CHANGELOG.asciidoc

@@ -32,6 +32,7 @@
 - Fix make sure the collected logs or metrics include streams information. {pull}18261[18261]
 - Fix version to 7.8 {pull}18286[18286]
 - Fix an issue where the checkin_frequency, jitter, and backoff options where not configurable. {pull}17843[17843]
+- Ensure that the beats uses the params prefer_v2_templates on bulk request. {pull}18318[18318]
 
 ==== New features
 

x-pack/elastic-agent/pkg/agent/application/decorators.go

@@ -0,0 +1,65 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package application
+
+import (
+	"github.com/elastic/beats/v7/libbeat/common"
+	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/program"
+	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/transpiler"
+)
+
+func injectPreferV2Template(
+	_ string,
+	_ *transpiler.AST,
+	programsToRun []program.Program,
+) ([]program.Program, error) {
+
+	const (
+		outputKey              = "output"
+		elasticsearchKey       = "elasticsearch"
+		paramsKey              = "parameters"
+		elasticsearchOutputKey = outputKey + "." + elasticsearchKey
+	)
+
+	params := common.MapStr{
+		"output": common.MapStr{
+			"elasticsearch": common.MapStr{
+				"parameters": common.MapStr{
+					"prefer_v2_templates": true,
+				},
+			},
+		},
+	}
+
+	programList := make([]program.Program, 0, len(programsToRun))
+
+	for _, program := range programsToRun {
+		if _, found := transpiler.Lookup(program.Config, elasticsearchOutputKey); !found {
+			programList = append(programList, program)
+			continue
+		}
+
+		m, err := program.Config.Map()
+		if err != nil {
+			return programsToRun, err
+		}
+
+		// Add prefer_v2_templates to every bulk request on the elasticsearch output.
+		// without this Elasticsearch will fallback to v1 templates even if the index matches existing v2.
+		mStr := common.MapStr(m)
+		mStr.DeepUpdate(params)
+
+		a, err := transpiler.NewAST(mStr)
+		if err != nil {
+			return programsToRun, err
+		}
+
+		program.Config = a
+
+		programList = append(programList, program)
+	}
+
+	return programList, nil
+}

x-pack/elastic-agent/pkg/agent/application/decorators_test.go

@@ -0,0 +1,183 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package application
+
+import (
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/program"
+	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/transpiler"
+)
+
+func TestInjectV2Templates(t *testing.T) {
+	const outputGroup = "default"
+	t.Run("inject parameters on elasticsearch output", func(t *testing.T) {
+
+		config := map[string]interface{}{
+			"outputs": map[string]interface{}{
+				"default": map[string]interface{}{
+					"type":     "elasticsearch",
+					"username": "foo",
+					"password": "secret",
+				},
+			},
+			"datasources": []map[string]interface{}{
+				map[string]interface{}{
+					"inputs": []map[string]interface{}{
+						map[string]interface{}{
+							"type": "log",
+							"streams": []map[string]interface{}{
+								map[string]interface{}{"paths": "/xxxx"},
+							},
+							"processors": []interface{}{
+								map[string]interface{}{
+									"dissect": map[string]interface{}{
+										"tokenizer": "---",
+									},
+								},
+							},
+						},
+					},
+				},
+				map[string]interface{}{
+					"inputs": []map[string]interface{}{
+						map[string]interface{}{
+							"type": "system/metrics",
+							"streams": []map[string]interface{}{
+								map[string]interface{}{
+									"id":      "system/metrics-system.core",
+									"enabled": true,
+									"dataset": "system.core",
+									"period":  "10s",
+									"metrics": []string{"percentages"},
+								},
+							},
+						},
+					},
+					"use_output": "default",
+				},
+			},
+		}
+
+		ast, err := transpiler.NewAST(config)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		programsToRun, err := program.Programs(ast)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		programsWithOutput, err := injectPreferV2Template(outputGroup, ast, programsToRun[outputGroup])
+		require.NoError(t, err)
+
+		assert.Equal(t, len(programsToRun[outputGroup]), len(programsWithOutput))
+
+		for _, program := range programsWithOutput {
+			m, err := program.Config.Map()
+			require.NoError(t, err)
+
+			want := map[string]interface{}{
+				"elasticsearch": map[string]interface{}{
+					"username": "foo",
+					"password": "secret",
+					"parameters": map[string]interface{}{
+						"prefer_v2_templates": true,
+					},
+				},
+			}
+
+			got, _ := m["output"]
+			if diff := cmp.Diff(want, got); diff != "" {
+				t.Errorf("output mismatch (-want +got):\n%s", diff)
+			}
+		}
+	})
+
+	t.Run("dont do anything on logstash output", func(t *testing.T) {
+		config := map[string]interface{}{
+			"outputs": map[string]interface{}{
+				"default": map[string]interface{}{
+					"type":     "logstash",
+					"username": "foo",
+					"password": "secret",
+				},
+			},
+			"datasources": []map[string]interface{}{
+				map[string]interface{}{
+					"inputs": []map[string]interface{}{
+						map[string]interface{}{
+							"type": "log",
+							"streams": []map[string]interface{}{
+								map[string]interface{}{"paths": "/xxxx"},
+							},
+							"processors": []interface{}{
+								map[string]interface{}{
+									"dissect": map[string]interface{}{
+										"tokenizer": "---",
+									},
+								},
+							},
+						},
+					},
+				},
+				map[string]interface{}{
+					"inputs": []map[string]interface{}{
+						map[string]interface{}{
+							"type": "system/metrics",
+							"streams": []map[string]interface{}{
+								map[string]interface{}{
+									"id":      "system/metrics-system.core",
+									"enabled": true,
+									"dataset": "system.core",
+									"period":  "10s",
+									"metrics": []string{"percentages"},
+								},
+							},
+						},
+					},
+					"use_output": "default",
+				},
+			},
+		}
+
+		ast, err := transpiler.NewAST(config)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		programsToRun, err := program.Programs(ast)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		programsWithOutput, err := injectPreferV2Template(outputGroup, ast, programsToRun[outputGroup])
+		require.NoError(t, err)
+
+		assert.Equal(t, len(programsToRun[outputGroup]), len(programsWithOutput))
+
+		for _, program := range programsWithOutput {
+			m, err := program.Config.Map()
+			require.NoError(t, err)
+
+			want := map[string]interface{}{
+				"logstash": map[string]interface{}{
+					"username": "foo",
+					"password": "secret",
+				},
+			}
+
+			got, _ := m["output"]
+			if diff := cmp.Diff(want, got); diff != "" {
+				t.Errorf("output mismatch (-want +got):\n%s", diff)
+			}
+		}
+	})
+}

x-pack/elastic-agent/pkg/agent/application/introspect_output_cmd.go

@@ -181,7 +181,7 @@ func getProgramsFromConfig(log *logger.Logger, cfg *config.Config) (map[string][
 		log,
 		router,
 		&configModifiers{
-			Decorators: []decoratorFunc{injectMonitoring},
+			Decorators: []decoratorFunc{injectMonitoring, injectPreferV2Template},
 			Filters:    []filterFunc{filters.ConstraintFilter},
 		},
 		monitor,

x-pack/elastic-agent/pkg/agent/application/local_mode.go

@@ -92,7 +92,15 @@ func newLocal(
 	}
 
 	discover := discoverer(pathConfigFile, c.Management.Path)
-	emit := emitter(log, router, &configModifiers{Decorators: []decoratorFunc{injectMonitoring}, Filters: []filterFunc{filters.ConstraintFilter}}, monitor)
+	emit := emitter(
+		log,
+		router,
+		&configModifiers{
+			Decorators: []decoratorFunc{injectMonitoring, injectPreferV2Template},
+			Filters:    []filterFunc{filters.ConstraintFilter},
+		},
+		monitor,
+	)
 
 	var cfgSource source
 	if !c.Management.Reload.Enabled {

x-pack/elastic-agent/pkg/agent/application/managed_mode.go

@@ -136,7 +136,7 @@ func newManaged(
 		log,
 		router,
 		&configModifiers{
-			Decorators: []decoratorFunc{injectMonitoring},
+			Decorators: []decoratorFunc{injectMonitoring, injectPreferV2Template},
 			Filters:    []filterFunc{filters.ConstraintFilter},
 		},
 		monitor,