分布式调用追踪系统Zipkin集成

发表于 分类于

首先给出demo地址，有问题也可以直接看代码。https://github.com/licheng-xd/dubbo-sleuth

准备

sleuth是springcloud对zipkin接入的封装，从2.0开始直接采用了brave提供的zipkin接入实现。

由于sleuth对dubbo的支持是从2.0开始的，所以一咬牙直接把整个项目升级成了springboot2.0，springboot2.0和springboot1.x有很多不兼容的地方，这里就不一一细说了，同时springboot升级到2.0以后，dubbo也必须要从0.1.0升级到0.2.0（Apache旗下的dubbo版本），整个过踩坑踩了一个星期。

因为springcloud的微服务框架是基于http构建的，所以sleuth默认是只支持http。在2.0版本中提供了对dubbo的支持，其实看一下源码就知道，就只是一个DubboFilter。后面实现自己的rpc接入sleuth就是参考了它的实现。zipkin本身支持多种collector和storage，默认采用异步http的collector，存储默认在内存中。考虑到对应用本身的性能影响，我们采用kafka来做collector，最大程度的解耦以及减少性能影响。存储选用es，用mysql的话当数据量较大时会影响zipkin-server的查询速度。

本地测试环境部署：
本地采用vagrant管理开发环境，在debian的虚拟机中装好，zookeeper，kafka，es，其中es建议用docker镜像直接运行，不需要修改很多系统参数。

启动zipkinserver，采用kafka-collector和es-storage。

我本地的vagrant虚拟机ip是192.168.100.101，首先下载zipkin-server，启动命令：

1
KAFKA_BOOTSTRAP_SERVERS=192.168.100.101:9092 STORAGE_TYPE=elasticsearch ES_HOSTS=http://192.168.100.101:9200  java -jar zipkin-server-2.9.4-exec.jar

项目添加sleuth依赖

由于采用kafka作为collector所以需要原来spring-kafka，同时添加了brave-dubbo的依赖。

1
2
3
4
5
6
7
8
9
10
// sleuth
compile "org.springframework.kafka:spring-kafka:2.1.7.RELEASE"
compile "org.springframework.cloud:spring-cloud-starter-sleuth:${SleuthVersion}"
compile "org.springframework.cloud:spring-cloud-starter-zipkin:${SleuthVersion}"
compile "org.springframework.cloud:spring-cloud-sleuth-core:${SleuthVersion}"
compile "org.springframework.cloud:spring-cloud-sleuth-zipkin:${SleuthVersion}"
compile "org.springframework.cloud:spring-cloud-sleuth-dependencies:${SleuthVersion}"
compile ("io.zipkin.brave:brave-instrumentation-dubbo-rpc:5.1.0") {
    exclude(group: 'com.alibaba', module: 'dubbo')
}

配置

1
2
3
4
5
6
7
8
9
10
11
12
13
spring:
  application:
    name: nim-bima-service-user
  zipkin:
    sender:
      type: kafka
  sleuth:
    enabled: true
    traceId128: true
    sampler:
      probability: 1.0
  kafka:
    bootstrap-servers: xxx.xxx.xxx.xxx:9092,xxx.xxx.xxx.xxx:9092

dubbo服务和消费端都需要加上对应的tracing filter

1
2
3
4
5
dubbo:
  provider:
    filter: 'tracing'
  consumer:
    filter: 'tracing'

扩展

自研RPC通过filter接入sleuth追踪。filter逻辑的核心在于filterchain的实现，而filterchain的核心逻辑其实就是一个链式的递归调用：

1
2
3
4
5
6
7
8
if (this.currentPosition == this.realFilters.size()) {
    invocation.invoke();
}
else {
    this.currentPosition++;
    NrpcFilter nextFilter = this.realFilters.get(this.currentPosition - 1);
    nextFilter.doFilter(invocation, this);
}

然后通过实现filter来加入追踪功能

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
@FilterActivate
public class TracingNrpcFilter implements NrpcFilter {
    private static final Logger logger = LoggerFactory.getLogger(TracingNrpcFilter.class);

    private Tracer tracer;

    private TraceContext.Injector<Map<String, String>> injector;

    @Autowired
    public void setTracing(Tracing tracing) {
        tracer = tracing.tracer();
        injector = tracing.propagation().injector(SETTER);
    }

    @Override
    public void doFilter(TaskInvocation invocation, NrpcFilterChain filterChain) throws Exception {
        if (tracer == null) {
            filterChain.doFilter(invocation, filterChain);
            return;
        }

        logger.debug("NrpcTracingFilter hit ...");
        String className = invocation.getTask().getMethod().getDeclaringClass().getSimpleName();
        String methodName = invocation.getTask().getMethod().getName();
        final Span span = tracer.nextSpan();
        injector.inject(span.context(), invocation.getAttachments());
        if (!span.isNoop()) {
            span.kind(Span.Kind.SERVER).start();
            span.name(className + "/" + methodName);
            InetSocketAddress remoteAddress = invocation.getRemoteAddress();
            Endpoint.Builder remoteEndpoint = Endpoint.newBuilder().port(remoteAddress.getPort());
            if (!remoteEndpoint.parseIp(remoteAddress.getAddress())) {
                remoteEndpoint.parseIp(remoteAddress.getHostName());
            }
            span.remoteEndpoint(remoteEndpoint.build());
        }
        String traceid = span.context().traceIdString();
        DubboTraceFilter.initTraceid(traceid);
        invocation.getTask().setTraceid(traceid);
        try (Tracer.SpanInScope scope = tracer.withSpanInScope(span)) {
            filterChain.doFilter(invocation, filterChain);
        } catch (Error | RuntimeException e) {
            onError(e, span);
            throw e;
        } finally {
            span.finish();
        }
    }

    static void onError(Throwable error, Span span) {
        span.error(error);
        if (error instanceof RpcException) {
            span.tag("nrpc.error_code", Integer.toString(((RpcException) error).getCode()));
        }
    }

    static final Propagation.Setter<Map<String, String>, String> SETTER =
            new Propagation.Setter<Map<String, String>, String>() {
                @Override
                public void put(Map<String, String> carrier, String key, String value) {
                    carrier.put(key, value);
                }

                @Override
                public String toString() {
                    return "Map::set";
                }
            };
}

小技巧：通过tracer.currentSpan().context().traceIdString();把tracing中的traceid拿出来放到业务日志中，结合elk日志收集可以非常容易定位问题。

结果

可以从zipkin的数据分析中清楚的看出来整个请求中的rpc调用链，以及每次rpc消耗的时间。

依赖分析

你会发现依赖分析页面没有数据。对了，只有当数据在内存中存储的时候，zipkin-server才能直接分析出调用依赖关系。当采用外部存储的时候，就需要单独的依赖分析来实现了。下载zipkin-dependencies.jar并在crontab中创建定时任务，定时对存储数据进行分析，这样就可以看到依赖关系了。

奇怪的小问题

  1. 由于引入了springcloud-sleuth，而springcloud会在进程启动时启动一个自己的bootstrap context，作为当前应用ApplicationContext的父ApplicationContext，由于每初始化一个ApplicationContext就会加载一遍spring.factories配置文件中的ApplicationListener，所以配置在spring.factories中的listener都会被执行两次。因此dubbo的WelcomeLogoListener会被执行两次。

  2. 使用feign作为http客户端时，build传入的client必须是spring的bean，否则sleuth无法做拦截。