opentracting+jager分布式鏈路追蹤探索實踐

一、Opentracing

opentracing通過提供平台無關、廠商無關的API，使得開發人員可以方便地實現追蹤系統。opentracing提供了用於運營支撐系統和針對特定平台的輔助程序庫，被跟蹤的服務只需要調用這套接口，就可以被任何實現這套接口的跟蹤后台（比如Zipkin, Jaeger等等）支持，而作為一個跟蹤后台，只要實現了個這套接口，就可以跟蹤到任何調用這套接口的服務。

 

二、Jaeger

Jaeger是Uber開源基於golang的分布式跟蹤系統，使用Jaeger可以非常直觀的展示整個分布式系統的調用鏈，由此可以很好發現和解決問題。

Jaeger的整體架構如下：

 

Jaeger組件

jaeger-agent

jaeger-agent是一個網絡守護進程，監聽通過UDP發送過來的Span，它會將其批量發送給collector。按照設計，Agent要被部署到所有主機上，作為基礎設施。Agent將collector和客戶端之間的路由與發現機制抽象出來。

jaeger-collector

jaeger-collector從Jaeger Agent接收Trace，並通過一個處理管道對其進行處理。目前的管道會校驗Trace、建立索引、執行轉換並最終進行存儲。存儲是一個可插入的組件，目前支持Cassandra和elasticsearch。

jaeger-query

jaeger-query服務會從存儲中檢索Trace並通過UI界面進行展現，通過UI界面可以展現Trace的詳細信息。

 

三、 安裝使用

1.下載安裝jaeger

https://www.jaegertracing.io/download/

下載並解壓

增加執行權限：

chmod a+x jaeger-*

 

2. 安裝golang package

go get github.com/opentracing/opentracing-go

go get github.com/uber/jaeger-client-go

 

3. 指定用ES存儲

export SPAN_STORAGE_TYPE=elasticsearch

export ES_SERVER_URLS=http://10.20.xx.xx:9200

 

4. 運行jarger

 cd jaeger-1.18.0-linux-amd64

./jaeger-all-in-one

在瀏覽器輸入http://10.20.xx.xx:16686/即可打開jaeger頁面：

 

使用docker運行jaeger的各組件：

docker-compose.yaml

version: '3'
services:
  jaeger-agent:
    image: jaegertracing/jaeger-agent:1.18
    stdin_open: true
    tty: true
    links:
    - jaeger-collector:jaeger-collector
    ports:
    - 6831:6831/udp
    command:
    - --reporter.grpc.host-port=jaeger-collector:14250
  jaeger-collector:
    image: jaegertracing/jaeger-collector:1.18
    environment:
      SPAN_STORAGE_TYPE: elasticsearch
      ES_SERVER_URLS: http://10.20.xx.xx:9200
    stdin_open: true
    tty: true
  jaeger-query:
    image: jaegertracing/jaeger-query:1.18
    environment:
      SPAN_STORAGE_TYPE: elasticsearch
      ES_SERVER_URLS: http://10.20.xx.xx:9200
    stdin_open: true
    tty: true
    ports:
    - 16686:16686/tcp

 

四、 在gin中使用jaeger

以wire依賴注入的方式引入jaeger：

package jaeger

import (
	"fmt"
	"github.com/google/wire"
	"github.com/opentracing/opentracing-go"
	"github.com/pkg/errors"
	"github.com/spf13/viper"
	"github.com/uber/jaeger-client-go"
	"github.com/uber/jaeger-client-go/config"
	"go.uber.org/zap"
	"io"
)

// ClientType 定義jaeger client 結構體
type ClientType struct {
	Tracer opentracing.Tracer
	Closer io.Closer
}

// Client  jaeger連接類型
var Client ClientType

// Options jaeger option
type Options struct {
	Type               string  // const
	Param              float64 // 1
	LogSpans           bool    // true
	LocalAgentHostPort string  // host:port
	Service            string  // service name
}

// NewOptions for jaeger
func NewOptions(v *viper.Viper, logger *zap.Logger) (*Options, error) {
	var (
		err error
		o   = new(Options)
	)
	if err = v.UnmarshalKey("jaeger", o); err != nil {
		return nil, errors.Wrap(err, "unmarshal redis option error")
	}

	logger.Info("load jaeger options success", zap.Any("jaeger options", o))
	return o, err
}

// New returns an instance of Jaeger Tracer that samples 100% of traces and logs all spans to stdout.
func New(o *Options) (opentracing.Tracer, error) {
	cfg := &config.Configuration{
		Sampler: &config.SamplerConfig{
			Type:  o.Type,
			Param: o.Param,
		},
		Reporter: &config.ReporterConfig{
			LogSpans: o.LogSpans,
			// 注意：填下地址不能加http://
			LocalAgentHostPort: o.LocalAgentHostPort,
		},
	}
	tracer, closer, err := cfg.New(o.service, config.Logger(jaeger.StdLogger))
	if err != nil {
		panic(fmt.Sprintf("ERROR: cannot init Jaeger: %v\n", err))
	}
	Client.Tracer = tracer
	Client.Closer = closer

	return tracer, err
}

// ProviderSet inject jaeger settings
var ProviderSet = wire.NewSet(New, NewOptions) 

初始化全局jaeger客戶端，調用jaeger

// Article 發布文章
func (pc *ArticleController) Article(c *gin.Context) {
	var req requests.Article
	if err := c.ShouldBind(&req); err != nil {
		pc.logger.Error("參數錯誤", zap.Error(err))
		c.JSON(http.StatusBadRequest, httputil.Error(nil, "參數校驗失敗"))
		return
	}
	tracer := jaeger.Client.Tracer
	opentracing.SetGlobalTracer(tracer)
	span := tracer.StartSpan("Article")
	defer span.Finish()
	ctx := context.Background()
	ctx = opentracing.ContextWithSpan(ctx, span)
	span.SetTag("http.url", c.Request.URL.Path)
	article, err := pc.service.Article(ctx, &req)
	if err != nil {
		pc.logger.Error("發表文章失敗", zap.Error(err))
		c.JSON(http.StatusInternalServerError, httputil.Error(nil, "發表主題失敗"))
		return
	}
	span.LogFields(
		log.String("event", "info"),
		log.Int("article", article.ID),
	)
	c.JSON(http.StatusOK, gin.H{"code": 0, "msg": "success", "data": article})
}

通過span和context傳遞tracer

// Article 發表文章
func (s *DefaultArticleService) Article(ctx context.Context, req *requests.Article) (responses.Article, error) {
	var result responses.Article
	cateInfo := s.Repository.GetCategory(req.CategoryID)
	if cateInfo.ID == 0 {
		s.logger.Error("categoryId 不存在", zap.Any("category_id", req.CategoryID))
		return result, gorm.ErrRecordNotFound
	}
	span, _ := opentracing.StartSpanFromContext(ctx, "Article")
	defer span.Finish()
	span.LogFields(
		log.String("event", "insert article service"),
	)

	res, err := s.Repository.Article(req)

	if err != nil {
		s.logger.Error("發表文章失敗", zap.Error(err))
		return result, errors.New("發表文章失敗")
	}

	result = responses.Article{
		ID:    res.ID,
		Title: res.Title,
	}
	return result, err
}

jaeger調用詳情：

 

五、 在go micro中使用jaeger

go micro中各rpc服務之間的調用定位異常較為困難，可以在網關處初始化jaeger，通過傳遞context的方式記錄追蹤調用的service

具體代碼實現如下：

func initJaeger(service string) (opentracing.Tracer, io.Closer) {
	cfg := &jaegercfg.Configuration{
		Sampler: &jaegercfg.SamplerConfig{
			Type:  "const",
			Param: 1,
		},
		Reporter: &jaegercfg.ReporterConfig{
			LogSpans: true,
			// 注意：填下地址不能加http:
			LocalAgentHostPort: "192.168.33.16:6831",
		},
	}
	tracer, closer, err := cfg.New(service, jaegercfg.Logger(jaeger.StdLogger))
	if err != nil {
		panic(fmt.Sprintf("ERROR: cannot init Jaeger: %v\n", err))
	}
	return tracer, closer
}

func main() {
	userRpcFlag := cli.StringFlag{
		Name:  "f",
		Value: "./config/config_api.json",
		Usage: "please use xxx -f config_rpc.json",
	}
	configFile := flag.String(userRpcFlag.Name, userRpcFlag.Value, userRpcFlag.Usage)
	flag.Parse()
	conf := new(gateWayConfig.ApiConfig)

	if err := config.LoadFile(*configFile); err != nil {
		log.Fatal(err)
	}
	if err := config.Scan(conf); err != nil {
		log.Fatal(err)
	}

	tracer, _ := initJaeger("micro-message-system.gateway")
	opentracing.SetGlobalTracer(tracer)

	engineGateWay, err := gorm.Open(conf.Engine.Name, conf.Engine.DataSource)
	if err != nil {
		log.Fatal(err)
	}
	etcdRegisty := etcdv3.NewRegistry(
		func(options *registry.Options) {
			options.Addrs = conf.Etcd.Address
		});

	// Create a new service. Optionally include some options here.
	rpcService := micro.NewService(
		micro.Name(conf.Server.Name),
		micro.Registry(etcdRegisty),
		micro.Transport(grpc.NewTransport()),
		micro.WrapClient(
			hystrix.NewClientWrapper(),
			wrapperTrace.NewClientWrapper(tracer),
		), // 客戶端熔斷、鏈路追蹤
		micro.WrapHandler(wrapperTrace.NewHandlerWrapper(tracer)),
		micro.Flags(userRpcFlag),
	)
	rpcService.Init()

	// 創建用戶服務客戶端 直接可以通過它調用user rpc的服務
	userRpcModel := userProto.NewUserService(conf.UserRpcServer.ServerName, rpcService.Client())

	// 創建IM服務客戶端 直接可以通過它調用im prc的服務
	imRpcModel := imProto.NewImService(conf.ImRpcServer.ServerName, rpcService.Client())

	gateWayModel := models.NewGateWayModel(engineGateWay)

	// 把用戶服務客戶端、IM服務客戶端 注冊到網關
	gateLogic := logic.NewGateWayLogic(userRpcModel, gateWayModel, conf.ImRpcServer.ImServerList, imRpcModel)
	gateWayController := controller.NewGateController(gateLogic)
	// web.NewService會在啟動web server的同時將rpc服務注冊進去
	service := web.NewService(
		web.Name(conf.Server.Name),
		web.Registry(etcdRegisty),
		web.Version(conf.Version),
		web.Flags(userRpcFlag),
		web.Address(conf.Port),
	)
	router := gin.Default()

	userRouterGroup := router.Group("/gateway")
	// 中間件驗證
	userRouterGroup.Use(middleware.ValidAccessToken)
	{
		userRouterGroup.POST("/send", gateWayController.SendHandle)
		userRouterGroup.POST("/address", gateWayController.GetServerAddressHandle)
	}
	service.Handle("/", router)
	if err := service.Run(); err != nil {
		log.Fatal(err)
	}

}

 

調用效果如下：