本文出處shenyifengtk.github.io 轉(zhuǎn)載請(qǐng)說(shuō)明

概念

Span

Span 是分布式跟蹤系統(tǒng)中一個(gè)重要且常用的概念. 可從 Google Dapper Paper 和 OpenTracing 學(xué)習(xí)更多與 Span 相關(guān)的知識(shí).

SkyWalking 從 2017 年開(kāi)始支持 OpenTracing 和 OpenTracing-Java API, 我們的 Span 概念與論文和 OpenTracing 類似. 我們也擴(kuò)展了 Span.

Span 有三種類型

1.1 EntrySpan

EntrySpan 代表服務(wù)提供者, 也是服務(wù)器端的端點(diǎn). 作為一個(gè) APM 系統(tǒng), 我們的目標(biāo)是應(yīng)用服務(wù)器. 所以幾乎所有的服務(wù)和 MQ-消費(fèi)者 都是 EntrySpan听绳⌒┡颍可以理解一個(gè)進(jìn)程處理第一個(gè)span就是EntrySpan衣形，意思為entiry span 進(jìn)入服務(wù)span峭判。

1.2 LocalSpan

LocalSpan 表示普通的 Java 方法, 它與遠(yuǎn)程服務(wù)無(wú)關(guān), 也不是 MQ 生產(chǎn)者/消費(fèi)者, 也不是服務(wù)（例如 HTTP 服務(wù)）提供者/消費(fèi)者族吻。所有本地方法調(diào)用都是localSpan,包括異步線程調(diào)用捏境，線程池提交任務(wù)都是峻村。

1.3 ExitSpan

ExitSpan 代表一個(gè)服務(wù)客戶端或MQ的生產(chǎn)者, 在 SkyWalking 的早期命名為 LeafSpan. 例如 通過(guò) JDBC 訪問(wèn)DB, 讀取 Redis/Memcached 被歸類為 ExitSpan.

image.png

上下文載體 (ContextCarrier)

為了實(shí)現(xiàn)分布式跟蹤, 需要綁定跨進(jìn)程的追蹤, 并且上下文應(yīng)該在整個(gè)過(guò)程中隨之傳播. 這就是 ContextCarrier 的職責(zé).

以下是有關(guān)如何在 A -> B 分布式調(diào)用中使用 ContextCarrier 的步驟.

1. 在客戶端, 創(chuàng)建一個(gè)新的空的 ContextCarrier.
2. 通過(guò) ContextManager#createExitSpan 創(chuàng)建一個(gè) ExitSpan 或者使用 ContextManager#inject 來(lái)初始化 ContextCarrier.
3. 將 ContextCarrier 所有信息放到請(qǐng)求頭 (如 HTTP HEAD), 附件(如 Dubbo RPC 框架), 或者消息 (如 Kafka) 中,詳情可以看官方給出跨進(jìn)程傳輸協(xié)議sw8
4. 通過(guò)服務(wù)調(diào)用, 將 ContextCarrier 傳遞到服務(wù)端.
5. 在服務(wù)端, 在對(duì)應(yīng)組件的頭部, 附件或消息中獲取 ContextCarrier 所有內(nèi)容.
6. 通過(guò) ContestManager#createEntrySpan 創(chuàng)建 EntrySpan 或者使用 ContextManager#extract 將服務(wù)端和客戶端的綁定.

讓我們通過(guò) Apache HttpComponent client 插件和 Tomcat 7 服務(wù)器插件演示, 步驟如下:

1. 客戶端 Apache HttpComponent client 插件

span = ContextManager.createExitSpan("/span/operation/name", contextCarrier, "ip:port");
CarrierItem next = contextCarrier.items();
while (next.hasNext()) {
    next = next.next();
    httpRequest.setHeader(next.getHeadKey(), next.getHeadValue());
}

2. 服務(wù)端 Tomcat 7 服務(wù)器插件

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

span = ContextManager.createEntrySpan("/span/operation/name", contextCarrier);

上下文快照 (ContextSnapshot)

除了跨進(jìn)程, 跨線程也是需要支持的, 例如異步線程（內(nèi)存中的消息隊(duì)列）和批處理在 Java 中很常見(jiàn), 跨進(jìn)程和跨線程十分相似, 因?yàn)槎际切枰獋鞑ド舷挛? 唯一的區(qū)別是, 不需要跨線程序列化.

以下是有關(guān)跨線程傳播的三個(gè)步驟：

1. 使用 ContextManager#capture 方法獲取 ContextSnapshot 對(duì)象.
2. 讓子線程以任何方式, 通過(guò)方法參數(shù)或由現(xiàn)有參數(shù)攜帶來(lái)訪問(wèn) ContextSnapshot
3. 在子線程中使用 ContextManager#continued田巴。

跨進(jìn)程Span傳輸原理

public class CarrierItem implements Iterator<CarrierItem> {
    private String headKey;
    private String headValue;
    private CarrierItem next;

    public CarrierItem(String headKey, String headValue) {
        this(headKey, headValue, null);
    }

    public CarrierItem(String headKey, String headValue, CarrierItem next) {
        this.headKey = headKey;
        this.headValue = headValue;
        this.next = next;
    }

    public String getHeadKey() {
        return headKey;
    }

    public String getHeadValue() {
        return headValue;
    }

    public void setHeadValue(String headValue) {
        this.headValue = headValue;
    }

    @Override
    public boolean hasNext() {
        return next != null;
    }

    @Override
    public CarrierItem next() {
        return next;
    }

    @Override
    public void remove() {

    }
}

CarrierItem 類似Map key value的數(shù)據(jù)接口钠糊，通過(guò)一個(gè)單向連接將K/V連接起來(lái)。
看下 ContextCarrier.items()方法如何創(chuàng)建CarrierItem

    public CarrierItem items() {
       //內(nèi)置一個(gè) sw8-x key
        SW8ExtensionCarrierItem sw8ExtensionCarrierItem = new SW8ExtensionCarrierItem(extensionContext, null); 
       //內(nèi)置  sw8-correlation key
        SW8CorrelationCarrierItem sw8CorrelationCarrierItem = new SW8CorrelationCarrierItem(
            correlationContext, sw8ExtensionCarrierItem);
       //內(nèi)置 sw8 key 
        SW8CarrierItem sw8CarrierItem = new SW8CarrierItem(this, sw8CorrelationCarrierItem);
        return new CarrierItemHead(sw8CarrierItem);
    }

創(chuàng)建一個(gè)鏈接CarrierItemHead->SW8CarrierItem ->SW8CorrelationCarrierItem->SW8ExtensionCarrierItem
在看下上面tomcat7 遍歷CarrierItem壹哺，調(diào)用key從http header獲取值設(shè)置到對(duì)象內(nèi)置值抄伍，這樣就可以做到將上一個(gè)進(jìn)程header 值設(shè)置到下一個(gè)進(jìn)程里，在調(diào)用

    ContextCarrier deserialize(String text, HeaderVersion version) {
        if (text == null) {
            return this;
        }
        if (HeaderVersion.v3.equals(version)) {
            String[] parts = text.split("-", 8);
            if (parts.length == 8) {
                try {
                    // parts[0] is sample flag, always trace if header exists.
                    this.traceId = Base64.decode2UTFString(parts[1]);
                    this.traceSegmentId = Base64.decode2UTFString(parts[2]);
                    this.spanId = Integer.parseInt(parts[3]);
                    this.parentService = Base64.decode2UTFString(parts[4]);
                    this.parentServiceInstance = Base64.decode2UTFString(parts[5]);
                    this.parentEndpoint = Base64.decode2UTFString(parts[6]);
                    this.addressUsedAtClient = Base64.decode2UTFString(parts[7]);
                } catch (IllegalArgumentException ignored) {

                }
            }
        }
        return this;
    }

這樣剛剛new 出來(lái)ContextCarrier就可以從上一個(gè)調(diào)用者上繼承所有的屬性管宵，新創(chuàng)建span就可以跟上一個(gè)span 關(guān)聯(lián)起來(lái)了了截珍。

開(kāi)發(fā)插件

知識(shí)點(diǎn)

追蹤的基本方法是攔截 Java 方法, 使用字節(jié)碼操作技術(shù)（byte-buddy）和 AOP 概念. SkyWalking 包裝了字節(jié)碼操作技術(shù)并追蹤上下文的傳播, 所以你只需要定義攔截點(diǎn)（換句話說(shuō)就是 Spring 的切面）。

ClassInstanceMethodsEnhancePluginDefine定義了構(gòu)造方法 Contructor 攔截點(diǎn)和 instance method 實(shí)例方法攔截點(diǎn)啄糙，主要有三個(gè)方法需要被重寫

     /**
     * 需要被攔截Class
     * @return
     */
    @Override
    protected ClassMatch enhanceClass() {
        return null;
    }

    /**
     * 構(gòu)造器切點(diǎn)
     * @return
     */
    @Override
    public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
        return new ConstructorInterceptPoint[0];
    }

    /**
     * 方法切點(diǎn)
     * @return InstanceMethodsInterceptPoint 里面會(huì)聲明攔截按個(gè)方法
     */
    @Override
    public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
        return new InstanceMethodsInterceptPoint[0];
    }

ClassMatch 以下有四種方法表示如何去匹配目標(biāo)類:

• NameMatch.byName, 通過(guò)類的全限定名(Fully Qualified Class Name, 即 包名 + . + 類名).
• ClassAnnotationMatch.byClassAnnotationMatch, 根據(jù)目標(biāo)類是否存在某些注解.
• MethodAnnotationMatchbyMethodAnnotationMatch, 根據(jù)目標(biāo)類的方法是否存在某些注解.
• HierarchyMatch.byHierarchyMatch, 根據(jù)目標(biāo)類的父類或接口

ClassStaticMethodsEnhancePluginDefine 定義了類方法 class 靜態(tài)method 攔截點(diǎn)笛臣。

public abstract class ClassStaticMethodsEnhancePluginDefine extends ClassEnhancePluginDefine {

    /**
     * 構(gòu)造器切點(diǎn)
     * @return null, means enhance no constructors.
     */
    @Override
    public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
        return null;
    }

    /**
     * 方法切點(diǎn)
     * @return null, means enhance no instance methods.
     */
    @Override
    public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
        return null;
    }
}

InstanceMethodsInterceptPoint 普通方法接口切點(diǎn)有哪些方法

public interface InstanceMethodsInterceptPoint {
    /**
     * class instance methods matcher.
     *  可以理解成功對(duì)Class 那些方法進(jìn)行增強(qiáng)
     *  ElementMatcher 是bytebuddy 類庫(kù)一個(gè)方法匹配器，里面封裝了各種方法匹配
     * @return methods matcher
     */
    ElementMatcher<MethodDescription> getMethodsMatcher();

    /**
     * @return represents a class name, the class instance must instanceof InstanceMethodsAroundInterceptor.
     *  返回一個(gè)攔截器全類名隧饼，所有攔截器必須實(shí)現(xiàn)    InstanceMethodsAroundInterceptor 接口
     */
    String getMethodsInterceptor();

    /**
     *  是否要覆蓋原方法入?yún)?     * @return
     */
    boolean isOverrideArgs();
}

在看下攔截器有那些方法

/**
 * A interceptor, which intercept method's invocation. The target methods will be defined in {@link
 * ClassEnhancePluginDefine}'s subclass, most likely in {@link ClassInstanceMethodsEnhancePluginDefine}
 */
public interface InstanceMethodsAroundInterceptor {
    /**
     * called before target method invocation.
     * 前置通知
     * @param result change this result, if you want to truncate the method.
     */
    void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
        MethodInterceptResult result) throws Throwable;

    /**
     * called after target method invocation. Even method's invocation triggers an exception.
     * 后置通知
     * @param ret the method's original return value. May be null if the method triggers an exception.
     * @return the method's actual return value.
     */
    Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
        Object ret) throws Throwable;

    /**
     * called when occur exception.
     * 異常通知
     * @param t the exception occur.
     */
    void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
        Class<?>[] argumentsTypes, Throwable t);
}

開(kāi)發(fā)Skywalking實(shí)戰(zhàn)

項(xiàng)目maven環(huán)境配置

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>tk.shenyifeng</groupId>
    <artifactId>skywalking-plugin</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <skywalking.version>8.10.0</skywalking.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.skywalking</groupId>
            <artifactId>apm-agent-core</artifactId>
            <version>${skywalking.version}</version>
            <scope>provided</scope>
        </dependency>

        <dependency>
            <groupId>org.apache.skywalking</groupId>
            <artifactId>java-agent-util</artifactId>
            <version>${skywalking.version}</version>
            <scope>provided</scope>
        </dependency>
    </dependencies>


    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>8</source>
                    <target>8</target>
                </configuration>
            </plugin>

            <plugin>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.2.4</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <shadedArtifactAttached>false</shadedArtifactAttached>
                            <createDependencyReducedPom>true</createDependencyReducedPom>
                            <createSourcesJar>true</createSourcesJar>
                            <shadeSourcesContent>true</shadeSourcesContent>
                            <relocations>
                                <relocation>
                                    <pattern>net.bytebuddy</pattern>
                                    <shadedPattern>org.apache.skywalking.apm.dependencies.net.bytebuddy</shadedPattern>
                                </relocation>
                            </relocations>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

為了更有代表性一些沈堡，使用Skywalking官方開(kāi)發(fā)的ES插件來(lái)做一個(gè)例子。為了兼容不同版本框架燕雁，Skywalking 官方使用witnessClasses诞丽，當(dāng)前框架Jar存在這個(gè)Class就會(huì)任務(wù)是某個(gè)版本、同樣witnessMethods當(dāng)Class存在某個(gè)Method拐格。

public class AdapterActionFutureInstrumentation extends ClassEnhancePluginDefine {

    @Override
    public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
        return new ConstructorInterceptPoint[0];
    }

    @Override
    public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
        return new InstanceMethodsInterceptPoint[] {
            new InstanceMethodsInterceptPoint() {
                @Override
                public ElementMatcher<MethodDescription> getMethodsMatcher() {
                    return named("actionGet"); //攔截方法
                }

                @Override
                public String getMethodsInterceptor() {  //攔截器全類名
                    return "org.apache.skywalking.apm.plugin.elasticsearch.v7.interceptor.AdapterActionFutureActionGetMethodsInterceptor";
                }

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

    @Override
    public StaticMethodsInterceptPoint[] getStaticMethodsInterceptPoints() {
        return new StaticMethodsInterceptPoint[0];
    }

    @Override
    protected ClassMatch enhanceClass() { //增強(qiáng)Class
        return byName("org.elasticsearch.action.support.AdapterActionFuture");
    }

    @Override
    protected String[] witnessClasses() {//ES7 存在Class
        return new String[] {"org.elasticsearch.transport.TaskTransportChannel"};
    }

    @Override
    protected List<WitnessMethod> witnessMethods() { //ES7 SearchHits 存在方法
        return Collections.singletonList(new WitnessMethod(
            "org.elasticsearch.search.SearchHits",
          named("getTotalHits").and(takesArguments(0)).and(returns(named("org.apache.lucene.search.TotalHits")))
        ));
    }
}

創(chuàng)建一個(gè)給定類名的攔截器僧免，實(shí)現(xiàn)InstanceMethodsAroundInterceptor接口。創(chuàng)建一個(gè)EntrySpan

public class TomcatInvokeInterceptor implements InstanceMethodsAroundInterceptor {

    private static boolean IS_SERVLET_GET_STATUS_METHOD_EXIST;
    private static final String SERVLET_RESPONSE_CLASS = "javax.servlet.http.HttpServletResponse";
    private static final String GET_STATUS_METHOD = "getStatus";

    static {
        IS_SERVLET_GET_STATUS_METHOD_EXIST = MethodUtil.isMethodExist(
            TomcatInvokeInterceptor.class.getClassLoader(), SERVLET_RESPONSE_CLASS, GET_STATUS_METHOD);
    }

    /**
     * * The {@link TraceSegment#ref} of current trace segment will reference to the trace segment id of the previous
     * level if the serialized context is not null.
     *
     * @param result change this result, if you want to truncate the method.
     */
    @Override
    public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
                             MethodInterceptResult result) throws Throwable {
        Request request = (Request) allArguments[0];
        ContextCarrier contextCarrier = new ContextCarrier();

        CarrierItem next = contextCarrier.items();
       //如果 HTTP 請(qǐng)求頭中有符合sw8 傳輸協(xié)議的請(qǐng)求頭則 取出來(lái)設(shè)置到上下文ContextCarrier
        while (next.hasNext()) {
            next = next.next();
            next.setHeadValue(request.getHeader(next.getHeadKey()));
        }
        String operationName =  String.join(":", request.getMethod(), request.getRequestURI());
        AbstractSpan span = ContextManager.createEntrySpan(operationName, contextCarrier);//關(guān)聯(lián)起來(lái)
        Tags.URL.set(span, request.getRequestURL().toString()); //添加 span 參數(shù)
        Tags.HTTP.METHOD.set(span, request.getMethod());
        span.setComponent(ComponentsDefine.TOMCAT);
        SpanLayer.asHttp(span);

        if (TomcatPluginConfig.Plugin.Tomcat.COLLECT_HTTP_PARAMS) {
            collectHttpParam(request, span);
        }
    }

    @Override
    public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
                              Object ret) throws Throwable {
        Request request = (Request) allArguments[0];
        HttpServletResponse response = (HttpServletResponse) allArguments[1];

        AbstractSpan span = ContextManager.activeSpan();
        if (IS_SERVLET_GET_STATUS_METHOD_EXIST && response.getStatus() >= 400) {
            span.errorOccurred();
            Tags.HTTP_RESPONSE_STATUS_CODE.set(span, response.getStatus());
        }
        // Active HTTP parameter collection automatically in the profiling context.
        if (!TomcatPluginConfig.Plugin.Tomcat.COLLECT_HTTP_PARAMS && span.isProfiling()) {
            collectHttpParam(request, span);
        }
        ContextManager.getRuntimeContext().remove(Constants.FORWARD_REQUEST_FLAG);
        ContextManager.stopSpan();
        return ret;
    }

    @Override
    public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
                                      Class<?>[] argumentsTypes, Throwable t) {
        AbstractSpan span = ContextManager.activeSpan();
        span.log(t);
    }

    private void collectHttpParam(Request request, AbstractSpan span) {
        final Map<String, String[]> parameterMap = new HashMap<>();
        final org.apache.coyote.Request coyoteRequest = request.getCoyoteRequest();
        final Parameters parameters = coyoteRequest.getParameters();
        for (final Enumeration<String> names = parameters.getParameterNames(); names.hasMoreElements(); ) {
            final String name = names.nextElement();
            parameterMap.put(name, parameters.getParameterValues(name));
        }

        if (!parameterMap.isEmpty()) {
            String tagValue = CollectionUtil.toString(parameterMap);
            tagValue = TomcatPluginConfig.Plugin.Http.HTTP_PARAMS_LENGTH_THRESHOLD > 0 ?
                StringUtil.cut(tagValue, TomcatPluginConfig.Plugin.Http.HTTP_PARAMS_LENGTH_THRESHOLD) :
                tagValue;
            Tags.HTTP.PARAMS.set(span, tagValue);
        }
    }
}

開(kāi)發(fā)完成攔截器后捏浊，一定要在類路徑上添加skywalking-plugin.def文件懂衩，將開(kāi)發(fā)后的全類名添加到配置。

xxxName = tk.shenyifeng.skywalking.plugin.RepladInstrumentation

如果jar 里面沒(méi)有這個(gè)文件，插件不會(huì)被Skywalking加載的浊洞。
最后將打包的jar 放到Skywalking的plugin或者activations目錄就可以了牵敷。

xml配置插件

<?xml version="1.0" encoding="UTF-8"?>
<enhanced>
    <class class_name="test.apache.skywalking.apm.testcase.customize.service.TestService1">
        <method method="staticMethod()" operation_name="/is_static_method" static="true"></method>
        <method method="staticMethod(java.lang.String,int.class,java.util.Map,java.util.List,[Ljava.lang.Object;)"
                operation_name="/is_static_method_args" static="true">
            <operation_name_suffix>arg[0]</operation_name_suffix>
            <operation_name_suffix>arg[1]</operation_name_suffix>
            <operation_name_suffix>arg[3].[0]</operation_name_suffix>
            <tag key="tag_1">arg[2].['k1']</tag>
            <tag key="tag_2">arg[4].[1]</tag>
            <log key="log_1">arg[4].[2]</log>
        </method>
        <method method="method()" static="false"></method>
        <method method="method(java.lang.String,int.class)" operation_name="/method_2" static="false">
            <operation_name_suffix>arg[0]</operation_name_suffix>
            <tag key="tag_1">arg[0]</tag>
            <log key="log_1">arg[1]</log>
        </method>
        <method
            method="method(test.apache.skywalking.apm.testcase.customize.model.Model0,java.lang.String,int.class)"
            operation_name="/method_3" static="false">
            <operation_name_suffix>arg[0].id</operation_name_suffix>
            <operation_name_suffix>arg[0].model1.name</operation_name_suffix>
            <operation_name_suffix>arg[0].model1.getId()</operation_name_suffix>
            <tag key="tag_os">arg[0].os.[1]</tag>
            <log key="log_map">arg[0].getM().['k1']</log>
        </method>
        <method method="retString(java.lang.String)" operation_name="/retString" static="false">
            <tag key="tag_ret">returnedObj</tag>
            <log key="log_map">returnedObj</log>
        </method>
        <method method="retModel0(test.apache.skywalking.apm.testcase.customize.model.Model0)"
          operation_name="/retModel0" static="false">
            <tag key="tag_ret">returnedObj.model1.id</tag>
            <log key="log_map">returnedObj.model1.getId()</log>
        </method>
    </class>
    
</enhanced>

通過(guò)xml配置可以省去編寫Java代碼，打包jar步驟法希。
xml規(guī)則

在配置文件agent.config中添加配置：

plugin.customize.enhance_file=customize_enhance.xml的絕對(duì)路徑

引用資料
https://www.itmuch.com/skywalking/apm-customize-enhance-plugin/
https://skyapm.github.io/document-cn-translation-of-skywalking/zh/6.1.0/guides/Java-Plugin-Development-Guide.html