Add Undertow 2.1.7.final+ worker thread pool metrics. (#744)

CHANGES.md

@@ -26,6 +26,7 @@ Release Notes.
 * Change context and parent entry span propagation mechanism from gRPC ThreadLocal context to SkyWalking native dynamic
   field as new propagation mechanism, to better support async scenarios. 
 * Add Caffeine plugin as optional.
+* Add Undertow 2.1.7.final+ worker thread pool metrics.
 
 All issues and pull requests are [here](https://github.com/apache/skywalking/milestone/222?closed=1)
 

apm-sniffer/apm-sdk-plugin/undertow-worker-thread-pool-plugin/pom.xml

@@ -29,5 +29,12 @@
     <name>undertow-worker-thread-pool-plugin</name>
     <url>http://maven.apache.org</url>
 
-
+    <dependencies>
+        <dependency>
+            <groupId>org.jboss.xnio</groupId>
+            <artifactId>xnio-api</artifactId>
+            <version>3.8.4.Final</version>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
 </project>

apm-sniffer/apm-sdk-plugin/undertow-worker-thread-pool-plugin/src/main/java/org/apache/skywalking/apm/plugin/undertow/worker/thread/pool/XnioWorkerConstructorInterceptor.java

@@ -0,0 +1,65 @@
+/*
+ * 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.undertow.worker.thread.pool;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.Function;
+import java.util.function.Supplier;
+import org.apache.skywalking.apm.agent.core.meter.MeterFactory;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceConstructorInterceptor;
+import org.apache.skywalking.apm.plugin.undertow.worker.thread.pool.util.XnioWorkerTaskPoolAccessor;
+import org.xnio.XnioWorker;
+
+public class XnioWorkerConstructorInterceptor implements InstanceConstructorInterceptor {
+
+    private static final String THREAD_POOL_NAME = "undertow_worker_pool";
+
+    private static final Map<String, Function<XnioWorkerTaskPoolAccessor, Supplier<Double>>> METRIC_MAP = new HashMap<String, Function<XnioWorkerTaskPoolAccessor, Supplier<Double>>>() {{
+        put("core_pool_size", (XnioWorkerTaskPoolAccessor threadPoolExecutor) -> () -> (double) threadPoolExecutor.getCorePoolSize());
+        put("max_pool_size", (XnioWorkerTaskPoolAccessor threadPoolExecutor) -> () -> (double) threadPoolExecutor.getMaximumPoolSize());
+        put("pool_size", (XnioWorkerTaskPoolAccessor threadPoolExecutor) -> () -> (double) threadPoolExecutor.getPoolSize());
+        put("queue_size", (XnioWorkerTaskPoolAccessor threadPoolExecutor) -> () -> (double) threadPoolExecutor.getQueueSize());
+        put("active_size", (XnioWorkerTaskPoolAccessor threadPoolExecutor) -> () -> (double) threadPoolExecutor.getActiveCount());
+    }};
+
+    @Override
+    public void onConstruct(EnhancedInstance objInst, Object[] allArguments) throws Throwable {
+        buildThreadPoolMeterMetric(new XnioWorkerTaskPoolAccessor((XnioWorker) objInst));
+    }
+
+    private void buildThreadPoolMeterMetric(XnioWorkerTaskPoolAccessor xnioWorkerTaskPoolAccessor) {
+        String threadPoolMeterName = "thread_pool";
+        String poolNameTag = "pool_name";
+        String metricTypeTag = "metric_type";
+        METRIC_MAP.forEach((key, value) -> {
+            if (Objects.equals(key, "pool_size")) {
+                if (xnioWorkerTaskPoolAccessor.isContainsGetPoolSizeMethod()) {
+                    MeterFactory.gauge(threadPoolMeterName, value.apply(xnioWorkerTaskPoolAccessor))
+                                .tag(poolNameTag, THREAD_POOL_NAME).tag(metricTypeTag, key).build();
+                }
+            } else {
+                MeterFactory.gauge(threadPoolMeterName, value.apply(xnioWorkerTaskPoolAccessor))
+                            .tag(poolNameTag, THREAD_POOL_NAME).tag(metricTypeTag, key).build();
+            }
+        });
+    }
+}

apm-sniffer/apm-sdk-plugin/undertow-worker-thread-pool-plugin/src/main/java/org/apache/skywalking/apm/plugin/undertow/worker/thread/pool/define/UndertowWorkerThreadPoolInstrumentation.java

@@ -19,32 +19,45 @@
 package org.apache.skywalking.apm.plugin.undertow.worker.thread.pool.define;
 
 import static net.bytebuddy.matcher.ElementMatchers.any;
-import static org.apache.skywalking.apm.agent.core.plugin.match.HierarchyMatch.byHierarchyMatch;
-import static org.apache.skywalking.apm.agent.core.plugin.match.PrefixMatch.nameStartsWith;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
 
+import java.util.Collections;
+import java.util.List;
 import net.bytebuddy.description.method.MethodDescription;
 import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.skywalking.apm.agent.core.plugin.WitnessMethod;
 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.StaticMethodsInterceptPoint;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassEnhancePluginDefine;
 import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
-import org.apache.skywalking.apm.agent.core.plugin.match.logical.LogicalMatchOperation;
 
+/**
+ * ThreadPoolExecutor implemented xnio worker task pool before 3.6.0
+ */
 public class UndertowWorkerThreadPoolInstrumentation extends ClassEnhancePluginDefine {
 
-    private static final String THREAD_POOL_EXECUTOR_CLASS = "java.util.concurrent.ThreadPoolExecutor";
+    private static final String THREAD_POOL_EXECUTOR_CLASS = "org.xnio.XnioWorker$TaskPool";
 
     private static final String UNDERTOW_WORKER_THREAD_POOL_INTERCEPT = "org.apache.skywalking.apm.plugin.undertow.worker.thread.pool.UndertowWorkerThreadPoolConstructorIntercept";
 
+    @Override
+    protected List<WitnessMethod> witnessMethods() {
+        return Collections.singletonList(new WitnessMethod(
+            "org.xnio.XnioWorker$TaskPool",
+            named("terminated")
+        ));
+    }
+
     @Override
     protected ClassMatch enhanceClass() {
-        return LogicalMatchOperation.and(nameStartsWith("org.xnio"), byHierarchyMatch(THREAD_POOL_EXECUTOR_CLASS));
+        return byName(THREAD_POOL_EXECUTOR_CLASS);
     }
 
     @Override
     public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
-        return new ConstructorInterceptPoint[]{
+        return new ConstructorInterceptPoint[] {
             new ConstructorInterceptPoint() {
                 @Override
                 public ElementMatcher<MethodDescription> getConstructorMatcher() {

apm-sniffer/apm-sdk-plugin/undertow-worker-thread-pool-plugin/src/main/java/org/apache/skywalking/apm/plugin/undertow/worker/thread/pool/define/XnioWorkerConstructorInstrumentation.java

@@ -0,0 +1,78 @@
+/*
+ * 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.undertow.worker.thread.pool.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.StaticMethodsInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassEnhancePluginDefine;
+import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
+
+import static net.bytebuddy.matcher.ElementMatchers.any;
+import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
+
+/**
+ * xnio task pool new implementation since 3.6.0
+ * https://github.com/xnio/xnio/commit/071800e0a85c9da9b88a976ac7ecb85760924dbf
+ */
+public class XnioWorkerConstructorInstrumentation extends ClassEnhancePluginDefine {
+
+    private static final String XNIO_WORKER_CLASS = "org.xnio.XnioWorker";
+
+    private static final String UNDERTOW_WORKER_THREAD_POOL_INTERCEPT = "org.apache.skywalking.apm.plugin.undertow.worker.thread.pool.XnioWorkerConstructorInterceptor";
+
+    @Override
+    protected String[] witnessClasses() {
+        return new String[] {"org.xnio.XnioWorker$EnhancedQueueExecutorTaskPool"};
+    }
+
+    @Override
+    protected ClassMatch enhanceClass() {
+        return byName(XNIO_WORKER_CLASS);
+    }
+
+    @Override
+    public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
+        return new ConstructorInterceptPoint[] {
+            new ConstructorInterceptPoint() {
+                @Override
+                public ElementMatcher<MethodDescription> getConstructorMatcher() {
+                    return any();
+                }
+
+                @Override
+                public String getConstructorInterceptor() {
+                    return UNDERTOW_WORKER_THREAD_POOL_INTERCEPT;
+                }
+            }
+        };
+    }
+
+    @Override
+    public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
+        return new InstanceMethodsInterceptPoint[0];
+    }
+
+    @Override
+    public StaticMethodsInterceptPoint[] getStaticMethodsInterceptPoints() {
+        return new StaticMethodsInterceptPoint[0];
+    }
+}

apm-sniffer/apm-sdk-plugin/undertow-worker-thread-pool-plugin/src/main/java/org/apache/skywalking/apm/plugin/undertow/worker/thread/pool/util/XnioWorkerTaskPoolAccessor.java

@@ -0,0 +1,117 @@
+/*
+ * 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.undertow.worker.thread.pool.util;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import lombok.Getter;
+import org.xnio.XnioWorker;
+
+public class XnioWorkerTaskPoolAccessor {
+
+    private final Object taskPool;
+    @Getter
+    private boolean containsGetPoolSizeMethod;
+
+    private Method getCorePoolSizeMethod;
+    private Method getMaximumPoolSizeMethod;
+    private Method getActiveCountMethod;
+    private Method getPoolSizeMethod;
+    private Method getQueueSizeMethod;
+
+    public XnioWorkerTaskPoolAccessor(final XnioWorker worker) throws NoSuchFieldException, IllegalAccessException {
+        Field field = worker.getClass().getSuperclass().getDeclaredField("taskPool");
+        field.setAccessible(true);
+        this.taskPool = field.get(worker);
+
+        try {
+            getCorePoolSizeMethod = taskPool.getClass().getDeclaredMethod("getCorePoolSize");
+            getCorePoolSizeMethod.setAccessible(true);
+        } catch (NoSuchMethodException e) {
+            // ignore
+        }
+        try {
+            getMaximumPoolSizeMethod = taskPool.getClass().getDeclaredMethod("getMaximumPoolSize");
+            getMaximumPoolSizeMethod.setAccessible(true);
+        } catch (NoSuchMethodException e) {
+            // ignore
+        }
+        try {
+            getActiveCountMethod = taskPool.getClass().getDeclaredMethod("getActiveCount");
+            getActiveCountMethod.setAccessible(true);
+        } catch (NoSuchMethodException e) {
+            // ignore
+        }
+        try {
+            // getPoolSize add since 3.8.0
+            getPoolSizeMethod = taskPool.getClass().getDeclaredMethod("getPoolSize");
+            getPoolSizeMethod.setAccessible(true);
+            containsGetPoolSizeMethod = true;
+        } catch (NoSuchMethodException e) {
+            containsGetPoolSizeMethod = false;
+        }
+        try {
+            getQueueSizeMethod = taskPool.getClass().getDeclaredMethod("getQueueSize");
+            getQueueSizeMethod.setAccessible(true);
+        } catch (NoSuchMethodException e) {
+            // ignore
+        }
+    }
+
+    public int getCorePoolSize() {
+        try {
+            return (int) getCorePoolSizeMethod.invoke(taskPool);
+        } catch (IllegalAccessException | InvocationTargetException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public int getMaximumPoolSize() {
+        try {
+            return (int) getMaximumPoolSizeMethod.invoke(taskPool);
+        } catch (IllegalAccessException | InvocationTargetException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public int getActiveCount() {
+        try {
+            return (int) getActiveCountMethod.invoke(taskPool);
+        } catch (IllegalAccessException | InvocationTargetException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public int getPoolSize() {
+        try {
+            return (int) getPoolSizeMethod.invoke(taskPool);
+        } catch (IllegalAccessException | InvocationTargetException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public int getQueueSize() {
+        try {
+            return (int) getQueueSizeMethod.invoke(taskPool);
+        } catch (IllegalAccessException | InvocationTargetException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}

apm-sniffer/apm-sdk-plugin/undertow-worker-thread-pool-plugin/src/main/resources/skywalking-plugin.def

@@ -15,3 +15,4 @@
 # limitations under the License.
 
 undertow-worker-thread-pool=org.apache.skywalking.apm.plugin.undertow.worker.thread.pool.define.UndertowWorkerThreadPoolInstrumentation
+undertow-worker-thread-pool=org.apache.skywalking.apm.plugin.undertow.worker.thread.pool.define.XnioWorkerConstructorInstrumentation

test/plugin/scenarios/undertow-worker-thread-pool-scenario/config/expectedData.yaml

@@ -22,29 +22,29 @@ meterItems:
           tags:
             - {name: metric_type, value: core_pool_size}
             - {name: pool_name, value: undertow_worker_pool}
-        singleValue: ge 1
+        singleValue: ge -1
       - meterId:
           name: thread_pool
           tags:
             - {name: metric_type, value: max_pool_size}
             - {name: pool_name, value: undertow_worker_pool}
-        singleValue: ge 1
+        singleValue: ge -1
       - meterId:
           name: thread_pool
           tags:
             - {name: metric_type, value: pool_size}
             - {name: pool_name, value: undertow_worker_pool}
-        singleValue: ge 0
+        singleValue: ge -1
       - meterId:
           name: thread_pool
           tags:
             - {name: metric_type, value: active_size}
             - {name: pool_name, value: undertow_worker_pool}
-        singleValue: ge 0
+        singleValue: ge -1
       - meterId:
           name: thread_pool
           tags:
             - {name: metric_type, value: queue_size}
             - {name: pool_name, value: undertow_worker_pool}
-        singleValue: ge 0
+        singleValue: ge -1
 

test/plugin/scenarios/undertow-worker-thread-pool-scenario/pom.xml

@@ -41,7 +41,7 @@
             <dependency>
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot-dependencies</artifactId>
-                <version>${spring-boot-version}</version>
+                <version>${test.framework.version}</version>
                 <type>pom</type>
                 <scope>import</scope>
             </dependency>