spring cloud sleuth源码分析

作者: ZMMWMY | 来源:发表于2018-04-11 17:32 被阅读176次

目前,微服务非常流行,有Dubbo,spring cloud等其他优秀框架。在当我们的服务到一定数量的时候,体积越来越庞大。这时候梳理服务之间的关系,定位问题变的十分困难。这时候就需要工具来对服务进行监控。其中有几个杰出的第三方框架

pinpoint

skywalking

zipkin

阿里的鹰眼,美团的CAT

我们使用的是zipkin,它是基于google发布的一篇Dapper论文来进行设计,对于它的介绍请看官网,这里介绍一下zikpin的客户端实现之spring cloud sleuth,另外它也有官网的java客户端Brave。列一下它的实现库。还有一点需要说明,最新版的spring cloud sleuth使用了Brave组件来实现,因此不需要担心存储的上下文(之前的版本塔门不同的上下文,所以会造成数据不能共用,说的直白一点就是一套系统中如果使用了Brave和Spring cloud sleuth,那你就会发现本来应该是一条链路分成了两段,我就体验过一把。。。)。

Starting with version 2.0.0, Spring Cloud Sleuth uses Brave as the tracing library. Consequently, Sleuth no longer takes care of storing the context but delegates that work to Brave.

这就很尴尬,因为我使用的不是最新版,而且它也没有出RELEASE版本。废话不多说,下面开始介绍一下spring cloud sleuth,它的代码量也不多。我使用的是1.2.0.RELEASE

自动配置

spring boot 一大亮点就是自动配置,只需要引入pom.xml加application.yml。(其实也需要一定文件的配合,那就是spring.factories)它就能加载相应的配置,spring boot的自动加载这里不介绍。我们来看代码

TraceAutoConfiguration

@Configuration
//会读取配置文件,默认是true,因此这个类在启动的时候就会加载了
@ConditionalOnProperty(value="spring.sleuth.enabled", matchIfMissing=true)
//读取配置类,并且注入spring bean
@EnableConfigurationProperties({TraceKeys.class, SleuthProperties.class})
public class TraceAutoConfiguration {
    @Autowired
    SleuthProperties properties;

    @Bean
    @ConditionalOnMissingBean
    //spanid生成策略,就是随机的
    public Random randomForSpanIds() {
        return new Random();
    }

    //是否需要传输到zipkin服务端
    @Bean
    @ConditionalOnMissingBean
    public Sampler defaultTraceSampler() {
        return NeverSampler.INSTANCE;
    }

    @Bean
    @ConditionalOnMissingBean(Tracer.class)
    /**
    *sampler 这个类,如果没有引入spring-cloud-sleuth-zikpin,它就是NeverSampler
    * 相反,它会读取spring.sleuth.sampler.percentage来构造一个百分比
    * spanReporter一样道理
    * Tracer：这是一个非常关键的类,下面单独讲一下。就是通过它来进行一系列的创建span,保存,更新等
    */
    public DefaultTracer sleuthTracer(Sampler sampler, Random random,
            SpanNamer spanNamer, SpanLogger spanLogger,
            SpanReporter spanReporter, TraceKeys traceKeys) {
        return new DefaultTracer(sampler, random, spanNamer, spanLogger,
                spanReporter, this.properties.isTraceId128(), traceKeys);
    }

    @Bean
    @ConditionalOnMissingBean
    public SpanNamer spanNamer() {
        return new DefaultSpanNamer();
    }

    @Bean
    @ConditionalOnMissingBean
    public SpanReporter defaultSpanReporter() {
        return new NoOpSpanReporter();
    }

}

TraceWebAutoConfiguration
这个另外一个关键的类,在一般情况下我们都是使用web来进行调用,其他的拓展这里也不讲了,有兴趣可以自己了解一下,大致思想类似

@Configuration
//默认自动加载
//这里一个重点就是加载了Filter和skip的类
@ConditionalOnProperty(value = "spring.sleuth.web.enabled", matchIfMissing = true)
@ConditionalOnWebApplication
@ConditionalOnBean(Tracer.class)
@AutoConfigureAfter(TraceHttpAutoConfiguration.class)
public class TraceWebAutoConfiguration {

    /**
     * Nested config that configures Web MVC if it's present (without adding a runtime
     * dependency to it)
     */
    @Configuration
    @ConditionalOnClass(WebMvcConfigurerAdapter.class)
    @Import(TraceWebMvcConfigurer.class)
    protected static class TraceWebMvcAutoConfiguration {
    }

    @Bean
    public TraceWebAspect traceWebAspect(Tracer tracer, TraceKeys traceKeys,
            SpanNamer spanNamer) {
        return new TraceWebAspect(tracer, spanNamer, traceKeys);
    }

    @Bean
    @ConditionalOnClass(name = "org.springframework.data.rest.webmvc.support.DelegatingHandlerMapping")
    public TraceSpringDataBeanPostProcessor traceSpringDataBeanPostProcessor(
            BeanFactory beanFactory) {
        return new TraceSpringDataBeanPostProcessor(beanFactory);
    }

    /**
    *创建并注册一个Filter,web环境下都会进入filter来
    */
    @Bean
    public FilterRegistrationBean traceWebFilter(TraceFilter traceFilter) {
        FilterRegistrationBean filterRegistrationBean = new FilterRegistrationBean(
                traceFilter);
        filterRegistrationBean.setDispatcherTypes(ASYNC, ERROR, FORWARD, INCLUDE,
                REQUEST);
        filterRegistrationBean.setOrder(TraceFilter.ORDER);
        return filterRegistrationBean;
    }

    @Bean
    public TraceFilter traceFilter(Tracer tracer, TraceKeys traceKeys,
            SkipPatternProvider skipPatternProvider, SpanReporter spanReporter,
            HttpSpanExtractor spanExtractor,
            HttpTraceKeysInjector httpTraceKeysInjector) {
        return new TraceFilter(tracer, traceKeys, skipPatternProvider.skipPattern(),
                spanReporter, spanExtractor, httpTraceKeysInjector);
    }

    @Configuration
    @ConditionalOnClass(ManagementServerProperties.class)
    @ConditionalOnMissingBean(SkipPatternProvider.class)
    @EnableConfigurationProperties(SleuthWebProperties.class)
    //skip：正则表达式检测进来的请求是否需要将span 传输到服务端去。
    //（下面实现代码可以不看）
    protected static class SkipPatternProviderConfig {

        @Bean
        @ConditionalOnBean(ManagementServerProperties.class)
        public SkipPatternProvider skipPatternForManagementServerProperties(
                final ManagementServerProperties managementServerProperties,
                final SleuthWebProperties sleuthWebProperties) {
            return new SkipPatternProvider() {
                @Override
                public Pattern skipPattern() {
                    return getPatternForManagementServerProperties(
                            managementServerProperties,
                            sleuthWebProperties);
                }
            };
        }

        /**
         * Sets or appends {@link ManagementServerProperties#getContextPath()} to the skip
         * pattern. If neither is available then sets the default one
         */
        static Pattern getPatternForManagementServerProperties(
                ManagementServerProperties managementServerProperties,
                SleuthWebProperties sleuthWebProperties) {
            String skipPattern = sleuthWebProperties.getSkipPattern();
            if (StringUtils.hasText(skipPattern)
                    && StringUtils.hasText(managementServerProperties.getContextPath())) {
                return Pattern.compile(skipPattern + "|"
                        + managementServerProperties.getContextPath() + ".*");
            }
            else if (StringUtils.hasText(managementServerProperties.getContextPath())) {
                return Pattern
                        .compile(managementServerProperties.getContextPath() + ".*");
            }
            return defaultSkipPattern(skipPattern);
        }

        @Bean
        @ConditionalOnMissingBean(ManagementServerProperties.class)
        public SkipPatternProvider defaultSkipPatternBeanIfManagementServerPropsArePresent(SleuthWebProperties sleuthWebProperties) {
            return defaultSkipPatternProvider(sleuthWebProperties.getSkipPattern());
        }
    }

    @Bean
    @ConditionalOnMissingClass("org.springframework.boot.actuate.autoconfigure.ManagementServerProperties")
    @ConditionalOnMissingBean(SkipPatternProvider.class)
    public SkipPatternProvider defaultSkipPatternBean(SleuthWebProperties sleuthWebProperties) {
        return defaultSkipPatternProvider(sleuthWebProperties.getSkipPattern());
    }

    private static SkipPatternProvider defaultSkipPatternProvider(
            final String skipPattern) {
        return new SkipPatternProvider() {
            @Override
            public Pattern skipPattern() {
                return defaultSkipPattern(skipPattern);
            }
        };
    }

    private static Pattern defaultSkipPattern(String skipPattern) {
        return StringUtils.hasText(skipPattern) ? Pattern.compile(skipPattern)
                : Pattern.compile(SleuthWebProperties.DEFAULT_SKIP_PATTERN);
    }

    interface SkipPatternProvider {
        Pattern skipPattern();
    }
}

TraceHttpAutoConfiguration
这个类的用途是将span信息注入到carrier（这里是http）,进行传递作用

@Configuration
@ConditionalOnBean(Tracer.class)
@AutoConfigureAfter(TraceAutoConfiguration.class)
@EnableConfigurationProperties({ TraceKeys.class, SleuthWebProperties.class })
public class TraceHttpAutoConfiguration {

    @Bean
    @ConditionalOnMissingBean
    public HttpTraceKeysInjector httpTraceKeysInjector(Tracer tracer, TraceKeys traceKeys) {
        return new HttpTraceKeysInjector(tracer, traceKeys);
    }

    @Bean
    @ConditionalOnMissingBean
    public HttpSpanExtractor httpSpanExtractor(SleuthWebProperties sleuthWebProperties) {
        return new ZipkinHttpSpanExtractor(Pattern.compile(sleuthWebProperties.getSkipPattern()));
    }

    @Bean
    @ConditionalOnMissingBean
    public HttpSpanInjector httpSpanInjector() {
        return new ZipkinHttpSpanInjector();
    }
}

链路监控的实现

TraceFilter 直接看关键方法
1. createSpan()

    /**
     * Creates a span and appends it as the current request's attribute
     */
    private Span createSpan(HttpServletRequest request,
            boolean skip, Span spanFromRequest, String name) {
        if (spanFromRequest != null) {
            if (log.isDebugEnabled()) {
                log.debug("Span has already been created - continuing with the previous one");
            }
            return spanFromRequest;
        }
        //从请求中获取信息,上一步是否有信息传入,有就进行解析
        //
        Span parent = this.spanExtractor.joinTrace(new HttpServletRequestTextMap(request));
        if (parent != null) {
            if (log.isDebugEnabled()) {
                log.debug("Found a parent span " + parent + " in the request");
            }
            addRequestTagsForParentSpan(request, parent);
            spanFromRequest = parent;
            //更新当前线程的Span
            this.tracer.continueSpan(spanFromRequest);
            if (parent.isRemote()) {
            //记录当前步骤SR
                parent.logEvent(Span.SERVER_RECV);
            }
            request.setAttribute(TRACE_REQUEST_ATTR, spanFromRequest);
            if (log.isDebugEnabled()) {
                log.debug("Parent span is " + parent + "");
            }
        } else {
            //carrier中没有span信息
            if (skip) {
                 //不需要上传
                spanFromRequest = this.tracer.createSpan(name, NeverSampler.INSTANCE);
            }
            else {
                String header = request.getHeader(Span.SPAN_FLAGS);
                if (Span.SPAN_SAMPLED.equals(header)) {
                    spanFromRequest = this.tracer.createSpan(name, new AlwaysSampler());
                } else {
                    //创建一个新的Span,代码就不贴了,就是random产生Id,然后放到当前线程中
                    spanFromRequest = this.tracer.createSpan(name);
                }
            }
            spanFromRequest.logEvent(Span.SERVER_RECV);
            request.setAttribute(TRACE_REQUEST_ATTR, spanFromRequest);
            if (log.isDebugEnabled()) {
                log.debug("No parent span present - creating a new span");
            }
        }
        return spanFromRequest;
    }

    /**
    * 构造一个ParentSpan
    *
    /
    @Override
    public Span joinTrace(SpanTextMap textMap) {
        Map<String, String> carrier = TextMapUtil.asMap(textMap);
        boolean debug = Span.SPAN_SAMPLED.equals(carrier.get(Span.SPAN_FLAGS));
        if (debug) {
            // we're only generating Trace ID since if there's no Span ID will assume
            // that it's equal to Trace ID
            generateIdIfMissing(carrier, Span.TRACE_ID_NAME);
        } else if (carrier.get(Span.TRACE_ID_NAME) == null) {
            // can't build a Span without trace id
            return null;
        }
        try {
            String uri = carrier.get(URI_HEADER);
            boolean skip = this.skipPattern.matcher(uri).matches()
                    || Span.SPAN_NOT_SAMPLED.equals(carrier.get(Span.SAMPLED_NAME));
            long spanId = spanId(carrier);
            return buildParentSpan(carrier, uri, skip, spanId);
        } catch (Exception e) {
            log.error("Exception occurred while trying to extract span from carrier", e);
            return null;
        }
    }
    
    private Span buildParentSpan(Map<String, String> carrier, String uri, boolean skip, long spanId) {
        String traceId = carrier.get(Span.TRACE_ID_NAME);
        Span.SpanBuilder span = Span.builder()
                .traceIdHigh(traceId.length() == 32 ? Span.hexToId(traceId, 0) : 0)
                .traceId(Span.hexToId(traceId))
                .spanId(spanId);
        String processId = carrier.get(Span.PROCESS_ID_NAME);
        String parentName = carrier.get(Span.SPAN_NAME_NAME);
        if (StringUtils.hasText(parentName)) {
            span.name(parentName);
        }  else {
            span.name(HTTP_COMPONENT + ":/parent" + uri);
        }
        if (StringUtils.hasText(processId)) {
            span.processId(processId);
        }
        if (carrier.containsKey(Span.PARENT_ID_NAME)) {
            span.parent(Span.hexToId(carrier.get(Span.PARENT_ID_NAME)));
        }
        span.remote(true);
        boolean debug = Span.SPAN_SAMPLED.equals(carrier.get(Span.SPAN_FLAGS));
        //是否要上传span
        if (debug) {
            span.exportable(true);
        } else if (skip) {
            span.exportable(false);
        }
        for (Map.Entry<String, String> entry : carrier.entrySet()) {
            if (entry.getKey().startsWith(Span.SPAN_BAGGAGE_HEADER_PREFIX + HEADER_DELIMITER)) {
                span.baggage(unprefixedKey(entry.getKey()), entry.getValue());
            }
        }
        return span.build();
    }

addErrorTag()

//如果请求期间发生了异常,将异常信息记录到Span中
catch (Throwable e) {
            exception = e;
            this.tracer.addTag(Span.SPAN_ERROR_TAG_NAME, ExceptionUtils.getExceptionMessage(e));
            throw e;
        }

closeSpan() 关闭Span并上传

finally {
            if (isAsyncStarted(request) || request.isAsyncStarted()) {
                if (log.isDebugEnabled()) {
                    log.debug("The span " + spanFromRequest + " will get detached by a HandleInterceptor");
                }
                // TODO: how to deal with response annotations and async?
                return;
            }
            spanFromRequest = createSpanIfRequestNotHandled(request, spanFromRequest, name, skip);
            detachOrCloseSpans(request, response, spanFromRequest, exception);
        }
        
        /**
        * 
        */
        private void recordParentSpan(Span parent) {
        if (parent == null) {
            return;
        }
        if (parent.isRemote()) {
            if (log.isDebugEnabled()) {
                log.debug("Trying to send the parent span " + parent + " to Zipkin");
            }
            parent.stop();
            // should be already done by HttpServletResponse wrappers
            SsLogSetter.annotateWithServerSendIfLogIsNotAlreadyPresent(parent);
            this.spanReporter.report(parent);
        } else {
            // should be already done by HttpServletResponse wrappers
            SsLogSetter.annotateWithServerSendIfLogIsNotAlreadyPresent(parent);
        }
    }
    
    /**
    *
    */
    @Override
    public Span close(Span span) {
        if (span == null) {
            return null;
        }
        Span cur = SpanContextHolder.getCurrentSpan();
        final Span savedSpan = span.getSavedSpan();
        if (!span.equals(cur)) {
            ExceptionUtils.warn(
                    "Tried to close span but it is not the current span: " + span
                            + ".  You may have forgotten to close or detach " + cur);
        }
        else {
          //统计Span存在时间,也就是调用时间
            span.stop();
            if (savedSpan != null && span.getParents().contains(savedSpan.getSpanId())) {
                this.spanReporter.report(span);
                this.spanLogger.logStoppedSpan(savedSpan, span);
            }
            else {
                if (!span.isRemote()) {
                    //上传span,这是spring-sleuth-zikpin的活
                    this.spanReporter.report(span);
                    this.spanLogger.logStoppedSpan(null, span);
                }
            }
            //移除当前线程的Span
            SpanContextHolder.close(new SpanContextHolder.SpanFunction() {
                @Override public void apply(Span span) {
                    DefaultTracer.this.spanLogger.logStoppedSpan(savedSpan, span);
                }
            });
        }
        return savedSpan;
    }

TraceFeignClient 这里的代码很好理解,就是创建Span,并将信息传入到carrier
其他Client的实现不讲了,挑一个Feign,因为都差不多

    @Override
    public Response execute(Request request, Request.Options options) throws IOException {
        String spanName = getSpanName(request);
        Span span = getTracer().createSpan(spanName);
        if (log.isDebugEnabled()) {
            log.debug("Created new Feign span " + span);
        }
        try {
            AtomicReference<Request> feignRequest = new AtomicReference<>(request);
            spanInjector().inject(span, new FeignRequestTextMap(feignRequest));
            span.logEvent(Span.CLIENT_SEND);
            addRequestTags(request);
            Request modifiedRequest = feignRequest.get();
            if (log.isDebugEnabled()) {
                log.debug("The modified request equals " + modifiedRequest);
            }
            Response response = this.delegate.execute(modifiedRequest, options);
            logCr();
            return response;
        } catch (RuntimeException | IOException e) {
            logCr();
            logError(e);
            throw e;
        } finally {
            //关闭Span,并上传
            closeSpan(span);
        }
    }

总结

spring cloud 实现链路追踪非常简单,只需要引入对应的pom包即可,如果想到展示这些链路,也只需要引入spring-sleuth-zipkin的包。还有一点非常好用,那就是它引入了Log,在生成和销毁Span的同时,也对TraceID,SpanID进行了记录,如果需要展示这些信息在日志中只需要在项目中多引入一个property。如果集成了ELK,想要发送日志到日志平台上去添加一个Appender。官网上有例子。

<property name="CONSOLE_LOG_PATTERN"
              value="%clr(%d{yyyy-MM-dd HH:mm:ss.SSS}){faint} %clr(${LOG_LEVEL_PATTERN:-%5p}) %clr([${springAppName:-},%X{X-B3-TraceId:-},%X{X-B3-SpanId:-},%X{X-Span-Export:-}]){yellow} %clr(${PID:- }){magenta} %clr(---){faint} %clr([%15.15t]){faint} %clr(%-40.40logger{39}){cyan} %clr(:){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:-%wEx}"/>

关于注解的部分，下次有空会将

spring cloud sleuth源码分析

自动配置

链路监控的实现

总结

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

分布式追踪系统

我爱编程

spring cloud