add more UT for trace receiver.

Signed-off-by: jian.tan <[email protected]>
open-telemetry · jpkrohling · Mar 2, 2022 · Feb 11, 2022 · Feb 13, 2022 · Feb 16, 2022
commit b66a00caf335c4bdcad61b6baedb5be48363dfcc
@@ -23,7 +23,6 @@ import (
 	"net"
 	"net/http"
 	"sync"
-	"time"
 
 	"github.com/gorilla/mux"
 	"go.opentelemetry.io/collector/component"
@@ -34,7 +33,7 @@ import (
 	"go.opentelemetry.io/collector/obsreport"
 	"go.uber.org/multierr"
 	"google.golang.org/grpc"
-	v1 "skywalking.apache.org/repo/goapi/satellite/data/v1"
+	v3 "skywalking.apache.org/repo/goapi/collect/language/agent/v3"
 )
 
 // configuration defines the behavior and the ports that
@@ -63,6 +62,7 @@ type swReceiver struct {
 
 	grpcObsrecv *obsreport.Receiver
 	httpObsrecv *obsreport.Receiver
+	service     *traceSegmentReportService
 }
 
 const (
@@ -175,7 +175,8 @@ func (sr *swReceiver) startCollector(host component.Host) error {
 			return fmt.Errorf("failed to bind to gRPC address %q: %v", gaddr, gerr)
 		}
 
-		//TODO: RegisterCollectorServiceServer(sr.grpc, sr)
+		sr.service = &traceSegmentReportService{}
+		v3.RegisterTraceSegmentReportServiceServer(sr.grpc, sr.service)
 
 		sr.goroutines.Add(1)
 		go func() {
@@ -196,27 +197,14 @@ type Response struct {
 
 func (sr *swReceiver) httpHandler(rsp http.ResponseWriter, r *http.Request) {
 	rsp.Header().Set("Content-Type", "application/json")
-	b, err := ioutil.ReadAll(r.Body)
+	_, err := ioutil.ReadAll(r.Body)
 	if err != nil {
 		response := &Response{Status: failing, Msg: err.Error()}
 		ResponseWithJSON(rsp, response, http.StatusBadRequest)
 		return
 	}
 
-	e := &v1.SniffData{
-		Name:      httpEventName,
-		Timestamp: time.Now().UnixNano() / 1e6,
-		Meta:      nil,
-		Type:      v1.SniffType_TracingType,
-		Remote:    true,
-		Data: &v1.SniffData_Segment{
-			Segment: b,
-		},
-	}
-
 	//TODO: convert to otel trace.
-	sr.settings.Logger.Debug(e.String())
-
 }
 
 func ResponseWithJSON(rsp http.ResponseWriter, response *Response, code int) {

@@ -0,0 +1,156 @@
+// 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"
+	"fmt"
+	"strconv"
+	"testing"
+	"time"
+
+	"go.opentelemetry.io/collector/config/confighttp"
+
+	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/collector/consumer/consumertest"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	common "skywalking.apache.org/repo/goapi/collect/common/v3"
+
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/config"
+	agent "skywalking.apache.org/repo/goapi/collect/language/agent/v3"
+)
+
+var (
+	skywalkingReceiver = config.NewComponentIDWithName("skywalking", "receiver_test")
+)
+
+func TestTraceSource(t *testing.T) {
+	set := componenttest.NewNopReceiverCreateSettings()
+	jr := newSkywalkingReceiver(skywalkingReceiver, &configuration{}, nil, set)
+	require.NotNil(t, jr)
+}
+
+func TestStartAndShutdown(t *testing.T) {
+	port := 12800
+	config := &configuration{
+		CollectorHTTPPort: port,
+		CollectorHTTPSettings: confighttp.HTTPServerSettings{
+			Endpoint: fmt.Sprintf(":%d", port),
+		},
+	}
+	sink := new(consumertest.TracesSink)
+
+	set := componenttest.NewNopReceiverCreateSettings()
+	sr := newSkywalkingReceiver(skywalkingReceiver, config, sink, set)
+
+	require.NoError(t, sr.Start(context.Background(), componenttest.NewNopHost()))
+	t.Cleanup(func() { require.NoError(t, sr.Shutdown(context.Background())) })
+
+}
+
+func TestGRPCReception(t *testing.T) {
+	config := &configuration{
+		CollectorGRPCPort: 11800, // that's the only one used by this test
+	}
+
+	sink := new(consumertest.TracesSink)
+
+	set := componenttest.NewNopReceiverCreateSettings()
+
+	swReceiver := newSkywalkingReceiver(skywalkingReceiver, config, sink, set)
+
+	require.NoError(t, swReceiver.Start(context.Background(), componenttest.NewNopHost()))
+
+	t.Cleanup(func() { require.NoError(t, swReceiver.Shutdown(context.Background())) })
+
+	conn, err := grpc.Dial(fmt.Sprintf("0.0.0.0:%d", config.CollectorGRPCPort), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	require.NoError(t, err)
+	defer conn.Close()
+
+	data := mockGrpcTraceSegment(1)
+	segmentCollection := &agent.SegmentCollection{
+		Segments: []*agent.SegmentObject{
+			data,
+		},
+	}
+
+	//skywalking agent client send trace data to otel/skywalkingreceiver
+	client := agent.NewTraceSegmentReportServiceClient(conn)
+	commands, err := client.CollectInSync(context.Background(), segmentCollection)
+	if err != nil {
+		t.Fatalf("cannot send data in sync mode: %v", err)
+	}
+	// verify
+	assert.NoError(t, err, "send skywalking segment successful.")
+	assert.NotNil(t, commands)
+}
+
+func mockGrpcTraceSegment(sequence int) *agent.SegmentObject {
+	seq := strconv.Itoa(sequence)
+	return &agent.SegmentObject{
+		TraceId:         "trace" + seq,
+		TraceSegmentId:  "trace-segment" + seq,
+		Service:         "demo-service" + seq,
+		ServiceInstance: "demo-instance" + seq,
+		IsSizeLimited:   false,
+		Spans: []*agent.SpanObject{
+			{
+				SpanId:        1,
+				ParentSpanId:  0,
+				StartTime:     time.Now().Unix(),
+				EndTime:       time.Now().Unix() + 10,
+				OperationName: "operation" + seq,
+				Peer:          "127.0.0.1:6666",
+				SpanType:      agent.SpanType_Entry,
+				SpanLayer:     agent.SpanLayer_Http,
+				ComponentId:   1,
+				IsError:       false,
+				SkipAnalysis:  false,
+				Tags: []*common.KeyStringValuePair{
+					{
+						Key:   "mock-key" + seq,
+						Value: "mock-value" + seq,
+					},
+				},
+				Logs: []*agent.Log{
+					{
+						Time: time.Now().Unix(),
+						Data: []*common.KeyStringValuePair{
+							{
+								Key:   "log-key" + seq,
+								Value: "log-value" + seq,
+							},
+						},
+					},
+				},
+				Refs: []*agent.SegmentReference{
+					{
+						RefType:                  agent.RefType_CrossThread,
+						TraceId:                  "trace" + seq,
+						ParentTraceSegmentId:     "parent-trace-segment" + seq,
+						ParentSpanId:             0,
+						ParentService:            "parent" + seq,
+						ParentServiceInstance:    "parent" + seq,
+						ParentEndpoint:           "parent" + seq,
+						NetworkAddressUsedAtPeer: "127.0.0.1:6666",
+					},
+				},
+			},
+		},
+	}
+}
@@ -0,0 +1,56 @@
+// 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 "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/skywalkingreceiver"
+
+import (
+	"context"
+	"fmt"
+	"io"
+
+	"google.golang.org/protobuf/proto"
+
+	common "skywalking.apache.org/repo/goapi/collect/common/v3"
+	agent "skywalking.apache.org/repo/goapi/collect/language/agent/v3"
+)
+
+type traceSegmentReportService struct {
+	agent.UnimplementedTraceSegmentReportServiceServer
+}
+
+func (s *traceSegmentReportService) Collect(stream agent.TraceSegmentReportService_CollectServer) error {
+	for {
+		var recData []byte
+		err := stream.RecvMsg(recData)
+		if err == io.EOF {
+			return stream.SendAndClose(&common.Commands{})
+		}
+		if err != nil {
+			return err
+		}
+		//TODO: convert to otel trace.
+	}
+}
+
+func (s *traceSegmentReportService) CollectInSync(ctx context.Context, segments *agent.SegmentCollection) (*common.Commands, error) {
+	for _, segment := range segments.Segments {
+		marshaledSegment, err := proto.Marshal(segment)
+		if err != nil {
+			fmt.Printf("cannot marshal segemnt from sync, %v", err)
+		}
+		//TODO: convert to otel trace.
+		fmt.Printf("receivec data:%s", marshaledSegment)
+	}
+	return &common.Commands{}, nil
+}