feat: add skywalking trace

cmd/layotto/main.go

@@ -32,6 +32,7 @@ import (
 	"mosn.io/layotto/diagnostics"
 	"mosn.io/layotto/pkg/grpc/default_api"
 	secretstores_loader "mosn.io/layotto/pkg/runtime/secretstores"
+	"mosn.io/mosn/pkg/trace/skywalking"
 	"os"
 	"strconv"
 	"time"
@@ -162,6 +163,8 @@ import (
 	_ "mosn.io/pkg/buffer"
 
 	_ "mosn.io/layotto/diagnostics/exporter_iml"
+	lprotocol "mosn.io/layotto/diagnostics/protocol"
+	lsky "mosn.io/layotto/diagnostics/skywalking"
 )
 
 // loggerForDaprComp is constructed for reusing dapr's components.
@@ -498,7 +501,8 @@ func ExtensionsRegister(c *cli.Context) {
 	// 4. register tracer
 	xtrace.RegisterDelegate(bolt.ProtocolName, tracebolt.Boltv1Delegate)
 	trace.RegisterTracerBuilder("SOFATracer", protocol.HTTP1, tracehttp.NewTracer)
-	trace.RegisterTracerBuilder("SOFATracer", "layotto", diagnostics.NewTracer)
+	trace.RegisterTracerBuilder("SOFATracer", lprotocol.Layotto, diagnostics.NewTracer)
+	trace.RegisterTracerBuilder(skywalking.SkyDriverName, lprotocol.Layotto, lsky.NewGrpcSkyTracer)
 
 }
 

cmd/layotto_multiple_api/main.go

@@ -24,6 +24,7 @@ import (
 	l8_grpc "mosn.io/layotto/pkg/grpc"
 	"mosn.io/layotto/pkg/grpc/dapr"
 	"mosn.io/layotto/pkg/grpc/default_api"
+	"mosn.io/mosn/pkg/trace/skywalking"
 	"os"
 	"strconv"
 	"time"
@@ -152,6 +153,8 @@ import (
 	_ "mosn.io/pkg/buffer"
 
 	_ "mosn.io/layotto/diagnostics/exporter_iml"
+	lprotocol "mosn.io/layotto/diagnostics/protocol"
+	lsky "mosn.io/layotto/diagnostics/skywalking"
 )
 
 // loggerForDaprComp is constructed for reusing dapr's components.
@@ -464,6 +467,7 @@ func ExtensionsRegister(c *cli.Context) {
 	xtrace.RegisterDelegate(bolt.ProtocolName, tracebolt.Boltv1Delegate)
 	trace.RegisterTracerBuilder("SOFATracer", protocol.HTTP1, tracehttp.NewTracer)
 	trace.RegisterTracerBuilder("SOFATracer", "layotto", diagnostics.NewTracer)
+	trace.RegisterTracerBuilder(skywalking.SkyDriverName, lprotocol.Layotto, lsky.NewGrpcSkyTracer)
 
 }
 

codecov.yml

@@ -27,4 +27,5 @@ ignore:
   - "cmd/**/*"
   - "components/lock/consul/consul_lock_task.go"
   - "components/rpc/invoker/mosn/channel/xchannel.go"
+  - "diagnostics/**/*"
   - "components/pkg/mock/**/*"

components/trace/span.go

@@ -51,6 +51,7 @@ func (span *Span) SpanId() string {
 func (span *Span) SetParentSpanId(id string) {
 	span.parentSpanId = id
 }
+
 func (span *Span) ParentSpanId() string {
 	return span.parentSpanId
 }

configs/config_trace_skywalking.json

@@ -0,0 +1,81 @@
+{
+  "servers": [
+    {
+      "default_log_path": "stdout",
+      "default_log_level": "INFO",
+      "listeners": [
+        {
+          "name": "grpc",
+          "address": "0.0.0.0:34904",
+          "bind_port": true,
+          "filter_chains": [
+            {
+              "filters": [
+                {
+                  "type": "grpc",
+                  "config": {
+                    "server_name": "runtime",
+                    "grpc_config": {
+                      "hellos": {
+                        "helloworld": {
+                          "hello": "greeting"
+                        }
+                      },
+                      "config_store": {
+                        "etcd": {
+                          "address": [
+                            "127.0.0.1:2379"
+                          ],
+                          "timeout": "10"
+                        }
+                      }
+                    }
+                  }
+                }
+              ]
+            }
+          ],
+          "stream_filters": [
+            {
+              "type": "flowControlFilter",
+              "config": {
+                "global_switch": true,
+                "limit_key_type": "PATH",
+                "rules": [
+                  {
+                    "resource": "/spec.proto.runtime.v1.Runtime/SayHello",
+                    "grade": 1,
+                    "threshold": 5
+                  }
+                ]
+              }
+            },
+            {
+              "type": "grpc_metric"
+            }
+          ]
+        }
+      ]
+    }
+  ],
+  "tracing": {
+    "enable": true,
+    "driver": "SkyWalking",
+    "config": {
+      "reporter": "gRPC",
+      "backend_service": "127.0.0.1:11800",
+      "service_name": "layotto"
+    }
+  },
+  "metrics": {
+    "sinks": [
+      {
+        "type": "prometheus",
+        "config": {
+          "port": 34903
+        }
+      }
+    ]
+  }
+}
+

diagnostics/grpc/grpc_request.go

@@ -0,0 +1,5 @@
+package grpc
+
+type RequestInfo struct {
+	FullMethod string
+}

diagnostics/grpc_tracing.go

@@ -8,6 +8,8 @@ import (
 
 	"google.golang.org/grpc"
 	ltrace "mosn.io/layotto/components/trace"
+	lgrpc "mosn.io/layotto/diagnostics/grpc"
+	"mosn.io/layotto/diagnostics/protocol"
 	"mosn.io/mosn/pkg/trace"
 )
 
@@ -18,14 +20,19 @@ func UnaryInterceptorFilter(ctx context.Context, req interface{}, info *grpc.Una
 		return resp, err
 	}
 	// get tracer
-	tracer := trace.Tracer("layotto")
+	tracer := trace.Tracer(protocol.Layotto)
 	// start a span
-	span := tracer.Start(ctx, req, time.Now())
+	span := tracer.Start(ctx, &lgrpc.RequestInfo{
+		FullMethod: info.FullMethod,
+	}, time.Now())
 	defer span.FinishSpan()
+
 	span.SetTag(ltrace.LAYOTTO_METHOD_NAME, info.FullMethod)
 	span.SetTag(ltrace.LAYOTTO_REQUEST_RESULT, "0")
+
 	// construct a new context which contains the span
 	ctx = GetNewContext(ctx, span)
+
 	// handle request
 	resp, err = handler(ctx, req)
 	if err != nil {
@@ -40,13 +47,14 @@ func StreamInterceptorFilter(srv interface{}, ss grpc.ServerStream, info *grpc.S
 		return err
 	}
 	// get tracer
-	tracer := trace.Tracer("layotto")
+	tracer := trace.Tracer(protocol.Layotto)
 	ctx := ss.Context()
 	// start a span
-	span := tracer.Start(ctx, nil, time.Now())
+	span := tracer.Start(ctx, &lgrpc.RequestInfo{
+		FullMethod: info.FullMethod,
+	}, time.Now())
 	defer span.FinishSpan()
-	span.SetTag(ltrace.LAYOTTO_METHOD_NAME, info.FullMethod)
-	span.SetTag(ltrace.LAYOTTO_REQUEST_RESULT, "0")
+
 	// construct a new context which contains the span
 	wrapped := grpc_middleware.WrapServerStream(ss)
 	ctx = GetNewContext(ctx, span)
@@ -56,5 +64,6 @@ func StreamInterceptorFilter(srv interface{}, ss grpc.ServerStream, info *grpc.S
 	if err != nil {
 		span.SetTag(ltrace.LAYOTTO_REQUEST_RESULT, "1")
 	}
+
 	return err
 }

diagnostics/protocol/protocol.go

@@ -0,0 +1,5 @@
+package protocol
+
+const (
+	Layotto = "layotto"
+)

diagnostics/skywalking/grpc_tracer.go

@@ -0,0 +1,109 @@
+/*
+ * Copyright 2021 Layotto 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 skywalking
+
+import (
+	"context"
+	"mosn.io/layotto/diagnostics/grpc"
+	"time"
+
+	"github.com/SkyAPM/go2sky"
+	language_agent "github.com/SkyAPM/go2sky/reporter/grpc/language-agent"
+	"mosn.io/api"
+	ltrace "mosn.io/layotto/components/trace"
+	"mosn.io/layotto/diagnostics/protocol"
+	"mosn.io/mosn/pkg/log"
+	"mosn.io/mosn/pkg/trace"
+	"mosn.io/mosn/pkg/trace/skywalking"
+	"mosn.io/mosn/pkg/types"
+)
+
+func init() {
+	trace.RegisterTracerBuilder(skywalking.SkyDriverName, protocol.Layotto, NewGrpcSkyTracer)
+}
+
+func NewGrpcSkyTracer(_ map[string]interface{}) (api.Tracer, error) {
+	return &grpcSkyTracer{}, nil
+}
+
+type grpcSkyTracer struct {
+	*go2sky.Tracer
+}
+
+func (tracer *grpcSkyTracer) SetGO2SkyTracer(t *go2sky.Tracer) {
+	tracer.Tracer = t
+}
+
+func (tracer *grpcSkyTracer) Start(ctx context.Context, request interface{}, _ time.Time) api.Span {
+	info, ok := request.(*grpc.RequestInfo)
+	if !ok {
+		log.DefaultLogger.Debugf("[SkyWalking] [tracer] [layotto] unable to get request header, downstream trace ignored")
+		return skywalking.NoopSpan
+	}
+
+	// create entry span (downstream)
+	entry, nCtx, err := tracer.CreateEntrySpan(ctx, info.FullMethod, func() (sw8 string, err error) {
+		return
+	})
+
+	if err != nil {
+		log.DefaultLogger.Errorf("[SkyWalking] [tracer] [http1] create entry span error, err: %v", err)
+		return skywalking.NoopSpan
+	}
+	entry.Tag(go2sky.TagHTTPMethod, "POST")
+	entry.Tag(go2sky.TagURL, info.FullMethod)
+	entry.SetComponent(skywalking.MOSNComponentID)
+	entry.SetSpanLayer(language_agent.SpanLayer_Http)
+
+	return &grpcSkySpan{
+		tracer: tracer,
+		ctx:    nCtx,
+		carrier: &skywalking.SpanCarrier{
+			EntrySpan: entry,
+		},
+		Span: &ltrace.Span{},
+	}
+}
+
+type grpcSkySpan struct {
+	*ltrace.Span
+	tracer  *grpcSkyTracer
+	ctx     context.Context
+	carrier *skywalking.SpanCarrier
+}
+
+func (h *grpcSkySpan) TraceId() string {
+	return go2sky.TraceID(h.ctx)
+}
+
+func (h *grpcSkySpan) InjectContext(requestHeaders types.HeaderMap, requestInfo api.RequestInfo) {
+}
+
+func (h *grpcSkySpan) SetRequestInfo(requestInfo api.RequestInfo) {
+}
+
+func (h *grpcSkySpan) FinishSpan() {
+	entry := h.carrier.EntrySpan
+	if h.Tag(ltrace.LAYOTTO_REQUEST_RESULT) == "1" {
+		entry.Error(time.Now(), skywalking.ErrorLog)
+		entry.Tag(go2sky.TagStatusCode, "500")
+	} else {
+		entry.Tag(go2sky.TagStatusCode, "200")
+	}
+
+	entry.End()
+}

diagnostics/skywalking/skywalking-docker-compose.yaml

@@ -0,0 +1,38 @@
+#Licensed to the Apache Software Foundation (ASF) under one or more
+#contributor license agreements.  See the NOTICE file distributed with
+#this work for additional information regarding copyright ownership.
+#The ASF licenses this file to You 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.
+
+version: '3.3'
+services:
+  skywalking-oap:
+    image: apache/skywalking-oap-server:8.0.1-es7
+    container_name: skywalking-oap
+    restart: always
+    ports:
+      - 11800:11800
+      - 12800:12800
+    environment:
+      SW_STORAGE: h2
+  skywalking-ui:
+    image: apache/skywalking-ui:8.0.1
+    container_name: skywalking-ui
+    depends_on:
+      - skywalking-oap
+    links:
+      - skywalking-oap
+    restart: always
+    ports:
+      - 8080:8080
+    environment:
+      SW_OAP_ADDRESS: skywalking-oap:12800

diagnostics/tracing.go

@@ -9,6 +9,7 @@ import (
 
 	"mosn.io/api"
 	ltrace "mosn.io/layotto/components/trace"
+	lgrpc "mosn.io/layotto/diagnostics/grpc"
 )
 
 const (
@@ -45,6 +46,11 @@ func getActiveExportersFromConfig(config map[string]interface{}) []string {
 
 func (tracer *grpcTracer) Start(ctx context.Context, request interface{}, startTime time.Time) api.Span {
 	span := NewSpan(ctx, startTime, tracer.config)
+
+	if reqInfo, ok := request.(*lgrpc.RequestInfo); ok {
+		span.SetTag(ltrace.LAYOTTO_METHOD_NAME, reqInfo.FullMethod)
+		span.SetTag(ltrace.LAYOTTO_REQUEST_RESULT, "0")
+	}
 	return span
 }