如何写一个 Skywalking trace 插件

javaagent 原理

美团技术团队 -Java 动静调试技术原理及实际

类图

实现

ConsumeMessageConcurrentlyInstrumentation

public class ConsumeMessageConcurrentlyInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
    // 须要加强的类
    private static final String ENHANCE_CLASS = "com.aliyun.openservices.shade.com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently";
    // 须要加强的办法
    private static final String CONSUMER_MESSAGE_METHOD = "consumeMessage";
    // 减少的办法对应的拦截器
    private static final String INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.ons.v1.MessageConcurrentlyConsumeInterceptor";
 
    // 结构器不须要拦挡
    @Override
    public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {return new ConstructorInterceptPoint[0];
    }
 
    @Override
    public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {return new InstanceMethodsInterceptPoint[] {
            // 新增一个拦截器
            new InstanceMethodsInterceptPoint() {
                @Override
                public ElementMatcher<MethodDescription> getMethodsMatcher() {
                    // 办法匹配
                    return named(CONSUMER_MESSAGE_METHOD);
                }
 
                @Override
                public String getMethodsInterceptor() {return INTERCEPTOR_CLASS;}
 
                @Override
                public boolean isOverrideArgs() {return false;}
            }
        };
    }
 
    @Override
    protected ClassMatch enhanceClass() {
        // 须要加强的类
        return HierarchyMatch.byHierarchyMatch(new String[] {ENHANCE_CLASS});
    }
}

AbstractMessageConsumeInterceptor

public abstract class AbstractMessageConsumeInterceptor implements InstanceMethodsAroundInterceptor {
 
    public static final String CONSUMER_OPERATION_NAME_PREFIX = "OnsRocketMQ/";
 
    // 在办法前加强
    @Override
    public final void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
        Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
        // 拿到办法参数，转换成音讯列表
        List<MessageExt> msgs = (List<MessageExt>) allArguments[0];
 
        // 从音讯中中获取 TraceId 等 Context 信息
        ContextCarrier contextCarrier = getContextCarrierFromMessage(msgs.get(0));
        
        // 创立一个 entry span
        AbstractSpan span = ContextManager.createEntrySpan(CONSUMER_OPERATION_NAME_PREFIX + msgs.get(0)
                                                                                                .getTopic() + "/Consumer", contextCarrier);
        span.setComponent(ComponentsDefine.ROCKET_MQ_CONSUMER);
        SpanLayer.asMQ(span);
        for (int i = 1; i < msgs.size(); i++) {ContextManager.extract(getContextCarrierFromMessage(msgs.get(i)));
        }
 
    }
 
    // 异样解决
    @Override
    public final void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
        Class<?>[] argumentsTypes, Throwable t) {ContextManager.activeSpan().log(t);
    }
 
    private ContextCarrier getContextCarrierFromMessage(MessageExt message) {ContextCarrier contextCarrier = new ContextCarrier();
 
        CarrierItem next = contextCarrier.items();
        while (next.hasNext()) {next = next.next();
            next.setHeadValue(message.getUserProperty(next.getHeadKey()));
        }
 
        return contextCarrier;
    }
}

MessageConcurrentlyConsumeInterceptor

public class MessageConcurrentlyConsumeInterceptor extends AbstractMessageConsumeInterceptor {
    // 在办法后处理
    @Override
    public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
        Object ret) throws Throwable {
        // 获取生产状态
        ConsumeConcurrentlyStatus status = (ConsumeConcurrentlyStatus) ret;
        if (status == ConsumeConcurrentlyStatus.RECONSUME_LATER) {
            // 生产状态为重试，则设置 span 呈现谬误
            AbstractSpan activeSpan = ContextManager.activeSpan();
            activeSpan.errorOccurred();
            Tags.MQ_STATUS.set(activeSpan, status.name());
        }
        // 进行 span
        ContextManager.stopSpan();
        return ret;
    }
}

我的项目：apm-ons-1.x-plugin

参考文档

1. apm-ons-1.x-plugin
2. 美团技术团队 -Java 动静调试技术原理及实际

分享并记录所学所见