蚂蚁金服分布式链路组件 SOFATracer 埋点机制解析 | SOFAChannel#15 直播整理数据库m046945278的博客-

和标准 Servlet 规范实现一样，所有基于 javax.sql.DataSource 实现的 DataSource 均可以使用 SOFATracer 进行埋点。因为 DataSource 并没有提供像 Servlet 那样的过滤器或者拦截器，所以 SOFATracer 中没法直接通过常规的方式（Filter/SPI 扩展拦截/拦截器等）进行埋点，而是使用了代理模式的方式来实现的。

上图为 SOFATracer 中 DataSource 代理类实现的类继承结构体系；可以看出，SOFATracer 中自定义了一个 BaseDataSource 抽象类，该抽象类继承 javax.sql.DataSource 接口，SmartDataSource 作为 BaseDataSource 的唯一子类，也就是 SOFATracer 中所使用的代理类。所以如果你使用了 sofa-tracer-datasource-plugin 插件的话，可以看到最终运行时的 Datasource 类型是 com.alipay.sofa.tracer.plugins.datasource.SmartDataSource。

 public abstract class BaseDataSource implements DataSource {     // 实际被代理的 datasource     protected DataSource        delegate;     //  sofatracer 中自定义的拦截器，用于对连接操作、db操作等进行拦截埋点     protected List<Interceptor> interceptors;     protected List<Interceptor> dataSourceInterceptors; }

Interceptor 主要包括以下三种类型：

以 StatementTracerInterceptor 为例 StatementTracerInterceptor 将将会拦截到所有 PreparedStatement 接口的方法，代码如下：

 public class StatementTracerInterceptor implements Interceptor {     // tracer 类型为 client      private DataSourceClientTracer clientTracer;     public void setClientTracer(DataSourceClientTracer clientTracer) {         // tracer 对象实例         this.clientTracer = clientTracer;     }      @Override     public Object intercept(Chain chain) throws Exception {         // 记录当前系统时间         long start = System.currentTimeMillis();         String resultCode = SofaTracerConstant.RESULT_SUCCESS;         try {             // 开始一个 span             clientTracer.startTrace(chain.getOriginalSql());             // 执行             return chain.proceed();         } catch (Exception e) {             resultCode = SofaTracerConstant.RESULT_FAILED;             throw e;         } finally {             // 这里计算执行时间 System.currentTimeMillis() - start             // 结束一个 span             clientTracer.endTrace(System.currentTimeMillis() - start, resultCode);         }     } }

总体思路是，DataSource 通过组合的方式自定义一个代理类（实际上也可以理解为适配器模式中的对象适配模型方式），对所有目标对象的方式进行代理拦截，在执行具体的 SQL 或者连接操作之前创建 DataSource 的 Span，在操作结束之后结束 Span，并进行上报。

消息埋点

消息框架组件包括很多，像常见的 RocketMQ、Kafka 等；除了各个组件自己提供的客户端之外，像 Spring 就提供了很多消息组件的封装，包括 Spring Cloud Stream、Spring Integration、Spring Message 等等。SOFATracer 基于 Spring Message 标准实现了对常见消息组件和 Spring Cloud Stream 的埋点支持，同时也提供了基于 RocketMQ 客户端模式的埋点实现。

Spring Messaging 埋点实现原理

spring-messaging 模块为集成 Messaging API 和消息协议提供支持。这里我们先看一个 pipes-and-filters 架构模型：

spring-messaging 的 support 模块中提供了各种不同的 Message Channel 实现和 Channel Interceptor 支持，因此在对 spring-messaging 进行埋点时我们自然就会想到去使用 Channel Interceptor。

 // SOFATracer 实现的基于 spring-messaging 消息拦截器 public class SofaTracerChannelInterceptor implements ChannelInterceptor, ExecutorChannelInterceptor {     // todo trace }  // THIS IS ChannelInterceptor public interface ChannelInterceptor {     // 发送之前     @Nullable     default Message<?> preSend(Message<?> message, MessageChannel channel) {         return message;     }     // 发送后     default void postSend(Message<?> message, MessageChannel channel, boolean sent) {     }     // 完成发送之后     default void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent, @Nullable Exception ex) {     }     // 接收消息之前     default boolean preReceive(MessageChannel channel) {         return true;     }     // 接收后     @Nullable     default Message<?> postReceive(Message<?> message, MessageChannel channel) {         return message;     }     // 完成接收消息之后     default void afterReceiveCompletion(@Nullable Message<?> message, MessageChannel channel, @Nullable Exception ex) {     } }

可以看到 ChannelInterceptor 实现了消息传递全生命周期的管控，通过暴露出来的方法，可以轻松的实现各个阶段的扩展埋点。

RocketMQ 埋点实现原理

RocketMQ 本身是提供了对 Opentracing 规范支持的，由于其支持的版本较高，与 SOFATracer 所实现的 Opentracing 版本不一致，所以在一定程度上不兼容；因此 SOFATracer（opentracing 0.22.0 版本）自身又单独提供了 RocketMQ  的插件。

RocketMQ 埋点其实是通过两个 hook 接口来完成，实际上在 RocketMQ 的官方文档中貌似并没有提到这两个点。

 // RocketMQ 消息消费端 hook 接口埋点实现类 public class SofaTracerConsumeMessageHook implements ConsumeMessageHook { } // RocketMQ 消息发送端 hook 接口埋点实现类 public class SofaTracerSendMessageHook implements SendMessageHook {}

首先是 SendMessageHook 接口，SendMessageHook 接口提供了两个方法，sendMessageBefore 和 sendMessageAfter，SOFATracer 在实现埋点时，sendMessageBefore 中用来解析和构建 Span，sendMessageAfter 中用于拿到结果然后结束 Span。

同样的，ConsumeMessageHook 中也提供了两个方法（consumeMessageBefore 和 consumeMessageAfter），可以提供给 SOFATracer 来从消息中解析出透传的 SOFATracer 信息然后再将 SOFATracer 信息透传到下游链路中去。

Redis 埋点原理

SOFATracer 中的 Redis 埋点是基于 spring-data-redis 实现的，没有针对具体的 Redis 客户端来埋点。另外  Redis 埋点部分参考的是开源社区 opentracing-spring-cloud-redis-starter 中的实现逻辑。

Redis 的埋点实现与 DataSource 的埋点实现基本思路是一致的，都是通过一层代理来是实现的拦截。sofa-tracer-redis-plugin 中对所有的 Redis 操作都通过 RedisActionWrapperHelper 进行了一层包装，在执行具体的命令前后通过 SOFATracer 自己提供的 API 进行埋点操作。代码如下：

 public <T> T doInScope(String command, Supplier<T> supplier) {     // 构建 span     Span span = buildSpan(command);     return activateAndCloseSpan(span, supplier); }  // 在 span 的生命周期内执行具体命令 private <T> T activateAndCloseSpan(Span span, Supplier<T> supplier) {     Throwable candidateThrowable = null;     try {         // 执行命令         return supplier.get();     } catch (Throwable t) {         candidateThrowable = t;         throw t;     } finally {         if (candidateThrowable != null) {             // ...         } else {             // ...         }         // 通过 tracer api 结束一个span         redisSofaTracer.clientReceiveTagFinish((SofaTracerSpan) span, "00");     } }

除此之外 MongoDB 的埋点也是基于 Spring Data 实现，埋点的实现思路和 Redis 基本相同，这里就不在单独分析。

总结

本次主要对蚂蚁金服分布式链路组件 SOFATracer 以及其埋点机制做了简要的介绍；从各个组件的埋点机制来看，整体思路就是对组件操作进行包装，在请求或者命令执行的前后进行 Span 构建和上报。目前一些主流的链路跟踪组件像 Brave 也是基于此思路，区别在于 Brave 并非是直接基于 OpenTracing 规范进行编码，而是其自己封装了一整套 API ，然后通过面向 OpenTracing API 进行一层适配；另外一个非常流行的 SkyWalking 则是基于 Java agent 实现，埋点实现的机制上与 SOFATracer 和 Brave 不同。

以上就是本期的全部内容，如果大家对 SOFATracer 感兴趣，也可以在群内或者 Github 上与我们交流。