使用opentelemetry 搭建新一代可视化分布式监控系统_getotlpprocessor()方法

1、分布式监控系统介绍

随着SOA，微服务架构及PaaS，Devops等技术的兴起，线上问题的追踪和排查变得更加困难。对线上业务的可观测性得到了越来越多企业的重视，由此涌现出了许多优秀的链路追踪及服务监控中间件。比较流行的有Spring Cloud全家桶自带的Zipkin，点评的CAT, 华为的skywalking，Uber的Jaeger, naver的Pinpoint。

一个典型的应用，通常有三种类型的数据需要被监控系统记录：Metric, logs and traces。让我们先了解下它们都是什么。

Metrics

提供进行运行时的指标信息。比如CPU使用率，内存使用情况，GC情况，网站流量等。

Logging

可以监控程序进程中的日志，比如集成Log4j记录的日志，或者程序运行中发生的事件或通知。

Tracing

也叫做分布式追踪，包含请求中每个子操作的开始和结束时间，传递的参数，请求间的调用链路，请求在各个链路上的耗时等信息。Tracing可以包含消息发送和接收，数据库访问，负载均衡等各种信息，让我们可以深入了解请求的执行情况。Tracing为我们提供了获取请求的时间主要消耗在哪里，请求的参数都是什么，如果发生了异常，那么异常是在哪个环节产生的等能力。

2、opentelemetry简介

opentelemetry是一款数据收集中间件。我们可以使用它来生成，收集和导出监测数据（Metrics,Logs and traces），这些数据可供支持OpenTelemetry的中间件存储，查询和显示，用以实现数据观测，性能分析，系统监控，服务告警等能力。

opentelemetry项目开始于2019年，旨在提供基于云环境的可观测性软件的标准化方案，提供与三方无关的监控服务体系。项目迄今为止已获得了Zipkin, Jaeger, skywalking, Prometheus等众多知名中间件的支持。

3、sample项目

本例中，我们使用spring cloud搭建一个简单的微服务，来体验下如何使用opentelemetry来进行系统监控，并在两个不同的监控系统（Zipkin，Jaeger）进行快速切换。项目由2个微服务，2个可视化监控系统，并使用opentelemetry 来集成微服务和监控系统。

• gateway-service -使用spring cloud gateway搭建的服务网关
• cloud-user-service -用户微服务,使用Spring boot + spring mvc
• Zipkin - Zipkin监控系统服务端
• Jaeger - Jaeger监控系统服务端
  

4、使用opentelemetry 集成Zipkin

示例中使用到的组件的版本：
java: 1.8
spring-cloud: 2020.0.2
spring-boot: 2.4.5
opentelemetry: 1.1.0
grpc: 1.36.1

4.1、cloud-user-service服务maven配置

引入Spring cloud 和 opentelemetry

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-dependencies</artifactId>
            <version>${spring-cloud.version}</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
        <dependency>
            <groupId>io.opentelemetry</groupId>
            <artifactId>opentelemetry-bom</artifactId>
            <version>${opentelemetry.version}</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

加入opentelemetry依赖项

<dependency>
    <groupId>io.opentelemetry</groupId>
    <artifactId>opentelemetry-api</artifactId>
</dependency>
<dependency>
    <groupId>io.opentelemetry</groupId>
    <artifactId>opentelemetry-sdk</artifactId>
</dependency>
<dependency>
    <groupId>io.opentelemetry</groupId>
    <artifactId>opentelemetry-exporter-otlp</artifactId>
</dependency>
<dependency>
    <groupId>io.opentelemetry</groupId>
    <artifactId>opentelemetry-semconv</artifactId>
    <version>1.1.0-alpha</version>
</dependency>
<dependency>
    <groupId>io.grpc</groupId>
    <artifactId>grpc-protobuf</artifactId>
    <version>${grpc.version}</version>
</dependency>
<dependency>
    <groupId>io.grpc</groupId>
    <artifactId>grpc-netty-shaded</artifactId>
    <version>${grpc.version}</version>
</dependency>
<dependency>
    <groupId>io.opentelemetry</groupId>
    <artifactId>opentelemetry-exporter-zipkin</artifactId>
</dependency>
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

4.2、配置opentelemetry

@Configuration
public class TraceConfig {
    private static final String ENDPOINT_V2_SPANS = "/api/v2/spans";
    private final AppConfig appConfig;

    @Autowired
    public TraceConfig(AppConfig appConfig) {
        this.appConfig = appConfig;
    }

    @Bean
    public OpenTelemetry openTelemetry() {
        SpanProcessor spanProcessor = getOtlpProcessor();
        Resource serviceNameResource = Resource.create(Attributes.of(ResourceAttributes.SERVICE_NAME, appConfig.getApplicationName()));

        // Set to process the spans by the Zipkin Exporter
        SdkTracerProvider tracerProvider =
                SdkTracerProvider.builder()
                        .addSpanProcessor(spanProcessor)
                        .setResource(Resource.getDefault().merge(serviceNameResource))
                        .build();
        OpenTelemetrySdk openTelemetry =
                OpenTelemetrySdk.builder().setTracerProvider(tracerProvider)
                        .setPropagators(ContextPropagators.create(W3CTraceContextPropagator.getInstance()))
                        .buildAndRegisterGlobal();

        // add a shutdown hook to shut down the SDK
        Runtime.getRuntime().addShutdownHook(new Thread(tracerProvider::close));

        // return the configured instance so it can be used for instrumentation.
        return openTelemetry;
    }

    private SpanProcessor getZipkinProcessor() {
        String host = "localhost";
        int port = 9411;
        String httpUrl = String.format("http://%s:%s", host, port);
        ZipkinSpanExporter zipkinExporter = ZipkinSpanExporter.builder().setEndpoint(httpUrl + ENDPOINT_V2_SPANS).build();
        return SimpleSpanProcessor.create(zipkinExporter);
    }
}
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

4.3、在cloud-user-service中，使用opentelemetry

当我们完成了配置后，就可以在spring boot项目中，通过autowired来使用opentelemetry。

接下来我们定制一个WebFilter来拦截所有的Http请求，并在Filter类中进行埋点。

@Component
public class TracingFilter implements Filter {
    private final AppConfig appConfig;
    private final OpenTelemetry openTelemetry;

    @Autowired
    public TracingFilter(AppConfig appConfig, OpenTelemetry openTelemetry) {
        this.appConfig = appConfig;
        this.openTelemetry = openTelemetry;
    }

    @Override
    public void doFilter(ServletRequest servletRequest, ServletResponse servletResponse, FilterChain filterChain) throws IOException, ServletException {
        HttpServletRequest httpServletRequest = (HttpServletRequest)servletRequest;
        Span span = getServerSpan(openTelemetry.getTracer(appConfig.getApplicationName()), httpServletRequest);
        try (Scope scope = span.makeCurrent()) {
            filterChain.doFilter(servletRequest, servletResponse);
        } catch (Exception ex) {
            span.setStatus(StatusCode.ERROR, "HTTP Code: " + ((HttpServletResponse)servletResponse).getStatus());
            span.recordException(ex);
            throw ex;
        } finally {
            span.end();
        }
    }

    private Span getServerSpan(Tracer tracer, HttpServletRequest httpServletRequest) {

        TextMapPropagator textMapPropagator = openTelemetry.getPropagators().getTextMapPropagator();
        Context context = textMapPropagator.extract(Context.current(), httpServletRequest, new TextMapGetter<HttpServletRequest>() {
            @Override
            public Iterable<String> keys(HttpServletRequest request) {
                List<String> headers = new ArrayList();
                for (Enumeration names = request.getHeaderNames(); names.hasMoreElements();) {
                    String name = (String)names.nextElement();
                    headers.add(name);
                }
                return headers;
            }

            @Override
            public String get(HttpServletRequest request, String s) {
                return request.getHeader(s);
            }
        });


        return tracer.spanBuilder(httpServletRequest.getRequestURI()).setParent(context).setSpanKind(SpanKind.SERVER).setAttribute(SemanticAttributes.HTTP_METHOD, httpServletRequest.getMethod()).startSpan();
    }
}
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

在示例代码中，我们实现了一个匿名类来从HttpServletRequest中解析tracing上下文信息。

在创建Span的同时，我们在Span中写入了Http请求的一些关键属性，并且为所有的异常做了跟踪记录。

4.4、编写服务代码

接下来我们通过一段简单的代码来模拟查询用户以及抛出异常

@GetMapping("/{id}")
public ResponseEntity<User> get(@PathVariable("id") Long id) {
    if (0 >= id) {
        throw new IllegalArgumentException("Illegal argument value");
    }
    return ResponseEntity.ok(userService.get(id));
}
1
2
3
4
5
6
7

4.5、配置gateway-service

我们使用和cloud-user-service同样的配置来配置gateway-service。

##4.6、在gateway-service中，集成opentelemetry
这里和cloud-user-service有些不同，由于gateway-service是基于webflux构建的。我们这次使用WebFilter和GlobalFilter来拦截网关上的http请求。

在WebFilter中，添加opentelemetry来记录收到的http请求

@Override
public Mono<Void> filter(ServerWebExchange serverWebExchange, WebFilterChain webFilterChain) {
    ServerHttpRequest serverHttpRequest = serverWebExchange.getRequest();
    Span span = getServerSpan(openTelemetry.getTracer(appConfig.getApplicationName()), serverHttpRequest);
    Scope scope = span.makeCurrent();

    serverWebExchange.getResponse().getHeaders().add("traceId", span.getSpanContext().getTraceId());
    span.setAttribute("params", serverHttpRequest.getQueryParams().toString());

    return webFilterChain.filter(serverWebExchange)
            .doFinally((signalType) -> {
                scope.close();
                span.end();
            })
            .doOnError(span::recordException);
}

private Span getServerSpan(Tracer tracer, ServerHttpRequest serverHttpRequest) {
    return tracer.spanBuilder(serverHttpRequest.getPath().toString()).setNoParent().setSpanKind(SpanKind.SERVER).setAttribute(SemanticAttributes.HTTP_METHOD, serverHttpRequest.getMethod().name()).startSpan();
}
接下来在GlobalFilter中，记录路由到微服务的http请求

@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain gatewayFilterChain) {
    Span span = getClientSpan(openTelemetry.getTracer(appConfig.getApplicationName()), exchange);
    Scope scope = span.makeCurrent();
    inject(exchange);
    return gatewayFilterChain.filter(exchange)
            .then(Mono.fromRunnable(() -> {
                        scope.close();
                        span.end();
                    })
            );
}

private void inject(ServerWebExchange serverWebExchange) {
    HttpHeaders httpHeaders = new HttpHeaders();
    TextMapPropagator textMapPropagator = openTelemetry.getPropagators().getTextMapPropagator();
    textMapPropagator.inject(Context.current(), httpHeaders, HttpHeaders::add);
    ServerHttpRequest request = serverWebExchange.getRequest().mutate()
            .headers(headers -> headers.addAll(httpHeaders))
            .build();
    serverWebExchange.mutate().request(request).build();
}

private Span getClientSpan(Tracer tracer, ServerWebExchange serverWebExchange) {
    ServerHttpRequest serverHttpRequest = serverWebExchange.getRequest();
    URI routeUri = serverWebExchange.getAttribute(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR);
    return tracer.spanBuilder(routeUri.getPath()).setSpanKind(SpanKind.CLIENT).setAttribute(SemanticAttributes.HTTP_METHOD, serverHttpRequest.getMethod().name()).startSpan();
}
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

为了传递tracing的上下文信息，我们需要调用inject方法，把tracing上下文信息写入到路由请求的头信息里面。

5、运行服务

现在，让我们访问网关http://localhost:8080/user/0 来观察Zipkin对于服务访问和异常的记录情况。

可以看到在Tracing方面，Zikin整体表现还不错，有异常的链路也使用红色做了标记。Zipkin没有打印出异常的堆栈信息，我们需要为此做额外的处理才行。

6、使用Jaeger对接opentelemetry

使用otlp exporter来替换之前使用的zipkin exporter。

<dependency>
    <groupId>io.opentelemetry</groupId>
    <artifactId>opentelemetry-exporter-otlp</artifactId>
</dependency>
1
2
3
4

在配置类中，使用otlp processor替换之前的zipkin processor。这样就完成了Zipkin到Jaeger的切换。

private SpanProcessor getOtlpProcessor(){
    OtlpGrpcSpanExporter spanExporter = OtlpGrpcSpanExporter.builder().setTimeout(2, TimeUnit.SECONDS).build();
    return BatchSpanProcessor.builder(spanExporter)
            .setScheduleDelay(100, TimeUnit.MILLISECONDS)
            .build();
}
1
2
3
4
5
6

7、再次运行服务

我们再次运行服务并访问网关http://localhost:8080/user/0 来观察Jaeger对于服务访问和异常的记录情况。

首先看主界面，Jaeger直接标记了请求中包含异常。

再看下访问的详情，Jaeger记录并显示了异常的堆栈信息。这对我们分析线上异常非常有帮助。

对比Zipkin，Jaeger提供了更加丰富的功能和更美观的可视化界面。

8、总结

本文介绍了使用opentelemetry 来搭建监控系统，以及如何集成到Zipkin和Jaeger。

利用opentelemetry的标准化能力，我们可以方便地记录更加详细的链路监控信息。

opentelemetry自推出以来，得到了越来越多厂商的关注和支持。对于分布式监控系统这个新生事物，opentelemetry是否能成为最终的事实标准，让我们拭目以待。

References

opentelemetry官网

Metrics, tracing, and logging

5 Reasons why OpenTelemetry will boost Observability and Monitoring

Exporting Open Telemetry Data to Jaeger