[v2] Implement sanitizers to operate on OTLP data

cmd/jaeger/README.md

@@ -1,6 +1,8 @@
 # jaeger
 
 This is experimental Jaeger V2 based on OpenTelemetry collector.
+See https://github.com/jaegertracing/jaeger/issues/4843.
+
 Read the [blog post](https://medium.com/jaegertracing/towards-jaeger-v2-moar-opentelemetry-2f8239bee48e).
 
 Tracking issue: https://github.com/jaegertracing/jaeger/issues/4843.
@@ -42,3 +44,9 @@ flowchart LR
   * `JAEGER_VERSION=1.59 docker compose -f path-to-yml-file-v2 up`
 * Access Jaeger UI at http://localhost:16686 and HotROD app at http://localhost:8080
 * Shutdown / cleanup with `docker compose -f path-to-yml-file down`
+
+## Compatibility
+
+### Service Name Sanitizer
+
+In v1, there was a `serviceNameSanitizer` that sanitized the service names in span annotations using a source of truth alias to service cache. This functionality has been removed in v2. If your implementation relies on this sanitizer, you will need to find a different way to integrate this functionality, such as implementing a custom processor.

cmd/jaeger/internal/exporters/storageexporter/sanitizer/empty_service_name_sanitizer.go

@@ -0,0 +1,38 @@
+// Copyright (c) 2024 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package sanitizer
+
+import (
+	"go.opentelemetry.io/collector/pdata/ptrace"
+)
+
+// Constants for the replacement names
+const (
+	emptyServiceName   = "empty-service-name"
+	missingServiceName = "missing-service-name"
+)
+
+// NewEmptyServiceNameSanitizer returns a function that replaces empty service names
+// with a predefined string.
+func NewEmptyServiceNameSanitizer() SanitizeTraces {
+	return sanitizeEmptyServiceName
+}
+
+// sanitizeEmptyServiceName sanitizes the service names in the resource attributes.
+func sanitizeEmptyServiceName(traces ptrace.Traces) ptrace.Traces {
+	resourceSpans := traces.ResourceSpans()
+	for i := 0; i < resourceSpans.Len(); i++ {
+		resourceSpan := resourceSpans.At(i)
+		attributes := resourceSpan.Resource().Attributes()
+		serviceNameAttr, ok := attributes.Get("service.name")
+		if !ok {
+			// If service.name is missing, set it to nullProcessServiceName
+			attributes.PutStr("service.name", missingServiceName)
+		} else if serviceNameAttr.Str() == "" {
+			// If service.name is empty, replace it with serviceNameReplacement
+			attributes.PutStr("service.name", emptyServiceName)
+		}
+	}
+	return traces
+}

cmd/jaeger/internal/exporters/storageexporter/sanitizer/empty_service_name_sanitizer_test.go

@@ -0,0 +1,4 @@
+// Copyright (c) 2024 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package sanitizer

cmd/jaeger/internal/exporters/storageexporter/sanitizer/package_test.go

@@ -0,0 +1,14 @@
+// Copyright (c) 2024 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package sanitizer
+
+import (
+	"testing"
+
+	"github.com/jaegertracing/jaeger/pkg/testutils"
+)
+
+func TestMain(m *testing.M) {
+	testutils.VerifyGoLeaks(m)
+}

cmd/jaeger/internal/exporters/storageexporter/sanitizer/sanitizer.go

@@ -0,0 +1,33 @@
+// Copyright (c) 2024 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package sanitizer
+
+import (
+	"go.opentelemetry.io/collector/pdata/ptrace"
+)
+
+// SanitizeTraces is a function that performs enrichment, clean-up, or normalization of trace data.
+type SanitizeTraces func(traces ptrace.Traces) ptrace.Traces
+
+// NewStandardSanitizers are automatically applied by SpanProcessor.
+func NewStandardSanitizers() []SanitizeTraces {
+	return []SanitizeTraces{
+		NewEmptyServiceNameSanitizer(),
+		NewUTF8Sanitizer(),
+	}
+}
+
+// NewChainedSanitizer creates a Sanitizer from the variadic list of passed Sanitizers.
+// If the list only has one element, it is returned directly to minimize indirection.
+func NewChainedSanitizer(sanitizers ...SanitizeTraces) SanitizeTraces {
+	if len(sanitizers) == 1 {
+		return sanitizers[0]
+	}
+	return func(traces ptrace.Traces) ptrace.Traces {
+		for _, s := range sanitizers {
+			traces = s(traces)
+		}
+		return traces
+	}
+}

cmd/jaeger/internal/exporters/storageexporter/sanitizer/utf8_sanitizer.go

@@ -0,0 +1,92 @@
+// Copyright (c) 2024 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package sanitizer
+
+import (
+	"unicode/utf8"
+
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+)
+
+const (
+	invalidOperation = "InvalidOperationName"
+	invalidTagKey    = "InvalidTagKey"
+	badUTF8Prefix    = "bad_utf8_"
+)
+
+// NewUTF8Sanitizer creates a UTF8 sanitizer.
+func NewUTF8Sanitizer() SanitizeTraces {
+	return sanitizeUTF8
+}
+
+// sanitizeUTF8 sanitizes the UTF8 in the traces.
+func sanitizeUTF8(traces ptrace.Traces) ptrace.Traces {
+	resourceSpans := traces.ResourceSpans()
+	for i := 0; i < resourceSpans.Len(); i++ {
+		resourceSpan := resourceSpans.At(i)
+
+		// Sanitize resource attributes
+		sanitizeAttributes(resourceSpan.Resource().Attributes())
+
+		scopeSpans := resourceSpan.ScopeSpans()
+		for j := 0; j < scopeSpans.Len(); j++ {
+			scopeSpan := scopeSpans.At(j)
+
+			// Sanitize scope attributes
+			sanitizeAttributes(scopeSpan.Scope().Attributes())
+
+			spans := scopeSpan.Spans()
+			for k := 0; k < spans.Len(); k++ {
+				span := spans.At(k)
+
+				// Sanitize operation name
+				if !utf8.ValidString(span.Name()) {
+					originalName := span.Name()
+					span.SetName(invalidOperation)
+					binaryAttr := span.Attributes().PutEmptyBytes(badUTF8Prefix + "operation_name")
+					binaryAttr.FromRaw([]byte(originalName))
+				}
+
+				// Sanitize span attributes
+				sanitizeAttributes(span.Attributes())
+			}
+		}
+	}
+	return traces
+}
+
+// sanitizeAttributes sanitizes attributes to ensure UTF8 validity.
+func sanitizeAttributes(attributes pcommon.Map) {
+	// Collect invalid keys and values during iteration
+	var invalidKeys []string
+	invalidValues := make(map[string]string)
+
+	attributes.Range(func(k string, v pcommon.Value) bool {
+		// Handle invalid UTF-8 in attribute keys
+		if !utf8.ValidString(k) {
+			invalidKeys = append(invalidKeys, k)
+		}
+		// Handle invalid UTF-8 in attribute values
+		if v.Type() == pcommon.ValueTypeStr && !utf8.ValidString(v.Str()) {
+			invalidValues[k] = v.Str()
+		}
+		return true
+	})
+
+	// Apply collected changes after iteration
+	for _, k := range invalidKeys {
+		originalKey := k
+		attributes.PutStr(invalidTagKey, k)
+		binaryAttr := attributes.PutEmptyBytes(badUTF8Prefix + originalKey)
+		binaryAttr.FromRaw([]byte(originalKey))
+	}
+
+	for k, v := range invalidValues {
+		originalValue := v
+		attributes.PutStr(k, invalidTagKey)
+		binaryAttr := attributes.PutEmptyBytes(badUTF8Prefix + k)
+		binaryAttr.FromRaw([]byte(originalValue))
+	}
+}