skywalking agent 插件开发基本逻辑

插件工程结构

代码结构分为以下部分

定义拦截形式
实现拦截形式的拦截器
在 resources 目录下定义 skywalking-plugin.def 文件, 让 agent 发现并加载探针

基本规范:

定义拦截形式使用 *Instrumentation 定义
实现拦截形式使用 *Interceptor

核心api

核心对象的相关API
本节主要介绍java探针开发过程中涉及重要的API使用, 进而使用正确的API 完成探针开发

(1) ContextCarrier#items

在跨进程链路追踪的案例中, 我们使用 ContextCarrier#items 完成两个进程链路数据的管理, 以HTTP请求为例, 我们需要处理一下两个场景

场景一, 将发送进程的链路绑定到header中并通过客户端发送出去, 具体代码如下

CarrierItem next = contextCarrier.items();
while (next.hasNext())) {
    next = next.next();
    httpRequest.setHeader(next.getHeadKey(), next.getHeadValue))
}

常见而, 接受服务器解析header并将链路绑定到本次接受处理中, 具体如下

CarrierItem next = contextCarrier.items();
while (next.hasNext())) {
    next = next.next();
    next.setHeaderValue(request.getHeader(next.getHeadKey())));
}

(2) ContextManager#createEntrySpan

一个应用服务的提供端或服务端的接收端点, 如web容器的服务端入口, RPC服务或消息队列的消费者, 在被调用时, 都需要创建 EntrySpan, 这是需要使用 ContextManager#createEntrySpan 完成

ContextManager#createEntrySpan(operationName, contextCarrier)

有两个关键入参

operationName: 定义 EntrySpan 的操作方法名称如 http 接口的请求路径, 注意, operationName 必须是有穷尽的, , 比如 restful 接口匹配 /path/{id}, 一定不要将id真实值记录, 因为 skywalking 上报数据时, 处于减少 operationName长度和链路消息传输性能的考虑, 将 operationName 缓存在本地映射字典中, 因此需要保证 operationName 是有穷尽的, 否则导致 map 过大
contextCarrier: 为了绑定跨进程的追踪, 需要将上游的追踪消息通过 ContextCarrier#items绑定到本链路中

(3) ContextManager#createExitSpan

在一个应用服务的客户端或消息队列生产者的发送端, 如redis客户端访问, mysql查询, rpc组件请求, 客户端都需要使用createExitSpan 来创建 ExitSpan

createExitSpan(operationName, contextCarrier, peer);

有三个参数

operationName: 和 EntrySpan 一样
contextCarrier: 为了绑定跨进程追踪, 需要将链路信息放入header中, 具体看 ContextCarrierItems()
peer: 下游地址: 具体格式为 ip:port, 若系统下游无法下探针, 如 reids,mysql, 则需要将下游地址写入peer中, 格式为 ip:port,ip:port

定义拦截形式

拦截形式定义一般通过继承 ClassInstanceMethodsEnhancePluginDefine 实现

需要增强哪些类, 通过 ClassMatch 类匹配, 支持以下几种方法
- byName: 通过类路径+类名, 通过常量指定, 不要用 *.class.getName()
- byClassAnnotationMatch: 类注解匹配, 不支持父类继承注解
- byHierarchyMatch 父类或接口, 在多层继承情况会导致多次拦截, 一般不用
需要增强的构造方法切入点, 需要指定以下几个部分
- getConstructorMatcher 构造方法匹配器
- getConstructorInterceptor 构造方法探针插件拦截器
需要增强的实例方法切入点需要指定以下几个部分
- getMethodsMatcher 拦截的方法
- getMethodsInterceptor 方法的拦截器
- isOverrideArgs 是否重写参数
需要增强的静态方法切入点, 静态方法基本和实例方法一致

实现拦截形式的拦截器

可以对匹配的方法, 在执行前, 执行后, 执行异常, 进行无侵入的拦截, 通过调用 agent 核心 api 来完成链路追踪

实例方法拦截器接口 InstanceMethodsAroundInterceptor 定义

public interface InstanceMethodsAroundInterceptor {
    // 方法前
    void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
        MethodInterceptResult result) throws Throwable;

    // 方法后
    Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
        Object ret) throws Throwable;

    // 方法异常
    void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
        Class<?>[] argumentsTypes, Throwable t);
}

想要修改入参是, 将 isOverrideArgs 改为true, 否则修改参数不会生效

dubbo-2.7.x 插件实例

拦截形式: DubboInstrumentation 继承 ClassInstanceMethodsEnhancePluginDefine

// 定义增强的类
@Override
protected ClassMatch enhanceClass() {
    // 通过类名匹配 org.apache.dubbo.monitor.support.MonitorFilter
    return NameMatch.byName(ENHANCE_CLASS);
}

@Override
public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
    return new InstanceMethodsInterceptPoint[] {
        new InstanceMethodsInterceptPoint() {
            // 定义拦截的方法, 这里是 MonitorFilter#invoke
            @Override
            public ElementMatcher<MethodDescription> getMethodsMatcher() {
                return named("invoke");
            }

            // 定义实现拦截形式的拦截器 DubboInterceptor
            @Override
            public String getMethodsInterceptor() {
                return INTERCEPT_CLASS;
            }

            @Override
            public boolean isOverrideArgs() {
                return false;
            }
        }
    };
}

拦截方法 DubboInterceptor

@Override
public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
                         MethodInterceptResult result) throws Throwable {
    Invoker invoker = (Invoker) allArguments[0];
    Invocation invocation = (Invocation) allArguments[1];
    RpcContext rpcContext = RpcContext.getContext();
    boolean isConsumer = rpcContext.isConsumerSide();
    URL requestURL = invoker.getUrl();

    AbstractSpan span;

    final String host = requestURL.getHost();
    final int port = requestURL.getPort();

    boolean needCollectArguments;
    int argumentsLengthThreshold;
    if (isConsumer) {
        // 调用方创建 ExitSpan
        final ContextCarrier contextCarrier = new ContextCarrier();
        span = ContextManager.createExitSpan(generateOperationName(requestURL, invocation), contextCarrier, host + ":" + port);
        // 将上下文序列化放入 dubbo attachments
        CarrierItem next = contextCarrier.items();
        while (next.hasNext()) {
            next = next.next();
            rpcContext.getAttachments().put(next.getHeadKey(), next.getHeadValue());
            if (invocation.getAttachments().containsKey(next.getHeadKey())) {
                invocation.getAttachments().remove(next.getHeadKey());
            }
        }
        needCollectArguments = DubboPluginConfig.Plugin.Dubbo.COLLECT_CONSUMER_ARGUMENTS;
        argumentsLengthThreshold = DubboPluginConfig.Plugin.Dubbo.CONSUMER_ARGUMENTS_LENGTH_THRESHOLD;
    } else {
        // 将数据反序列化回 ContextCarrier
        ContextCarrier contextCarrier = new ContextCarrier();
        CarrierItem next = contextCarrier.items();
        while (next.hasNext()) {
            next = next.next();
            next.setHeadValue(rpcContext.getAttachment(next.getHeadKey()));
        }
        // 服务方创建 EntrySpan
        span = ContextManager.createEntrySpan(generateOperationName(requestURL, invocation), contextCarrier);
        span.setPeer(rpcContext.getRemoteAddressString());
        needCollectArguments = DubboPluginConfig.Plugin.Dubbo.COLLECT_PROVIDER_ARGUMENTS;
        argumentsLengthThreshold = DubboPluginConfig.Plugin.Dubbo.PROVIDER_ARGUMENTS_LENGTH_THRESHOLD;
    }

    Tags.URL.set(span, generateRequestURL(requestURL, invocation));
    // 收集参数
    collectArguments(needCollectArguments, argumentsLengthThreshold, span, invocation);
    // 将 span 设置为 dubbo 以及 RPC 调用
    span.setComponent(ComponentsDefine.DUBBO);
    SpanLayer.asRPCFramework(span);
}

skywalking-plugin.def 定义

dubbo=org.apache.skywalking.apm.plugin.asf.dubbo.DubboInstrumentation

toolkit 工具箱

截止8.7, 目前 toolkit 支持, 功能上为 agent 提供各丰富的自定义实现

apm-toolkit-kafka: kafka plugin 抓的是 spring 的 KafkaTemplate, 如果自定义则用它, 使用 @KafkaPollAndInvoke 注解实现
apm-toolkit-log4j: log4j日志收集
apm-toolkit-logback: logback 日志收集
apm-toolkit-meter
apm-toolkit-micrometer-registry
apm-toolkit-opentracing
apm-toolkit-trace: 通过代码方式对 trace 信息进行补充

skywalking agent 中有对应的激活包, 在 /agent/activations 目录下

apm-toolkit-kafka-activation-8.7.0
apm-toolkit-log4j-1.x-activation-8.7.0
...

下面以 trace 包方式解释下处理流程

trace 包处理方式

首先代码一般会导入 toolkit 包

<dependency>
    <groupId>org.apache.skywalking</groupId>
    <artifactId>apm-toolkit-trace</artifactId>
    <version>${project.version}</version>
</dependency>

挑选 AcitveSpan 来进行说明

/**
 * provide custom api that set tag for current active span.
 */
public class ActiveSpan {
    /**
     * 为 span 增加自定义属性
     * @param key   tag key
     * @param value tag value
     */
    public static void tag(String key, String value) {
    }

    ...
}

可以看到每个方法内容都为空, 具体的业务在 apm-toolkit-trace-activation-8.7.0 包中 ActiveSpanActivation实现

ActiveSpanActivation 激活类继承了用于插件开发的 ClassStaticMethodsEnhancePluginDefine, 并定义 ActiveSpan 中每个静态方法的 StaticMethodsInterceptPoint 用于处理

 @Override
public StaticMethodsInterceptPoint[] getStaticMethodsInterceptPoints() {
    return new StaticMethodsInterceptPoint[] {
        new StaticMethodsInterceptPoint() {
            @Override
            public ElementMatcher<MethodDescription> getMethodsMatcher() {
                // tag 方法
                return named(TAG_INTERCEPTOR_METHOD_NAME);
            }

            @Override
            public String getMethodsInterceptor() {
                // tag 方法处理类 ActiveSpanTagInterceptor
                return TAG_INTERCEPTOR_CLASS;
            }

            @Override
            public boolean isOverrideArgs() {
                return false;
            }
        },
        ...
}

实际业务处理类 ActiveSpanTagInterceptor 用于自定义属性添加


public class ActiveSpanTagInterceptor implements StaticMethodsAroundInterceptor {
    @Override
    public void beforeMethod(Class clazz, Method method, Object[] allArguments, Class<?>[] parameterTypes,
        MethodInterceptResult result) {
        try {
            // 获取当前在用的 Span
            AbstractSpan activeSpan = ContextManager.activeSpan();
            // 为 Span 添加自定义属性
            activeSpan.tag(Tags.ofKey(String.valueOf(allArguments[0])), String.valueOf(allArguments[1]));
        } catch (NullPointerException ignored) {
        }
    }

    @Override
    public Object afterMethod(Class clazz, Method method, Object[] allArguments, Class<?>[] parameterTypes,
        Object ret) {
        return ret;
    }

    @Override
    public void handleMethodException(Class clazz, Method method, Object[] allArguments, Class<?>[] parameterTypes,
        Throwable t) {

    }
}