[receiver/skywalking] add Skywalking tracing receiver impl

CHANGELOG.md

@@ -26,6 +26,7 @@
 - `coralogixexporter` Allow exporter timeout to be configured (#7957)
 - `prometheusremotewriteexporter` support adding trace id and span id attached to exemplars (#8380)
 - `influxdbexporter`: accept histogram metric missing infinity bucket. (#8462)
+- `skywalkingreceiver`: Added implementation of Skywalking receiver. (#8549)
 
 ### 🛑 Breaking changes 🛑
 

cmd/configschema/go.mod

@@ -471,7 +471,7 @@ require (
 	sigs.k8s.io/json v0.0.0-20211208200746-9f7c6b3444d2 // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.2.1 // indirect
 	sigs.k8s.io/yaml v1.3.0 // indirect
-	skywalking.apache.org/repo/goapi v0.0.0-20211122071111-ffc517fbfe21 // indirect
+	skywalking.apache.org/repo/goapi v0.0.0-20220121092418-9c455d0dda3f // indirect
 )
 
 // Replace references to modules that are in this repository with their relateive paths
@@ -751,6 +751,8 @@ replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/signa
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/simpleprometheusreceiver => ../../receiver/simpleprometheusreceiver
 
+replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/skywalkingreceiver => ../../receiver/skywalkingreceiver
+
 replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/splunkhecreceiver => ../../receiver/splunkhecreceiver
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/statsdreceiver => ../../receiver/statsdreceiver

go.mod

@@ -112,6 +112,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/sapmreceiver v0.47.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/signalfxreceiver v0.47.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/simpleprometheusreceiver v0.47.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/skywalkingreceiver v0.47.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/splunkhecreceiver v0.47.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/statsdreceiver v0.47.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/syslogreceiver v0.47.0
@@ -477,7 +478,7 @@ require (
 	sigs.k8s.io/json v0.0.0-20211208200746-9f7c6b3444d2 // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.2.1 // indirect
 	sigs.k8s.io/yaml v1.3.0 // indirect
-	skywalking.apache.org/repo/goapi v0.0.0-20211122071111-ffc517fbfe21 // indirect
+	skywalking.apache.org/repo/goapi v0.0.0-20220121092418-9c455d0dda3f // indirect
 )
 
 // Replace references to modules that are in this repository with their relateive paths

internal/components/components.go

@@ -132,6 +132,7 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/sapmreceiver"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/signalfxreceiver"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/simpleprometheusreceiver"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/skywalkingreceiver"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/splunkhecreceiver"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/statsdreceiver"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/syslogreceiver"
@@ -207,6 +208,7 @@ func Components() (component.Factories, error) {
 		sapmreceiver.NewFactory(),
 		signalfxreceiver.NewFactory(),
 		simpleprometheusreceiver.NewFactory(),
+		skywalkingreceiver.NewFactory(),
 		splunkhecreceiver.NewFactory(),
 		statsdreceiver.NewFactory(),
 		wavefrontreceiver.NewFactory(),

internal/components/receivers_test.go

@@ -199,6 +199,9 @@ func TestDefaultReceivers(t *testing.T) {
 		{
 			receiver: "prometheus_simple",
 		},
+		{
+			receiver: "skywalking",
+		},
 		{
 			receiver: "splunk_hec",
 		},

receiver/skywalkingreceiver/README.md

@@ -6,7 +6,7 @@ Supported pipeline types: traces
 
 ## ⚠️ Warning
 
-Note: This component is currently work in progress, and traces receiver is not yet fully functional.
+Note: This component is experimental and is not recommended for production environments.
 
 ## Getting Started
 

receiver/skywalkingreceiver/skywalkingproto_to_traces.go

@@ -0,0 +1,250 @@
+// Copyright The 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 skywalkingreceiver
+
+import (
+	"encoding/binary"
+	"fmt"
+	"reflect"
+	"time"
+	"unsafe"
+
+	"go.opentelemetry.io/collector/model/pdata"
+	conventions "go.opentelemetry.io/collector/model/semconv/v1.8.0"
+	common "skywalking.apache.org/repo/goapi/collect/common/v3"
+	agentV3 "skywalking.apache.org/repo/goapi/collect/language/agent/v3"
+)
+
+var OtSpanTagsMapping = map[string]string{
+	"url":         conventions.AttributeHTTPURL,
+	"status_code": conventions.AttributeHTTPStatusCode,
+	"db.type":     conventions.AttributeDBSystem,
+	"db.instance": conventions.AttributeDBName,
+	"mq.broker":   conventions.AttributeNetPeerName,
+}
+
+func SkywalkingToOtlpTraces(segment *agentV3.SegmentObject) pdata.Traces {
+	traceData := pdata.NewTraces()
+
+	swSpans := segment.Spans
+	if swSpans == nil && len(swSpans) == 0 {
+		return traceData
+	}
+
+	resourceSpan := traceData.ResourceSpans().AppendEmpty()
+	rs := resourceSpan.Resource()
+	for _, span := range swSpans {
+		swTagsToInternalResource(span, rs)
+		rs.Attributes().Insert(conventions.AttributeServiceName, pdata.NewAttributeValueString(segment.GetService()))
+		rs.Attributes().Insert(conventions.AttributeServiceInstanceID, pdata.NewAttributeValueString(segment.GetServiceInstance()))
+	}
+
+	il := resourceSpan.InstrumentationLibrarySpans().AppendEmpty()
+	swSpansToOtlpSpans(segment.GetTraceId(), swSpans, il.Spans())
+
+	return traceData
+}
+
+func swTagsToInternalResource(span *agentV3.SpanObject, dest pdata.Resource) {
+	if span == nil {
+		return
+	}
+
+	attrs := dest.Attributes()
+	attrs.Clear()
+
+	tags := span.Tags
+	if tags == nil {
+		return
+	}
+
+	for _, tag := range tags {
+		otKey, ok := OtSpanTagsMapping[tag.Key]
+		if ok {
+			attrs.UpsertString(otKey, tag.Value)
+		}
+	}
+}
+
+func swSpansToOtlpSpans(traceId string, spans []*agentV3.SpanObject, dest pdata.SpanSlice) {
+	if len(spans) == 0 {
+		return
+	}
+
+	dest.EnsureCapacity(len(spans))
+	for _, span := range spans {
+		if span == nil {
+			continue
+		}
+		swSpanToOtelSpan(traceId, span, dest.AppendEmpty())
+	}
+}
+
+func swSpanToOtelSpan(traceId string, span *agentV3.SpanObject, dest pdata.Span) {
+	dest.SetTraceID(stringToTraceID(traceId))
+	dest.SetSpanID(uInt32ToSpanID(uint32(span.GetSpanId())))
+
+	// parent spanid = -1, means(root span) no parent span in skywalking,so just make otlp's parent span id empty.
+	if span.ParentSpanId != -1 {
+		dest.SetParentSpanID(uInt32ToSpanID(uint32(span.GetParentSpanId())))
+	}
+
+	dest.SetName(span.OperationName)
+	dest.SetStartTimestamp(microsecondsToTimestamp(span.GetStartTime()))
+	dest.SetEndTimestamp(microsecondsToTimestamp(span.GetEndTime()))
+
+	attrs := dest.Attributes()
+	attrs.EnsureCapacity(len(span.Tags))
+	swKvPairsToInternalAttributes(span.Tags, attrs)
+	// drop the attributes slice if all of them were replaced during translation
+	if attrs.Len() == 0 {
+		attrs.Clear()
+	}
+
+	setInternalSpanStatus(span, dest.Status())
+
+	switch {
+	case span.SpanLayer == agentV3.SpanLayer_MQ:
+		if span.SpanType == agentV3.SpanType_Entry {
+			dest.SetKind(pdata.SpanKindConsumer)
+		} else if span.SpanType == agentV3.SpanType_Exit {
+			dest.SetKind(pdata.SpanKindProducer)
+		}
+	case span.GetSpanType() == agentV3.SpanType_Exit:
+		dest.SetKind(pdata.SpanKindClient)
+	case span.GetSpanType() == agentV3.SpanType_Entry:
+		dest.SetKind(pdata.SpanKindServer)
+	case span.GetSpanType() == agentV3.SpanType_Local:
+		dest.SetKind(pdata.SpanKindInternal)
+	default:
+		dest.SetKind(pdata.SpanKindUnspecified)
+	}
+
+	swLogsToSpanEvents(span.GetLogs(), dest.Events())
+	// swkyalking: In the across thread and across process, these references targeting the parent segments.
+	swReferencesToSpanLinks(span.Refs, int64(span.ParentSpanId), dest.Links())
+}
+
+func swReferencesToSpanLinks(refs []*agentV3.SegmentReference, excludeParentID int64, dest pdata.SpanLinkSlice) {
+	if len(refs) == 0 {
+		return
+	}
+
+	dest.EnsureCapacity(len(refs))
+
+	for _, ref := range refs {
+		parentTraceSegmentID := ""
+		for _, part := range ref.ParentTraceSegmentId {
+			parentTraceSegmentID += fmt.Sprintf("%d", part)
+		}
+		link := dest.AppendEmpty()
+		link.SetTraceID(stringToTraceID(ref.TraceId))
+		link.SetSpanID(stringToParentSpanId(ref.ParentTraceSegmentId))
+	}
+}
+
+func setInternalSpanStatus(span *agentV3.SpanObject, dest pdata.SpanStatus) {
+	statusCode := pdata.StatusCodeUnset
+	statusMessage := ""
+
+	if span.GetIsError() {
+		statusCode = pdata.StatusCodeError
+		statusMessage = "ERROR"
+	} else {
+		statusCode = pdata.StatusCodeOk
+		statusMessage = "SUCCESS"
+	}
+
+	dest.SetCode(statusCode)
+	dest.SetMessage(statusMessage)
+}
+
+func swLogsToSpanEvents(logs []*agentV3.Log, dest pdata.SpanEventSlice) {
+	if len(logs) == 0 {
+		return
+	}
+	dest.EnsureCapacity(len(logs))
+
+	for i, log := range logs {
+		var event pdata.SpanEvent
+		if dest.Len() > i {
+			event = dest.At(i)
+		} else {
+			event = dest.AppendEmpty()
+		}
+
+		event.SetName("logs")
+		event.SetTimestamp(microsecondsToTimestamp(log.GetTime()))
+		if len(log.GetData()) == 0 {
+			continue
+		}
+
+		attrs := event.Attributes()
+		attrs.Clear()
+		attrs.EnsureCapacity(len(log.GetData()))
+		swKvPairsToInternalAttributes(log.GetData(), attrs)
+	}
+}
+
+func swKvPairsToInternalAttributes(pairs []*common.KeyStringValuePair, dest pdata.AttributeMap) {
+	if pairs == nil {
+		return
+	}
+
+	for _, pair := range pairs {
+		dest.UpsertString(pair.Key, pair.Value)
+	}
+}
+
+// microsecondsToTimestamp converts epoch microseconds to pdata.Timestamp
+func microsecondsToTimestamp(ms int64) pdata.Timestamp {
+	return pdata.NewTimestampFromTime(time.UnixMilli(ms))
+}
+
+func stringToTraceID(traceId string) pdata.TraceID {
+	return pdata.NewTraceID(stringToBytes(traceId))
+}
+
+func stringToParentSpanId(traceId string) pdata.SpanID {
+	traceID := stringTo8Bytes(traceId)
+	return pdata.NewSpanID(traceID)
+}
+
+// uInt32ToSpanID converts the uint64 representation of a SpanID to pdata.SpanID.
+func uInt32ToSpanID(id uint32) pdata.SpanID {
+	spanID := [8]byte{}
+	binary.BigEndian.PutUint32(spanID[:], id)
+	return pdata.NewSpanID(spanID)
+}
+
+func stringToBytes(s string) [16]byte {
+	sh := (*reflect.StringHeader)(unsafe.Pointer(&s))
+	bh := reflect.SliceHeader{
+		Data: sh.Data,
+		Len:  sh.Len,
+		Cap:  sh.Len,
+	}
+	return *(*[16]byte)(unsafe.Pointer(&bh))
+}
+
+func stringTo8Bytes(s string) [8]byte {
+	sh := (*reflect.StringHeader)(unsafe.Pointer(&s))
+	bh := reflect.SliceHeader{
+		Data: sh.Data,
+		Len:  sh.Len,
+		Cap:  sh.Len,
+	}
+	return *(*[8]byte)(unsafe.Pointer(&bh))
+}

receiver/skywalkingreceiver/skywalkingproto_to_traces_test.go

@@ -0,0 +1,30 @@
+// Copyright The 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 skywalkingreceiver
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSwProtoToTraces(t *testing.T) {
+	swSpan := mockGrpcTraceSegment(1)
+	td := SkywalkingToOtlpTraces(swSpan)
+
+	assert.Equal(t, 1, td.ResourceSpans().Len())
+	assert.Equal(t, 2, td.ResourceSpans().At(0).InstrumentationLibrarySpans().At(0).Spans().Len())
+
+}

receiver/skywalkingreceiver/sw_dummy_clr_service.go

@@ -0,0 +1,30 @@
+// Copyright  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 skywalkingreceiver
+
+import (
+	"context"
+
+	common "skywalking.apache.org/repo/goapi/collect/common/v3"
+	agent "skywalking.apache.org/repo/goapi/collect/language/agent/v3"
+)
+
+type clrService struct {
+	agent.UnimplementedCLRMetricReportServiceServer
+}
+
+func (c *clrService) Collect(ctx context.Context, req *agent.CLRMetricCollection) (*common.Commands, error) {
+	return &common.Commands{}, nil
+}