Opentracing基本模型
image.png
如图，在跟踪链中有以下几个比较重要的数据结构和概念：
span：标识一次分布式调用，其自身包含了id，parentId(指向上级Span的id)， traceIds，服务名称等重要属性，其应尽量保持精简；
trace：标识整个请求链，即一些列Span的组合。其自身的ID将贯穿整个调用链，其中的每个Span都必须携带这个traceId，因此traceId将在整个调用链中传递；
cs：客户端发起请求，标志Span的开始；
sr：服务端接收到请求，并开始处理内部事务，其中sr - cs则为网络延迟和时钟抖动；
ss：服务端处理完请求，返回响应内容，其中ss - sr则为服务端处理请求耗时；
cr：客户端接收到服务端响应内容，标志着Span的结束，其中cr - ss则为网络延迟和时钟抖动。
客户端调用时间=cr-cs
服务端处理时间=sr-ss
分布式系统调用跟踪的基本架构要求
低侵入性，高性能，高可用容错，低丢失率等。
基于GRPC的分布式系统调用跟踪实践
创建TraceContext
TraceContext通过Threadlocal对span进行保存，并且将traceid和spanid向底层服务传递，zebra对线程上下文传递进行了封装，具体参照GRPC如何实现公共参数与业务参数分离传递下面是TraceContext具体代码
public class TraceContext{
private static final String SPAN_LIST_KEY = "spanList";
public static final String TRACE_ID_KEY = "traceId";
public static final String SPAN_ID_KEY = "spanId";
public static final String ANNO_CS = "cs";
public static final String ANNO_CR = "cr";
public static final String ANNO_SR = "sr";
public static final String ANNO_SS = "ss";
private TraceContext(){}
public static void setTraceId(String traceId) {
RpcContext.getContext().set(TRACE_ID_KEY, traceId);
}
public static String getTraceId() {
return (String) RpcContext.getContext().get(TRACE_ID_KEY);
}
public static String getSpanId() {
return (String) RpcContext.getContext().get(SPAN_ID_KEY);
}
public static void setSpanId(String spanId) {
RpcContext.getContext().set(SPAN_ID_KEY, spanId);
}
@SuppressWarnings("unchecked")
public static void addSpan(Span span){
((List)RpcContext.getContext().get(SPAN_LIST_KEY)).add(span);
}
@SuppressWarnings("unchecked")
public static List getSpans(){
return (List) RpcContext.getContext().get(SPAN_LIST_KEY);
}
public static void clear(){
RpcContext.getContext().remove(TRACE_ID_KEY);
RpcContext.getContext().remove(SPAN_ID_KEY);
RpcContext.getContext().remove(SPAN_LIST_KEY);
}
public static void start(){
clear();
RpcContext.getContext().set(SPAN_LIST_KEY, new ArrayList());
}
}
创建TraceAgent
TraceAgent将span信息上传至kafka，代码如下：
public class TraceAgent {
private GrpcProperties grpcProperties;
private KafkaSender sender;
private AsyncReporter report;
public TraceAgent() {
grpcProperties = SpringContextUtils.getBean(GrpcProperties.class);
sender = KafkaSender.newBuilder().bootstrapServers(grpcProperties.getCallChainUpdAddr()).topic("zipkin").encoding(Encoding.JSON).build();
report = AsyncReporter.builder(sender).build();
}
public void send(final List spans){
spans.forEach(item ->{
report.report(item);
});
}
}
创建ZebraClientTracing
ZebraClientTracing用于记录调用端的span信息，具体代码如下：
@Component
public class ZebraClientTracing {
public Span startTrace(String method) {
String id = IdUtils.get() + "";
String traceId = null;
if (null == TraceContext.getTraceId()) {
TraceContext.start();
traceId = id;
} else {
traceId = TraceContext.getTraceId();
}
long timestamp = System.currentTimeMillis() * 1000;
// 注册本地信息
Endpoint endpoint = Endpoint.newBuilder().ip(NetUtils.getLocalHost()).serviceName(EtcdRegistry.serviceName)
.port(50003).build();
// 初始化span
Span consumerSpan = Span.newBuilder().localEndpoint(endpoint).id(id).traceId(traceId)
.parentId(TraceContext.getSpanId() + "").name(EtcdRegistry.serviceName).timestamp(timestamp)
.addAnnotation(timestamp, TraceContext.ANNO_CS).putTag("method", method)
.putTag("pkgId", RpcContext.getContext().getAttachment("pkg")).build();
// 将tracing id和spanid放到上下文
RpcContext.getContext().get().put(TraceContext.TRACE_ID_KEY, consumerSpan.traceId());
RpcContext.getContext().get().put(TraceContext.SPAN_ID_KEY, String.valueOf(consumerSpan.id()));
return consumerSpan;
}
public void endTrace(Span span, Stopwatch watch,int code) {
span = span.toBuilder().addAnnotation(System.currentTimeMillis() * 1000, TraceContext.ANNO_CR)
.duration(watch.stop().elapsed(TimeUnit.MICROSECONDS)).putTag("code", code+"").build();
TraceAgent traceAgent = new TraceAgent();
traceAgent.send(TraceContext.getSpans());
}
}
创建ZebraServerTracing
ZebraServerTracing用于记录服务端的span信息，具体代码如下：
@Component
public class ZebraServerTracing {
public Span startTrace(String method) {
String traceId = (String) RpcContext.getContext().get(TraceContext.TRACE_ID_KEY);
String parentSpanId = (String) RpcContext.getContext().get(TraceContext.SPAN_ID_KEY);
String id = IdUtils.get() + "";
TraceContext.start();
TraceContext.setTraceId(traceId);
TraceContext.setSpanId(parentSpanId);
long timestamp = System.currentTimeMillis() * 1000;
Endpoint endpoint = Endpoint.newBuilder().ip(NetUtils.getLocalHost()).serviceName(EtcdRegistry.serviceName)
.port(50003).build();
Span providerSpan = Span.newBuilder().id(id).parentId(parentSpanId).traceId(traceId)
.name(EtcdRegistry.serviceName).timestamp(timestamp).localEndpoint(endpoint)
.addAnnotation(timestamp, TraceContext.ANNO_SR).putTag("method", method)
.putTag("pkgId", RpcContext.getContext().getAttachment("pkg"))
.build();
TraceContext.addSpan(providerSpan);
return providerSpan;
}
public void endTrace(Span span, Stopwatch watch,int code) {
span = span.toBuilder().addAnnotation(System.currentTimeMillis() * 1000, TraceContext.ANNO_SS)
.duration(watch.stop().elapsed(TimeUnit.MICROSECONDS)).putTag("code", code+"").build();
TraceAgent traceAgent = new TraceAgent();
traceAgent.send(TraceContext.getSpans());
}
}
创建grpc client拦截器
public class HeaderClientInterceptor implements ClientInterceptor {
private static final Logger log = LogManager.getLogger(HeaderClientInterceptor.class);
private final ZebraClientTracing clientTracing;
public static ClientInterceptor instance() {
return new HeaderClientInterceptor();
}
private HeaderClientInterceptor() {
clientTracing = SpringContextUtils.getBean(ZebraClientTracing.class);
}
@Override
public ClientCall interceptCall(MethodDescriptor method,
CallOptions callOptions, Channel next) {
return new SimpleForwardingClientCall(next.newCall(method, callOptions)) {
//判断API网关是否要打开调用链
boolean isGatewayTracing = "1".equals(RpcContext.getContext().getAttachment(ZebraConstants.ZEBRA_OPEN_TRACING))?true:false;
boolean isSubTracing = RpcContext.getContext().get(TraceContext.TRACE_ID_KEY)!=null?true:false;
Stopwatch watch =null;
Span span =null;
@Override
public void start(Listener responseListener, Metadata headers) {
if(isSubTracing||isGatewayTracing){
span =clientTracing.startTrace(method.getFullMethodName());
watch = Stopwatch.createStarted();
}
copyThreadLocalToMetadata(headers);
super.start(new SimpleForwardingClientCallListener(responseListener) {
@Override
public void onHeaders(Metadata headers) {
super.onHeaders(headers);
}
@Override
public void onClose(Status status, Metadata trailers) {
super.onClose(status, trailers);
if(isSubTracing||isGatewayTracing)
clientTracing.endTrace(span, watch,status.getCode().value());
}
}, headers);
}
};
}
private void copyThreadLocalToMetadata(Metadata headers) {
Map attachments = RpcContext.getContext().getAttachments();
Map values = RpcContext.getContext().get();
try {
if (!attachments.isEmpty()) {
headers.put(GrpcUtil.GRPC_CONTEXT_ATTACHMENTS, SerializerUtil.toJson(attachments));
}
if (!values.isEmpty()) {
headers.put(GrpcUtil.GRPC_CONTEXT_VALUES, SerializerUtil.toJson(values));
}
} catch (Throwable e) {
log.error(e.getMessage(), e);
}
}
}
创建grpc server拦截器
public class HeaderServerInterceptor implements ServerInterceptor {
private static final Logger log = LogManager.getLogger(HeaderServerInterceptor.class);
private final ZebraServerTracing serverTracing;
public static ServerInterceptor instance() {
return new HeaderServerInterceptor();
}
private HeaderServerInterceptor() {
serverTracing = SpringContextUtils.getBean(ZebraServerTracing.class);
}
@Override
public Listener interceptCall(ServerCall call, final Metadata headers,
ServerCallHandler next) {
return next.startCall(new SimpleForwardingServerCall(call) {
boolean isSubTracing = RpcContext.getContext().get(TraceContext.TRACE_ID_KEY) != null ? true : false;
Stopwatch watch = null;
Span span = null;
@Override
public void request(int numMessages) {
if (isSubTracing) {
span = serverTracing.startTrace(call.getMethodDescriptor().getFullMethodName());
watch = Stopwatch.createStarted();
}
InetSocketAddress remoteAddress = (InetSocketAddress) call.getAttributes()
.get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR);
RpcContext.getContext().setAttachment(ZebraConstants.REMOTE_ADDRESS, remoteAddress.getHostString());
copyMetadataToThreadLocal(headers);
log.debug("FullMethodName:{},RemoteAddress={},attachments={},context={}",
call.getMethodDescriptor().getFullMethodName(), remoteAddress.getHostString(),
headers.get(GrpcUtil.GRPC_CONTEXT_ATTACHMENTS), headers.get(GrpcUtil.GRPC_CONTEXT_VALUES));
super.request(numMessages);
}
@Override
public void close(Status status, Metadata trailers) {
delegate().close(status, trailers);
if(isSubTracing)
serverTracing.endTrace(span, watch,status.getCode().value());
}
}, headers);
}
private void copyMetadataToThreadLocal(Metadata headers) {
String attachments = headers.get(GrpcUtil.GRPC_CONTEXT_ATTACHMENTS);
String values = headers.get(GrpcUtil.GRPC_CONTEXT_VALUES);
try {
if (attachments != null) {
Map attachmentsMap = SerializerUtil.fromJson(attachments,
new TypeToken>() {
}.getType());
RpcContext.getContext().setAttachments(attachmentsMap);
}
if (values != null) {
Map valuesMap = SerializerUtil.fromJson(values, new TypeToken>() {
}.getType());
for (Map.Entry entry : valuesMap.entrySet()) {
RpcContext.getContext().set(entry.getKey(), entry.getValue());
}
}
} catch (Throwable e) {
log.error(e.getMessage(), e);
}
}
}
简单用法)
)
