Merge pull request #20 from liuxp0827/master

add: 增加skyWalking中间件的支持

.env.example

@@ -1,8 +1,10 @@
 # toml配置文件
 # Wiki：https://github.com/toml-lang/toml
+ServiceName = "snow"
 Debug = true
 Env = "local" # local-本地 develop-开发 beta-预发布 production-线上
 PrometheusCollectEnable = true
+SkyWalkingOapServer = "127.0.0.1:11800"
 
 [Log]
 Handler = "file"

CHANGLOG.md

@@ -1,3 +1,9 @@
+## Version 1.2.3 (2020-08-20)
+
+### Changes
+- 引入skyWalking服务追踪的中间件
+- 二次封装 github.com/go-resty/resty/v2 支持skyWalking链路追踪,作为默认Http Client
+
 ## Version 1.2.2 (2020-07-02)
 
 ### Changes

README.md

@@ -69,6 +69,7 @@ curl "http://127.0.0.1:8080/hello"
 - xiongwilee ([@xiongwilee](https://github.com/xiongwilee))
 - KEL ([@deathkel](https://github.com/deathkel))
 - peterwu ([@peterwu](https://github.com/yukunwu))
+- liuxp0827 ([@liuxp0827](https://github.com/liuxp0827))
 
 
 

app/http/controllers/test.go

@@ -1,19 +1,29 @@
 package controllers
 
 import (
-	"github.com/gin-gonic/gin"
-	"github.com/qit-team/snow-core/log/logger"
+	"strconv"
+	"time"
+
 	"github.com/qit-team/snow/app/constants/errorcode"
 	"github.com/qit-team/snow/app/http/entities"
 	"github.com/qit-team/snow/app/http/formatters/bannerformatter"
 	"github.com/qit-team/snow/app/services/bannerservice"
-	"strconv"
-	"time"
+	"github.com/qit-team/snow/app/utils/httpclient"
+
+	"github.com/gin-gonic/gin"
+	"github.com/qit-team/snow-core/log/logger"
 )
 
 // hello示例
 func HandleHello(c *gin.Context) {
 	logger.Debug(c, "hello", "test message")
+	client := httpclient.NewClient(c.Request.Context())
+	resposne, err := client.R().Get("https://www.baidu.com")
+	if err != nil {
+		Error(c, errorcode.SystemError, err.Error())
+		return
+	}
+	logger.Info(c, "HandleHello", resposne.String())
 	Success(c, "hello world!")
 	return
 }

app/http/middlewares/tracer.go

@@ -0,0 +1,54 @@
+package middlewares
+
+import (
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/SkyAPM/go2sky"
+	"github.com/SkyAPM/go2sky/propagation"
+	v3 "github.com/SkyAPM/go2sky/reporter/grpc/language-agent"
+	"github.com/gin-gonic/gin"
+	"github.com/qit-team/snow-core/log/logger"
+	"github.com/qit-team/snow/app/http/trace"
+)
+
+const (
+	componentIDGOHttpServer = 5004
+)
+
+func Trace() gin.HandlerFunc {
+	return func(c *gin.Context) {
+		tracer, err := trace.Tracer()
+		if err != nil {
+			logger.Error(c, "Trace", err.Error())
+			c.Next()
+			return
+		}
+		r := c.Request
+		operationName := fmt.Sprintf("/%s%s", r.Method, r.URL.Path)
+		span, ctx, err := tracer.CreateEntrySpan(c, operationName, func() (string, error) {
+			// 从http头部捞取上一层的调用链信息, 当前使用v3版本的协议
+			// https://github.com/apache/skywalking/blob/master/docs/en/protocols/Skywalking-Cross-Process-Propagation-Headers-Protocol-v3.md
+			return r.Header.Get(propagation.Header), nil
+		})
+		if err != nil {
+			logger.Error(c, "Trace", err.Error())
+			c.Next()
+			return
+		}
+		span.SetComponent(componentIDGOHttpServer)
+		// 可以自定义tag
+		span.Tag(go2sky.TagHTTPMethod, r.Method)
+		span.Tag(go2sky.TagURL, fmt.Sprintf("%s%s", r.Host, r.URL.Path))
+		span.SetSpanLayer(v3.SpanLayer_Http)
+		c.Request = c.Request.WithContext(ctx)
+		c.Next()
+		code := c.Writer.Status()
+		if code >= 400 {
+			span.Error(time.Now(), fmt.Sprintf("Error on handling request, statusCode: %d", code))
+		}
+		span.Tag(go2sky.TagStatusCode, strconv.Itoa(code))
+		span.End()
+	}
+}

app/http/routes/route.go

@@ -6,11 +6,13 @@ package routes
 import (
 	"github.com/qit-team/snow/app/http/controllers"
 	"github.com/qit-team/snow/app/http/middlewares"
+	"github.com/qit-team/snow/app/http/trace"
 	"github.com/qit-team/snow/app/utils/metric"
 	"github.com/qit-team/snow/config"
 
 	"github.com/gin-gonic/gin"
 	"github.com/qit-team/snow-core/http/middleware"
+	"github.com/qit-team/snow-core/log/logger"
 	"github.com/swaggo/gin-swagger"
 	"github.com/swaggo/gin-swagger/swaggerFiles"
 )
@@ -29,6 +31,15 @@ func RegisterRoute(router *gin.Engine) {
 		})
 	}
 
+	if len(config.GetConf().SkyWalkingOapServer) > 0 && config.IsEnvEqual(config.ProdEnv) {
+		err := trace.InitTracer(config.GetConf().ServiceName, config.GetConf().SkyWalkingOapServer)
+		if err != nil {
+			logger.Error(nil, "InitTracer", err.Error())
+		} else {
+			router.Use(middlewares.Trace())
+		}
+	}
+
 	router.NoRoute(controllers.Error404)
 	router.GET("/hello", controllers.HandleHello)
 	router.POST("/test", controllers.HandleTest)

app/http/trace/trace.go

@@ -0,0 +1,41 @@
+package trace
+
+import (
+	"sync"
+
+	"github.com/SkyAPM/go2sky"
+	"github.com/SkyAPM/go2sky/reporter"
+	"github.com/qit-team/snow/config"
+)
+
+var (
+	tracer *go2sky.Tracer
+	lock   sync.Mutex
+)
+
+func Tracer() (*go2sky.Tracer, error) {
+	if tracer == nil {
+		// 有err, 不适合用sync.Once做单例
+		lock.Lock()
+		defer lock.Unlock()
+		if tracer == nil {
+			err := InitTracer(config.GetConf().ServiceName, config.GetConf().SkyWalkingOapServer)
+			if err != nil {
+				return nil, err
+			}
+		}
+	}
+	return tracer, nil
+}
+
+func InitTracer(serviceName, skyWalkingOapServer string) error {
+	report, err := reporter.NewGRPCReporter(skyWalkingOapServer)
+	if err != nil {
+		return err
+	}
+	tracer, err = go2sky.NewTracer(serviceName, go2sky.WithReporter(report))
+	if err != nil {
+		return err
+	}
+	return nil
+}

app/utils/httpclient/httpclient.go

@@ -0,0 +1,189 @@
+package httpclient
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"strconv"
+	"time"
+
+	"github.com/SkyAPM/go2sky"
+	"github.com/SkyAPM/go2sky/propagation"
+	v3 "github.com/SkyAPM/go2sky/reporter/grpc/language-agent"
+	"github.com/go-resty/resty/v2"
+	"github.com/qit-team/snow-core/log/logger"
+	"github.com/qit-team/snow/app/http/trace"
+	"github.com/qit-team/snow/config"
+)
+
+const (
+	RetryCounts   = 2
+	RetryInterval = 3 * time.Second
+)
+
+const componentIDGOHttpClient = 5005
+
+type ClientConfig struct {
+	ctx       context.Context
+	client    *resty.Client
+	tracer    *go2sky.Tracer
+	extraTags map[string]string
+}
+
+type ClientOption func(*ClientConfig)
+
+func WithClientTag(key string, value string) ClientOption {
+	return func(c *ClientConfig) {
+		if c.extraTags == nil {
+			c.extraTags = make(map[string]string)
+		}
+		c.extraTags[key] = value
+	}
+}
+
+func WithClient(client *resty.Client) ClientOption {
+	return func(c *ClientConfig) {
+		c.client = client
+	}
+}
+
+func WithContext(ctx context.Context) ClientOption {
+	return func(c *ClientConfig) {
+		c.ctx = ctx
+	}
+}
+
+type transport struct {
+	*ClientConfig
+	delegated http.RoundTripper
+}
+
+func (t *transport) RoundTrip(req *http.Request) (resp *http.Response, err error) {
+	span, err := t.tracer.CreateExitSpan(t.ctx, fmt.Sprintf("/%s%s", req.Method, req.URL.Path), req.Host, func(header string) error {
+		// 将本层的调用链信息写入http头部, 传入到下一层调用, 当前使用v3版本的协议
+		// https://github.com/apache/skywalking/blob/master/docs/en/protocols/Skywalking-Cross-Process-Propagation-Headers-Protocol-v3.md
+		req.Header.Set(propagation.Header, header)
+		return nil
+	})
+	if err != nil {
+		return t.delegated.RoundTrip(req)
+	}
+	defer span.End()
+	span.SetComponent(componentIDGOHttpClient)
+	for k, v := range t.extraTags {
+		span.Tag(go2sky.Tag(k), v)
+	}
+	span.Tag(go2sky.TagHTTPMethod, req.Method)
+	span.Tag(go2sky.TagURL, req.URL.String())
+	span.SetSpanLayer(v3.SpanLayer_Http)
+	resp, err = t.delegated.RoundTrip(req)
+	if err != nil {
+		span.Error(time.Now(), err.Error())
+		return
+	}
+	span.Tag(go2sky.TagStatusCode, strconv.Itoa(resp.StatusCode))
+	if resp.StatusCode >= http.StatusBadRequest {
+		span.Error(time.Now(), "Errors on handling client")
+	}
+	return resp, nil
+}
+
+func NewClient(ctx context.Context, options ...ClientOption) (client *resty.Client) {
+	client = resty.New()
+	if config.IsDebug() {
+		client.SetDebug(true).EnableTrace()
+	}
+
+	var (
+		tracer *go2sky.Tracer
+		err    error
+	)
+	if len(config.GetConf().SkyWalkingOapServer) > 0 && config.IsEnvEqual(config.ProdEnv) {
+		tracer, err = trace.Tracer()
+		if err != nil {
+			logger.Error(ctx, "NewClient:Tracer", err.Error())
+		}
+	}
+	if tracer != nil {
+		co := &ClientConfig{ctx: ctx, tracer: tracer}
+		for _, option := range options {
+			option(co)
+		}
+		if co.client == nil {
+			co.client = client
+		}
+		tp := &transport{
+			ClientConfig: co,
+			delegated:    http.DefaultTransport,
+		}
+		if co.client.GetClient().Transport != nil {
+			tp.delegated = co.client.GetClient().Transport
+		}
+		co.client.SetTransport(tp)
+	}
+
+	client.OnBeforeRequest(func(ct *resty.Client, req *resty.Request) error {
+		//req.SetContext(c)
+		logger.Info(ctx, "OnBeforeRequest", logger.NewWithField("url", req.URL))
+		return nil // if its success otherwise return error
+	})
+	// Registering Response Middleware
+	client.OnAfterResponse(func(ct *resty.Client, resp *resty.Response) error {
+		logger.Info(ctx, "OnAfterResponse", logger.NewWithField("url", resp.Request.URL), logger.NewWithField("request", resp.Request.RawRequest), logger.NewWithField("response", resp.String()))
+		return nil
+	})
+	return client
+}
+
+func NewClientWithRetry(ctx context.Context, retryCounts int, retryInterval time.Duration, options ...ClientOption) (client *resty.Client) {
+	client = resty.New()
+	if config.IsDebug() {
+		client.SetDebug(true).EnableTrace()
+	}
+	if retryCounts == 0 {
+		retryCounts = RetryCounts
+	}
+	if retryInterval.Seconds() == 0.0 {
+		retryInterval = RetryInterval
+	}
+	client.SetRetryCount(retryCounts).SetRetryMaxWaitTime(retryInterval)
+
+	var (
+		tracer *go2sky.Tracer
+		err    error
+	)
+	if len(config.GetConf().SkyWalkingOapServer) > 0 && config.IsEnvEqual(config.ProdEnv) {
+		tracer, err = trace.Tracer()
+		if err != nil {
+			logger.Error(ctx, "NewClient:Tracer", err.Error())
+		}
+	}
+	if tracer != nil {
+		co := &ClientConfig{ctx: ctx, tracer: tracer}
+		for _, option := range options {
+			option(co)
+		}
+		if co.client == nil {
+			co.client = client
+		}
+		tp := &transport{
+			ClientConfig: co,
+			delegated:    http.DefaultTransport,
+		}
+		if co.client.GetClient().Transport != nil {
+			tp.delegated = co.client.GetClient().Transport
+		}
+		co.client.SetTransport(tp)
+	}
+
+	client.OnBeforeRequest(func(ct *resty.Client, req *resty.Request) error {
+		logger.Info(ctx, "OnBeforeRequest", logger.NewWithField("url", req.URL))
+		return nil // if its success otherwise return error
+	})
+	// Registering Response Middleware
+	client.OnAfterResponse(func(ct *resty.Client, resp *resty.Response) error {
+		logger.Info(ctx, "OnAfterResponse", logger.NewWithField("url", resp.Request.URL), logger.NewWithField("request", resp.Request.RawRequest), logger.NewWithField("response", resp.String()))
+		return nil
+	})
+	return client
+}

config/config.go

@@ -17,9 +17,11 @@ var srvConf *Config
 
 // ------------------------配置文件解析
 type Config struct {
+	ServiceName             string             `toml:"ServiceName"`
 	Env                     string             `toml:"Env"`
 	Debug                   bool               `toml:"Debug"`
 	PrometheusCollectEnable bool               `toml:"PrometheusCollectEnable"`
+	SkyWalkingOapServer     string             `toml:"SkyWalkingOapServer"`
 	Log                     config.LogConfig   `toml:"Log"`
 	Redis                   config.RedisConfig `toml:"Redis"`
 	Mns                     config.MnsConfig   `toml:"AliMns"`