Support active thread metric for Executors.newVirtualThreadPerTaskExecutor() (#6008)

Support active thread metrics for Executors.newVirtualThreadPerTaskExecutor(), if reflection is allowed via the `--add-opens` option.

Closes gh-5488

Signed-off-by: Johnny Lim <[email protected]>

Author: izeye

Diff for: micrometer-core/src/main/java/io/micrometer/core/instrument/binder/jvm/ExecutorServiceMetrics.java

@@ -28,7 +28,10 @@
 import io.micrometer.core.instrument.internal.TimedExecutorService;
 import io.micrometer.core.instrument.internal.TimedScheduledExecutorService;
 
+import java.lang.invoke.MethodHandle;
+import java.lang.invoke.MethodHandles;
 import java.lang.reflect.Field;
+import java.lang.reflect.Method;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.*;
@@ -56,6 +59,11 @@
 @NonNullFields
 public class ExecutorServiceMetrics implements MeterBinder {
 
+    private static final String CLASS_NAME_THREAD_PER_TASK_EXECUTOR = "java.util.concurrent.ThreadPerTaskExecutor";
+
+    @Nullable
+    private static final MethodHandle METHOD_HANDLE_THREAD_COUNT_FROM_THREAD_PER_TASK_EXECUTOR = getMethodHandleForThreadCountFromThreadPerTaskExecutor();
+
     private static boolean allowIllegalReflectiveAccess = true;
 
     private static final InternalLogger log = InternalLoggerFactory.getInstance(ExecutorServiceMetrics.class);
@@ -315,6 +323,9 @@ else if (className.equals("java.util.concurrent.Executors$FinalizableDelegatedEx
                 monitor(registry,
                         unwrapThreadPoolExecutor(executorService, executorService.getClass().getSuperclass()));
             }
+            else if (className.equals(CLASS_NAME_THREAD_PER_TASK_EXECUTOR)) {
+                monitorThreadPerTaskExecutor(registry, executorService);
+            }
             else {
                 log.warn("Failed to bind as {} is unsupported.", className);
             }
@@ -439,6 +450,39 @@ private void monitor(MeterRegistry registry, ForkJoinPool fj) {
         registeredMeterIds.addAll(meters.stream().map(Meter::getId).collect(toSet()));
     }
 
+    private void monitorThreadPerTaskExecutor(MeterRegistry registry, ExecutorService executorService) {
+        List<Meter> meters = asList(Gauge
+            .builder(metricPrefix + "executor.active", executorService,
+                    ExecutorServiceMetrics::getThreadCountFromThreadPerTaskExecutor)
+            .tags(tags)
+            .description("The approximate number of threads that are actively executing tasks")
+            .baseUnit(BaseUnits.THREADS)
+            .register(registry));
+        registeredMeterIds.addAll(meters.stream().map(Meter::getId).collect(toSet()));
+    }
+
+    private static long getThreadCountFromThreadPerTaskExecutor(ExecutorService executorService) {
+        try {
+            return (long) METHOD_HANDLE_THREAD_COUNT_FROM_THREAD_PER_TASK_EXECUTOR.invoke(executorService);
+        }
+        catch (Throwable e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Nullable
+    private static MethodHandle getMethodHandleForThreadCountFromThreadPerTaskExecutor() {
+        try {
+            Class<?> clazz = Class.forName(CLASS_NAME_THREAD_PER_TASK_EXECUTOR);
+            Method method = clazz.getMethod("threadCount");
+            method.setAccessible(true);
+            return MethodHandles.lookup().unreflect(method);
+        }
+        catch (Throwable e) {
+            return null;
+        }
+    }
+
     /**
      * Disable illegal reflective accesses.
      *

Diff for: micrometer-java21/src/test/java/io/micrometer/core/instrument/binder/jvm/ExecutorServiceMetricsReflectiveTests.java

@@ -19,14 +19,17 @@
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 
+import java.time.Duration;
 import java.util.concurrent.*;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 
 /**
  * Tests for {@link ExecutorServiceMetrics} with reflection enabled.
  *
  * @author Tommy Ludwig
+ * @author Johnny Lim
  */
 @Tag("reflective")
 class ExecutorServiceMetricsReflectiveTests {
@@ -45,4 +48,25 @@ void threadPoolMetricsWith_AutoShutdownDelegatedExecutorService() throws Interru
         assertThat(registry.get("executor.completed").tag("name", "test").functionCounter().count()).isEqualTo(1L);
     }
 
+    @Test
+    void monitorWithExecutorsNewVirtualThreadPerTaskExecutor() {
+        CountDownLatch latch = new CountDownLatch(1);
+        ExecutorService unmonitored = Executors.newVirtualThreadPerTaskExecutor();
+        assertThat(unmonitored.getClass().getName()).isEqualTo("java.util.concurrent.ThreadPerTaskExecutor");
+        ExecutorService monitored = ExecutorServiceMetrics.monitor(registry, unmonitored, "test");
+        monitored.execute(() -> {
+            try {
+                latch.await(1, TimeUnit.SECONDS);
+            }
+            catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            }
+        });
+        await().atMost(Duration.ofSeconds(1))
+            .untilAsserted(() -> assertThat(registry.get("executor.active").gauge().value()).isEqualTo(1));
+        latch.countDown();
+        await().atMost(Duration.ofSeconds(1))
+            .untilAsserted(() -> assertThat(registry.get("executor.active").gauge().value()).isEqualTo(0));
+    }
+
 }