Merge pull request #758 from lytscu/develop

[Agent] Provide plugins for Apache httpcomponent AsyncClient 4.1 #588

apm-protocol/apm-network/src/main/java/org/apache/skywalking/apm/network/trace/component/ComponentsDefine.java

@@ -76,6 +76,7 @@ public class ComponentsDefine {
   
     public static final OfficialComponent ROCKET_MQ = new OfficialComponent(25, "RocketMQ");
 
+    public static final OfficialComponent HTTP_ASYNC_CLIENT = new OfficialComponent(26, "httpasyncclient");
 
     private static ComponentsDefine INSTANCE = new ComponentsDefine();
 
@@ -86,7 +87,7 @@ public class ComponentsDefine {
     }
 
     public ComponentsDefine() {
-        components = new String[26];
+        components = new String[27];
         addComponent(TOMCAT);
         addComponent(HTTPCLIENT);
         addComponent(DUBBO);
@@ -112,6 +113,7 @@ public class ComponentsDefine {
         addComponent(GRPC);
         addComponent(ELASTIC_JOB);
         addComponent(ROCKET_MQ);
+        addComponent(HTTP_ASYNC_CLIENT);
     }
 
     private void addComponent(OfficialComponent component) {

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/pom.xml

+<?xml version="1.0"?>
+<!--
+  ~ Licensed to the Apache Software Foundation (ASF) under one or more
+  ~ contributor license agreements.  See the NOTICE file distributed with
+  ~ this work for additional information regarding copyright ownership.
+  ~ The ASF licenses this file to You under the Apache License, Version 2.0
+  ~ (the "License"); you may not use this file except in compliance with
+  ~ the License.  You may obtain a copy of the License at
+  ~
+  ~     http://www.apache.org/licenses/LICENSE-2.0
+  ~
+  ~ Unless required by applicable law or agreed to in writing, software
+  ~ distributed under the License is distributed on an "AS IS" BASIS,
+  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  ~ See the License for the specific language governing permissions and
+  ~ limitations under the License.
+  ~
+  -->
+
+<project
+    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
+    xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.skywalking</groupId>
+        <artifactId>apm-sdk-plugin</artifactId>
+        <version>5.0.0-alpha</version>
+    </parent>
+
+    <artifactId>apm-httpasyncclient-4.x-plugin</artifactId>
+
+    <name>httpasyncclient-4.x-plugin</name>
+    <url>http://maven.apache.org</url>
+
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpasyncclient</artifactId>
+            <version>4.1.1</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpasyncclient-cache</artifactId>
+            <version>4.1.1</version>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
+    <build>
+        <plugins>
+            <plugin>
+                <!-- 源码插件 -->
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-source-plugin</artifactId>
+                <!-- 发布时自动将源码同时发布的配置 -->
+                <executions>
+                    <execution>
+                        <id>attach-sources</id>
+                        <goals>
+                            <goal>jar</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/ConnectIterceptor.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4;
+
+import java.lang.reflect.Method;
+import java.net.InetSocketAddress;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
+
+/**
+ * Pass ref accross thread by SessionRequest.
+ *
+ * @author liyuntao
+ */
+
+public class ConnectIterceptor implements InstanceMethodsAroundInterceptor {
+
+    @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
+
+    }
+
+    @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Object ret) throws Throwable {
+        ((EnhancedInstance)ret).setSkyWalkingDynamicField(ContextManager.capture());
+
+        InetSocketAddress remoteAddress = (InetSocketAddress)allArguments[0];
+        String peer = remoteAddress.toString().substring(1);
+
+        Object[] cacheValue = new Object[3];
+        cacheValue[0] = ContextManager.capture();
+        cacheValue[1] = peer;
+        objInst.setSkyWalkingDynamicField(cacheValue);
+        return ret;
+    }
+
+    @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Throwable t) {
+        AbstractSpan activeSpan = ContextManager.activeSpan();
+        activeSpan.errorOccurred();
+        activeSpan.log(t);
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/DefaultConnectingIOReactorIterceptor.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4;
+
+import java.lang.reflect.Method;
+import org.apache.skywalking.apm.agent.core.context.ContextCarrier;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.context.ContextSnapshot;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
+import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
+
+/**
+ * Create local span :httpasyncclient/SocketChannel, to showcase the ability to connect to the remote host.
+ *
+ * @author liyuntao
+ */
+
+public class DefaultConnectingIOReactorIterceptor implements InstanceMethodsAroundInterceptor {
+
+    @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
+        Object[] cacheValue = (Object[])objInst.getSkyWalkingDynamicField();
+        final ContextCarrier contextCarrier = new ContextCarrier();
+        AbstractSpan span = ContextManager.createExitSpan("httpasyncclient/" + method.getName(), contextCarrier, cacheValue[1].toString());
+        ContextManager.continued((ContextSnapshot)cacheValue[0]);
+        span.setComponent(ComponentsDefine.HTTP_ASYNC_CLIENT).setLayer(SpanLayer.HTTP);
+    }
+
+    @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Object ret) throws Throwable {
+        ContextManager.stopSpan();
+        return ret;
+    }
+
+    @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Throwable t) {
+        AbstractSpan activeSpan = ContextManager.activeSpan();
+        activeSpan.errorOccurred();
+        activeSpan.log(t);
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/HttpAsyncResponseConsumerInterceptor.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4;
+
+import java.lang.reflect.Method;
+import org.apache.http.nio.protocol.HttpAsyncRequestProducer;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.context.tag.Tags;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
+import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
+
+/**
+ * End a local span for  {@link org.apache.http.impl.nio.client.CloseableHttpAsyncClient#execute} called by
+ * application.
+ *
+ * @author liyuntao
+ */
+
+public class HttpAsyncResponseConsumerInterceptor implements InstanceMethodsAroundInterceptor {
+
+    @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
+
+        HttpAsyncRequestProducer producer = (HttpAsyncRequestProducer)allArguments[0];
+        String uri = producer.generateRequest().getRequestLine().getUri();
+        String requestMethod = producer.generateRequest().getRequestLine().getMethod();
+        AbstractSpan span = ContextManager.createLocalSpan("httpasyncclient/" + method.getName());
+        Tags.HTTP.METHOD.set(span, requestMethod);
+        span.setComponent(ComponentsDefine.HTTP_ASYNC_CLIENT).setLayer(SpanLayer.HTTP);
+        Tags.URL.set(span, uri);
+
+    }
+
+    @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Object ret) throws Throwable {
+        ContextManager.stopSpan();
+        return ret;
+    }
+
+    @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Throwable t) {
+        AbstractSpan activeSpan = ContextManager.activeSpan();
+        activeSpan.errorOccurred();
+        activeSpan.log(t);
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/HttpHostInterceptor.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4;
+
+import java.lang.reflect.Method;
+import org.apache.http.HttpHost;
+import org.apache.http.HttpRequest;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.context.tag.Tags;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
+import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
+
+/**
+ * End a local span for  {@link org.apache.http.impl.nio.client.CloseableHttpAsyncClient#execute} called by
+ * application.
+ *
+ * @author liyuntao
+ */
+
+public class HttpHostInterceptor implements InstanceMethodsAroundInterceptor {
+
+    @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
+
+        HttpHost producer = (HttpHost)allArguments[0];
+        String uri = producer.toURI();
+        AbstractSpan span = ContextManager.createLocalSpan("httpasyncclient/" + method.getName());
+        span.setComponent(ComponentsDefine.HTTP_ASYNC_CLIENT).setLayer(SpanLayer.HTTP);
+        Tags.HTTP.METHOD.set(span, ((HttpRequest)allArguments[1]).getRequestLine().getMethod());
+        Tags.URL.set(span, uri);
+
+    }
+
+    @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Object ret) throws Throwable {
+        ContextManager.stopSpan();
+        return ret;
+    }
+
+    @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Throwable t) {
+        AbstractSpan activeSpan = ContextManager.activeSpan();
+        activeSpan.errorOccurred();
+        activeSpan.log(t);
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/ProcessResponseInterceptor.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4;
+
+import java.lang.reflect.Method;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
+import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
+
+/**
+ * 1.End exit span.
+ * 2.Create a local span of  callback.
+ * 3.End local span:AsyncThread/execute.
+ *
+ * @author liyuntao
+ */
+
+public class ProcessResponseInterceptor implements InstanceMethodsAroundInterceptor {
+
+    @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
+        AbstractSpan activeSpan = ContextManager.activeSpan();
+        String uri = activeSpan.getOperationName();
+        //stop exitSpan
+        ContextManager.stopSpan();
+        AbstractSpan localSpan = ContextManager.createLocalSpan("callback:" + uri);
+        localSpan.setComponent(ComponentsDefine.HTTP_ASYNC_CLIENT).setLayer(SpanLayer.HTTP);
+    }
+
+    @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Object ret) throws Throwable {
+        //stop local span:callback
+        ContextManager.stopSpan();
+        //stop local span:AsyncThread/execute
+        ContextManager.stopSpan();
+
+        return ret;
+    }
+
+    @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Throwable t) {
+        AbstractSpan activeSpan = ContextManager.activeSpan();
+        activeSpan.errorOccurred();
+        activeSpan.log(t);
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/SessionRequestImplIterceptor.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4;
+
+import java.lang.reflect.Method;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
+
+/**
+ * Set local span false When  connect to the remote host  failed .
+ *
+ * @author liyuntao
+ */
+
+public class SessionRequestImplIterceptor implements InstanceMethodsAroundInterceptor {
+
+    @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
+        AbstractSpan activeSpan = ContextManager.activeSpan();
+        activeSpan.errorOccurred();
+
+    }
+
+    @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Object ret) throws Throwable {
+        return ret;
+    }
+
+    @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Throwable t) {
+        AbstractSpan activeSpan = ContextManager.activeSpan();
+        activeSpan.errorOccurred();
+        activeSpan.log(t);
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/SetResponseInterceptor.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4;
+
+import java.lang.reflect.Method;
+import org.apache.http.HttpResponse;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.context.tag.Tags;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
+
+/**
+ * End exit span and create  a local span of  future/Callback.
+ *
+ * @author liyuntao
+ */
+
+public class SetResponseInterceptor implements InstanceMethodsAroundInterceptor {
+
+    @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
+        if (null == allArguments[0]) {
+            return;
+        }
+        AbstractSpan span = ContextManager.activeSpan();
+        int statusCode = ((HttpResponse)allArguments[0]).getStatusLine().getStatusCode();
+        if (statusCode >= 400) {
+            span.errorOccurred();
+            Tags.STATUS_CODE.set(span, Integer.toString(statusCode));
+        }
+
+    }
+
+    @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Object ret) throws Throwable {
+        return ret;
+    }
+
+    @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Throwable t) {
+        AbstractSpan activeSpan = ContextManager.activeSpan();
+        activeSpan.errorOccurred();
+        activeSpan.log(t);
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/StateInterceptor.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4;
+
+import java.lang.reflect.Method;
+import java.net.MalformedURLException;
+import java.net.URL;
+import org.apache.http.client.methods.HttpRequestWrapper;
+import org.apache.skywalking.apm.agent.core.context.CarrierItem;
+import org.apache.skywalking.apm.agent.core.context.ContextCarrier;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.context.tag.Tags;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
+import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
+
+/**
+ * Create exit span of httpasyncclient.
+ *
+ * @author liyuntao
+ */
+
+public class StateInterceptor implements InstanceMethodsAroundInterceptor {
+
+    @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
+        if (null == allArguments[0]) {
+            return;
+        }
+        HttpRequestWrapper httpRequest = (HttpRequestWrapper)allArguments[0];
+        String uri = httpRequest.getOriginal().getRequestLine().getUri();
+        AbstractSpan span = null;
+        final ContextCarrier contextCarrier = new ContextCarrier();
+        try {
+            URL url = new URL(httpRequest.getOriginal().getRequestLine().getUri());
+            String remotePeer = url.getHost() + ":" + url.getPort();
+            span = ContextManager.createExitSpan(url.getPath(), contextCarrier, remotePeer);
+        } catch (MalformedURLException e) {
+            throw e;
+        }
+        span.setComponent(ComponentsDefine.HTTP_ASYNC_CLIENT);
+        Tags.URL.set(span, uri);
+        Tags.HTTP.METHOD.set(span, httpRequest.getOriginal().getRequestLine().getMethod());
+        SpanLayer.asHttp(span);
+
+        CarrierItem next = contextCarrier.items();
+        while (next.hasNext()) {
+            next = next.next();
+            httpRequest.setHeader(next.getHeadKey(), next.getHeadValue());
+        }
+
+    }
+
+    @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Object ret) throws Throwable {
+        return ret;
+    }
+
+    @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Throwable t) {
+        AbstractSpan activeSpan = ContextManager.activeSpan();
+        activeSpan.errorOccurred();
+        activeSpan.log(t);
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/SuccessInterceptor.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4;
+
+import java.lang.reflect.Method;
+import org.apache.http.nio.reactor.SessionRequest;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.context.ContextSnapshot;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
+import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
+
+/**
+ * Create a local sapn and passing ref accross thread by SessionRequest.
+ *
+ * @author liyuntao
+ */
+
+public class SuccessInterceptor implements InstanceMethodsAroundInterceptor {
+
+    @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
+        SessionRequest request = (SessionRequest)allArguments[0];
+        AbstractSpan localSpan = ContextManager.createLocalSpan("AsyncThread/execute");
+        localSpan.setComponent(ComponentsDefine.HTTP_ASYNC_CLIENT).setLayer(SpanLayer.HTTP);
+        Object cacheValue = ((EnhancedInstance)request).getSkyWalkingDynamicField();
+        ContextManager.continued((ContextSnapshot)cacheValue);
+
+    }
+
+    @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Object ret) throws Throwable {
+        return ret;
+    }
+
+    @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Throwable t) {
+        AbstractSpan activeSpan = ContextManager.activeSpan();
+        activeSpan.errorOccurred();
+        activeSpan.log(t);
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/define/AbstractNIOConnPoolInstrumentation.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4.define;
+
+import net.bytebuddy.description.method.MethodDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine;
+import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
+
+/**
+ * {@link AbstractNIOConnPoolInstrumentation} presents that skywalking intercept {@link
+ * org.apache.http.nio.protocol.AbstractNIOConnPool #requestCompleted}.
+ *
+ * @author liyuntao
+ */
+
+public class AbstractNIOConnPoolInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
+
+    private static final String ENHANCE_CLASS = "org.apache.http.nio.pool.AbstractNIOConnPool";
+    private static final String START_LOCAL_SUCCESS_INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.httpasyncclient.v4.SuccessInterceptor";
+
+    @Override
+    public ClassMatch enhanceClass() {
+        return byName(ENHANCE_CLASS);
+    }
+
+    @Override
+    protected ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
+        return null;
+    }
+
+    @Override
+    protected InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
+        return new InstanceMethodsInterceptPoint[] {
+
+            new InstanceMethodsInterceptPoint() {
+                @Override
+                public ElementMatcher<MethodDescription> getMethodsMatcher() {
+                    return named("requestCompleted");
+                }
+
+                @Override
+                public String getMethodsInterceptor() {
+                    return START_LOCAL_SUCCESS_INTERCEPT_CLASS;
+                }
+
+                @Override
+                public boolean isOverrideArgs() {
+                    return false;
+                }
+            }
+
+        };
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/define/DefaultConnectingIOReactorInstrumentation.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4.define;
+
+import net.bytebuddy.description.method.MethodDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine;
+import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
+
+/**
+ * {@link DefaultConnectingIOReactorInstrumentation} presents that skywalking intercepts {@link
+ * org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor#processEvent}
+ *
+ * @author liyuntao
+ */
+
+public class DefaultConnectingIOReactorInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
+
+    private static final String ENHANCE_CLASS = "org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor";
+    private static final String INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.httpasyncclient.v4.DefaultConnectingIOReactorIterceptor";
+    private static final String LOCAL_INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.httpasyncclient.v4.ConnectIterceptor";
+
+    @Override
+    public ClassMatch enhanceClass() {
+        return byName(ENHANCE_CLASS);
+    }
+
+    @Override
+    protected ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
+        return null;
+    }
+
+    @Override
+    protected InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
+        return new InstanceMethodsInterceptPoint[] {
+            new InstanceMethodsInterceptPoint() {
+                @Override
+                public ElementMatcher<MethodDescription> getMethodsMatcher() {
+                    return named("processEvent");
+                }
+
+                @Override
+                public String getMethodsInterceptor() {
+                    return INTERCEPT_CLASS;
+                }
+
+                @Override
+                public boolean isOverrideArgs() {
+                    return false;
+                }
+            },
+            new InstanceMethodsInterceptPoint() {
+                @Override
+                public ElementMatcher<MethodDescription> getMethodsMatcher() {
+                    return named("connect");
+                }
+
+                @Override
+                public String getMethodsInterceptor() {
+                    return LOCAL_INTERCEPT_CLASS;
+                }
+
+                @Override
+                public boolean isOverrideArgs() {
+                    return false;
+                }
+            }
+        };
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/define/ExecuteInstrumentation.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4.define;
+
+import net.bytebuddy.description.method.MethodDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine;
+import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
+import static org.apache.skywalking.apm.agent.core.plugin.bytebuddy.ArgumentTypeNameMatch.takesArgumentWithType;
+import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
+
+/**
+ * {@link ExecuteInstrumentation} presents that skywalking intercepts {@link org.apache.http.impl.nio.client.CloseableHttpAsyncClient#execute}
+ *
+ * @author liyuntao
+ */
+
+public class ExecuteInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
+
+    private static final String ENHANCE_CLASS = "org.apache.http.impl.nio.client.CloseableHttpAsyncClient";
+    private static final String CONSUMER_INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.httpasyncclient.v4.HttpAsyncResponseConsumerInterceptor";
+    private static final String HOST_INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.httpasyncclient.v4.HttpHostInterceptor";
+
+    @Override
+    public ClassMatch enhanceClass() {
+        return byName(ENHANCE_CLASS);
+    }
+
+    @Override
+    protected ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
+        return null;
+    }
+
+    @Override
+    protected InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
+        return new InstanceMethodsInterceptPoint[] {
+            new InstanceMethodsInterceptPoint() {
+                @Override
+                public ElementMatcher<MethodDescription> getMethodsMatcher() {
+                    return named("execute").and(takesArgumentWithType(0, "org.apache.http.nio.protocol.HttpAsyncRequestProducer")).and(takesArguments(3));
+                }
+
+                @Override
+                public String getMethodsInterceptor() {
+                    return CONSUMER_INTERCEPT_CLASS;
+                }
+
+                @Override
+                public boolean isOverrideArgs() {
+                    return false;
+                }
+            },
+            new InstanceMethodsInterceptPoint() {
+                @Override
+                public ElementMatcher<MethodDescription> getMethodsMatcher() {
+                    return named("execute").and(takesArguments(4)).and(takesArgumentWithType(0, "org.apache.http.HttpHost"));
+                }
+
+                @Override
+                public String getMethodsInterceptor() {
+                    return HOST_INTERCEPT_CLASS;
+                }
+
+                @Override
+                public boolean isOverrideArgs() {
+                    return false;
+                }
+            }
+        };
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/define/ProcessResponseInstrumentation.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4.define;
+
+import net.bytebuddy.description.method.MethodDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine;
+import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
+
+/**
+ * {@link ProcessResponseInstrumentation} presents that skywalking intercept {@link
+ * org.apache.http.nio.protocol.HttpAsyncRequestExecutor#processResponse,#connected} .
+ *
+ * @author liyuntao
+ */
+
+public class ProcessResponseInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
+
+    private static final String ENHANCE_CLASS = "org.apache.http.nio.protocol.HttpAsyncRequestExecutor";
+    private static final String END_EXIT_INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.httpasyncclient.v4.ProcessResponseInterceptor";
+
+    @Override
+    public ClassMatch enhanceClass() {
+        return byName(ENHANCE_CLASS);
+    }
+
+    @Override
+    protected ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
+        return null;
+    }
+
+    @Override
+    protected InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
+        return new InstanceMethodsInterceptPoint[] {
+            new InstanceMethodsInterceptPoint() {
+                @Override
+                public ElementMatcher<MethodDescription> getMethodsMatcher() {
+                    return named("processResponse");
+                }
+
+                @Override
+                public String getMethodsInterceptor() {
+                    return END_EXIT_INTERCEPT_CLASS;
+                }
+
+                @Override
+                public boolean isOverrideArgs() {
+                    return false;
+                }
+            }
+
+        };
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/define/SessionRequestImplInstrumentation.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4.define;
+
+import net.bytebuddy.description.method.MethodDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine;
+import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
+
+/**
+ * {@link SessionRequestImplInstrumentation} presents that skywalking intercepts {@link
+ * org.apache.http.impl.nio.reactor.SessionRequestImpl#failed(final IOException exception)}
+ *
+ * @author liyuntao
+ */
+
+public class SessionRequestImplInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
+
+    private static final String ENHANCE_CLASS = "org.apache.http.impl.nio.reactor.SessionRequestImpl";
+    private static final String INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.httpasyncclient.v4.SessionRequestImplIterceptor";
+
+    @Override
+    public ClassMatch enhanceClass() {
+        return byName(ENHANCE_CLASS);
+    }
+
+    @Override
+    protected ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
+        return null;
+    }
+
+    @Override
+    protected InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
+        return new InstanceMethodsInterceptPoint[] {
+            new InstanceMethodsInterceptPoint() {
+                @Override
+                public ElementMatcher<MethodDescription> getMethodsMatcher() {
+                    return named("failed");
+                }
+
+                @Override
+                public String getMethodsInterceptor() {
+                    return INTERCEPT_CLASS;
+                }
+
+                @Override
+                public boolean isOverrideArgs() {
+                    return false;
+                }
+            }
+        };
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/define/StateInstrumentation.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4.define;
+
+import net.bytebuddy.description.method.MethodDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine;
+import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
+
+/**
+ * {@link StateInstrumentation} presents that skywalking intercept {@link org.apache.http.nio.protocol.HttpAsyncRequestExecutor$State#setRequest
+ * #setResponse} .
+ *
+ * @author liyuntao
+ */
+
+public class StateInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
+
+    private static final String ENHANCE_CLASS = "org.apache.http.nio.protocol.HttpAsyncRequestExecutor$State";
+    private static final String START_EXIT_INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.httpasyncclient.v4.StateInterceptor";
+    private static final String END_EXIT_INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.httpasyncclient.v4.SetResponseInterceptor";
+
+    @Override
+    public ClassMatch enhanceClass() {
+        return byName(ENHANCE_CLASS);
+    }
+
+    @Override
+    protected ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
+        return null;
+    }
+
+    @Override
+    protected InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
+        return new InstanceMethodsInterceptPoint[] {
+            new InstanceMethodsInterceptPoint() {
+                @Override
+                public ElementMatcher<MethodDescription> getMethodsMatcher() {
+                    return named("setRequest");
+                }
+
+                @Override
+                public String getMethodsInterceptor() {
+                    return START_EXIT_INTERCEPT_CLASS;
+                }
+
+                @Override
+                public boolean isOverrideArgs() {
+                    return false;
+                }
+            },
+            new InstanceMethodsInterceptPoint() {
+                @Override
+                public ElementMatcher<MethodDescription> getMethodsMatcher() {
+                    return named("setResponse");
+                }
+
+                @Override
+                public String getMethodsInterceptor() {
+                    return END_EXIT_INTERCEPT_CLASS;
+                }
+
+                @Override
+                public boolean isOverrideArgs() {
+                    return false;
+                }
+            }
+
+        };
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/resources/skywalking-plugin.def

+httpasyncclient-4.x=org.apache.skywalking.apm.plugin.httpasyncclient.v4.define.ExecuteInstrumentation
+httpasyncclient-4.x=org.apache.skywalking.apm.plugin.httpasyncclient.v4.define.DefaultConnectingIOReactorInstrumentation
+httpasyncclient-4.x=org.apache.skywalking.apm.plugin.httpasyncclient.v4.define.SessionRequestImplInstrumentation
+httpasyncclient-4.x=org.apache.skywalking.apm.plugin.httpasyncclient.v4.define.AbstractNIOConnPoolInstrumentation
+httpasyncclient-4.x=org.apache.skywalking.apm.plugin.httpasyncclient.v4.define.StateInstrumentation
+httpasyncclient-4.x=org.apache.skywalking.apm.plugin.httpasyncclient.v4.define.ProcessResponseInstrumentation
+
+

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/StateInterceptorTest.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4;
+
+import java.util.List;
+import org.apache.http.HttpHost;
+import org.apache.http.HttpRequest;
+import org.apache.http.HttpResponse;
+import org.apache.http.ProtocolVersion;
+import org.apache.http.RequestLine;
+import org.apache.http.StatusLine;
+import org.apache.http.client.methods.HttpRequestWrapper;
+import org.apache.skywalking.apm.agent.core.boot.ServiceManager;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment;
+import org.apache.skywalking.apm.agent.core.context.util.KeyValuePair;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.test.helper.SegmentHelper;
+import org.apache.skywalking.apm.agent.test.helper.SpanHelper;
+import org.apache.skywalking.apm.agent.test.tools.AgentServiceRule;
+import org.apache.skywalking.apm.agent.test.tools.SegmentStorage;
+import org.apache.skywalking.apm.agent.test.tools.SegmentStoragePoint;
+import org.apache.skywalking.apm.agent.test.tools.TracingSegmentRunner;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.powermock.modules.junit4.PowerMockRunnerDelegate;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@RunWith(PowerMockRunner.class)
+@PowerMockRunnerDelegate(TracingSegmentRunner.class)
+@PrepareForTest(HttpHost.class)
+public class StateInterceptorTest {
+
+    @SegmentStoragePoint
+    private SegmentStorage segmentStorage;
+
+    @Rule
+    public AgentServiceRule agentServiceRule = new AgentServiceRule();
+
+    private StateInterceptor stateInterceptor;
+
+    private SetResponseInterceptor setResponseInterceptor;
+
+    private ProcessResponseInterceptor processResponseInterceptor;
+    @Mock
+    private HttpHost httpHost;
+    @Mock
+    private HttpRequestWrapper request;
+    @Mock
+    private HttpRequest httpRequest;
+    @Mock
+    private HttpResponse httpResponse;
+    @Mock
+    private StatusLine statusLine;
+
+    private Object[] allArguments;
+    private Class[] argumentsType;
+
+    @Mock
+    private EnhancedInstance enhancedInstance;
+
+    @Before
+    public void setUp() throws Exception {
+        ServiceManager.INSTANCE.boot();
+        stateInterceptor = new StateInterceptor();
+        setResponseInterceptor = new SetResponseInterceptor();
+        processResponseInterceptor = new ProcessResponseInterceptor();
+
+        PowerMockito.mock(HttpHost.class);
+        when(statusLine.getStatusCode()).thenReturn(200);
+        when(httpResponse.getStatusLine()).thenReturn(statusLine);
+        when(httpHost.getHostName()).thenReturn("127.0.0.1");
+        when(httpHost.getSchemeName()).thenReturn("http");
+        when(request.getOriginal()).thenReturn(httpRequest);
+        when(httpRequest.getRequestLine()).thenReturn(new RequestLine() {
+            @Override
+            public String getMethod() {
+                return "GET";
+            }
+
+            @Override
+            public ProtocolVersion getProtocolVersion() {
+                return new ProtocolVersion("http", 1, 1);
+            }
+
+            @Override
+            public String getUri() {
+                return "http://127.0.0.1:8080/test-web/httpasync";
+            }
+        });
+        when(httpHost.getPort()).thenReturn(8080);
+
+        allArguments = new Object[] {request};
+        argumentsType = new Class[] {request.getClass()};
+    }
+
+    @Test
+    public void testHttpClient() throws Throwable {
+        AbstractSpan span = ContextManager.createLocalSpan("httpasyncclient/HttpAsyncRequestExecutor:");
+        stateInterceptor.beforeMethod(enhancedInstance, null, allArguments, argumentsType, null);
+        stateInterceptor.afterMethod(enhancedInstance, null, allArguments, argumentsType, httpResponse);
+        processResponseInterceptor.beforeMethod(enhancedInstance, null, allArguments, argumentsType, null);
+        processResponseInterceptor.afterMethod(enhancedInstance, null, allArguments, argumentsType, httpResponse);
+        Assert.assertThat(segmentStorage.getTraceSegments().size(), is(1));
+        TraceSegment traceSegment = segmentStorage.getTraceSegments().get(0);
+
+        List<AbstractTracingSpan> spans = SegmentHelper.getSpans(traceSegment);
+        assertHttpSpan(spans.get(0));
+        verify(request, times(1)).setHeader(anyString(), anyString());
+    }
+
+    @Test
+    public void testStatusCodeNotEquals200() throws Throwable {
+        when(statusLine.getStatusCode()).thenReturn(500);
+        AbstractSpan span = ContextManager.createLocalSpan("httpasyncclient/HttpAsyncRequestExecutor:");
+        stateInterceptor.beforeMethod(enhancedInstance, null, allArguments, argumentsType, null);
+        stateInterceptor.afterMethod(enhancedInstance, null, allArguments, argumentsType, httpResponse);
+        allArguments = new Object[] {httpResponse};
+        setResponseInterceptor.beforeMethod(enhancedInstance, null, allArguments, argumentsType, null);
+        setResponseInterceptor.afterMethod(enhancedInstance, null, allArguments, argumentsType, httpResponse);
+        processResponseInterceptor.beforeMethod(enhancedInstance, null, allArguments, argumentsType, null);
+        processResponseInterceptor.afterMethod(enhancedInstance, null, allArguments, argumentsType, httpResponse);
+
+        Assert.assertThat(segmentStorage.getTraceSegments().size(), is(1));
+        TraceSegment traceSegment = segmentStorage.getTraceSegments().get(0);
+        List<AbstractTracingSpan> spans = SegmentHelper.getSpans(traceSegment);
+
+        assertThat(spans.size(), is(3));
+
+        List<KeyValuePair> tags = SpanHelper.getTags(spans.get(0));
+        assertThat(tags.size(), is(3));
+        assertThat(tags.get(2).getValue(), is("500"));
+
+        assertHttpSpan(spans.get(0));
+        assertThat(SpanHelper.getErrorOccurred(spans.get(0)), is(true));
+        verify(request, times(1)).setHeader(anyString(), anyString());
+    }
+
+    private void assertHttpSpan(AbstractTracingSpan span) {
+        assertThat(span.getOperationName(), is("/test-web/httpasync"));
+        assertThat(SpanHelper.getComponentId(span), is(26));
+        List<KeyValuePair> tags = SpanHelper.getTags(span);
+        assertThat(tags.get(0).getValue(), is("http://127.0.0.1:8080/test-web/httpasync"));
+        assertThat(tags.get(1).getValue(), is("GET"));
+        assertThat(span.isExit(), is(true));
+    }
+
+}

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/TestException.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4;
+
+import java.util.List;
+import org.apache.http.HttpHost;
+import org.apache.http.HttpRequest;
+import org.apache.http.HttpResponse;
+import org.apache.http.ProtocolVersion;
+import org.apache.http.RequestLine;
+import org.apache.http.StatusLine;
+import org.apache.http.client.methods.HttpRequestWrapper;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.hamcrest.CoreMatchers;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.powermock.modules.junit4.PowerMockRunnerDelegate;
+import org.apache.skywalking.apm.agent.core.boot.ServiceManager;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.LogDataEntity;
+import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.test.helper.SegmentHelper;
+import org.apache.skywalking.apm.agent.test.helper.SpanHelper;
+import org.apache.skywalking.apm.agent.test.tools.AgentServiceRule;
+import org.apache.skywalking.apm.agent.test.tools.SegmentStorage;
+import org.apache.skywalking.apm.agent.test.tools.SegmentStoragePoint;
+import org.apache.skywalking.apm.agent.test.tools.TracingSegmentRunner;
+
+import static junit.framework.TestCase.assertNotNull;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * @auther lytscu
+ */
+@RunWith(PowerMockRunner.class)
+@PowerMockRunnerDelegate(TracingSegmentRunner.class)
+@PrepareForTest(HttpHost.class)
+public class TestException {
+    @SegmentStoragePoint
+    private SegmentStorage segmentStorage;
+
+    @Rule
+    public AgentServiceRule agentServiceRule = new AgentServiceRule();
+
+    private StateInterceptor stateInterceptor;
+
+    private SetResponseInterceptor setResponseInterceptor;
+
+    private ProcessResponseInterceptor processResponseInterceptor;
+    @Mock
+    private HttpHost httpHost;
+    @Mock
+    private HttpRequestWrapper request;
+    @Mock
+    private HttpRequest httpRequest;
+    @Mock
+    private HttpResponse httpResponse;
+    @Mock
+    private StatusLine statusLine;
+
+    private Object[] allArguments, setResponseInterceptorArguments;
+    private Class[] argumentsType;
+
+    @Mock
+    private EnhancedInstance enhancedInstance;
+
+    @Before
+    public void setUp() throws Exception {
+        ServiceManager.INSTANCE.boot();
+        stateInterceptor = new StateInterceptor();
+        setResponseInterceptor = new SetResponseInterceptor();
+        processResponseInterceptor = new ProcessResponseInterceptor();
+        PowerMockito.mock(HttpHost.class);
+        when(statusLine.getStatusCode()).thenReturn(200);
+        when(httpResponse.getStatusLine()).thenReturn(statusLine);
+        when(httpHost.getHostName()).thenReturn("127.0.0.1");
+        when(httpHost.getSchemeName()).thenReturn("http");
+        when(request.getOriginal()).thenReturn(httpRequest);
+        when(httpRequest.getRequestLine()).thenReturn(new RequestLine() {
+            @Override
+            public String getMethod() {
+                return "GET";
+            }
+
+            @Override
+            public ProtocolVersion getProtocolVersion() {
+                return new ProtocolVersion("http", 1, 1);
+            }
+
+            @Override
+            public String getUri() {
+                return "http://127.0.0.1:8080/test-web/httpasync";
+            }
+        });
+        when(httpHost.getPort()).thenReturn(8080);
+
+        allArguments = new Object[] {request};
+        setResponseInterceptorArguments = new Object[] {httpResponse};
+        argumentsType = new Class[] {request.getClass()};
+    }
+
+    @Test
+    public void testHttpClientWithException() throws Throwable {
+        AbstractSpan localSpan = ContextManager.createLocalSpan("httpasyncclient/HttpAsyncRequestExecutor:");
+        stateInterceptor.beforeMethod(enhancedInstance, null, allArguments, argumentsType, null);
+        stateInterceptor.handleMethodException(enhancedInstance, null, allArguments, argumentsType, new RuntimeException("testException"));
+        processResponseInterceptor.beforeMethod(enhancedInstance, null, allArguments, argumentsType, null);
+        processResponseInterceptor.afterMethod(enhancedInstance, null, allArguments, argumentsType, httpResponse);
+        Assert.assertThat(segmentStorage.getTraceSegments().size(), is(1));
+        TraceSegment traceSegment = segmentStorage.getTraceSegments().get(0);
+        List<AbstractTracingSpan> spans = SegmentHelper.getSpans(traceSegment);
+
+        assertThat(spans.size(), is(3));
+        AbstractTracingSpan span = spans.get(0);
+        assertThat(SpanHelper.getErrorOccurred(span), is(true));
+        assertHttpSpanErrorLog(SpanHelper.getLogs(span));
+        verify(request, times(1)).setHeader(anyString(), anyString());
+
+    }
+
+    private void assertHttpSpanErrorLog(List<LogDataEntity> logs) {
+        assertThat(logs.size(), is(1));
+        LogDataEntity logData = logs.get(0);
+        Assert.assertThat(logData.getLogs().size(), is(4));
+        Assert.assertThat(logData.getLogs().get(0).getValue(), CoreMatchers.<Object>is("error"));
+        Assert.assertThat(logData.getLogs().get(1).getValue(), CoreMatchers.<Object>is(RuntimeException.class.getName()));
+        Assert.assertThat(logData.getLogs().get(2).getValue(), is("testException"));
+        assertNotNull(logData.getLogs().get(3).getValue());
+    }
+}

apm-sniffer/apm-sdk-plugin/pom.xml

@@ -52,6 +52,7 @@
         <module>rocketMQ-4.x-plugin</module>
         <module>elastic-job-2.x-plugin</module>
         <module>mongodb-2.x-plugin</module>
+        <module>httpasyncclient-4.x-plugin</module>
     </modules>
     <packaging>pom</packaging>