基于 OpenTelemetry 的链路追踪

链路追踪的前世今生

分布式跟踪(也称为分布式请求跟踪)是一种用于分析和监控应用程序的方法,尤其是使用微服务架构构建的应用程序。分布式跟踪有助于精确定位故障发生的位置以及导致性能差的原因。

起源

链路追踪(Distributed Tracing) 一词最早出现于谷歌发布的论文 《Dapper, a Large-Scale Distributed Systems Tracing Infrastructure》 中,这篇论文对于实现链路追踪,对于后来出现的 Jaeger、Zipkin 等开源分布式追踪项目设计理念仍有很深的影响。

微服务架构是一个分布式的架构,会有很多个不同的服务。不同的服务之前相互调用,如果出现了错误由于一个请求经过了 N 个服务。随着业务的增加越来越多的服务之间的调用,如果没有一个工具去记录调用链,解决问题的时候就会像下面图片里小猫咪玩的毛线球一样,毫无头绪,无从下手

所以需要有一个工具能够清楚的了解一个请求经过了哪些服务,顺序是如何,从而能够轻易的定位问题。

百家争艳

从谷歌发布 Dapper 后,分布式链路追踪工具越来越多,以下简单列举了一些常用的链路追踪系统

  • Skywalking
  • 阿里 鹰眼
  • 大众点评 CAT
  • Twitter Zipkin
  • Naver pinpoint
  • Uber Jaeger

争锋相对?

随着链路追踪工具越来越多,开源领域主要分为两派,一派是以 CNCF技术委员 会为主的 OpenTracing 的规范,例如 jaeger zipkin 都是遵循了OpenTracing 的规范。而另一派则是谷歌作为发起者的 OpenCensus,而且谷歌本身还是最早提出链路追踪概念的公司,后期连微软也加入了 OpenCensus

OpenTelemetry 诞生

OpenTelemetric 是一组 API、SDK、模组和集成,专为创建和管理‎‎遥测数据‎‎(如追踪、指标和日志)而设

微软加入 OpenCensus 后,直接打破了之前平衡的局面,间接的导致了 OpenTelemetry 的诞生 谷歌和微软下定决心结束江湖之乱,首要的问题是如何整合两个两个社区已有的项目,OpenTelemetry 主要的理念就是,兼容 OpenCensusOpenTracing ,可以让使用者无需改动或者很小的改动就可以接入 OpenTelemetry

Kratos 的链路追踪实践

Kratos 一套轻量级 Go 微服务框架,包含大量微服务相关框架及工具。

tracing 中间件

kratos 框架提供的自带中间件中有一个名为 tracing 中间件,它基于 Opentelemetry 实现了kratos 框架的链路追踪功能,中间件的代码可以从 middleware/tracing 中看到。

实现原理

kratos 的链路追踪中间件由三个文件组成 carrie.go,tracer.go,tracing.go。client和 server 的实现原理基本相同,本文以 server 实现进行原理解析。

  1. 首先当请求进入时,tracing 中间件会被调用,首先调用了 tracer.go 中的 NewTracer 方法
1// Server returns a new server middleware for OpenTelemetry.
2func Server(opts ...Option) middleware.Middleware {
3        // 调用 tracer.go 中的 NewTracer 传入了一个 SpanKindServer 和配置项
4	tracer := NewTracer(trace.SpanKindServer, opts...)
5        // ... 省略代码
6}
  1. tracer.go 中的 NewTracer 方法被调用后会返回一个 Tracer,实现如下
 1func NewTracer(kind trace.SpanKind, opts ...Option) *Tracer {
 2	options := options{}
 3	for _, o := range opts {
 4		o(&options)
 5	}
 6	// 判断是否存在 otel 追踪提供者配置,如果存在则设置
 7	if options.TracerProvider != nil {
 8		otel.SetTracerProvider(options.TracerProvider)
 9	}
10	/*
11	判断是否存在 Propagators 设置,如果存在设置则覆盖,不存在则设置一个默认的TextMapPropagator
12	注意如果没有设置默认的TextMapPropagator,链路信息则无法正确的传递
13	*/
14	if options.Propagators != nil {
15		otel.SetTextMapPropagator(options.Propagators)
16	} else {	otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(propagation.Baggage{}, propagation.TraceContext{}))
17	}
18
19
20	var name string
21	// 判断当前中间件的类型,是 server 还是 client
22	if kind == trace.SpanKindServer {
23		name = "server"
24	} else if kind == trace.SpanKindClient {
25		name = "client"
26	} else {
27		panic(fmt.Sprintf("unsupported span kind: %v", kind))
28	}
29	// 调用 otel包的 Tracer 方法 传入 name 用来创建一个 tracer 实例
30	tracer := otel.Tracer(name)
31	return &Tracer{tracer: tracer, kind: kind}
32}
  1. 判断当前请求类型,处理需要采集的数据,并调用 tracer.go 中的 Start 方法
 1var (
 2	component string
 3	operation string
 4	carrier   propagation.TextMapCarrier
 5)
 6// 判断请求类型
 7if info, ok := http.FromServerContext(ctx); ok {
 8	// HTTP
 9	component = "HTTP"
10	// 取出请求的地址
11	operation = info.Request.RequestURI
12	// 调用 otel/propagation包中的 HeaderCarrier,会处理 http.Header 以用来满足TextMapCarrier interface
13	// TextMapCarrier 是一个文本映射载体,用于承载信息
14	carrier = propagation.HeaderCarrier(info.Request.Header)
15	// otel.GetTextMapPropagator().Extract() 方法用于将文本映射载体,读取到上下文中
16	ctx = otel.GetTextMapPropagator().Extract(ctx, propagation.HeaderCarrier(info.Request.Header))
17} else if info, ok := grpc.FromServerContext(ctx); ok {
18	// Grpc
19	component = "gRPC"
20	operation = info.FullMethod
21	//
22	// 调用 grpc/metadata包中metadata.FromIncomingContext(ctx)传入 ctx,转换 grpc 的元数据
23	if md, ok := metadata.FromIncomingContext(ctx); ok {
24		// 调用carrier.go 中的 MetadataCarrier 将 MD 转换 成文本映射载体
25		carrier = MetadataCarrier(md)
26	}
27}
28// 调用 tracer.Start 方法
29ctx, span := tracer.Start(ctx, component, operation, carrier)
30// ... 省略代码
31}
  1. 调用 tracing.go 中的 Start 方法
 1func (t *Tracer) Start(ctx context.Context, component string, operation string, carrier propagation.TextMapCarrier) (context.Context, trace.Span) {
 2	// 判断当前中间件如果是 server则将 carrier 注入到上下文中
 3	if t.kind == trace.SpanKindServer {
 4		ctx = otel.GetTextMapPropagator().Extract(ctx, carrier)
 5	}
 6	// 调用otel/tracer 包中的 start 方法,用来创建一个 span
 7	ctx, span := t.tracer.Start(ctx,
 8		// tracing.go 中声明的请求路由作为 spanName
 9		operation,
10		// 设置 span 的属性,设置了一个 component,component的值为请求类型
11		trace.WithAttributes(attribute.String("component", component)),
12		// 设置 span种类
13		trace.WithSpanKind(t.kind),
14	)
15	// 判断如果当前中间件是 client 则将 carrier 注入到请求里面
16	if t.kind == trace.SpanKindClient {
17		otel.GetTextMapPropagator().Inject(ctx, carrier)
18	}
19	return ctx, span
20}
  1. defer 声明了一个闭包方法
1// 这个地方要注意,需要使用闭包,因为 defer 的参数是实时计算的如果异常发生,err 会一直为 nil
2// https://github.com/go-kratos/kratos/issues/927
3defer func() { tracer.End(ctx, span, err) }()
  1. 中间件继续执行
1// tracing.go 69行
2reply, err = handler(ctx, req)
  1. 中间件调用结束 defer 中的闭包被调用后执行了 tracer.go 中的 End 方法
 1func (t *Tracer) End(ctx context.Context, span trace.Span, err error) {
 2	// 判断是否有异常发生,如果有则设置一些异常信息
 3	if err != nil {
 4		// 记录异常
 5		span.RecordError(err)
 6		// 设置span 属性
 7		span.SetAttributes(
 8			// 设置事件为异常
 9			attribute.String("event", "error"),
10			// 设置 message 为 err.Error().
11			attribute.String("message", err.Error()),
12		)
13		//设置了 span 的状态
14		span.SetStatus(codes.Error, err.Error())
15	} else {
16		// 如果没有发生异常,span 状态则为 ok
17		span.SetStatus(codes.Ok, "OK")
18	}
19	// 中止 span
20	span.End()
21}

如何使用

tracing 中间件的使用示例可以从 kratos/examples/traces ,该示例简单的实现了跨服务间的链路追踪,以下代码片段包含部分示例代码。

 1// https://github.com/go-kratos/kratos/blob/7f835db398c9d0332e69b81bad4c652b4b45ae2e/examples/traces/app/message/main.go#L38
 2// 首先调用otel 库方法,得到一个 TracerProvider
 3func tracerProvider(url string) (*tracesdk.TracerProvider, error) {
 4	// examples/traces 中使用的是 jaeger,其他方式可以查看 opentelemetry 官方示例
 5	exp, err := jaeger.NewRawExporter(jaeger.WithCollectorEndpoint(jaeger.WithEndpoint(url)))
 6	if err != nil {
 7		return nil, err
 8	}
 9	tp := tracesdk.NewTracerProvider(
10		tracesdk.WithSampler(tracesdk.AlwaysSample()),
11		// 设置 Batcher,注册jaeger导出程序
12		tracesdk.WithBatcher(exp),
13		// 记录一些默认信息
14		tracesdk.WithResource(resource.NewWithAttributes(
15			semconv.ServiceNameKey.String(pb.User_ServiceDesc.ServiceName),
16			attribute.String("environment", "development"),
17			attribute.Int64("ID", 1),
18		)),
19	)
20	return tp, nil
21}

在 grpc/server 中使用

 1// https://github.com/go-kratos/kratos/blob/main/examples/traces/app/message/main.go
 2grpcSrv := grpc.NewServer(
 3	grpc.Address(":9000"),
 4	grpc.Middleware(
 5		// Configuring tracing Middleware
 6		tracing.Server(
 7			tracing.WithTracerProvider(tp),
 8		),
 9	),
10)

在 grpc/client 中使用

 1// https://github.com/go-kratos/kratos/blob/149fc0195eb62ee1fbc2728adb92e1bcd1a12c4e/examples/traces/app/user/main.go#L63
 2conn, err := grpc.DialInsecure(ctx,
 3	grpc.WithEndpoint("127.0.0.1:9000"),
 4	grpc.WithMiddleware(
 5		tracing.Client(
 6			tracing.WithTracerProvider(s.tracer),
 7			tracing.WithPropagators(
 8				propagation.NewCompositeTextMapPropagator(propagation.Baggage{}, propagation.TraceContext{}),
 9			),
10		)
11	),
12	grpc.WithTimeout(2*time.Second),
13)

在 http/server 中使用

 1// https://github.com/go-kratos/kratos/blob/main/examples/traces/app/user/main.go
 2httpSrv := http.NewServer(http.Address(":8000"))
 3httpSrv.HandlePrefix("/", pb.NewUserHandler(s,
 4	http.Middleware(
 5		// Configuring tracing middleware
 6		tracing.Server(
 7			tracing.WithTracerProvider(tp),
 8			tracing.WithPropagators(
 9				propagation.NewCompositeTextMapPropagator(propagation.Baggage{}, propagation.TraceContext{}),
10			),
11		),
12	),
13)

在 http/client 中使用

1http.NewClient(ctx, http.WithMiddleware(
2	tracing.Client(
3		tracing.WithTracerProvider(s.tracer),
4	),
5))

如何实现一个其他场景的 tracing

我们可以借鉴 kratostracing 中间件的代码来实现例如数据库的 tracing,如下面的代码片段,作者借鉴了tracing 中间件,实现了 qmgo 库操作 MongoDB 数据库的 tracing

 1func mongoTracer(ctx context.Context,tp trace.TracerProvider, command interface{}) {
 2	var (
 3		commandName string
 4		failure     string
 5		nanos       int64
 6		reply       bson.Raw
 7		queryId     int64
 8		eventName   string
 9	)
10	otel.SetTracerProvider(tp)
11	reply = bson.Raw{}
12	switch value := command.(type) {
13	case *event.CommandStartedEvent:
14		commandName = value.CommandName
15		reply = value.Command
16		queryId = value.RequestID
17		eventName = "CommandStartedEvent"
18	case *event.CommandSucceededEvent:
19		commandName = value.CommandName
20		nanos = value.DurationNanos
21		queryId = value.RequestID
22		eventName = "CommandSucceededEvent"
23	case *event.CommandFailedEvent:
24		commandName = value.CommandName
25		failure = value.Failure
26		nanos = value.DurationNanos
27		queryId = value.RequestID
28		eventName = "CommandFailedEvent"
29	}
30	duration, _ := time.ParseDuration(strconv.FormatInt(nanos, 10) + "ns")
31	tracer := otel.Tracer("mongodb")
32	kind := trace.SpanKindServer
33	ctx, span := tracer.Start(ctx,
34		commandName,
35		trace.WithAttributes(
36			attribute.String("event", eventName),
37			attribute.String("command", commandName),
38			attribute.String("query", reply.String()),
39			attribute.Int64("queryId", queryId),
40			attribute.String("ms", duration.String()),
41		),
42		trace.WithSpanKind(kind),
43	)
44	if failure != "" {
45		span.RecordError(errors.New(failure))
46	}
47	span.End()
48}

文章转自