[exporter/elasticsearch] Add sanitization utils for datastream fields

.chloggen/elasticsearchexporter_sanitize-datastream-fields.yaml

@@ -0,0 +1,28 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# 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: Sanitize datastream routing fields
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [34285]
+
+# (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:
+  Sanitize the dataset and namespace fields according to https://www.elastic.co/guide/en/ecs/current/ecs-data_stream.html.
+
+# 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/data_stream_router.go

@@ -6,12 +6,39 @@ package elasticsearchexporter // import "github.com/open-telemetry/opentelemetry
 import (
 	"fmt"
 	"regexp"
+	"strings"
+	"unicode"
 
 	"go.opentelemetry.io/collector/pdata/pcommon"
 )
 
 var receiverRegex = regexp.MustCompile(`/receiver/(\w*receiver)`)
 
+const (
+	maxDataStreamBytes       = 100
+	disallowedNamespaceRunes = "\\/*?\"<>| ,#:"
+	disallowedDatasetRunes   = "-\\/*?\"<>| ,#:"
+)
+
+// Sanitize the datastream fields (dataset, namespace) to apply restrictions
+// as outlined in https://www.elastic.co/guide/en/ecs/current/ecs-data_stream.html
+// The suffix will be appended after truncation of max bytes.
+func sanitizeDataStreamField(field, disallowed, appendSuffix string) string {
+	field = strings.Map(func(r rune) rune {
+		if strings.ContainsRune(disallowed, r) {
+			return '_'
+		}
+		return unicode.ToLower(r)
+	}, field)
+
+	if len(field) > maxDataStreamBytes-len(appendSuffix) {
+		field = field[:maxDataStreamBytes-len(appendSuffix)]
+	}
+	field += appendSuffix
+
+	return field
+}
+
 func routeWithDefaults(defaultDSType string) func(
 	pcommon.Map,
 	pcommon.Map,
@@ -53,15 +80,20 @@ func routeWithDefaults(defaultDSType string) func(
 			dataset = receiverName
 		}
 
-		// The naming convention for datastream is expected to be "logs-[dataset].otel-[namespace]".
+		// For dataset, the naming convention for datastream is expected to be "logs-[dataset].otel-[namespace]".
 		// This is in order to match the built-in logs-*.otel-* index template.
+		var datasetSuffix string
 		if otel {
-			dataset += ".otel"
+			datasetSuffix += ".otel"
 		}
 
+		dataset = sanitizeDataStreamField(dataset, disallowedDatasetRunes, datasetSuffix)
+		namespace = sanitizeDataStreamField(namespace, disallowedNamespaceRunes, "")
+
 		recordAttr.PutStr(dataStreamDataset, dataset)
 		recordAttr.PutStr(dataStreamNamespace, namespace)
 		recordAttr.PutStr(dataStreamType, defaultDSType)
+
 		return fmt.Sprintf("%s-%s-%s", defaultDSType, dataset, namespace)
 	}
 }

exporter/elasticsearchexporter/exporter_test.go

@@ -216,7 +216,8 @@ func TestExporterLogs(t *testing.T) {
 		server := newESTestServer(t, func(docs []itemRequest) ([]itemResponse, error) {
 			rec.Record(docs)
 
-			assert.Equal(t, "logs-record.dataset-resource.namespace", actionJSONToIndex(t, docs[0].Action))
+			expected := "logs-record.dataset.____________-resource.namespace.-____________"
+			assert.Equal(t, expected, actionJSONToIndex(t, docs[0].Action))
 
 			return itemsAllOK(docs)
 		})
@@ -226,12 +227,12 @@ func TestExporterLogs(t *testing.T) {
 		})
 		logs := newLogsWithAttributes(
 			map[string]any{
-				dataStreamDataset: "record.dataset",
+				dataStreamDataset: "record.dataset.\\/*?\"<>| ,#:",
 			},
 			nil,
 			map[string]any{
 				dataStreamDataset:   "resource.dataset",
-				dataStreamNamespace: "resource.namespace",
+				dataStreamNamespace: "resource.namespace.-\\/*?\"<>| ,#:",
 			},
 		)
 		logs.ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().At(0).Body().SetStr("hello world")
@@ -582,7 +583,7 @@ func TestExporterMetrics(t *testing.T) {
 		server := newESTestServer(t, func(docs []itemRequest) ([]itemResponse, error) {
 			rec.Record(docs)
 
-			expected := "metrics-resource.dataset-data.point.namespace"
+			expected := "metrics-resource.dataset.____________-data.point.namespace.-____________"
 			assert.Equal(t, expected, actionJSONToIndex(t, docs[0].Action))
 
 			return itemsAllOK(docs)
@@ -594,11 +595,11 @@ func TestExporterMetrics(t *testing.T) {
 		})
 		metrics := newMetricsWithAttributes(
 			map[string]any{
-				dataStreamNamespace: "data.point.namespace",
+				dataStreamNamespace: "data.point.namespace.-\\/*?\"<>| ,#:",
 			},
 			nil,
 			map[string]any{
-				dataStreamDataset:   "resource.dataset",
+				dataStreamDataset:   "resource.dataset.\\/*?\"<>| ,#:",
 				dataStreamNamespace: "resource.namespace",
 			},
 		)
@@ -1223,7 +1224,7 @@ func TestExporterTraces(t *testing.T) {
 		server := newESTestServer(t, func(docs []itemRequest) ([]itemResponse, error) {
 			rec.Record(docs)
 
-			expected := "traces-span.dataset-default"
+			expected := "traces-span.dataset.____________-default"
 			assert.Equal(t, expected, actionJSONToIndex(t, docs[0].Action))
 
 			return itemsAllOK(docs)
@@ -1235,7 +1236,7 @@ func TestExporterTraces(t *testing.T) {
 
 		mustSendTraces(t, exporter, newTracesWithAttributes(
 			map[string]any{
-				dataStreamDataset: "span.dataset",
+				dataStreamDataset: "span.dataset.\\/*?\"<>| ,#:",
 			},
 			nil,
 			map[string]any{

exporter/elasticsearchexporter/model_test.go

@@ -960,6 +960,9 @@ func decodeOTelID(data []byte) ([]byte, error) {
 }
 
 func TestEncodeLogOtelMode(t *testing.T) {
+	randomString := strings.Repeat("abcdefghijklmnopqrstuvwxyz0123456789", 10)
+	maxLenNamespace := maxDataStreamBytes - len(disallowedNamespaceRunes)
+	maxLenDataset := maxDataStreamBytes - len(disallowedDatasetRunes) - len(".otel")
 
 	tests := []struct {
 		name   string
@@ -1044,6 +1047,20 @@ func TestEncodeLogOtelMode(t *testing.T) {
 				return assignDatastreamData(or, "", "third.otel")
 			},
 		},
+		{
+			name: "sanitize dataset/namespace",
+			rec: buildOTelRecordTestData(t, func(or OTelRecord) OTelRecord {
+				or.Attributes["data_stream.dataset"] = disallowedDatasetRunes + randomString
+				or.Attributes["data_stream.namespace"] = disallowedNamespaceRunes + randomString
+				return or
+			}),
+			wantFn: func(or OTelRecord) OTelRecord {
+				deleteDatasetAttributes(or)
+				ds := strings.Repeat("_", len(disallowedDatasetRunes)) + randomString[:maxLenDataset] + ".otel"
+				ns := strings.Repeat("_", len(disallowedNamespaceRunes)) + randomString[:maxLenNamespace]
+				return assignDatastreamData(or, "", ds, ns)
+			},
+		},
 	}
 
 	m := encodeModel{