feat(outputs.elasticsearch): Support data streams (#15613)

plugins/outputs/elasticsearch/README.md

@@ -286,6 +286,10 @@ to use them.
   # default_tag_value = "none"
   index_name = "telegraf-%Y.%m.%d" # required.
 
+  ## Optional Index Config
+  ## Set to true if Telegraf should use the "create" OpType while indexing
+  # use_optype_create = false
+
   ## Optional TLS Config
   # tls_ca = "/etc/telegraf/ca.pem"
   # tls_cert = "/etc/telegraf/cert.pem"
@@ -397,6 +401,9 @@ the `default_tag_value` will be used instead.
   `inf`s if `float_handling` is set to `replace`. Negative `inf` will be
   replaced by the negative value in this number to respect the sign of the
   field's original value.
+* `use_optype_create`: If set, the "create" operation type will be used when
+   indexing into Elasticsearch, which is needed when using the Elasticsearch
+   data streams feature.
 * `use_pipeline`: If set, the set value will be used as the pipeline to call
   when sending events to elasticsearch. Additionally, you can specify dynamic
   pipeline names by using tags with the notation ```{{tag_name}}```.  If the tag

plugins/outputs/elasticsearch/elasticsearch.go

@@ -43,6 +43,7 @@ type Elasticsearch struct {
 	IndexTemplate       map[string]interface{} `toml:"template_index_settings"`
 	ManageTemplate      bool                   `toml:"manage_template"`
 	OverwriteTemplate   bool                   `toml:"overwrite_template"`
+	UseOpTypeCreate     bool                   `toml:"use_optype_create"`
 	Username            config.Secret          `toml:"username"`
 	Password            config.Secret          `toml:"password"`
 	TemplateName        string                 `toml:"template_name"`
@@ -304,6 +305,10 @@ func (a *Elasticsearch) Write(metrics []telegraf.Metric) error {
 
 		br := elastic.NewBulkIndexRequest().Index(indexName).Doc(m)
 
+		if a.UseOpTypeCreate {
+			br.OpType("create")
+		}
+
 		if a.ForceDocumentID {
 			id := GetPointID(metric)
 			br.Id(id)

plugins/outputs/elasticsearch/elasticsearch_test.go

@@ -298,6 +298,50 @@ func TestTemplateManagementEmptyTemplateIntegration(t *testing.T) {
 	require.Error(t, err)
 }
 
+func TestUseOpTypeCreate(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	container := launchTestContainer(t)
+	defer container.Terminate()
+
+	urls := []string{
+		fmt.Sprintf("http://%s:%s", container.Address, container.Ports[servicePort]),
+	}
+
+	e := &Elasticsearch{
+		URLs:              urls,
+		IndexName:         "test-%Y.%m.%d",
+		Timeout:           config.Duration(time.Second * 5),
+		EnableGzip:        true,
+		ManageTemplate:    true,
+		TemplateName:      "telegraf",
+		OverwriteTemplate: true,
+		UseOpTypeCreate:   true,
+		Log:               testutil.Logger{},
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(e.Timeout))
+	defer cancel()
+
+	metrics := []telegraf.Metric{
+		testutil.TestMetric(1),
+	}
+
+	err := e.Connect()
+	require.NoError(t, err)
+
+	err = e.manageTemplate(ctx)
+	require.NoError(t, err)
+
+	// Verify that we can fail for metric with unhandled NaN/inf/-inf values
+	for _, m := range metrics {
+		err = e.Write([]telegraf.Metric{m})
+		require.NoError(t, err)
+	}
+}
+
 func TestTemplateManagementIntegration(t *testing.T) {
 	if testing.Short() {
 		t.Skip("Skipping integration test in short mode")

plugins/outputs/elasticsearch/sample.conf

@@ -40,6 +40,10 @@
   # default_tag_value = "none"
   index_name = "telegraf-%Y.%m.%d" # required.
 
+  ## Optional Index Config
+  ## Set to true if Telegraf should use the "create" OpType while indexing
+  # use_optype_create = false
+
   ## Optional TLS Config
   # tls_ca = "/etc/telegraf/ca.pem"
   # tls_cert = "/etc/telegraf/cert.pem"