[processor/elasticapmprocessor] Fix otlp span discrepancies

processor/elasticapmprocessor/internal/ecs/ecs_translation.go

@@ -23,8 +23,13 @@ import (
 	"github.com/elastic/opentelemetry-collector-components/internal/elasticattr"
 	"github.com/elastic/opentelemetry-collector-components/processor/elasticapmprocessor/internal/sanitize"
 	"go.opentelemetry.io/collector/pdata/pcommon"
+	semconv12 "go.opentelemetry.io/otel/semconv/v1.12.0"
+	semconv16 "go.opentelemetry.io/otel/semconv/v1.16.0"
+	semconv25 "go.opentelemetry.io/otel/semconv/v1.25.0"
 	semconv26 "go.opentelemetry.io/otel/semconv/v1.26.0"
 	semconv "go.opentelemetry.io/otel/semconv/v1.27.0"
+	semconv37 "go.opentelemetry.io/otel/semconv/v1.37.0"
+	semconv39 "go.opentelemetry.io/otel/semconv/v1.39.0"
 )
 
 // Supported ECS resource attributes
@@ -186,6 +191,128 @@ func TranslateLogRecordAttributes(attributes pcommon.Map) {
 	})
 }
 
+// TranslateSpanAttributes applies the apm-data OTLP span fallback behaviour for
+// ECS mode spans. The preserved attributes and fallback-to-label cases are based
+// on the OTLP span translation in apm-data's input/otlp/traces.go:
+// https://github.com/elastic/apm-data/blob/7da222dcc0320f9c812c5d72f65f830c838aae11/input/otlp/traces.go
+//
+// Attributes required by span enrichment or exporter-side ECS conversions are
+// preserved, while unsupported attributes are moved to labels.* /
+// numeric_labels.* with a sanitized key.
+func TranslateSpanAttributes(attributes pcommon.Map) {
+	var attrsToDrop []string
+	attributes.Range(func(k string, v pcommon.Value) bool {
+		if sanitizeExistingLabelAttribute(attributes, k, v) {
+			return true
+		}
+
+		switch k {
+		// data_stream.*
+		case elasticattr.DataStreamDataset,
+			elasticattr.DataStreamNamespace,
+			elasticattr.DataStreamType,
+			elasticattr.ServiceTargetName,
+			elasticattr.ServiceTargetType,
+			elasticattr.SpanDestinationServiceName,
+			elasticattr.SpanDestinationServiceType,
+			elasticattr.SpanDestinationServiceResource,
+			// miscellaneous
+			elasticattr.EventOutcome,
+			elasticattr.ProcessorEvent,
+			elasticattr.SessionID,
+			elasticattr.TransactionType,
+			"type",
+			"code.stacktrace",
+			// db.*
+			string(semconv25.DBNameKey),
+			string(semconv37.DBNamespaceKey),
+			string(semconv37.DBQueryTextKey),
+			string(semconv25.DBStatementKey),
+			string(semconv25.DBSystemKey),
+			string(semconv37.DBSystemNameKey),
+			string(semconv25.DBUserKey),
+			// gen_ai.*
+			string(semconv37.GenAIProviderNameKey),
+			string(semconv.GenAISystemKey),
+			// http.*
+			string(semconv25.HTTPFlavorKey),
+			string(semconv25.HTTPMethodKey),
+			string(semconv25.HTTPRequestMethodKey),
+			string(semconv.HTTPResponseBodySizeKey),
+			string(semconv25.HTTPResponseStatusCodeKey),
+			string(semconv25.HTTPSchemeKey),
+			string(semconv25.HTTPStatusCodeKey),
+			string(semconv25.HTTPTargetKey),
+			string(semconv12.HTTPHostKey),
+			string(semconv25.HTTPURLKey),
+			string(semconv25.HTTPUserAgentKey),
+			// messaging.*
+			string(semconv16.MessagingDestinationKey),
+			string(semconv25.MessagingDestinationNameKey),
+			string(semconv25.MessagingDestinationTemporaryKey),
+			string(semconv25.MessagingOperationKey),
+			string(semconv.MessagingOperationTypeKey),
+			string(semconv37.MessagingOperationNameKey),
+			string(semconv25.MessagingSystemKey),
+			// net.*
+			string(semconv25.NetHostNameKey),
+			string(semconv25.NetPeerNameKey),
+			string(semconv25.NetPeerPortKey),
+			string(semconv12.NetPeerIPKey),
+			string(semconv25.NetSockPeerAddrKey),
+			string(semconv.NetworkPeerAddressKey),
+			// network.*
+			string(semconv.NetworkCarrierIccKey),
+			string(semconv.NetworkCarrierMccKey),
+			string(semconv.NetworkCarrierMncKey),
+			string(semconv.NetworkCarrierNameKey),
+			string(semconv.NetworkConnectionSubtypeKey),
+			string(semconv.NetworkConnectionTypeKey),
+			// rpc.*
+			string(semconv25.PeerServiceKey),
+			string(semconv25.RPCGRPCStatusCodeKey),
+			string(semconv39.RPCMethodKey),
+			string(semconv39.RPCResponseStatusCodeKey),
+			string(semconv25.RPCServiceKey),
+			string(semconv25.RPCSystemKey),
+			string(semconv39.RPCSystemNameKey),
+			// server.*
+			string(semconv25.ServerAddressKey),
+			string(semconv25.ServerPortKey),
+			// service.*
+			string(semconv39.ServicePeerNameKey),
+			// url.*
+			string(semconv25.URLDomainKey),
+			string(semconv25.URLFullKey),
+			string(semconv25.URLPathKey),
+			string(semconv25.URLPortKey),
+			string(semconv25.URLQueryKey),
+			string(semconv25.URLSchemeKey),
+			// user_agent.*
+			string(semconv.UserAgentNameKey),
+			string(semconv.UserAgentOriginalKey),
+			string(semconv.UserAgentVersionKey):
+			return true
+		case "span.kind",
+			"message_bus.destination",
+			"peer.address",
+			"peer.port",
+			"peer.hostname",
+			"peer.ipv4",
+			"peer.ipv6":
+			// remove redundant aliases
+			attrsToDrop = append(attrsToDrop, k)
+			return true
+		default:
+			fallbackToLabelAttribute(attributes, k, v)
+			return true
+		}
+	})
+	for _, key := range attrsToDrop {
+		attributes.Remove(key)
+	}
+}
+
 // TranslateMetricDataPointAttributes applies the apm-data OTLP metric fallback
 // for raw metric datapoint attributes in ECS mode. Existing labels.* /
 // numeric_labels.* keys are sanitized in place, metric-specific special cases

processor/elasticapmprocessor/internal/ecs/ecs_translation_test.go

@@ -414,6 +414,143 @@ func TestTranslateLogRecordAttributes(t *testing.T) {
 	}
 }
 
+func TestTranslateSpanAttributes(t *testing.T) {
+	tests := []struct {
+		name       string
+		setAttrs   func(pcommon.Map)
+		want       map[string]any
+		wantAbsent []string
+	}{
+		{
+			name: "supported span attrs preserved",
+			setAttrs: func(attrs pcommon.Map) {
+				attrs.PutStr("http.request.method", "GET")
+				attrs.PutInt("http.response.status_code", 200)
+				attrs.PutStr("http.url", "https://api.example.com/users")
+				attrs.PutStr("db.query.text", "SELECT * FROM users")
+				attrs.PutStr(elasticattr.EventOutcome, "unknown")
+				attrs.PutStr(elasticattr.ProcessorEvent, "transaction")
+				attrs.PutStr(elasticattr.ServiceTargetName, "api.example.com:443")
+				attrs.PutStr(elasticattr.ServiceTargetType, "http")
+				attrs.PutStr(elasticattr.SpanDestinationServiceName, "https://api.example.com")
+				attrs.PutStr(elasticattr.SpanDestinationServiceType, "external")
+				attrs.PutStr(elasticattr.SpanDestinationServiceResource, "api.example.com:443")
+				attrs.PutStr("session.id", "session-123")
+				attrs.PutStr(string(semconv.NetworkConnectionTypeKey), "wifi")
+				attrs.PutStr(elasticattr.DataStreamDataset, "apm")
+			},
+			want: map[string]any{
+				"http.request.method":                      "GET",
+				"http.response.status_code":                int64(200),
+				"http.url":                                 "https://api.example.com/users",
+				"db.query.text":                            "SELECT * FROM users",
+				elasticattr.EventOutcome:                   "unknown",
+				elasticattr.ProcessorEvent:                 "transaction",
+				elasticattr.ServiceTargetName:              "api.example.com:443",
+				elasticattr.ServiceTargetType:              "http",
+				elasticattr.SpanDestinationServiceName:     "https://api.example.com",
+				elasticattr.SpanDestinationServiceType:     "external",
+				elasticattr.SpanDestinationServiceResource: "api.example.com:443",
+				"session.id":                               "session-123",
+				string(semconv.NetworkConnectionTypeKey):   "wifi",
+				elasticattr.DataStreamDataset:              "apm",
+			},
+		},
+		{
+			name: "unsupported span attrs moved to labels",
+			setAttrs: func(attrs pcommon.Map) {
+				attrs.PutStr("http.route", "/users")
+				attrs.PutInt("http.response_content_length", 1024)
+				attrs.PutStr("error.message", "Database connection failed")
+				attrs.PutStr("error.type", "DBError")
+				attrs.PutStr("exception.message", "Database connection failed")
+				attrs.PutStr("exception.type", "DBError")
+				attrs.PutStr("exception.stacktrace", "stacktrace")
+				attrs.PutBool("custom.flag", true)
+			},
+			want: map[string]any{
+				"labels.http_route":                           "/users",
+				"numeric_labels.http_response_content_length": float64(1024),
+				"labels.error_message":                        "Database connection failed",
+				"labels.error_type":                           "DBError",
+				"labels.exception_message":                    "Database connection failed",
+				"labels.exception_type":                       "DBError",
+				"labels.exception_stacktrace":                 "stacktrace",
+				"labels.custom_flag":                          "true",
+			},
+			wantAbsent: []string{
+				"http.route",
+				"http.response_content_length",
+				"error.message",
+				"error.type",
+				"exception.message",
+				"exception.type",
+				"exception.stacktrace",
+				"custom.flag",
+			},
+		},
+		{
+			name: "existing span label keys are sanitized",
+			setAttrs: func(attrs pcommon.Map) {
+				attrs.PutStr("labels.http.route", "/users")
+				attrs.PutDouble("numeric_labels.http.response_content_length", 1024)
+			},
+			want: map[string]any{
+				"labels.http_route":                           "/users",
+				"numeric_labels.http_response_content_length": 1024.0,
+			},
+			wantAbsent: []string{
+				"labels.http.route",
+				"numeric_labels.http.response_content_length",
+			},
+		},
+		{
+			name: "unsupported span map attr dropped",
+			setAttrs: func(attrs pcommon.Map) {
+				attrs.PutEmptyMap("http.request")
+			},
+			wantAbsent: []string{"http.request"},
+		},
+		{
+			name: "span kind dropped",
+			setAttrs: func(attrs pcommon.Map) {
+				attrs.PutStr("span.kind", "client")
+			},
+			wantAbsent: []string{"span.kind"},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			attrs := pcommon.NewMap()
+			tc.setAttrs(attrs)
+
+			TranslateSpanAttributes(attrs)
+
+			for key, want := range tc.want {
+				got, ok := attrs.Get(key)
+				if !assert.True(t, ok, "expected %q to be present", key) {
+					continue
+				}
+				switch want := want.(type) {
+				case string:
+					assert.Equal(t, want, got.Str())
+				case int64:
+					assert.Equal(t, want, got.Int())
+				case float64:
+					assert.InDelta(t, want, got.Double(), 1e-9)
+				default:
+					t.Fatalf("unsupported want type %T", want)
+				}
+			}
+			for _, key := range tc.wantAbsent {
+				_, ok := attrs.Get(key)
+				assert.False(t, ok, "expected %q to be absent", key)
+			}
+		})
+	}
+}
+
 func TestTranslateMetricDataPointAttributes(t *testing.T) {
 	tests := []struct {
 		name       string

processor/elasticapmprocessor/internal/enrichments/config/config.go

@@ -87,19 +87,20 @@ type ElasticSpanConfig struct {
 	// TimestampUs is a temporary attribute to enable higher
 	// resolution timestamps in Elasticsearch. For more details see:
 	// https://github.com/elastic/opentelemetry-dev/issues/374.
-	TimestampUs         AttributeConfig `mapstructure:"timestamp_us"`
-	ProcessorEvent      AttributeConfig `mapstructure:"processor_event"`
-	RepresentativeCount AttributeConfig `mapstructure:"representative_count"`
-	Action              AttributeConfig `mapstructure:"action"`
-	TypeSubtype         AttributeConfig `mapstructure:"type_subtype"`
-	DurationUs          AttributeConfig `mapstructure:"duration_us"`
-	EventOutcome        AttributeConfig `mapstructure:"event_outcome"`
-	ServiceTarget       AttributeConfig `mapstructure:"service_target"`
-	DestinationService  AttributeConfig `mapstructure:"destination_service"`
-	InferredSpans       AttributeConfig `mapstructure:"inferred_spans"`
-	UserAgent           AttributeConfig `mapstructure:"user_agent"`
-	RemoveMessaging     AttributeConfig `mapstructure:"remove_messaging"`
-	MessageQueueName    AttributeConfig `mapstructure:"message_queue_name"`
+	TimestampUs                    AttributeConfig `mapstructure:"timestamp_us"`
+	ProcessorEvent                 AttributeConfig `mapstructure:"processor_event"`
+	RepresentativeCount            AttributeConfig `mapstructure:"representative_count"`
+	Action                         AttributeConfig `mapstructure:"action"`
+	TypeSubtype                    AttributeConfig `mapstructure:"type_subtype"`
+	DurationUs                     AttributeConfig `mapstructure:"duration_us"`
+	EventOutcome                   AttributeConfig `mapstructure:"event_outcome"`
+	ServiceTarget                  AttributeConfig `mapstructure:"service_target"`
+	DestinationService             AttributeConfig `mapstructure:"destination_service"`
+	InferredSpans                  AttributeConfig `mapstructure:"inferred_spans"`
+	UserAgent                      AttributeConfig `mapstructure:"user_agent"`
+	RemoveMessaging                AttributeConfig `mapstructure:"remove_messaging"`
+	MessageQueueName               AttributeConfig `mapstructure:"message_queue_name"`
+	TranslateUnsupportedAttributes AttributeConfig `mapstructure:"translate_unsupported_attributes"`
 }
 
 // SpanEventConfig configures enrichment attributes for the span events.

processor/elasticapmprocessor/internal/enrichments/enricher.go

@@ -52,16 +52,22 @@ type Enricher struct {
 func (e *Enricher) EnrichTraces(pt ptrace.Traces) {
 	resSpans := pt.ResourceSpans()
 	for i := 0; i < resSpans.Len(); i++ {
-		resSpan := resSpans.At(i)
-		EnrichResource(resSpan.Resource(), e.Config.Resource)
-		scopeSpans := resSpan.ScopeSpans()
-		for j := 0; j < scopeSpans.Len(); j++ {
-			scopeSpan := scopeSpans.At(j)
-			EnrichScope(scopeSpan.Scope(), e.Config)
-			spans := scopeSpan.Spans()
-			for k := 0; k < spans.Len(); k++ {
-				EnrichSpan(spans.At(k), e.Config, e.userAgentParser)
-			}
+		e.EnrichResourceSpans(resSpans.At(i))
+	}
+}
+
+// EnrichResourceSpans enriches a single ResourceSpans with the current enricher
+// configuration. This is used when ECS batches need per-resource enrichment
+// policies based on the origin of each ResourceSpans.
+func (e *Enricher) EnrichResourceSpans(resSpan ptrace.ResourceSpans) {
+	EnrichResource(resSpan.Resource(), e.Config.Resource)
+	scopeSpans := resSpan.ScopeSpans()
+	for j := 0; j < scopeSpans.Len(); j++ {
+		scopeSpan := scopeSpans.At(j)
+		EnrichScope(scopeSpan.Scope(), e.Config)
+		spans := scopeSpan.Spans()
+		for k := 0; k < spans.Len(); k++ {
+			EnrichSpan(spans.At(k), e.Config, e.userAgentParser)
 		}
 	}
 }