源码cloud分支
假设已经了解了Dubbo的Filter，Spring Cloud Sleuth，ZipKin。
其实openZipKin已经提供了该功能,请切换到dubbo分支
但它是基于最新的spring cloud sleuth。spring cloud sleuth 2.0.0，我也没去了解最新的版本的代码，因为公司版本不是想升就升，所以造了一个轮子，其思想和源码差不多。
另外使用spring cloud sleuth来链路追踪而不使用其他库，比如Brave。那就是它集成了将链路信息记录到LogBack的功能。
基于Brave的Demo 请看源码master分支
本文直接贴代码，有空会进行讲解
- 定义Filter
@Activate(group = {Constants.PROVIDER, Constants.CONSUMER})
public class MyTracingFilter implements Filter {
private static final Logger LOGGER = LoggerFactory.getLogger(MyTracingFilter.class);
protected void addRequestTags(RpcContext context) {
this.httpTraceKeysInjector.addRequestTags(context.getUrl().getAddress(),
context.getUrl().getHost(),
context.getUrl().getPath(),
context.getMethodName(),
Collections.emptyMap());
}
HttpTraceKeysInjector httpTraceKeysInjector;
Tracer tracer;
DubboExtractor dubboExtractor;
DubboInject injector;
SpanReporter spanReporter;
@Override
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
tracer = ApplicationBeanHolder.getBean(Tracer.class);
if (tracer == null) {
return invoker.invoke(invocation);
} else {
injectBean();
}
RpcContext rpcContext = RpcContext.getContext();
Span span;
String service = invoker.getInterface().getSimpleName();
String method = RpcUtils.getMethodName(invocation);
String spanName = service + "/" + method;
if (rpcContext.isConsumerSide()) {
span = tracer.createSpan(spanName);
injector.inject(span, new DubboRequestTextMap(RpcContext.getContext()));
addRequestTags(RpcContext.getContext());
span.logEvent(Span.CLIENT_SEND);
Result result;
try {
result = invoker.invoke(invocation);
} finally {
closeSpan(span,true);
}
return result;
} else {
Span parentSpan = dubboExtractor.joinTrace(new DubboRequestTextMap(RpcContext.getContext()));
if (parentSpan != null) {
span = parentSpan;
tracer.continueSpan(span);
span.logEvent(Span.SERVER_RECV);
} else {
String header = RpcContext.getContext().getAttachment(Span.SPAN_FLAGS);
if (Span.SPAN_SAMPLED.equals(header)) {
span = tracer.createSpan(spanName, new AlwaysSampler());
} else {
span = tracer.createSpan(spanName);
}
span.logEvent(Span.SERVER_RECV);
}
Result result;
try {
result = invoker.invoke(invocation);
} finally {
recordParentSpan(span);
closeSpan(span,false);
}
return result;
}
}
private void recordParentSpan(Span parent) {
if (parent == null) {
return;
}
if (parent.isRemote()) {
tracer.getCurrentSpan().logEvent(Span.SERVER_SEND);
parent.stop();
this.spanReporter.report(parent);
}
}
private void injectBean() {
injector = ApplicationBeanHolder.getBean(DubboInject.class);
dubboExtractor = ApplicationBeanHolder.getBean(DubboExtractor.class);
httpTraceKeysInjector = ApplicationBeanHolder.getBean(HttpTraceKeysInjector.class);
spanReporter = ApplicationBeanHolder.getBean(SpanReporter.class);
}
private void closeSpan(Span span, Boolean type) {
if (type) {
tracer.getCurrentSpan().logEvent(Span.CLIENT_RECV);
}
if (span != null) {
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Closing Dubbo span " + span);
}
tracer.close(span);
}
}
}
- DubboInject
public class DubboInject implements SpanInjector<SpanTextMap> {
private static final DubboSpanMapper SPAN_CARRIER_MAPPER = new DubboSpanMapper();
@Override
public void inject(Span span, SpanTextMap map) {
Map<String, String> carrier = SPAN_CARRIER_MAPPER.convert(map);
setHeader(map, carrier, Span.TRACE_ID_NAME, span.traceIdString());
setIdHeader(map, carrier, Span.SPAN_ID_NAME, span.getSpanId());
setHeader(map, carrier, Span.SAMPLED_NAME, span.isExportable() ? Span.SPAN_SAMPLED : Span.SPAN_NOT_SAMPLED);
setHeader(map, carrier, Span.SPAN_NAME_NAME, span.getName());
setIdHeader(map, carrier, Span.PARENT_ID_NAME, getParentId(span));
setHeader(map, carrier, Span.PROCESS_ID_NAME, span.getProcessId());
for (Map.Entry<String, String> entry : span.baggageItems()) {
map.put(prefixedKey(entry.getKey()), entry.getValue());
}
}
private String prefixedKey(String key) {
if (key.startsWith(Span.SPAN_BAGGAGE_HEADER_PREFIX
+ DubboSpanMapper.HEADER_DELIMITER)) {
return key;
}
return Span.SPAN_BAGGAGE_HEADER_PREFIX + DubboSpanMapper.HEADER_DELIMITER
+ key;
}
private Long getParentId(Span span) {
return !span.getParents().isEmpty() ? span.getParents().get(0) : null;
}
private void setIdHeader(SpanTextMap map, Map<String, String> carrier, String name, Long value) {
if (value != null) {
setHeader(map, carrier, name, Span.idToHex(value));
}
}
private void setHeader(SpanTextMap map, Map<String, String> carrier, String name, String value) {
if (StringUtils.hasText(value) && !carrier.containsKey(name)) {
map.put(name, value);
}
}
}
- DubboExtractor
public class DubboExtractor implements SpanExtractor<SpanTextMap> {
private static final org.apache.commons.logging.Log log = LogFactory.getLog(
MethodHandles.lookup().lookupClass());
private static final String HTTP_COMPONENT = "http";
private static final DubboSpanMapper SPAN_CARRIER_MAPPER = new DubboSpanMapper();
private final Pattern skipPattern;
private final Random random;
public DubboExtractor(Pattern skipPattern) {
this.skipPattern = skipPattern;
this.random = new Random();
}
@Override
public Span joinTrace(SpanTextMap textMap) {
Map<String, String> carrier = SPAN_CARRIER_MAPPER.convert(textMap);
boolean debug = Span.SPAN_SAMPLED.equals(carrier.get(Span.SPAN_FLAGS));
boolean idToBeGenerated = debug && onlySpanIdIsPresent(carrier);
if (!idToBeGenerated && traceIdIsMissing(carrier)) {
return null;
}
try {
return buildParentSpan(carrier, idToBeGenerated);
} catch (Exception e) {
log.error("Exception occurred while trying to extract span from carrier", e);
return null;
}
}
private boolean onlySpanIdIsPresent(Map<String, String> carrier) {
return traceIdIsMissing(carrier) && spanIdIsPresent(carrier);
}
private boolean traceIdIsMissing(Map<String, String> carrier) {
return carrier.get(Span.TRACE_ID_NAME) == null;
}
private boolean spanIdIsPresent(Map<String, String> carrier) {
return carrier.get(Span.SPAN_ID_NAME) != null;
}
private String generateId() {
return Span.idToHex(this.random.nextLong());
}
private long spanId(String spanId, String traceId) {
if (spanId == null) {
if (log.isDebugEnabled()) {
log.debug("Request is missing a span id but it has a trace id. We'll assume that this is "
+ "a root span with span id equal to the lower 64-bits of the trace id");
}
return Span.hexToId(traceId);
} else {
return Span.hexToId(spanId);
}
}
private Span buildParentSpan(Map<String, String> carrier, boolean idToBeGenerated) {
String traceId = carrier.get(Span.TRACE_ID_NAME);
if (traceId == null) {
traceId = generateId();
}
Span.SpanBuilder span = Span.builder()
.traceIdHigh(traceId.length() == 32 ? Span.hexToId(traceId, 0) : 0)
.traceId(Span.hexToId(traceId))
.spanId(spanId(carrier.get(Span.SPAN_ID_NAME), traceId));
String parentName = carrier.get(Span.SPAN_NAME_NAME);
if (StringUtils.hasText(parentName)) {
span.name(parentName);
} else {
span.name(HTTP_COMPONENT + ":/parent"
+ carrier.get(DubboSpanMapper.URI_HEADER));
}
String processId = carrier.get(Span.PROCESS_ID_NAME);
if (StringUtils.hasText(processId)) {
span.processId(processId);
}
String parentId = carrier.get(Span.PARENT_ID_NAME);
if (parentId != null) {
span.parent(Span.hexToId(parentId));
}
span.remote(true);
// boolean skip = this.skipPattern
// .matcher(carrier.get(DubboSpanMapper.URI_HEADER)).matches()
// || Span.SPAN_NOT_SAMPLED.equals(carrier.get(Span.SAMPLED_NAME));
boolean skip = false;
// trace, span id were retrieved from the headers and span is sampled
span.exportable(!(skip || idToBeGenerated));
boolean debug = Span.SPAN_SAMPLED.equals(carrier.get(Span.SPAN_FLAGS));
if (debug) {
span.exportable(true);
} else if (skip) {
span.exportable(false);
}
for (Map.Entry<String, String> entry : carrier.entrySet()) {
if (entry.getKey().toLowerCase()
.startsWith(DubboSpanMapper.BAGGAGE_PREFIX)) {
span.baggage(unprefixedKey(entry.getKey()), entry.getValue());
}
}
return span.build();
}
private String unprefixedKey(String key) {
return key.substring(key.indexOf(DubboSpanMapper.HEADER_DELIMITER) + 1)
.toLowerCase();
}
}
- DubboRequestTextMap
public class DubboRequestTextMap implements SpanTextMap {
private final RpcContext delegate;
public DubboRequestTextMap(RpcContext delegate) {
this.delegate = delegate;
}
@Override
public Iterator<Map.Entry<String, String>> iterator() {
return this.delegate.getAttachments().entrySet().iterator();
}
@Override
public void put(String key, String value) {
if (!StringUtils.hasText(value)) {
return;
}
this.delegate.getAttachments().put(key, value);
}
}
- DubboSpanMapper
public class DubboSpanMapper {
static final String HEADER_DELIMITER = "-";
static final String BAGGAGE_PREFIX = Span.SPAN_BAGGAGE_HEADER_PREFIX
+ HEADER_DELIMITER;
static final String URI_HEADER = "X-Span-Uri";
private static Comparator<String> IGNORE_CASE_COMPARATOR = new Comparator<String>() {
@Override
public int compare(String o1, String o2) {
return o1.toLowerCase().compareTo(o2.toLowerCase());
}
};
/**
* Acceptable span fields
*/
private static final Set<String> SPAN_FIELDS;
static {
TreeSet<String> fields = new TreeSet<>(IGNORE_CASE_COMPARATOR);
Collections.addAll(fields, Span.SPAN_FLAGS, Span.TRACE_ID_NAME, Span.SPAN_ID_NAME,
Span.PROCESS_ID_NAME, Span.SPAN_NAME_NAME, Span.PARENT_ID_NAME,
Span.SAMPLED_NAME, URI_HEADER);
SPAN_FIELDS = Collections.unmodifiableSet(fields);
}
/**
* Create new Map of carrier values
*/
Map<String, String> convert(SpanTextMap textMap) {
Map<String, String> carrier = new TreeMap<>(IGNORE_CASE_COMPARATOR);
for (Map.Entry<String, String> entry : textMap) {
if (isAcceptable(entry.getKey())) {
carrier.put(entry.getKey(), entry.getValue());
}
}
return Collections.unmodifiableMap(carrier);
}
private boolean isAcceptable(String key) {
return SPAN_FIELDS.contains(key) || key.startsWith(BAGGAGE_PREFIX);
}
}
网友评论