Add Java Flight Recorder support

spectator-ext-jvm/build.gradle

@@ -1,12 +1,58 @@
+sourceSets {
+  java17 {
+    java {
+      srcDirs = ['src/main/java17']
+      compileClasspath = configurations.compileClasspath
+      runtimeClasspath = configurations.runtimeClasspath
+    }
+  }
+  java17Test {
+    java {
+      srcDirs = ['src/test/java17']
+      compileClasspath = jar.outputs.files + configurations.testCompileClasspath
+      runtimeClasspath = jar.outputs.files + runtimeClasspath + configurations.testRuntimeClasspath
+    }
+  }
+}
+
 dependencies {
   api project(':spectator-api')
   implementation 'com.typesafe:config'
 }
 
-jar {
+def java17Compiler = javaToolchains.compilerFor {
+  languageVersion = JavaLanguageVersion.of(17)
+}
+
+tasks.named('compileJava17Java', JavaCompile).configure {
+  javaCompiler = java17Compiler
+}
+
+tasks.named('compileJava17TestJava', JavaCompile).configure {
+  javaCompiler = java17Compiler
+}
+
+tasks.named('jar').configure {
+  into('META-INF/versions/17') {
+    from sourceSets.java17.output
+  }
   manifest {
     attributes(
-      "Automatic-Module-Name": "com.netflix.spectator.jvm"
+      'Automatic-Module-Name': 'com.netflix.spectator.jvm',
+      'Multi-Release': 'true'
     )
   }
 }
+
+def testJava17 = tasks.register('testJava17', Test) {
+  description = "Runs tests for java17Test sourceset."
+  group = 'verification'
+
+  testClassesDirs = sourceSets.java17Test.output.classesDirs
+  classpath = sourceSets.java17Test.runtimeClasspath
+
+  javaLauncher = javaToolchains.launcherFor {
+    languageVersion = JavaLanguageVersion.of(17)
+  }
+}
+check.dependsOn testJava17

spectator-ext-jvm/src/main/java/com/netflix/spectator/jvm/JavaFlightRecorder.java

@@ -0,0 +1,34 @@
+package com.netflix.spectator.jvm;
+
+import com.netflix.spectator.api.Registry;
+
+import java.util.concurrent.Executor;
+
+/**
+ * Helpers supporting continuous monitoring with Java Flight Recorder.
+ */
+public class JavaFlightRecorder {
+
+  /**
+   * Return if Java Flight Recorder continuous monitoring is supported on the current JVM.
+   */
+  public static boolean isSupported() {
+    return false;
+  }
+
+ /**
+   * Collect low-overhead Java Flight Recorder events, using the provided
+   * {@link java.util.concurrent.Executor} to execute a single task to collect events.
+   * <p>
+   * These measures provide parity with {@link Jmx#registerStandardMXBeans} and the
+   * `spectator-ext-gc` module.
+   *
+   * @param registry the registry
+   * @param executor the executor to execute the task for streaming events
+   * @return an {@link AutoCloseable} allowing the underlying event stream to be closed
+   */
+  public static AutoCloseable monitorDefaultEvents(Registry registry, Executor executor) {
+    throw new UnsupportedOperationException("Java Flight Recorder support is only available on Java 17 and later");
+  }
+
+}

spectator-ext-jvm/src/main/java17/com/netflix/spectator/jvm/JavaFlightRecorder.java

@@ -0,0 +1,136 @@
+package com.netflix.spectator.jvm;
+
+import com.netflix.spectator.api.Registry;
+import jdk.jfr.EventSettings;
+import jdk.jfr.consumer.RecordedEvent;
+import jdk.jfr.consumer.RecordingStream;
+
+import java.time.Duration;
+import java.util.Objects;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Consumer;
+
+public class JavaFlightRecorder {
+
+  private static final String PREFIX = "jdk.";
+  private static final String ClassLoadingStatistics = PREFIX + "ClassLoadingStatistics";
+  private static final String CompilerStatistics = PREFIX + "CompilerStatistics";
+  private static final String JavaThreadStatistics = PREFIX + "JavaThreadStatistics";
+  private static final String VirtualThreadPinned = PREFIX + "VirtualThreadPinned";
+  private static final String VirtualThreadSubmitFailed = PREFIX + "VirtualThreadSubmitFailed";
+  private static final String ZAllocationStall = PREFIX + "ZAllocationStall";
+  private static final String ZYoungGarbageCollection = PREFIX + "ZYoungGarbageCollection";
+  private static final String ZOldGarbageCollection = PREFIX + "ZOldGarbageCollection";
+
+  private JavaFlightRecorder() {
+  }
+
+  public static boolean isSupported() {
+    try {
+        Class.forName("jdk.jfr.consumer.RecordingStream");
+    } catch (ClassNotFoundException e) {
+        return false;
+    }
+    return true;
+  }
+
+  public static AutoCloseable monitorDefaultEvents(Registry registry, Executor executor) {
+    if (!isSupported()) {
+      throw new UnsupportedOperationException("This JVM does not support Java Flight Recorder event streaming");
+    }
+    Objects.requireNonNull(registry);
+    Objects.requireNonNull(executor);
+    RecordingStream rs = new RecordingStream();
+    collectClassLoadingStatistics(registry, rs);
+    collectCompilerStatistics(registry, rs);
+    collectThreadStatistics(registry, rs);
+    collectVirtualThreadEvents(registry, rs);
+    collectZgcEvents(registry, rs);
+    executor.execute(rs::start);
+    return rs::close;
+  }
+
+  private static void collectClassLoadingStatistics(Registry registry, RecordingStream rs) {
+    AtomicLong prevLoadedClassCount = new AtomicLong();
+    AtomicLong prevUnloadedClassCount = new AtomicLong();
+    consume(ClassLoadingStatistics, rs, event -> {
+      long classesLoaded = event.getLong("loadedClassCount");
+      classesLoaded = classesLoaded - prevLoadedClassCount.getAndSet(classesLoaded);
+      registry.counter("jvm.classloading.classesLoaded").increment(classesLoaded);
+
+      long classesUnloaded = event.getLong("unloadedClassCount");
+      classesUnloaded = classesUnloaded - prevUnloadedClassCount.getAndSet(classesUnloaded);
+      registry.counter("jvm.classloading.classesUnloaded").increment(classesUnloaded);
+    });
+  }
+
+  private static void collectCompilerStatistics(Registry registry, RecordingStream rs) {
+    AtomicLong prevTotalTimeSpent = new AtomicLong();
+    consume(CompilerStatistics, rs, event -> {
+      long totalTimeSpent = event.getLong("totalTimeSpent");
+      totalTimeSpent = totalTimeSpent - prevTotalTimeSpent.getAndAdd(totalTimeSpent);
+      registry.counter("jvm.compilation.compilationTime").add(totalTimeSpent / 1000.0);
+    });
+  }
+
+  private static void collectThreadStatistics(Registry registry, RecordingStream rs) {
+    AtomicLong prevAccumulatedCount = new AtomicLong();
+    consume(JavaThreadStatistics, rs, event -> {
+      long activeCount = event.getLong("activeCount");
+      long daemonCount = event.getLong("daemonCount");
+      long nonDaemonCount = activeCount - daemonCount;
+      registry.gauge("jvm.thread.threadCount", "id", "non-daemon").set(nonDaemonCount);
+      registry.gauge("jvm.thread.threadCount", "id", "daemon").set(daemonCount);
+      long accumulatedCount = event.getLong("accumulatedCount");
+      long threadsStarted = accumulatedCount - prevAccumulatedCount.getAndSet(accumulatedCount);
+      registry.counter("jvm.thread.threadsStarted").increment(threadsStarted);
+    });
+  }
+
+  private static void collectVirtualThreadEvents(Registry registry, RecordingStream rs) {
+    consume(VirtualThreadPinned, rs, event ->
+      registry.timer("jvm.vt.pinned").record(event.getDuration())
+    ).withThreshold(Duration.ofMillis(20));
+    consume(VirtualThreadSubmitFailed, rs, event ->
+      registry.counter("jvm.vt.submitFailed").increment()
+    );
+  }
+
+  private static void collectZgcEvents(Registry registry, RecordingStream rs) {
+    consume(ZYoungGarbageCollection, rs, event ->
+      registry.timer("jvm.zgc.youngCollection", "tenuringThreshold", event.getString("tenuringThreshold"))
+        .record(event.getDuration()));
+
+    consume(ZOldGarbageCollection, rs, event ->
+      registry.timer("jvm.zgc.oldCollection")
+        .record(event.getDuration()));
+
+    consume(ZAllocationStall, rs, event ->
+      registry.timer("jvm.zgc.allocationStall", "type", event.getString("type"))
+        .record(event.getDuration()));
+  }
+
+  /**
+   * Consume a given JFR event. For full event details see the event definitions and default/profiling configuration:
+   * <p>
+   * - <a href="https://github.com/openjdk/jdk/blob/master/src/hotspot/share/jfr/metadata/metadata.xml">metadata.xml</a>
+   * - <a href="https://github.com/openjdk/jdk/blob/master/src/jdk.jfr/share/conf/jfr/default.jfc">default.jfc</a>
+   * - <a href="https://github.com/openjdk/jdk/blob/master/src/jdk.jfr/share/conf/jfr/profile.jfc">profile.jfc</a>
+   * <p>
+   * We avoid the default event configurations because despite their claims of "low-overhead" there are
+   * situtations where they can impose significant overhead to the application.
+   */
+  private static EventSettings consume(String name, RecordingStream rs, Consumer<RecordedEvent> consumer) {
+    // Apply sensible defaults to settings to avoid the overhead of collecting unnecessary stacktraces
+    // and collecting periodic events at a finer interval than we require upstream
+    EventSettings settings = rs.enable(name)
+      .withoutStackTrace()
+      .withThreshold(Duration.ofMillis(0))
+      .withPeriod(Duration.ofSeconds(5));
+    rs.onEvent(name, consumer);
+    return settings;
+  }
+
+}

spectator-ext-jvm/src/test/java/com/netflix/spectator/jvm/JavaFlightRecorderUnsupportedTest.java

@@ -0,0 +1,20 @@
+package com.netflix.spectator.jvm;
+
+import com.netflix.spectator.api.NoopRegistry;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class JavaFlightRecorderUnsupportedTest {
+
+    @Test
+    public void isUnsupported() {
+        Assertions.assertFalse(JavaFlightRecorder.isSupported());
+    }
+
+    @Test
+    public void monitorThrowsUOE() {
+        Assertions.assertThrows(UnsupportedOperationException.class, () ->
+          JavaFlightRecorder.monitorDefaultEvents(new NoopRegistry(), Runnable::run));
+    }
+
+}

spectator-ext-jvm/src/test/java17/com/netflix/spectator/jvm/JavaFlightRecorderTest.java

@@ -0,0 +1,57 @@
+package com.netflix.spectator.jvm;
+
+import com.netflix.spectator.api.DefaultRegistry;
+import com.netflix.spectator.api.Id;
+import com.netflix.spectator.api.Measurement;
+import com.netflix.spectator.api.Registry;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.stream.Collectors;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+public class JavaFlightRecorderTest {
+
+  @Test
+  public void isSupported() {
+    assertTrue(JavaFlightRecorder.isSupported());
+  }
+
+  @Test
+  public void checkDefaultMeasures() throws Exception {
+    Registry registry = new DefaultRegistry();
+    ExecutorService executor = Executors.newSingleThreadExecutor();
+    try (var closable = JavaFlightRecorder.monitorDefaultEvents(registry, executor)) {
+      Thread.sleep(6000);
+    }
+    executor.shutdownNow();
+
+    Map<Id, Measurement> measures = registry.measurements()
+      .collect(Collectors.toMap(Measurement::id, m -> m));
+
+    Measurement classesLoaded = measures.get(Id.create("jvm.classloading.classesLoaded"));
+    Measurement classesUnloaded = measures.get(Id.create("jvm.classloading.classesUnloaded"));
+    assertNotEquals(null, classesLoaded);
+    assertNotEquals(null, classesUnloaded);
+    assertTrue(classesLoaded.value() > 3000 && classesLoaded.value() < 4000);
+    assertEquals(0, classesUnloaded.value());
+
+    Measurement compilationTime = measures.get(Id.create("jvm.compilation.compilationTime"));
+    assertNotEquals(null, compilationTime);
+
+    Measurement nonDaemonThreadCount = measures.get(Id.create("jvm.thread.threadCount").withTag("id", "non-daemon"));
+    Measurement daemonThreadCount = measures.get(Id.create("jvm.thread.threadCount").withTag("id", "daemon"));
+    Measurement threadsStarted = measures.get(Id.create("jvm.thread.threadsStarted"));
+    assertNotEquals(null, nonDaemonThreadCount);
+    assertEquals(5, nonDaemonThreadCount.value());
+    assertNotEquals(null, daemonThreadCount);
+    assertEquals(7, daemonThreadCount.value());
+    assertNotEquals(null, threadsStarted);
+    assertEquals(12, threadsStarted.value());
+  }
+
+}