Dubbo + Zipkin + Brave实现全链路追踪

最近写了一个链路追踪Demo分享下，实现了链路追踪过程中数据的记录，还有能扩展的地方，后期再继续补充。

实现链路追踪的目的

服务调用的流程信息，定位服务调用链

记录调用入参及返回值信息，方便问题重现

记录调用时间线，代码重构及调优处理

调用信息统计

分布式跟踪系统还有其他比较成熟的实现，例如：Naver的Pinpoint、Apache的HTrace、阿里的鹰眼Tracing、京东的Hydra、新浪的Watchman，美团点评的CAT，skywalking等。本次主要利用Dubbo数据传播特性扩展Filter接口来实现链路追踪的目的

重点主要是zipkin及brave使用及特性,当前brave版本为 5.2.0 为 2018年8月份发布的release版本 , zipkin版本为2.2.1 所需JDK为1.8

快速启动zipkin

下载最新的zipkin并启动

1wget -O zipkin.jar 'https://search.maven.org/remote_content?g=io.zipkin.java&a=zipkin-server&v=LATEST&c=exec'
2java -jar zipkin.jar

输入 http://localhost:9411/zipkin/ 进入WebUI界面如下 zipUI

核心源码

代码的初步版本：方便描述

1import brave.Span;
2import brave.Tracer;
3import brave.Tracing;
4import brave.propagation.*;
5import brave.sampler.Sampler;
6import com.alibaba.dubbo.common.Constants;
7import com.alibaba.dubbo.common.extension.Activate;
8import com.alibaba.dubbo.common.json.JSON;
9import com.alibaba.dubbo.common.logger.Logger;
10import com.alibaba.dubbo.common.logger.LoggerFactory;
11import com.alibaba.dubbo.remoting.exchange.ResponseCallback;
12import com.alibaba.dubbo.rpc.*;
13import com.alibaba.dubbo.rpc.protocol.dubbo.FutureAdapter;
14import com.alibaba.dubbo.rpc.support.RpcUtils;
15import zipkin2.codec.SpanBytesEncoder;
16import zipkin2.reporter.AsyncReporter;
17import zipkin2.reporter.Sender;
18import zipkin2.reporter.okhttp3.OkHttpSender;
19
20import java.io.IOException;
21import java.net.InetSocketAddress;
22import java.util.Map;
23import java.util.concurrent.Future;
24import java.util.concurrent.TimeUnit;
25
26/**
27 * Created with IntelliJ IDEA.
28 *
29 * @author: bakerZhu
30 * @description:
31 * @modifytime:
32 */
33@Activate(group = {Constants.PROVIDER, Constants.CONSUMER})
34public class TracingFilter  implements Filter {
35
36	private static final Logger log = LoggerFactory.getLogger(TracingFilter.class);
37
38	private static Tracing tracing;
39	private static Tracer tracer;
40	private static TraceContext.Extractor<Map<String, String>> extractor;
41	private static TraceContext.Injector<Map<String, String>> injector;
42
43	static final Propagation.Getter<Map<String, String>, String> GETTER =
44			new Propagation.Getter<Map<String, String>, String>() {
45				@Override
46				public String get(Map<String, String> carrier, String key) {
47					return carrier.get(key);
48				}
49
50				@Override
51				public String toString() {
52					return "Map::get";
53				}
54			};
55
56	static final Propagation.Setter<Map<String, String>, String> SETTER =
57			new Propagation.Setter<Map<String, String>, String>() {
58				@Override
59				public void put(Map<String, String> carrier, String key, String value) {
60					carrier.put(key, value);
61				}
62
63				@Override
64				public String toString() {
65					return "Map::set";
66				}
67			};
68
69	static {
70		// 1
71		Sender sender = OkHttpSender.create("http://localhost:9411/api/v2/spans");
72		// 2
73		AsyncReporter asyncReporter = AsyncReporter.builder(sender)
74				.closeTimeout(500, TimeUnit.MILLISECONDS)
75				.build(SpanBytesEncoder.JSON_V2);
76		// 3
77		tracing = Tracing.newBuilder()
78				.localServiceName("tracer-client")
79				.spanReporter(asyncReporter)
80				.sampler(Sampler.ALWAYS_SAMPLE)
81				.propagationFactory(ExtraFieldPropagation.newFactory(B3Propagation.FACTORY, "user-name"))
82				.build();
83		tracer = tracing.tracer();
84		// 4
85		// 4.1
86		extractor = tracing.propagation().extractor(GETTER);
87		// 4.2
88		injector = tracing.propagation().injector(SETTER);
89	}
90
91
92
93	public TracingFilter() {
94	}
95
96	@Override
97	public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
98
99
100		RpcContext rpcContext = RpcContext.getContext();
101		// 5
102		Span.Kind kind = rpcContext.isProviderSide() ? Span.Kind.SERVER : Span.Kind.CLIENT;
103		final Span span;
104		if (kind.equals(Span.Kind.CLIENT)) {
105			//6
106			span = tracer.nextSpan();
107			//7
108			injector.inject(span.context(), invocation.getAttachments());
109		} else {
110			//8
111			TraceContextOrSamplingFlags extracted = extractor.extract(invocation.getAttachments());
112			//9
113			span = extracted.context() != null ? tracer.joinSpan(extracted.context()) : tracer.nextSpan(extracted);
114		}
115
116		if (!span.isNoop()) {
117			span.kind(kind).start();
118			//10
119			String service = invoker.getInterface().getSimpleName();
120			String method = RpcUtils.getMethodName(invocation);
121			span.kind(kind);
122			span.name(service + "/" + method);
123			InetSocketAddress remoteAddress = rpcContext.getRemoteAddress();
124			span.remoteIpAndPort(
125					remoteAddress.getAddress() != null ? remoteAddress.getAddress().getHostAddress() : remoteAddress.getHostName(),remoteAddress.getPort());
126		}
127
128		boolean isOneway = false, deferFinish = false;
129		try (Tracer.SpanInScope scope = tracer.withSpanInScope(span)){
130			//11
131			collectArguments(invocation, span, kind);
132			Result result = invoker.invoke(invocation);
133
134			if (result.hasException()) {
135				onError(result.getException(), span);
136			}
137			// 12
138			isOneway = RpcUtils.isOneway(invoker.getUrl(), invocation);
139			// 13
140			Future<Object> future = rpcContext.getFuture();
141
142			if (future instanceof FutureAdapter) {
143				deferFinish = true;
144				((FutureAdapter) future).getFuture().setCallback(new FinishSpanCallback(span));// 14
145			}
146			return result;
147		} catch (Error | RuntimeException e) {
148			onError(e, span);
149			throw e;
150		} finally {
151			if (isOneway) { // 15
152				span.flush();
153			} else if (!deferFinish) { // 16
154				span.finish();
155			}
156		}
157	}
158
159	static void onError(Throwable error, Span span) {
160		span.error(error);
161		if (error instanceof RpcException) {
162			span.tag("dubbo.error_msg", RpcExceptionEnum.getMsgByCode(((RpcException) error).getCode()));
163		}
164	}
165
166	static void collectArguments(Invocation invocation, Span span, Span.Kind kind) {
167		if (kind == Span.Kind.CLIENT) {
168			StringBuilder fqcn = new StringBuilder();
169			Object[] args = invocation.getArguments();
170			if (args != null && args.length > 0) {
171				try {
172					fqcn.append(JSON.json(args));
173				} catch (IOException e) {
174					log.warn(e.getMessage(), e);
175				}
176			}
177			span.tag("args", fqcn.toString());
178		}
179	}
180
181
182
183	static final class FinishSpanCallback implements ResponseCallback {
184		final Span span;
185
186		FinishSpanCallback(Span span) {
187			this.span = span;
188		}
189
190		@Override
191		public void done(Object response) {
192			span.finish();
193		}
194
195		@Override
196		public void caught(Throwable exception) {
197			onError(exception, span);
198			span.finish();
199		}
200	}
201	// 17
202	private enum RpcExceptionEnum {
203		UNKNOWN_EXCEPTION(0, "unknown exception"),
204		NETWORK_EXCEPTION(1, "network exception"),
205		TIMEOUT_EXCEPTION(2, "timeout exception"),
206		BIZ_EXCEPTION(3, "biz exception"),
207		FORBIDDEN_EXCEPTION(4, "forbidden exception"),
208		SERIALIZATION_EXCEPTION(5, "serialization exception"),;
209
210		private int code;
211
212		private String msg;
213
214		RpcExceptionEnum(int code, String msg) {
215			this.code = code;
216			this.msg = msg;
217		}
218
219		public static String getMsgByCode(int code) {
220			for (RpcExceptionEnum error : RpcExceptionEnum.values()) {
221				if (code == error.code) {
222					return error.msg;
223				}
224			}
225			return null;
226		}
227	}
228}

构建客户端发送工具
构建异步reporter
构建tracing上下文
初始化injector 和 Extractor [tab]4.1 extractor 指数据提取对象,用于在carrier中提取TraceContext相关信息或者采样标记信息到TraceContextOrSamplingFlags 中 -4.2 injector 用于将TraceContext中的各种数据注入到carrier中,其中carrier一半是指数据传输中的载体,类似于Dubbo中Invocation中的attachment(附件集合)
判断此次调用是作为服务端还是客户端
rpc客户端调用会从ThreadLocal中获取parent的 TraceContext ,为新生成的Span指定traceId及 parentId如果没有parent traceContext 则生成的Span为 root span
将Span绑定的TraceContext中属性信息 Copy 到 Invocation中达到远程参数传递的作用
rpc服务提供端 , 从invocation中提取TraceContext相关信息及采样数据信息
生成span , 兼容初次服务端调用
记录接口信息及远程IP Port
将创建的Span 作为当前Span (可以通过Tracer.currentSpan 访问到它) 并设置查询范围
oneway调用即只请求不接受结果
如果future不为空则为 async 调用在回调中finish span
设置异步回调，回调代码执行span finish() .
oneway调用因为不需等待返回值即没有 cr (Client Receive) 需手动flush()
同步调用业务代码执行完毕后需手动finish()
设置枚举类与 Dubbo中RpcException保持对应

测试项

Dubbo sync async oneway 测试

RPC异常测试

普通业务异常测试

并发测试

配置方式

POM依赖添加

1<dependency>
2    <groupId>com.github.baker</groupId>
3    <artifactId>Tracing</artifactId>
4    <version>1.0-SNAPSHOT</version>
5</dependency>

资源目录根路径下添加tracing.properties文件一次调用信息调用链调用成功失败汇总 zipkinHost 指定zipkin服务器IP:PORT 默认为localhost:9411 serviceName 指定应用名称默认为trace-default

调用链： tracing.properties

待扩展项

抽象数据传输（扩展Kafka数据传输）

调用返回值数据打印

更灵活的配置方式

源码地址

Dubbo + Zipkin + Brave实现全链路追踪

Dubbo + Zipkin + Brave实现全链路追踪

快速启动zipkin

核心源码

配置方式

赞赏支持

评论区

相关推荐