如何寫一個Skywalking trace外掛

javaagent 原理

美團技術團隊-Java 動態除錯技術原理及實踐

類圖

實現

ConsumeMessageConcurrentlyInstrumentation

  1. public class ConsumeMessageConcurrentlyInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
  2.     // 需要增強的類
  3.     private static final String ENHANCE_CLASS = "com.aliyun.openservices.shade.com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently";
  4.     // 需要增強的方法
  5.     private static final String CONSUMER_MESSAGE_METHOD = "consumeMessage";
  6.     // 增加的方法對應的攔截器
  7.     private static final String INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.ons.v1.MessageConcurrentlyConsumeInterceptor";
  9.     // 構造器不需要攔截
  10.     @Override
  11.     public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
  12.         return new ConstructorInterceptPoint[0];
  13.     }
  15.     @Override
  16.     public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
  17.         return new InstanceMethodsInterceptPoint[] {
  18.             // 新增一個攔截器
  19.             new InstanceMethodsInterceptPoint() {
  20.                 @Override
  21.                 public ElementMatcher<MethodDescription> getMethodsMatcher() {
  22.                     // 方法匹配
  23.                     return named(CONSUMER_MESSAGE_METHOD);
  24.                 }
  26.                 @Override
  27.                 public String getMethodsInterceptor() {
  28.                     return INTERCEPTOR_CLASS;
  29.                 }
  31.                 @Override
  32.                 public boolean isOverrideArgs() {
  33.                     return false;
  34.                 }
  35.             }
  36.         };
  37.     }
  39.     @Override
  40.     protected ClassMatch enhanceClass() {
  41.         // 需要增強的類
  42.         return HierarchyMatch.byHierarchyMatch(new String[] {ENHANCE_CLASS});
  43.     }
  44. }

AbstractMessageConsumeInterceptor

  1. public abstract class AbstractMessageConsumeInterceptor implements InstanceMethodsAroundInterceptor {
  3.     public static final String CONSUMER_OPERATION_NAME_PREFIX = "OnsRocketMQ/";
  5.     // 在方法前增強
  6.     @Override
  7.     public final void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
  8.         Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
  9.         // 拿到方法引數，轉換成訊息列表
  10.         List<MessageExt> msgs = (List<MessageExt>) allArguments[0];
  12.         // 從訊息中中獲取TraceId等Context資訊
  13.         ContextCarrier contextCarrier = getContextCarrierFromMessage(msgs.get(0));
  15.         // 建立一個entry span
  16.         AbstractSpan span = ContextManager.createEntrySpan(CONSUMER_OPERATION_NAME_PREFIX + msgs.get(0)
  17.                                                                                                 .getTopic() + "/Consumer", contextCarrier);
  18.         span.setComponent(ComponentsDefine.ROCKET_MQ_CONSUMER);
  19.         SpanLayer.asMQ(span);
  20.         for (int i = 1; i < msgs.size(); i++) {
  21.             ContextManager.extract(getContextCarrierFromMessage(msgs.get(i)));
  22.         }
  24.     }
  26.     // 異常處理
  27.     @Override
  28.     public final void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
  29.         Class<?>[] argumentsTypes, Throwable t) {
  30.         ContextManager.activeSpan().log(t);
  31.     }
  33.     private ContextCarrier getContextCarrierFromMessage(MessageExt message) {
  34.         ContextCarrier contextCarrier = new ContextCarrier();
  36.         CarrierItem next = contextCarrier.items();
  37.         while (next.hasNext()) {
  38.             next = next.next();
  39.             next.setHeadValue(message.getUserProperty(next.getHeadKey()));
  40.         }
  42.         return contextCarrier;
  43.     }
  44. }

MessageConcurrentlyConsumeInterceptor

  1. public class MessageConcurrentlyConsumeInterceptor extends AbstractMessageConsumeInterceptor {
  2.     // 在方法後處理
  3.     @Override
  4.     public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
  5.         Object ret) throws Throwable {
  6.         // 獲取消費狀態
  7.         ConsumeConcurrentlyStatus status = (ConsumeConcurrentlyStatus) ret;
  8.         if (status == ConsumeConcurrentlyStatus.RECONSUME_LATER) {
  9. 			// 消費狀態為重試，則設定span出現錯誤
  10.             AbstractSpan activeSpan = ContextManager.activeSpan();
  11.             activeSpan.errorOccurred();
  12.             Tags.MQ_STATUS.set(activeSpan, status.name());
  13.         }
  14.         // 停止span
  15.         ContextManager.stopSpan();
  16.         return ret;
  17.     }
  18. }

專案：apm-ons-1.x-plugin

參考文件

  1. apm-ons-1.x-plugin
  2. 美團技術團隊-Java 動態除錯技術原理及實踐

分享並記錄所學所見