VirtualThreadSchedulerMXBean metrics (#6104)

Adds virtual thread metrics based on the newly introduced MBean in Java 24.

Author: shakuzen

Diff for: .circleci/config.yml

@@ -76,6 +76,7 @@ jobs:
     executor: circle-jdk-executor
     steps:
       - gradlew-build
+      - run: ./gradlew jdk24Test
       - run: ./gradlew shenandoahTest
       - run: ./gradlew zgcTest
       - run: ./gradlew zgcGenerationalTest

Diff for: docs/modules/ROOT/pages/reference/jvm.adoc

@@ -54,11 +54,42 @@ To use the following `ExecutorService` instances, `--add-opens java.base/java.ut
 
 == Java 21 Metrics
 
-Micrometer provides support for https://openjdk.org/jeps/444[virtual threads] released in Java 21. In order to utilize it, you need to add the `io.micrometer:micrometer-java21` dependency to your classpath to use the binder:
+=== Virtual Threads
+
+Micrometer provides metrics for https://openjdk.org/jeps/444[virtual threads] released in Java 21. In order to utilize it, you need to add the `io.micrometer:micrometer-java21` dependency to your classpath to use the binder:
 
 [source, java]
 ----
 new VirtualThreadMetrics().bindTo(registry);
 ----
 
-The binder measures the duration (and counts the number of events) of virtual threads being pinned; also counts the number of events when starting or unparking a virtual thread failed.
+The binder measures the duration (and counts the number of events) of virtual threads being pinned; it also counts the number of events when starting or unparking a virtual thread failed.
+
+If you are running your application with Java 24 or later on a JVM that has `jdk.management.VirtualThreadSchedulerMXBean` provided as a platform MXBean, the following additional virtual thread metrics will be provided.
+
+
+|===
+|Meter name | Type | Tag(s) | Description
+
+|`jvm.threads.virtual.parallelism`
+|Gauge
+|
+|Virtual thread scheduler's target parallelism
+
+|`jvm.threads.virtual.pool.size`
+|Gauge
+|
+|Current number of platform threads that the scheduler has started but have not terminated; -1 if not known.
+
+|`jvm.threads.virtual.live`
+|Gauge
+|`scheduling.status` = `mounted`
+|Approximate current number of virtual threads that are unfinished and mounted to a platform thread by the scheduler
+
+|`jvm.threads.virtual.live`
+|Gauge
+|`scheduling.status` = `queued`
+|Approximate current number of virtual threads that are unfinished and queued waiting to be scheduled
+|===
+
+Note that aggregating the values of `jvm.threads.virtual.live` across the different tags gives the total number of virtual threads started but not ended.

Diff for: micrometer-java21/build.gradle

@@ -17,12 +17,22 @@ java {
 }
 
 tasks.withType(JavaCompile).configureEach {
-    sourceCompatibility = JavaVersion.VERSION_21
-    targetCompatibility = JavaVersion.VERSION_21
+    sourceCompatibility = 21
+    targetCompatibility = 21
     options.release = 21
 }
 
-task reflectiveTests(type: Test) {
+tasks.register('jdk24Test', Test) {
+    useJUnitPlatform {
+        includeTags 'jdk24'
+    }
+
+    javaLauncher = javaToolchains.launcherFor {
+        languageVersion = JavaLanguageVersion.of(24)
+    }
+}
+
+tasks.register('reflectiveTests', Test) {
     useJUnitPlatform {
         includeTags 'reflective'
     }
@@ -39,6 +49,6 @@ task reflectiveTests(type: Test) {
 test {
     dependsOn reflectiveTests
     useJUnitPlatform {
-        excludeTags 'reflective'
+        excludeTags 'reflective', 'jdk24'
     }
 }

Diff for: micrometer-java21/src/main/java/io/micrometer/java21/instrument/binder/jdk/VirtualThreadMetrics.java

@@ -15,24 +15,28 @@
  */
 package io.micrometer.java21.instrument.binder.jdk;
 
-import io.micrometer.core.instrument.Counter;
-import io.micrometer.core.instrument.MeterRegistry;
-import io.micrometer.core.instrument.Tag;
-import io.micrometer.core.instrument.Timer;
+import io.micrometer.core.instrument.*;
+import io.micrometer.core.instrument.binder.BaseUnits;
 import io.micrometer.core.instrument.binder.MeterBinder;
 import jdk.jfr.consumer.RecordingStream;
 
 import java.io.Closeable;
+import java.lang.invoke.MethodHandle;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.MethodType;
+import java.lang.management.ManagementFactory;
+import java.lang.management.PlatformManagedObject;
 import java.time.Duration;
 import java.util.Objects;
 
 import static java.util.Collections.emptyList;
 
 /**
- * Instrumentation support for Virtual Threads, see:
- * https://openjdk.org/jeps/425#JDK-Flight-Recorder-JFR
+ * Metrics instrumentation for Java Virtual Threads.
  *
  * @author Artyom Gabeev
+ * @see <a href="https://openjdk.org/jeps/425#JDK-Flight-Recorder-JFR">JEP 425</a>
+ * @see VirtualThreadSchedulerMXBean
  * @since 1.14.0
  */
 public class VirtualThreadMetrics implements MeterBinder, Closeable {
@@ -41,6 +45,10 @@ public class VirtualThreadMetrics implements MeterBinder, Closeable {
 
     private static final String SUBMIT_FAILED_EVENT = "jdk.VirtualThreadSubmitFailed";
 
+    private static final String LIVE_THREADS_DESCRIPTION = "Approximate current number of virtual threads that are unfinished";
+
+    private static final String METER_NAME_PREFIX = "jvm.threads.virtual.";
+
     private final RecordingStream recordingStream;
 
     private final Iterable<Tag> tags;
@@ -60,18 +68,71 @@ private VirtualThreadMetrics(RecordingConfig config, Iterable<Tag> tags) {
 
     @Override
     public void bindTo(MeterRegistry registry) {
-        Timer pinnedTimer = Timer.builder("jvm.threads.virtual.pinned")
+        Timer pinnedTimer = Timer.builder(METER_NAME_PREFIX + "pinned")
             .description("The duration while the virtual thread was pinned without releasing its platform thread")
             .tags(tags)
             .register(registry);
 
-        Counter submitFailedCounter = Counter.builder("jvm.threads.virtual.submit.failed")
+        Counter submitFailedCounter = Counter.builder(METER_NAME_PREFIX + "submit.failed")
             .description("The number of events when starting or unparking a virtual thread failed")
             .tags(tags)
             .register(registry);
 
         recordingStream.onEvent(PINNED_EVENT, event -> pinnedTimer.record(event.getDuration()));
         recordingStream.onEvent(SUBMIT_FAILED_EVENT, event -> submitFailedCounter.increment());
+
+        bindVirtualThreadSchedulerMXBean(registry);
+    }
+
+    private void bindVirtualThreadSchedulerMXBean(MeterRegistry registry) {
+        try {
+            Class clazz = Class.forName("jdk.management.VirtualThreadSchedulerMXBean");
+            PlatformManagedObject platformMXBean = ManagementFactory.getPlatformMXBean(clazz);
+            Object ignored = clazz.cast(platformMXBean);
+
+            MethodHandle getParallelism = MethodHandles.publicLookup()
+                .findVirtual(clazz, "getParallelism", MethodType.methodType(int.class));
+            MethodHandle getPoolSize = MethodHandles.publicLookup()
+                .findVirtual(clazz, "getPoolSize", MethodType.methodType(int.class));
+            MethodHandle getMountedVirtualThreadCount = MethodHandles.publicLookup()
+                .findVirtual(clazz, "getMountedVirtualThreadCount", MethodType.methodType(int.class));
+            MethodHandle getQueuedVirtualThreadCount = MethodHandles.publicLookup()
+                .findVirtual(clazz, "getQueuedVirtualThreadCount", MethodType.methodType(long.class));
+
+            Gauge.builder(METER_NAME_PREFIX + "parallelism", platformMXBean, o -> invoke(getParallelism, o))
+                .description("Virtual thread scheduler's target parallelism")
+                .register(registry);
+
+            Gauge.builder(METER_NAME_PREFIX + "pool.size", platformMXBean, o -> invoke(getPoolSize, o))
+                .baseUnit(BaseUnits.THREADS)
+                .description(
+                        "Current number of platform threads that the scheduler has started but have not terminated; -1 if not known.")
+                .register(registry);
+
+            Gauge.builder(METER_NAME_PREFIX + "live", platformMXBean, o -> invoke(getMountedVirtualThreadCount, o))
+                .tag("scheduling.status", "mounted")
+                .baseUnit(BaseUnits.THREADS)
+                .description(LIVE_THREADS_DESCRIPTION)
+                .register(registry);
+
+            Gauge.builder(METER_NAME_PREFIX + "live", platformMXBean, o -> invoke(getQueuedVirtualThreadCount, o))
+                .tag("scheduling.status", "queued")
+                .baseUnit(BaseUnits.THREADS)
+                .description(LIVE_THREADS_DESCRIPTION)
+                .register(registry);
+        }
+        catch (ClassNotFoundException | ClassCastException | NoSuchMethodException | IllegalAccessException ignored) {
+            // cannot instrument VirtualThreadSchedulerMXBean
+        }
+    }
+
+    private <T> double invoke(MethodHandle methodHandle, T arg) {
+        try {
+            return (double) methodHandle.invoke(arg);
+        }
+        catch (Throwable t) {
+            throw new RuntimeException(t);
+        }
     }
 
     private RecordingStream createRecordingStream(RecordingConfig config) {

Diff for: micrometer-java21/src/test/java/io/micrometer/java21/instrument/binder/jdk/VirtualThreadMetricsJdk24Tests.java

@@ -0,0 +1,102 @@
+/*
+ * Copyright 2025 VMware, Inc.
+ *
+ * Licensed 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
+ *
+ * https://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 io.micrometer.java21.instrument.binder.jdk;
+
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@Tag("jdk24")
+class VirtualThreadMetricsJdk24Tests {
+
+    MeterRegistry registry = new SimpleMeterRegistry();
+
+    VirtualThreadMetrics virtualThreadMetrics = new VirtualThreadMetrics();
+
+    @BeforeEach
+    void setUp() {
+        virtualThreadMetrics.bindTo(registry);
+    }
+
+    @AfterEach
+    void tearDown() {
+        virtualThreadMetrics.close();
+    }
+
+    @Test
+    void parallelism() {
+        int expectedParallelism = Runtime.getRuntime().availableProcessors();
+        assertThat(registry.get("jvm.threads.virtual.parallelism").gauge().value()).isEqualTo(expectedParallelism);
+    }
+
+    @Test
+    void poolSize() throws InterruptedException {
+        assertThat(registry.get("jvm.threads.virtual.pool.size").gauge().value()).isGreaterThanOrEqualTo(0);
+        Thread.ofVirtual()
+            .start(() -> assertThat(registry.get("jvm.threads.virtual.pool.size").gauge().value())
+                .isGreaterThanOrEqualTo(1))
+            .join(Duration.ofMillis(100));
+    }
+
+    @Test
+    void mountedThreads() throws InterruptedException {
+        Thread.ofVirtual()
+            .start(() -> assertThat(
+                    registry.get("jvm.threads.virtual.live").tag("scheduling.status", "mounted").gauge().value())
+                .isEqualTo(1d))
+            .join(Duration.ofMillis(100));
+    }
+
+    @Test
+    void queuedThreads() throws InterruptedException {
+        AtomicBoolean spin = new AtomicBoolean(true);
+        Runnable spinWait = () -> {
+            while (spin.get()) {
+                Thread.onSpinWait();
+            }
+        };
+        try (ExecutorService executorService = Executors.newVirtualThreadPerTaskExecutor()) {
+            int parallelism = Runtime.getRuntime().availableProcessors();
+            try {
+                for (int i = 0; i < parallelism; i++) {
+                    executorService.submit(spinWait);
+                }
+                int expectedQueuedThreads = 7;
+                for (int i = 0; i < expectedQueuedThreads; i++) {
+                    executorService.submit(() -> {
+                    });
+                }
+                assertThat(registry.get("jvm.threads.virtual.live").tag("scheduling.status", "queued").gauge().value())
+                    .isGreaterThanOrEqualTo(expectedQueuedThreads);
+            }
+            finally {
+                spin.set(false);
+                executorService.shutdown();
+                executorService.awaitTermination(100, TimeUnit.MILLISECONDS);
+            }
+        }
+    }
+
+}