translate otel messaging.* to ecs

changelogs/head.asciidoc

@@ -23,6 +23,7 @@ https://github.com/elastic/apm-server/compare/7.13\...master[View commits]
 * Support setting agent configuration from apm-server.yml {pull}5177[5177]
 * Add metric_type and unit to field metadata of system metrics {pull}5230[5230]
 * Upgrade Go to 1.15.12 {pull}[]
+* Translate otel messaging.* semantic conventions to ECS {pull}5334[5334]
 * Add support for dynamic histogram metrics {pull}5239[5239]
 
 [float]

processor/otel/consumer.go

@@ -313,10 +313,8 @@ func translateTransaction(
 			case conventions.AttributeNetHostName:
 				netHostName = stringval
 
-			// messaging
-			//
-			// TODO(axw) translate OpenTelemtry messaging conventions.
-			case "message_bus.destination":
+			// messaging.*
+			case "message_bus.destination", conventions.AttributeMessagingDestination:
 				message.QueueName = stringval
 				isMessaging = true
 
@@ -425,6 +423,11 @@ func translateSpan(span pdata.Span, metadata model.Metadata, event *model.Span)
 		httpScheme string = "http"
 	)
 
+	var (
+		messageSystem    string
+		messageOperation string
+	)
+
 	var http model.HTTP
 	var message model.Message
 	var db model.DB
@@ -517,12 +520,21 @@ func translateSpan(span pdata.Span, metadata model.Metadata, event *model.Span)
 					netPeerName = stringval
 				}
 
-			// messaging
-			//
-			// TODO(axw) translate OpenTelemtry messaging conventions.
-			case "message_bus.destination":
+			// messaging.*
+			case "message_bus.destination", conventions.AttributeMessagingDestination:
 				message.QueueName = stringval
 				isMessagingSpan = true
+			case conventions.AttributeMessagingOperation:
+				if stringval == "" && span.Kind() == pdata.SpanKindPRODUCER {
+					stringval = "send"
+				}
+				messageOperation = stringval
+				isMessagingSpan = true
+			case conventions.AttributeMessagingSystem:
+				messageSystem = stringval
+				destinationService.Resource = stringval
+				destinationService.Name = stringval
+				isMessagingSpan = true
 
 			// rpc.*
 			//
@@ -650,6 +662,12 @@ func translateSpan(span pdata.Span, metadata model.Metadata, event *model.Span)
 		event.DB = &db
 	case isMessagingSpan:
 		event.Type = "messaging"
+		event.Subtype = messageSystem
+		event.Action = messageOperation
+
+		if destinationService.Resource != "" && message.QueueName != "" {
+			destinationService.Resource += "/" + message.QueueName
+		}
 		event.Message = &message
 	case isRPCSpan:
 		event.Type = "external"

processor/otel/consumer_test.go

@@ -518,6 +518,42 @@ func TestRPCSpan(t *testing.T) {
 	}, span.DestinationService)
 }
 
+func TestMessagingTransaction(t *testing.T) {
+	tx := transformTransactionWithAttributes(t, map[string]pdata.AttributeValue{
+		"messaging.destination": pdata.NewAttributeValueString("myQueue"),
+	})
+	assert.Equal(t, "messaging", tx.Type)
+	assert.Empty(t, tx.Labels)
+	assert.Equal(t, &model.Message{
+		QueueName: "myQueue",
+	}, tx.Message)
+}
+
+func TestMessagingSpan(t *testing.T) {
+	span := transformSpanWithAttributes(t, map[string]pdata.AttributeValue{
+		"messaging.system":      pdata.NewAttributeValueString("kafka"),
+		"messaging.operation":   pdata.NewAttributeValueString(""),
+		"messaging.destination": pdata.NewAttributeValueString("myTopic"),
+		"net.peer.ip":           pdata.NewAttributeValueString("10.20.30.40"),
+		"net.peer.port":         pdata.NewAttributeValueInt(123),
+	}, func(s pdata.Span) {
+		s.SetKind(pdata.SpanKindPRODUCER)
+	})
+	assert.Equal(t, "messaging", span.Type)
+	assert.Equal(t, "kafka", span.Subtype)
+	assert.Equal(t, "send", span.Action)
+	assert.Empty(t, span.Labels)
+	assert.Equal(t, &model.Destination{
+		Address: "10.20.30.40",
+		Port:    123,
+	}, span.Destination)
+	assert.Equal(t, &model.DestinationService{
+		Type:     "messaging",
+		Name:     "kafka",
+		Resource: "kafka/myTopic",
+	}, span.DestinationService)
+}
+
 func TestArrayLabels(t *testing.T) {
 	stringArray := pdata.NewAttributeValueArray()
 	stringArray.ArrayVal().Append(pdata.NewAttributeValueString("string1"))
@@ -1066,12 +1102,15 @@ func transformTransactionWithAttributes(t *testing.T, attrs map[string]pdata.Att
 	return events.Transactions[0]
 }
 
-func transformSpanWithAttributes(t *testing.T, attrs map[string]pdata.AttributeValue) *model.Span {
+func transformSpanWithAttributes(t *testing.T, attrs map[string]pdata.AttributeValue, configFns ...func(pdata.Span)) *model.Span {
 	traces, spans := newTracesSpans()
 	otelSpan := pdata.NewSpan()
 	otelSpan.SetTraceID(pdata.NewTraceID([16]byte{1}))
 	otelSpan.SetSpanID(pdata.NewSpanID([8]byte{2}))
 	otelSpan.SetParentSpanID(pdata.NewSpanID([8]byte{3}))
+	for _, fn := range configFns {
+		fn(otelSpan)
+	}
 	otelSpan.Attributes().InitFromMap(attrs)
 	spans.Spans().Append(otelSpan)
 	events := transformTraces(t, traces)