Sanitize OTEL datastream

input/otlp/logs.go

@@ -195,12 +195,12 @@ func (c *Consumer) convertLogRecord(
 			if event.DataStream == nil {
 				event.DataStream = &modelpb.DataStream{}
 			}
-			event.DataStream.Dataset = v.Str()
+			event.DataStream.Dataset = sanitizeDataStreamField(v.Str())
 		case attributeDataStreamNamespace:
 			if event.DataStream == nil {
 				event.DataStream = &modelpb.DataStream{}
 			}
-			event.DataStream.Namespace = v.Str()
+			event.DataStream.Namespace = sanitizeDataStreamField(v.Str())
 		default:
 			setLabel(replaceDots(k), event, v)
 		}

input/otlp/logs_test.go

@@ -37,6 +37,7 @@ package otlp_test
 import (
 	"context"
 	"fmt"
+	"strings"
 	"testing"
 	"time"
 
@@ -532,6 +533,9 @@ func processLogEvents(t *testing.T, logs plog.Logs) modelpb.Batch {
 }
 
 func TestConsumerConsumeLogsDataStream(t *testing.T) {
+	randomString := strings.Repeat("abcdefghijklmnopqrstuvwxyz0123456789", 10)
+	maxLen := otlp.MaxDataStreamBytes - len(otlp.DisallowedDataStreamRunes)
+
 	for _, tc := range []struct {
 		resourceDataStreamDataset   string
 		resourceDataStreamNamespace string
@@ -567,6 +571,15 @@ func TestConsumerConsumeLogsDataStream(t *testing.T) {
 			expectedDataStreamDataset:   "1",
 			expectedDataStreamNamespace: "2",
 		},
+		// Test data sanitization: https://www.elastic.co/guide/en/ecs/current/ecs-data_stream.html
+		// 1. Replace all disallowed runes with _
+		// 2. Datastream length should not exceed otlp.MaxDataStreamBytes
+		{
+			resourceDataStreamDataset:   otlp.DisallowedDataStreamRunes + randomString,
+			resourceDataStreamNamespace: otlp.DisallowedDataStreamRunes + randomString,
+			expectedDataStreamDataset:   strings.Repeat("_", len(otlp.DisallowedDataStreamRunes)) + randomString[:maxLen],
+			expectedDataStreamNamespace: strings.Repeat("_", len(otlp.DisallowedDataStreamRunes)) + randomString[:maxLen],
+		},
 	} {
 		tcName := fmt.Sprintf("%s,%s", tc.expectedDataStreamDataset, tc.expectedDataStreamNamespace)
 		t.Run(tcName, func(t *testing.T) {

input/otlp/metadata.go

@@ -22,6 +22,7 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
+	"unicode"
 
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	semconv "go.opentelemetry.io/collector/semconv/v1.25.0"
@@ -30,7 +31,9 @@ import (
 )
 
 const (
-	AgentNameJaeger = "Jaeger"
+	AgentNameJaeger           = "Jaeger"
+	MaxDataStreamBytes        = 100
+	DisallowedDataStreamRunes = "-\\/*?\"<>| ,#:"
 )
 
 var (
@@ -316,12 +319,12 @@ func translateResourceMetadata(resource pcommon.Resource, out *modelpb.APMEvent)
 			if out.DataStream == nil {
 				out.DataStream = &modelpb.DataStream{}
 			}
-			out.DataStream.Dataset = v.Str()
+			out.DataStream.Dataset = sanitizeDataStreamField(v.Str())
 		case attributeDataStreamNamespace:
 			if out.DataStream == nil {
 				out.DataStream = &modelpb.DataStream{}
 			}
-			out.DataStream.Namespace = v.Str()
+			out.DataStream.Namespace = sanitizeDataStreamField(v.Str())
 		default:
 			if out.Labels == nil {
 				out.Labels = make(modelpb.Labels)
@@ -459,12 +462,12 @@ func translateScopeMetadata(scope pcommon.InstrumentationScope, out *modelpb.APM
 			if out.DataStream == nil {
 				out.DataStream = &modelpb.DataStream{}
 			}
-			out.DataStream.Dataset = v.Str()
+			out.DataStream.Dataset = sanitizeDataStreamField(v.Str())
 		case attributeDataStreamNamespace:
 			if out.DataStream == nil {
 				out.DataStream = &modelpb.DataStream{}
 			}
-			out.DataStream.Namespace = v.Str()
+			out.DataStream.Namespace = sanitizeDataStreamField(v.Str())
 		}
 		return true
 	})
@@ -545,3 +548,20 @@ func setLabel(key string, event *modelpb.APMEvent, v pcommon.Value) {
 		}
 	}
 }
+
+// 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 sanitizeDataStreamField(field string) string {
+	field = strings.Map(replaceReservedRune, field)
+	if len(field) > MaxDataStreamBytes {
+		return field[:MaxDataStreamBytes]
+	}
+	return field
+}
+
+func replaceReservedRune(r rune) rune {
+	if strings.ContainsRune(DisallowedDataStreamRunes, r) {
+		return '_'
+	}
+	return unicode.ToLower(r)
+}

input/otlp/metrics.go

@@ -180,12 +180,12 @@ func (c *Consumer) handleScopeMetrics(
 					if event.DataStream == nil {
 						event.DataStream = &modelpb.DataStream{}
 					}
-					event.DataStream.Dataset = v.Str()
+					event.DataStream.Dataset = sanitizeDataStreamField(v.Str())
 				case attributeDataStreamNamespace:
 					if event.DataStream == nil {
 						event.DataStream = &modelpb.DataStream{}
 					}
-					event.DataStream.Namespace = v.Str()
+					event.DataStream.Namespace = sanitizeDataStreamField(v.Str())
 
 				// The below fields are required by the Processes tab of the
 				// curated Kibana's hostmetrics UI. These fields are

input/otlp/metrics_test.go

@@ -729,6 +729,9 @@ func TestConsumeMetricsExportTimestamp(t *testing.T) {
 }
 
 func TestConsumeMetricsDataStream(t *testing.T) {
+	randomString := strings.Repeat("abcdefghijklmnopqrstuvwxyz0123456789", 10)
+	maxLen := otlp.MaxDataStreamBytes - len(otlp.DisallowedDataStreamRunes)
+
 	for _, tc := range []struct {
 		resourceDataStreamDataset   string
 		resourceDataStreamNamespace string
@@ -764,6 +767,15 @@ func TestConsumeMetricsDataStream(t *testing.T) {
 			expectedDataStreamDataset:   "1",
 			expectedDataStreamNamespace: "2",
 		},
+		// Test data sanitization: https://www.elastic.co/guide/en/ecs/current/ecs-data_stream.html
+		// 1. Replace all disallowed runes with _
+		// 2. Datastream length should not exceed otlp.MaxDataStreamBytes
+		{
+			resourceDataStreamDataset:   otlp.DisallowedDataStreamRunes + randomString,
+			resourceDataStreamNamespace: otlp.DisallowedDataStreamRunes + randomString,
+			expectedDataStreamDataset:   strings.Repeat("_", len(otlp.DisallowedDataStreamRunes)) + randomString[:maxLen],
+			expectedDataStreamNamespace: strings.Repeat("_", len(otlp.DisallowedDataStreamRunes)) + randomString[:maxLen],
+		},
 	} {
 		tcName := fmt.Sprintf("%s,%s", tc.expectedDataStreamDataset, tc.expectedDataStreamNamespace)
 		t.Run(tcName, func(t *testing.T) {

input/otlp/traces.go

@@ -481,12 +481,12 @@ func TranslateTransaction(
 				if event.DataStream == nil {
 					event.DataStream = &modelpb.DataStream{}
 				}
-				event.DataStream.Dataset = stringval
+				event.DataStream.Dataset = sanitizeDataStreamField(stringval)
 			case attributeDataStreamNamespace:
 				if event.DataStream == nil {
 					event.DataStream = &modelpb.DataStream{}
 				}
-				event.DataStream.Namespace = stringval
+				event.DataStream.Namespace = sanitizeDataStreamField(stringval)
 			default:
 				modelpb.Labels(event.Labels).Set(k, stringval)
 			}
@@ -824,12 +824,12 @@ func TranslateSpan(spanKind ptrace.SpanKind, attributes pcommon.Map, event *mode
 				if event.DataStream == nil {
 					event.DataStream = &modelpb.DataStream{}
 				}
-				event.DataStream.Dataset = stringval
+				event.DataStream.Dataset = sanitizeDataStreamField(stringval)
 			case attributeDataStreamNamespace:
 				if event.DataStream == nil {
 					event.DataStream = &modelpb.DataStream{}
 				}
-				event.DataStream.Namespace = stringval
+				event.DataStream.Namespace = sanitizeDataStreamField(stringval)
 			default:
 				setLabel(k, event, v)
 			}
@@ -1103,12 +1103,12 @@ func (c *Consumer) convertSpanEvent(
 				if event.DataStream == nil {
 					event.DataStream = &modelpb.DataStream{}
 				}
-				event.DataStream.Dataset = v.Str()
+				event.DataStream.Dataset = sanitizeDataStreamField(v.Str())
 			case attributeDataStreamNamespace:
 				if event.DataStream == nil {
 					event.DataStream = &modelpb.DataStream{}
 				}
-				event.DataStream.Namespace = v.Str()
+				event.DataStream.Namespace = sanitizeDataStreamField(v.Str())
 
 			default:
 				setLabel(replaceDots(k), event, v)
@@ -1144,12 +1144,12 @@ func (c *Consumer) convertSpanEvent(
 				if event.DataStream == nil {
 					event.DataStream = &modelpb.DataStream{}
 				}
-				event.DataStream.Dataset = v.Str()
+				event.DataStream.Dataset = sanitizeDataStreamField(v.Str())
 			case attributeDataStreamNamespace:
 				if event.DataStream == nil {
 					event.DataStream = &modelpb.DataStream{}
 				}
-				event.DataStream.Namespace = v.Str()
+				event.DataStream.Namespace = sanitizeDataStreamField(v.Str())
 			default:
 				k = replaceDots(k)
 				if isJaeger && k == "message" {
@@ -1202,12 +1202,12 @@ func (c *Consumer) convertJaegerErrorSpanEvent(event ptrace.SpanEvent, apmEvent
 			if apmEvent.DataStream == nil {
 				apmEvent.DataStream = &modelpb.DataStream{}
 			}
-			apmEvent.DataStream.Dataset = v.Str()
+			apmEvent.DataStream.Dataset = sanitizeDataStreamField(v.Str())
 		case attributeDataStreamNamespace:
 			if apmEvent.DataStream == nil {
 				apmEvent.DataStream = &modelpb.DataStream{}
 			}
-			apmEvent.DataStream.Namespace = v.Str()
+			apmEvent.DataStream.Namespace = sanitizeDataStreamField(v.Str())
 
 		default:
 			setLabel(replaceDots(k), apmEvent, v)

input/otlp/traces_test.go

@@ -873,6 +873,9 @@ func TestSpanNetworkAttributes(t *testing.T) {
 }
 
 func TestSpanDataStream(t *testing.T) {
+	randomString := strings.Repeat("abcdefghijklmnopqrstuvwxyz0123456789", 10)
+	maxLen := otlp.MaxDataStreamBytes - len(otlp.DisallowedDataStreamRunes)
+
 	for _, tc := range []struct {
 		resourceDataStreamDataset   string
 		resourceDataStreamNamespace string
@@ -908,6 +911,15 @@ func TestSpanDataStream(t *testing.T) {
 			expectedDataStreamDataset:   "1",
 			expectedDataStreamNamespace: "2",
 		},
+		// Test data sanitization: https://www.elastic.co/guide/en/ecs/current/ecs-data_stream.html
+		// 1. Replace all disallowed runes with _
+		// 2. Datastream length should not exceed otlp.MaxDataStreamBytes
+		{
+			resourceDataStreamDataset:   otlp.DisallowedDataStreamRunes + randomString,
+			resourceDataStreamNamespace: otlp.DisallowedDataStreamRunes + randomString,
+			expectedDataStreamDataset:   strings.Repeat("_", len(otlp.DisallowedDataStreamRunes)) + randomString[:maxLen],
+			expectedDataStreamNamespace: strings.Repeat("_", len(otlp.DisallowedDataStreamRunes)) + randomString[:maxLen],
+		},
 	} {
 		for _, isTxn := range []bool{false, true} {
 			tcName := fmt.Sprintf("%s,%s,txn=%v", tc.expectedDataStreamDataset, tc.expectedDataStreamNamespace, isTxn)