[Feature] Add JDK ThreadPoolExecutor plugin(#8743)

.github/workflows/plugins-test.3.yaml

@@ -98,6 +98,7 @@ jobs:
           - tomcat-thread-pool-scenario
           - guava-eventbus-scenario
           - shenyu-2.4.x-scenario
+          - jdk-threadpool-scenario
     steps:
       - uses: actions/checkout@v2
         with:

CHANGES.md

@@ -36,6 +36,7 @@ Release Notes.
 * Make sure the parent endpoint in tracing context from existing first ENTRY span, rather than first span only.
 * Fix the bug that maybe causing memory leak and repeated traceId when use gateway-2.1.x-plugin or gateway-3.x-plugin.
 * Fix Grpc 1.x plugin could leak context due to gRPC cancelled.
+* Add JDK ThreadPoolExecutor Plugin.
 
 #### Documentation
 

apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/logging/core/WriterFactory.java

@@ -22,6 +22,7 @@
 import org.apache.skywalking.apm.agent.core.boot.AgentPackagePath;
 import org.apache.skywalking.apm.agent.core.conf.Config;
 import org.apache.skywalking.apm.agent.core.conf.SnifferConfigInitializer;
+import org.apache.skywalking.apm.agent.core.plugin.PluginFinder;
 import org.apache.skywalking.apm.util.StringUtil;
 
 public class WriterFactory {
@@ -35,7 +36,9 @@ public static IWriter getLogWriter() {
                 if (WRITER != null) {
                     return WRITER;
                 }
-                if (SnifferConfigInitializer.isInitCompleted() && AgentPackagePath.isPathFound()) {
+                if (SnifferConfigInitializer.isInitCompleted()
+                        && PluginFinder.isPluginInitCompleted()
+                        && AgentPackagePath.isPathFound()) {
                     if (StringUtil.isEmpty(Config.Logging.DIR)) {
                         try {
                             Config.Logging.DIR = AgentPackagePath.getPath() + "/logs";

apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/plugin/PluginFinder.java

@@ -43,6 +43,7 @@ public class PluginFinder {
     private final Map<String, LinkedList<AbstractClassEnhancePluginDefine>> nameMatchDefine = new HashMap<String, LinkedList<AbstractClassEnhancePluginDefine>>();
     private final List<AbstractClassEnhancePluginDefine> signatureMatchDefine = new ArrayList<AbstractClassEnhancePluginDefine>();
     private final List<AbstractClassEnhancePluginDefine> bootstrapClassMatchDefine = new ArrayList<AbstractClassEnhancePluginDefine>();
+    private static boolean IS_PLUGIN_INIT_COMPLETED = false;
 
     public PluginFinder(List<AbstractClassEnhancePluginDefine> plugins) {
         for (AbstractClassEnhancePluginDefine plugin : plugins) {
@@ -107,4 +108,12 @@ public boolean matches(NamedElement target) {
     public List<AbstractClassEnhancePluginDefine> getBootstrapClassMatchDefine() {
         return bootstrapClassMatchDefine;
     }
+
+    public static void pluginInitCompleted() {
+        IS_PLUGIN_INIT_COMPLETED = true;
+    }
+
+    public static boolean isPluginInitCompleted() {
+        return IS_PLUGIN_INIT_COMPLETED;
+    }
 }

apm-sniffer/apm-agent-core/src/test/java/org/apache/skywalking/apm/agent/core/logging/core/WriterFactoryTest.java

@@ -21,6 +21,7 @@
 import org.apache.skywalking.apm.agent.core.boot.AgentPackagePath;
 import org.apache.skywalking.apm.agent.core.conf.Config;
 import org.apache.skywalking.apm.agent.core.conf.SnifferConfigInitializer;
+import org.apache.skywalking.apm.agent.core.plugin.PluginFinder;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.BDDMockito;
@@ -33,6 +34,7 @@
 @RunWith(PowerMockRunner.class)
 @PrepareForTest(value = {
     SnifferConfigInitializer.class,
+    PluginFinder.class,
     AgentPackagePath.class
 })
 public class WriterFactoryTest {
@@ -41,11 +43,14 @@ public class WriterFactoryTest {
     public void alwaysReturnSystemLogWriteWithSetLoggingDir() {
         Config.Logging.OUTPUT = LogOutput.CONSOLE;
         PowerMockito.mockStatic(SnifferConfigInitializer.class);
+        PowerMockito.mockStatic(PluginFinder.class);
         PowerMockito.mockStatic(AgentPackagePath.class);
         BDDMockito.given(SnifferConfigInitializer.isInitCompleted()).willReturn(true);
+        BDDMockito.given(PluginFinder.isPluginInitCompleted()).willReturn(true);
         BDDMockito.given(AgentPackagePath.isPathFound()).willReturn(true);
 
         assertTrue(SnifferConfigInitializer.isInitCompleted());
+        assertTrue(PluginFinder.isPluginInitCompleted());
         assertTrue(AgentPackagePath.isPathFound());
 
         IWriter logWriter = WriterFactory.getLogWriter();
@@ -56,11 +61,14 @@ public void alwaysReturnSystemLogWriteWithSetLoggingDir() {
     public void returnFileWriterWriteWithBlankLoggingDir() {
         Config.Logging.OUTPUT = LogOutput.FILE;
         PowerMockito.mockStatic(SnifferConfigInitializer.class);
+        PowerMockito.mockStatic(PluginFinder.class);
         PowerMockito.mockStatic(AgentPackagePath.class);
         BDDMockito.given(SnifferConfigInitializer.isInitCompleted()).willReturn(true);
+        BDDMockito.given(PluginFinder.isPluginInitCompleted()).willReturn(true);
         BDDMockito.given(AgentPackagePath.isPathFound()).willReturn(true);
 
         assertTrue(SnifferConfigInitializer.isInitCompleted());
+        assertTrue(PluginFinder.isPluginInitCompleted());
         assertTrue(AgentPackagePath.isPathFound());
 
         IWriter logWriter = WriterFactory.getLogWriter();

apm-sniffer/apm-agent/src/main/java/org/apache/skywalking/apm/agent/SkyWalkingAgent.java

@@ -130,6 +130,8 @@ public static void premain(String agentArgs, Instrumentation instrumentation) th
                     .with(new Listener())
                     .installOn(instrumentation);
 
+        PluginFinder.pluginInitCompleted();
+
         try {
             ServiceManager.INSTANCE.boot();
         } catch (Exception e) {

apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/pom.xml

@@ -0,0 +1,48 @@
+<!--
+  ~ 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 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">
+    <parent>
+        <artifactId>bootstrap-plugins</artifactId>
+        <groupId>org.apache.skywalking</groupId>
+        <version>8.10.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>apm-jdk-threadpool-plugin</artifactId>
+    <packaging>jar</packaging>
+
+    <name>apm-jdk-threadpool-plugin</name>
+    <url>http://maven.apache.org</url>
+
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
+
+
+    <build>
+        <plugins>
+            <plugin>
+                <artifactId>maven-deploy-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>

apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/java/org/apache/skywalking/apm/plugin/AbstractThreadingPoolInterceptor.java

@@ -0,0 +1,62 @@
+/*
+ * 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;
+
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+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 java.lang.reflect.Method;
+
+public abstract class AbstractThreadingPoolInterceptor implements InstanceMethodsAroundInterceptor {
+    @Override
+    public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
+        if (!ContextManager.isActive()) {
+            return;
+        }
+
+        if (allArguments == null || allArguments.length < 1) {
+            return;
+        }
+
+        Object argument = allArguments[0];
+
+        Object wrappedObject = wrap(argument);
+        if (wrappedObject != null) {
+            allArguments[0] = wrappedObject;
+        }
+    }
+
+    /**
+     * wrap the Callable or Runnable object if needed
+     * @param param  Callable or Runnable object
+     * @return Wrapped object or null if not needed to wrap
+     */
+    public abstract Object wrap(Object param);
+
+    @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) {
+        ContextManager.activeSpan().log(t);
+    }
+}

apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/java/org/apache/skywalking/apm/plugin/ThreadPoolExecuteMethodInterceptor.java

@@ -0,0 +1,45 @@
+/*
+ * 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;
+
+import org.apache.skywalking.apm.plugin.wrapper.SwRunnableWrapper;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import java.util.concurrent.RunnableFuture;
+
+public class ThreadPoolExecuteMethodInterceptor extends AbstractThreadingPoolInterceptor {
+
+    @Override
+    public Object wrap(Object param) {
+        if (param instanceof SwRunnableWrapper) {
+            return null;
+        }
+
+        if (param instanceof RunnableFuture) {
+            return null;
+        }
+
+        if (!(param instanceof Runnable)) {
+            return null;
+        }
+
+        Runnable runnable = (Runnable) param;
+        return new SwRunnableWrapper(runnable, ContextManager.capture());
+    }
+
+}

apm-sniffer/bootstrap-plugins/jdk-threadpool-plugin/src/main/java/org/apache/skywalking/apm/plugin/ThreadPoolSubmitMethodInterceptor.java

@@ -0,0 +1,47 @@
+/*
+ * 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;
+
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.plugin.wrapper.SwCallableWrapper;
+import org.apache.skywalking.apm.plugin.wrapper.SwRunnableWrapper;
+
+import java.util.concurrent.Callable;
+
+public class ThreadPoolSubmitMethodInterceptor extends AbstractThreadingPoolInterceptor {
+
+    @Override
+    public Object wrap(Object param) {
+        if (param instanceof SwRunnableWrapper || param instanceof SwCallableWrapper) {
+            return null;
+        }
+
+        if (param instanceof Callable) {
+            Callable callable = (Callable) param;
+            return new SwCallableWrapper(callable, ContextManager.capture());
+        }
+
+        if (param instanceof Runnable) {
+            Runnable runnable = (Runnable) param;
+            return new SwRunnableWrapper(runnable, ContextManager.capture());
+        }
+
+        return null;
+    }
+}