[exporter/awsxray] Add span links and messenger field translation to x-ray exporter.

.gitignore

@@ -1,4 +1,5 @@
 local/
+vendor/
 
 # GoLand IDEA
 /.idea/

exporter/awsxrayexporter/internal/translator/messaging.go

@@ -0,0 +1,40 @@
+// Copyright 2019, OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package translator // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/awsxrayexporter/internal/translator"
+
+import (
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+	"strings"
+)
+
+const MessagingPrefix = "messaging."
+
+func makeMessaging(span ptrace.Span, attributes map[string]pcommon.Value) (map[string]pcommon.Value, map[string]interface{}) {
+	var (
+		filtered  = make(map[string]pcommon.Value)
+		messaging = make(map[string]interface{})
+	)
+
+	for key, value := range attributes {
+		if strings.HasPrefix(key, MessagingPrefix) {
+			messaging[strings.TrimPrefix(key, MessagingPrefix)] = value.AsRaw()
+		} else {
+			filtered[key] = value
+		}
+	}
+
+	return filtered, messaging
+}

exporter/awsxrayexporter/internal/translator/messaging_test.go

@@ -0,0 +1,165 @@
+// Copyright 2019, OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package translator // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/awsxrayexporter/internal/translator"
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"go.opentelemetry.io/collector/pdata/ptrace"
+)
+
+func TestMessagingSimple(t *testing.T) {
+	spanName := "ProcessingMessage"
+	parentSpanID := newSegmentID()
+	attributes := make(map[string]interface{})
+	resource := constructDefaultResource()
+	span := constructServerSpan(parentSpanID, spanName, ptrace.StatusCodeOk, "OK", attributes)
+
+	span.Attributes().PutStr("messaging.operation", "process")
+	span.Attributes().PutStr("messaging.system", "AmazonSQS")
+	span.Attributes().PutStr("notMessaging", "myValue")
+
+	segment, _ := MakeSegment(span, resource, nil, false, nil)
+
+	assert.Equal(t, 2, len(segment.Messaging))
+	assert.Equal(t, "process", segment.Messaging["operation"])
+	assert.Equal(t, "AmazonSQS", segment.Messaging["system"])
+	assert.Equal(t, "myValue", segment.Metadata["default"]["notMessaging"])
+	assert.Equal(t, 0, len(segment.Annotations))
+
+	jsonStr, _ := MakeSegmentDocumentString(span, resource, nil, false, nil)
+
+	assert.True(t, strings.Contains(jsonStr, "messaging"))
+
+	assert.True(t, strings.Contains(jsonStr, "operation"))
+	assert.True(t, strings.Contains(jsonStr, "process"))
+
+	assert.True(t, strings.Contains(jsonStr, "system"))
+	assert.True(t, strings.Contains(jsonStr, "AmazonSQS"))
+
+	assert.True(t, strings.Contains(jsonStr, "notMessaging"))
+	assert.True(t, strings.Contains(jsonStr, "myValue"))
+}
+
+func TestMessagingEmpty(t *testing.T) {
+	spanName := "ProcessingMessage"
+	parentSpanID := newSegmentID()
+	attributes := make(map[string]interface{})
+	resource := constructDefaultResource()
+	span := constructServerSpan(parentSpanID, spanName, ptrace.StatusCodeOk, "OK", attributes)
+
+	segment, _ := MakeSegment(span, resource, nil, false, nil)
+
+	assert.Equal(t, 0, len(segment.Messaging))
+
+	jsonStr, _ := MakeSegmentDocumentString(span, resource, nil, false, nil)
+
+	assert.False(t, strings.Contains(jsonStr, "messaging"))
+}
+
+func TestMessagingComplex(t *testing.T) {
+	spanName := "ProcessingMessage"
+	parentSpanID := newSegmentID()
+	attributes := make(map[string]interface{})
+	resource := constructDefaultResource()
+	span := constructServerSpan(parentSpanID, spanName, ptrace.StatusCodeOk, "OK", attributes)
+
+	span.Attributes().PutStr("messaging.operation", "process")
+	span.Attributes().PutStr("messaging.system", "AmazonSQS")
+	span.Attributes().PutInt("messaging.message_count", 7)
+	span.Attributes().PutInt("messaging.payload_size_bytes", 2048)
+	span.Attributes().PutInt("messaging.payload_compressed_size_bytes", 1024)
+	span.Attributes().PutStr("messaging.conversation_id", "MyConversationId")
+	span.Attributes().PutStr("messaging.id", "452a7c7c7c7048c2f887f61572b18fc2")
+	var slice = span.Attributes().PutEmptySlice("messaging.sliceData")
+	slice.AppendEmpty().SetStr("alpha")
+	slice.AppendEmpty().SetStr("beta")
+
+	segment, _ := MakeSegment(span, resource, nil, false, nil)
+
+	assert.Equal(t, 8, len(segment.Messaging))
+	assert.Equal(t, "process", segment.Messaging["operation"])
+	assert.Equal(t, "AmazonSQS", segment.Messaging["system"])
+	assert.Equal(t, int64(7), segment.Messaging["message_count"])
+	assert.Equal(t, int64(2048), segment.Messaging["payload_size_bytes"])
+	assert.Equal(t, int64(1024), segment.Messaging["payload_compressed_size_bytes"])
+	assert.Equal(t, "MyConversationId", segment.Messaging["conversation_id"])
+	assert.Equal(t, "452a7c7c7c7048c2f887f61572b18fc2", segment.Messaging["id"])
+	assert.Equal(t, "alpha", segment.Messaging["sliceData"].([]interface{})[0])
+	assert.Equal(t, "beta", segment.Messaging["sliceData"].([]interface{})[1])
+
+	jsonStr, _ := MakeSegmentDocumentString(span, resource, nil, false, nil)
+
+	assert.True(t, strings.Contains(jsonStr, "messaging"))
+
+	assert.True(t, strings.Contains(jsonStr, "operation"))
+	assert.True(t, strings.Contains(jsonStr, "process"))
+
+	assert.True(t, strings.Contains(jsonStr, "system"))
+	assert.True(t, strings.Contains(jsonStr, "AmazonSQS"))
+
+	assert.True(t, strings.Contains(jsonStr, "message_count"))
+	assert.True(t, strings.Contains(jsonStr, "7"))
+
+	assert.True(t, strings.Contains(jsonStr, "payload_size_bytes"))
+	assert.True(t, strings.Contains(jsonStr, "2048"))
+
+	assert.True(t, strings.Contains(jsonStr, "payload_compressed_size_bytes"))
+	assert.True(t, strings.Contains(jsonStr, "1024"))
+
+	assert.True(t, strings.Contains(jsonStr, "conversation_id"))
+	assert.True(t, strings.Contains(jsonStr, "MyConversationId"))
+
+	assert.True(t, strings.Contains(jsonStr, "id"))
+	assert.True(t, strings.Contains(jsonStr, "452a7c7c7c7048c2f887f61572b18fc2"))
+
+	assert.True(t, strings.Contains(jsonStr, "sliceData"))
+	assert.True(t, strings.Contains(jsonStr, "alpha"))
+	assert.True(t, strings.Contains(jsonStr, "beta"))
+}
+
+func TestMessagingWithIndexedAttributes(t *testing.T) {
+	spanName := "ProcessingMessage"
+	parentSpanID := newSegmentID()
+	attributes := make(map[string]interface{})
+	resource := constructDefaultResource()
+	span := constructServerSpan(parentSpanID, spanName, ptrace.StatusCodeOk, "OK", attributes)
+
+	span.Attributes().PutStr("messaging.operation", "process")
+	span.Attributes().PutStr("messaging.system", "AmazonSQS")
+
+	segment, _ := MakeSegment(span, resource, []string{"messaging.system"}, false, nil)
+
+	assert.Equal(t, 2, len(segment.Messaging))
+	assert.Equal(t, "process", segment.Messaging["operation"])
+	assert.Equal(t, "AmazonSQS", segment.Messaging["system"])
+	assert.Equal(t, 1, len(segment.Annotations))
+	assert.Equal(t, "AmazonSQS", segment.Annotations["messaging_system"])
+
+	jsonStr, _ := MakeSegmentDocumentString(span, resource, []string{"messaging.system"}, false, nil)
+
+	assert.True(t, strings.Contains(jsonStr, "messaging"))
+
+	assert.True(t, strings.Contains(jsonStr, "operation"))
+	assert.True(t, strings.Contains(jsonStr, "process"))
+
+	assert.True(t, strings.Contains(jsonStr, "system"))
+	assert.True(t, strings.Contains(jsonStr, "AmazonSQS"))
+	assert.True(t, strings.Contains(jsonStr, "annotations"))
+	assert.True(t, strings.Contains(jsonStr, "messaging_system"))
+}

exporter/awsxrayexporter/internal/translator/segment.go

@@ -110,12 +110,18 @@ func MakeSegment(span ptrace.Span, resource pcommon.Resource, indexedAttrs []str
 		awsfiltered, aws                                   = makeAws(causefiltered, resource, logGroupNames)
 		service                                            = makeService(resource)
 		sqlfiltered, sql                                   = makeSQL(span, awsfiltered)
-		additionalAttrs                                    = addSpecialAttributes(sqlfiltered, indexedAttrs, attributes)
+		messagingFiltered, messaging                       = makeMessaging(span, sqlfiltered)
+		additionalAttrs                                    = addSpecialAttributes(messagingFiltered, indexedAttrs, attributes)
 		user, annotations, metadata                        = makeXRayAttributes(additionalAttrs, resource, storeResource, indexedAttrs, indexAllAttrs)
+		spanLinks, makeSpanLinkErr                         = makeSpanLinks(span.Links())
 		name                                               string
 		namespace                                          string
 	)
 
+	if makeSpanLinkErr != nil {
+		return nil, makeSpanLinkErr
+	}
+
 	// X-Ray segment names are service names, unlike span names which are methods. Try to find a service name.
 
 	// peer.service should always be prioritized for segment names when set because it is what the user decided.
@@ -211,6 +217,8 @@ func MakeSegment(span ptrace.Span, resource pcommon.Resource, indexedAttrs []str
 		Annotations: annotations,
 		Metadata:    metadata,
 		Type:        awsxray.String(segmentType),
+		Links:       spanLinks,
+		Messaging:   messaging,
 	}, nil
 }
 

exporter/awsxrayexporter/internal/translator/spanLinks.go

@@ -0,0 +1,53 @@
+// Copyright 2019, OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package translator // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/awsxrayexporter/internal/translator"
+
+import (
+	awsxray "github.com/open-telemetry/opentelemetry-collector-contrib/internal/aws/xray"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+)
+
+func makeSpanLinks(links ptrace.SpanLinkSlice) ([]awsxray.SpanLinkData, error) {
+	var spanLinkDataArray []awsxray.SpanLinkData
+
+	for i := 0; i < links.Len(); i++ {
+		var spanLinkData awsxray.SpanLinkData
+		var link = links.At(i)
+
+		var spanID = link.SpanID().String()
+		traceID, err := convertToAmazonTraceID(link.TraceID())
+
+		if err != nil {
+			return nil, err
+		}
+
+		spanLinkData.SpanID = &spanID
+		spanLinkData.TraceID = &traceID
+
+		if link.Attributes().Len() > 0 {
+			spanLinkData.Attributes = make(map[string]interface{})
+
+			link.Attributes().Range(func(k string, v pcommon.Value) bool {
+				spanLinkData.Attributes[k] = v.AsRaw()
+				return true
+			})
+		}
+
+		spanLinkDataArray = append(spanLinkDataArray, spanLinkData)
+	}
+
+	return spanLinkDataArray, nil
+}