[exporter/elasticsearch] Add sanitization utils for datastream fields

exporter/elasticsearchexporter/data_stream_router.go

@@ -44,7 +44,10 @@ func routeWithDefaults(defaultDSType string) func(
 			}
 		}
 
-		// Receiver-based routing
+		dataset = sanitizeDataStreamDataset(dataset)
+		namespace = sanitizeDataStreamNamespace(namespace)
+
+    // Receiver-based routing
 		// For example, hostmetricsreceiver (or hostmetricsreceiver.otel in the OTel output mode)
 		// for the scope name
 		// github.com/open-telemetry/opentelemetry-collector-contrib/receiver/hostmetricsreceiver/internal/scraper/cpuscraper

exporter/elasticsearchexporter/exporter_test.go

@@ -11,6 +11,7 @@ import (
 	"math"
 	"net/http"
 	"runtime"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -216,7 +217,10 @@ 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))
+			ds := "record.dataset" + strings.Repeat("_", len(disallowedDatasetRunes))
+			ns := "resource.namespace" + strings.Repeat("_", len(disallowedNamespaceRunes))
+
+			assert.Equal(t, fmt.Sprintf("logs-%s-%s", ds, ns), actionJSONToIndex(t, docs[0].Action))
 
 			return itemsAllOK(docs)
 		})
@@ -226,12 +230,12 @@ func TestExporterLogs(t *testing.T) {
 		})
 		logs := newLogsWithAttributes(
 			map[string]any{
-				dataStreamDataset: "record.dataset",
+				dataStreamDataset: "record.dataset" + disallowedDatasetRunes,
 			},
 			nil,
 			map[string]any{
 				dataStreamDataset:   "resource.dataset",
-				dataStreamNamespace: "resource.namespace",
+				dataStreamNamespace: "resource.namespace" + disallowedNamespaceRunes,
 			},
 		)
 		logs.ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().At(0).Body().SetStr("hello world")

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)
 
 	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{

exporter/elasticsearchexporter/util.go

@@ -6,11 +6,49 @@ package elasticsearchexporter // import "github.com/open-telemetry/opentelemetry
 import (
 	"bytes"
 	"fmt"
+	"strings"
 	"time"
+	"unicode"
 
 	"github.com/lestrrat-go/strftime"
 )
 
+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
+func sanitizeDataStreamDataset(field string) string {
+	field = strings.Map(replaceReservedRune(disallowedDatasetRunes), field)
+	if len(field) > maxDataStreamBytes {
+		return field[:maxDataStreamBytes]
+	}
+
+	return field
+}
+
+// Sanitize the datastream fields (dataset, namespace) to apply restrictions
+// as outlined in https://www.elastic.co/guide/en/ecs/current/ecs-data_stream.html
+func sanitizeDataStreamNamespace(field string) string {
+	field = strings.Map(replaceReservedRune(disallowedNamespaceRunes), field)
+	if len(field) > maxDataStreamBytes {
+		return field[:maxDataStreamBytes]
+	}
+	return field
+}
+
+func replaceReservedRune(disallowedRunes string) func(r rune) rune {
+	return func(r rune) rune {
+		if strings.ContainsRune(disallowedRunes, r) {
+			return '_'
+		}
+		return unicode.ToLower(r)
+	}
+}
+
 func generateIndexWithLogstashFormat(index string, conf *LogstashFormatSettings, t time.Time) (string, error) {
 	if conf.Enabled {
 		partIndex := fmt.Sprintf("%s%s", index, conf.PrefixSeparator)