[exporter/elasticsearch] Add _metric_names_hash to avoid metric rejections (#37511)

If metrics that have the same timestamp and dimensions aren't grouped
into the same document, ES will consider them to be a duplicate. This
adds a hash of the metric names that will be mapped as a dimension in
Elasticsearch. The tradeoff is that if the composition of the metrics
grouping changes over time, a new time series will be created. That has
an impact on the rate aggregation for counters.

ES mapping changes: elastic/elasticsearch#120952

---------

Co-authored-by: Carson Ip <[email protected]>

Author: felixbarny

.chloggen/elasticsearchexporter_metric-names-hash.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: elasticsearchexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add `_metric_names_hash` field to metric documents in `otel` mode to avoid metric rejections
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [37511]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

exporter/elasticsearchexporter/README.md

@@ -457,23 +457,44 @@ processors:
 
 ### version_conflict_engine_exception
 
-Symptom: elasticsearchexporter logs an error "failed to index document" with `error.type` "version_conflict_engine_exception" and `error.reason` containing "version conflict, document already exists".
-
-This happens when the target data stream is a TSDB metrics data stream (e.g. using OTel mapping mode sending to a 8.16+ Elasticsearch). See the following scenarios.
+Symptom: `elasticsearchexporter` logs an error "failed to index document" with `error.type` "version_conflict_engine_exception" and `error.reason` containing "version conflict, document already exists".
+
+This happens when the target data stream is a TSDB metrics data stream (e.g. using OTel mapping mode sending to a 8.16+ Elasticsearch).
+
+Elasticsearch [Time Series Data Streams](https://www.elastic.co/guide/en/elasticsearch/reference/current/tsds.html) requires that there must only be one document per timestamp with the same dimensions.
+The purpose is to avoid duplicate data when re-trying a batch of metrics that were previously sent but failed to be indexed.
+The dimensions are mostly made up of resource attributes, scope attributes, scope name, attributes, and the unit.
+
+The exporter can only group metrics with the same dimensions into the same document if they arrive in the same batch.
+To ensure metrics are not dropped even if they arrive in different batches in the exporter, the exporter adds a fingerprint of the metric names to the document in the `otel` mapping mode.
+Note that you'll need to be on a minimum version of Elasticsearch in order for this to take effect 8.16.5, 8.17.3, 8.19.0, 9.0.0.
+If you are on an earlier version, either update your Elasticsearch cluster or install this custom component template:
+
+```shell
+PUT _component_template/metrics-otel@custom
+{
+  "template": {
+    "mappings": {
+      "properties": {
+        "_metric_names_hash": {
+          "type": "keyword",
+          "time_series_dimension": true
+        }
+      }
+    }
+  }
+}
+```
 
-1. When sending different metrics with the same dimension (mostly made up of resource attributes, scope attributes, attributes),
-`version_conflict_engine_exception` is returned by Elasticsearch when these metrics are not grouped into the same document.
-It also means that they have to be in the same batch in the exporter, as metric grouping is done per-batch in elasticsearchexporter.
-To work around the issue, use a [transform processor](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/processor/transformprocessor/README.md) to ensure different metrics to never share the same set of dimensions. This is done at the expense of storage efficiency.
-This workaround will no longer be necessary once the limitation is lifted in Elasticsearch (see [issue](https://github.com/elastic/elasticsearch/issues/99123)).
+While in most situations, this error is just a sign that Elasticsearch's duplicate detection is working as intended, the data may be classified as a duplicate while it was not.
+This implies data is lost.
 
-```yaml
-processors:
-  transform/unique_dimensions:
-    metric_statements:
-      - context: datapoint
-        statements:
-          - set(attributes["metric_name"], metric.name)
-```
+1. If the data is not sent in `otel` mapping mode to `metrics-*.otel-*` data streams, the metrics name fingerprint is not applied.
+This can happen for OTel host and k8s metrics that the [`elasticinframetricsprocessor`](https://github.com/elastic/opentelemetry-collector-components/tree/main/processor/elasticinframetricsprocessor) has translated to the format the host and k8s dashboards in Kibana can consume.
+If these metrics arrive in the `elasticsearchexporter` in different batches, they will not be grouped to the same document.
+This can cause the `version_conflict_engine_exception` error.
+Try to remove the `batchprocessor` from the pipeline (or set `send_batch_max_size: 0`) to ensure metrics are not split into different batches.
+This gives the exporter the opportunity to group all related metrics into the same document.
 
 2. Otherwise, check your metrics pipeline setup for misconfiguration that causes an actual violation of the [single writer principle](https://opentelemetry.io/docs/specs/otel/metrics/data-model/#single-writer).
+ This means that the same metric with the same dimensions is sent from multiple sources, which is not allowed in the OTel metrics data model.

exporter/elasticsearchexporter/exporter_test.go

@@ -1176,19 +1176,19 @@ func TestExporterMetrics(t *testing.T) {
 		expected := []itemRequest{
 			{
 				Action:   []byte(`{"create":{"_index":"metrics-generic.otel-default","dynamic_templates":{"metrics.metric.foo":"histogram"}}}`),
-				Document: []byte(`{"@timestamp":"0.0","data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"metrics":{"metric.foo":{"counts":[1,2,3,4],"values":[0.5,1.5,2.5,3.0]}},"resource":{},"scope":{}}`),
+				Document: []byte(`{"@timestamp":"0.0","data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"metrics":{"metric.foo":{"counts":[1,2,3,4],"values":[0.5,1.5,2.5,3.0]}},"resource":{},"scope":{},"_metric_names_hash":"f7fdad9f"}`),
 			},
 			{
 				Action:   []byte(`{"create":{"_index":"metrics-generic.otel-default","dynamic_templates":{"metrics.metric.foo":"histogram"}}}`),
-				Document: []byte(`{"@timestamp":"3600000.0","data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"metrics":{"metric.foo":{"counts":[4,5,6,7],"values":[2.0,4.5,5.5,6.0]}},"resource":{},"scope":{}}`),
+				Document: []byte(`{"@timestamp":"3600000.0","data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"metrics":{"metric.foo":{"counts":[4,5,6,7],"values":[2.0,4.5,5.5,6.0]}},"resource":{},"scope":{},"_metric_names_hash":"f7fdad9f"}`),
 			},
 			{
 				Action:   []byte(`{"create":{"_index":"metrics-generic.otel-default","dynamic_templates":{"metrics.metric.sum":"gauge_double"}}}`),
-				Document: []byte(`{"@timestamp":"3600000.0","data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"metrics":{"metric.sum":1.5},"resource":{},"scope":{},"start_timestamp":"7200000.0"}`),
+				Document: []byte(`{"@timestamp":"3600000.0","data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"metrics":{"metric.sum":1.5},"resource":{},"scope":{},"start_timestamp":"7200000.0","_metric_names_hash":"6e599000"}`),
 			},
 			{
 				Action:   []byte(`{"create":{"_index":"metrics-generic.otel-default","dynamic_templates":{"metrics.metric.summary":"summary"}}}`),
-				Document: []byte(`{"@timestamp":"10800000.0","data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"metrics":{"metric.summary":{"sum":1.5,"value_count":1}},"resource":{},"scope":{},"start_timestamp":"10800000.0"}`),
+				Document: []byte(`{"@timestamp":"10800000.0","data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"metrics":{"metric.summary":{"sum":1.5,"value_count":1}},"resource":{},"scope":{},"start_timestamp":"10800000.0","_metric_names_hash":"45a9e3cb"}`),
 			},
 		}
 
@@ -1257,7 +1257,7 @@ func TestExporterMetrics(t *testing.T) {
 		expected := []itemRequest{
 			{
 				Action:   []byte(`{"create":{"_index":"metrics-generic.otel-default","dynamic_templates":{"metrics.sum":"gauge_long","metrics.summary":"summary"}}}`),
-				Document: []byte(`{"@timestamp":"0.0","_doc_count":10,"data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"metrics":{"sum":0,"summary":{"sum":1.0,"value_count":10}},"resource":{},"scope":{}}`),
+				Document: []byte(`{"@timestamp":"0.0","_doc_count":10,"data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"metrics":{"sum":0,"summary":{"sum":1.0,"value_count":10}},"resource":{},"scope":{},"_metric_names_hash":"7dc58200"}`),
 			},
 		}
 
@@ -1307,11 +1307,11 @@ func TestExporterMetrics(t *testing.T) {
 		expected := []itemRequest{
 			{
 				Action:   []byte(`{"create":{"_index":"metrics-generic.otel-default","dynamic_templates":{"metrics.histogram.summary":"summary"}}}`),
-				Document: []byte(`{"@timestamp":"0.0","_doc_count":10,"data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"attributes":{},"metrics":{"histogram.summary":{"sum":1.0,"value_count":10}},"resource":{},"scope":{}}`),
+				Document: []byte(`{"@timestamp":"0.0","_doc_count":10,"data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"attributes":{},"metrics":{"histogram.summary":{"sum":1.0,"value_count":10}},"resource":{},"scope":{},"_metric_names_hash":"acbaed6b"}`),
 			},
 			{
 				Action:   []byte(`{"create":{"_index":"metrics-generic.otel-default","dynamic_templates":{"metrics.exphistogram.summary":"summary"}}}`),
-				Document: []byte(`{"@timestamp":"3600000.0","_doc_count":10,"data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"attributes":{},"metrics":{"exphistogram.summary":{"sum":1.0,"value_count":10}},"resource":{},"scope":{}}`),
+				Document: []byte(`{"@timestamp":"3600000.0","_doc_count":10,"data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"attributes":{},"metrics":{"exphistogram.summary":{"sum":1.0,"value_count":10}},"resource":{},"scope":{},"_metric_names_hash":"29641c64"}`),
 			},
 		}
 
@@ -1350,7 +1350,7 @@ func TestExporterMetrics(t *testing.T) {
 		expected := []itemRequest{
 			{
 				Action:   []byte(`{"create":{"_index":"metrics-generic.otel-default","dynamic_templates":{"metrics.foo.bar":"gauge_long","metrics.foo":"gauge_long","metrics.foo.bar.baz":"gauge_long"}}}`),
-				Document: []byte(`{"@timestamp":"0.0","data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"metrics":{"foo":0,"foo.bar":0,"foo.bar.baz":0},"resource":{},"scope":{}}`),
+				Document: []byte(`{"@timestamp":"0.0","data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"metrics":{"foo":0,"foo.bar":0,"foo.bar.baz":0},"resource":{},"scope":{},"_metric_names_hash":"204c382a"}`),
 			},
 		}
 

exporter/elasticsearchexporter/internal/serializer/otelserializer/metrics.go

@@ -6,6 +6,9 @@ package otelserializer // import "github.com/open-telemetry/opentelemetry-collec
 import (
 	"bytes"
 	"fmt"
+	"hash/fnv"
+	"sort"
+	"strconv"
 
 	"github.com/elastic/go-structform"
 	"github.com/elastic/go-structform/json"
@@ -47,10 +50,11 @@ func serializeDataPoints(v *json.Visitor, dataPoints []datapoints.DataPoint, val
 
 	dynamicTemplates := make(map[string]string, len(dataPoints))
 	var docCount uint64
-	metricNames := make(map[string]bool, len(dataPoints))
+	metricNamesSet := make(map[string]bool, len(dataPoints))
+	metricNames := make([]string, 0, len(dataPoints))
 	for _, dp := range dataPoints {
 		metric := dp.Metric()
-		if _, present := metricNames[metric.Name()]; present {
+		if _, present := metricNamesSet[metric.Name()]; present {
 			*validationErrors = append(
 				*validationErrors,
 				fmt.Errorf(
@@ -61,7 +65,8 @@ func serializeDataPoints(v *json.Visitor, dataPoints []datapoints.DataPoint, val
 			)
 			continue
 		}
-		metricNames[metric.Name()] = true
+		metricNamesSet[metric.Name()] = true
+		metricNames = append(metricNames, metric.Name())
 		// TODO here's potential for more optimization by directly serializing the value instead of allocating a pcommon.Value
 		//  the tradeoff is that this would imply a duplicated logic for the ECS mode
 		value, err := dp.Value()
@@ -86,5 +91,14 @@ func serializeDataPoints(v *json.Visitor, dataPoints []datapoints.DataPoint, val
 	if docCount != 0 {
 		writeUIntField(v, "_doc_count", docCount)
 	}
+	sort.Strings(metricNames)
+	hasher := fnv.New32a()
+	for _, name := range metricNames {
+		_, _ = hasher.Write([]byte(name))
+	}
+	// workaround for https://github.com/elastic/elasticsearch/issues/99123
+	// should use a string field to benefit from run-length encoding
+	writeStringFieldSkipDefault(v, "_metric_names_hash", strconv.FormatUint(uint64(hasher.Sum32()), 16))
+
 	return dynamicTemplates
 }

exporter/elasticsearchexporter/internal/serializer/otelserializer/metrics_test.go

@@ -56,5 +56,6 @@ func TestSerializeMetricsConflict(t *testing.T) {
 		"metrics": map[string]any{
 			"foo": json.Number("42"),
 		},
+		"_metric_names_hash": "a9f37ed7",
 	}, result, eventAsJSON)
 }