Thread count and classes loaded handlers (#571)

**Description:**
Add handlers for thread count and loaded classes count metrics. These
are derived from `jdk.ClassLoadingStatistics` and
`jdk.JavaThreadStatistics` JFR events

**Testing:**
A new unit test was added to test these handlers in
`jfr-streaming/src/test/java/io/opentelemetry/contrib/jfr/metrics`.

Co-authored-by: Mateusz Rzeszutek <[email protected]>

jfr-streaming/src/main/java/io/opentelemetry/contrib/jfr/metrics/HandlerRegistry.java

@@ -8,6 +8,7 @@
 import io.opentelemetry.api.metrics.MeterProvider;
 import io.opentelemetry.contrib.jfr.metrics.internal.RecordedEventHandler;
 import io.opentelemetry.contrib.jfr.metrics.internal.ThreadGrouper;
+import io.opentelemetry.contrib.jfr.metrics.internal.classes.ClassesLoadedHandler;
 import io.opentelemetry.contrib.jfr.metrics.internal.container.ContainerConfigurationHandler;
 import io.opentelemetry.contrib.jfr.metrics.internal.cpu.ContextSwitchRateHandler;
 import io.opentelemetry.contrib.jfr.metrics.internal.cpu.LongLockHandler;
@@ -19,6 +20,7 @@
 import io.opentelemetry.contrib.jfr.metrics.internal.memory.ParallelHeapSummaryHandler;
 import io.opentelemetry.contrib.jfr.metrics.internal.network.NetworkReadHandler;
 import io.opentelemetry.contrib.jfr.metrics.internal.network.NetworkWriteHandler;
+import io.opentelemetry.contrib.jfr.metrics.internal.threads.ThreadCountHandler;
 import java.lang.management.ManagementFactory;
 import java.util.ArrayList;
 import java.util.HashSet;
@@ -68,7 +70,9 @@ static HandlerRegistry createDefault(MeterProvider meterProvider) {
             new ContextSwitchRateHandler(),
             new OverallCPULoadHandler(),
             new ContainerConfigurationHandler(),
-            new LongLockHandler(grouper));
+            new LongLockHandler(grouper),
+            new ThreadCountHandler(),
+            new ClassesLoadedHandler());
     handlers.addAll(basicHandlers);
 
     var meter =

jfr-streaming/src/main/java/io/opentelemetry/contrib/jfr/metrics/internal/Constants.java

@@ -27,6 +27,7 @@ private Constants() {}
   public static final String REGION_COUNT = "region.count";
   public static final String COMMITTED = "committed";
   public static final String RESERVED = "reserved";
+  public static final String DAEMON = "daemon";
 
   public static final String METRIC_NAME_NETWORK_BYTES = "process.runtime.jvm.network.io";
   public static final String METRIC_DESCRIPTION_NETWORK_BYTES = "Network read/write bytes";
@@ -43,4 +44,7 @@ private Constants() {}
   public static final AttributeKey<String> ATTR_ARENA_NAME = AttributeKey.stringKey("arena");
   public static final AttributeKey<String> ATTR_NETWORK_MODE = AttributeKey.stringKey("mode");
   public static final AttributeKey<String> ATTR_USAGE = AttributeKey.stringKey("usage.type");
+  public static final AttributeKey<Boolean> ATTR_DAEMON = AttributeKey.booleanKey(DAEMON);
+  public static final String UNIT_CLASSES = "{classes}";
+  public static final String UNIT_THREADS = "{threads}";
 }

jfr-streaming/src/main/java/io/opentelemetry/contrib/jfr/metrics/internal/classes/ClassesLoadedHandler.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.jfr.metrics.internal.classes;
+
+import static io.opentelemetry.contrib.jfr.metrics.internal.Constants.UNIT_CLASSES;
+import static io.opentelemetry.contrib.jfr.metrics.internal.RecordedEventHandler.defaultMeter;
+
+import io.opentelemetry.api.metrics.Meter;
+import io.opentelemetry.contrib.jfr.metrics.internal.RecordedEventHandler;
+import java.time.Duration;
+import java.util.Optional;
+import jdk.jfr.consumer.RecordedEvent;
+
+public final class ClassesLoadedHandler implements RecordedEventHandler {
+  /**
+   * process.runtime.jvm.classes.loaded is the total number of classes loaded since JVM start. See:
+   * https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/semantic_conventions/runtime-environment-metrics.md#jvm-metrics
+   */
+  private static final String METRIC_NAME_LOADED = "process.runtime.jvm.classes.loaded";
+
+  private static final String METRIC_NAME_UNLOADED = "process.runtime.jvm.classes.unloaded";
+  /**
+   * process.runtime.jvm.classes.current_loaded is the number of classes loaded at the time of
+   * jdk.ClassLoadingStatistics event emission.
+   */
+  private static final String METRIC_NAME_CURRENT = "process.runtime.jvm.classes.current_loaded";
+
+  private static final String EVENT_NAME = "jdk.ClassLoadingStatistics";
+  private static final String METRIC_DESCRIPTION_CURRENT = "Number of classes currently loaded";
+  private static final String METRIC_DESCRIPTION_LOADED =
+      "Number of classes loaded since JVM start";
+  private static final String METRIC_DESCRIPTION_UNLOADED =
+      "Number of classes unloaded since JVM start";
+  private volatile long loaded = 0;
+  private volatile long unloaded = 0;
+
+  public ClassesLoadedHandler() {
+    initializeMeter(defaultMeter());
+  }
+
+  @Override
+  public void accept(RecordedEvent ev) {
+    loaded = ev.getLong("loadedClassCount");
+    unloaded = ev.getLong("unloadedClassCount");
+  }
+
+  @Override
+  public String getEventName() {
+    return EVENT_NAME;
+  }
+
+  @Override
+  public void initializeMeter(Meter meter) {
+    meter
+        .upDownCounterBuilder(METRIC_NAME_CURRENT)
+        .setDescription(METRIC_DESCRIPTION_CURRENT)
+        .setUnit(UNIT_CLASSES)
+        .buildWithCallback(measurement -> measurement.record(loaded - unloaded));
+    meter
+        .counterBuilder(METRIC_NAME_LOADED)
+        .setDescription(METRIC_DESCRIPTION_LOADED)
+        .setUnit(UNIT_CLASSES)
+        .buildWithCallback(measurement -> measurement.record(loaded));
+    meter
+        .counterBuilder(METRIC_NAME_UNLOADED)
+        .setDescription(METRIC_DESCRIPTION_UNLOADED)
+        .setUnit(UNIT_CLASSES)
+        .buildWithCallback(measurement -> measurement.record(unloaded));
+  }
+
+  @Override
+  public Optional<Duration> getPollingDuration() {
+    return Optional.of(Duration.ofSeconds(1));
+  }
+}

jfr-streaming/src/main/java/io/opentelemetry/contrib/jfr/metrics/internal/threads/ThreadCountHandler.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.jfr.metrics.internal.threads;
+
+import static io.opentelemetry.contrib.jfr.metrics.internal.Constants.ATTR_DAEMON;
+import static io.opentelemetry.contrib.jfr.metrics.internal.Constants.UNIT_THREADS;
+import static io.opentelemetry.contrib.jfr.metrics.internal.RecordedEventHandler.defaultMeter;
+
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.metrics.Meter;
+import io.opentelemetry.contrib.jfr.metrics.internal.RecordedEventHandler;
+import java.time.Duration;
+import java.util.Optional;
+import jdk.jfr.consumer.RecordedEvent;
+
+public final class ThreadCountHandler implements RecordedEventHandler {
+  private static final String METRIC_NAME = "process.runtime.jvm.threads.count";
+  private static final String EVENT_NAME = "jdk.JavaThreadStatistics";
+  private static final String METRIC_DESCRIPTION = "Number of executing threads";
+  private static final Attributes ATTR_DAEMON_TRUE = Attributes.of(ATTR_DAEMON, true);
+  private static final Attributes ATTR_DAEMON_FALSE = Attributes.of(ATTR_DAEMON, false);
+  private volatile long activeCount = 0;
+  private volatile long daemonCount = 0;
+
+  public ThreadCountHandler() {
+    initializeMeter(defaultMeter());
+  }
+
+  @Override
+  public void accept(RecordedEvent ev) {
+    activeCount = ev.getLong("activeCount");
+    daemonCount = ev.getLong("daemonCount");
+  }
+
+  @Override
+  public String getEventName() {
+    return EVENT_NAME;
+  }
+
+  @Override
+  public void initializeMeter(Meter meter) {
+    meter
+        .upDownCounterBuilder(METRIC_NAME)
+        .setDescription(METRIC_DESCRIPTION)
+        .setUnit(UNIT_THREADS)
+        .buildWithCallback(
+            measurement -> {
+              long d = daemonCount;
+              measurement.record(d, ATTR_DAEMON_TRUE);
+              measurement.record(activeCount - d, ATTR_DAEMON_FALSE);
+            });
+  }
+
+  @Override
+  public Optional<Duration> getPollingDuration() {
+    return Optional.of(Duration.ofSeconds(1));
+  }
+}

jfr-streaming/src/test/java/io/opentelemetry/contrib/jfr/metrics/AbstractMetricsTest.java

@@ -22,9 +22,14 @@ public class AbstractMetricsTest {
 
   static SdkMeterProvider meterProvider;
   static InMemoryMetricReader metricReader;
+  static boolean isInitialized = false;
 
   @BeforeAll
   static void initializeOpenTelemetry() {
+    if (isInitialized) {
+      return;
+    }
+    isInitialized = true;
     metricReader = InMemoryMetricReader.create();
     meterProvider = SdkMeterProvider.builder().registerMetricReader(metricReader).build();
     GlobalOpenTelemetry.set(OpenTelemetrySdk.builder().setMeterProvider(meterProvider).build());

jfr-streaming/src/test/java/io/opentelemetry/contrib/jfr/metrics/JfrClassesLoadedCountTest.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.jfr.metrics;
+
+import static io.opentelemetry.contrib.jfr.metrics.internal.Constants.UNIT_CLASSES;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class JfrClassesLoadedCountTest extends AbstractMetricsTest {
+
+  @Test
+  void shouldHaveJfrLoadedClassesCountEvents() throws Exception {
+    Thread.sleep(2000);
+
+    waitAndAssertMetrics(
+        metric ->
+            metric
+                .hasName("process.runtime.jvm.classes.loaded")
+                .hasDescription("Number of classes loaded since JVM start")
+                .hasUnit(UNIT_CLASSES)
+                .hasLongSumSatisfying(
+                    sum ->
+                        sum.hasPointsSatisfying(
+                            point ->
+                                point.satisfies(
+                                    pointData -> Assertions.assertTrue(pointData.getValue() > 0)))),
+        metric ->
+            metric
+                .hasName("process.runtime.jvm.classes.current_loaded")
+                .hasDescription("Number of classes currently loaded")
+                .hasUnit(UNIT_CLASSES)
+                .hasLongSumSatisfying(
+                    sum ->
+                        sum.hasPointsSatisfying(
+                            point ->
+                                point.satisfies(
+                                    pointData ->
+                                        Assertions.assertTrue(pointData.getValue() >= 0)))),
+        metric ->
+            metric
+                .hasName("process.runtime.jvm.classes.unloaded")
+                .hasDescription("Number of classes unloaded since JVM start")
+                .hasUnit(UNIT_CLASSES)
+                .hasLongSumSatisfying(
+                    sum ->
+                        sum.hasPointsSatisfying(
+                            point ->
+                                point.satisfies(
+                                    pointData ->
+                                        Assertions.assertTrue(pointData.getValue() >= 0)))));
+  }
+}

jfr-streaming/src/test/java/io/opentelemetry/contrib/jfr/metrics/JfrThreadCountTest.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.jfr.metrics;
+
+import static io.opentelemetry.contrib.jfr.metrics.internal.Constants.DAEMON;
+import static io.opentelemetry.contrib.jfr.metrics.internal.Constants.UNIT_THREADS;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.sdk.metrics.data.LongPointData;
+import io.opentelemetry.sdk.metrics.data.SumData;
+import org.junit.jupiter.api.Test;
+
+class JfrThreadCountTest extends AbstractMetricsTest {
+  private static final int SAMPLING_INTERVAL = 1000;
+
+  private static void doWork() throws InterruptedException {
+    Thread.sleep(2 * SAMPLING_INTERVAL);
+  }
+
+  private static boolean isDaemon(LongPointData p) {
+    Boolean daemon = p.getAttributes().get(AttributeKey.booleanKey(DAEMON));
+    assertThat(daemon).isNotNull();
+    return daemon;
+  }
+
+  @Test
+  void shouldHaveJfrThreadCountEvents() throws Exception {
+    // This should generate some events
+    Runnable work =
+        () -> {
+          // create contention between threads for one lock
+          try {
+            doWork();
+          } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+          }
+        };
+    Thread userThread = new Thread(work);
+    userThread.setDaemon(false);
+    userThread.start();
+
+    Thread daemonThread = new Thread(work);
+    daemonThread.setDaemon(true);
+    daemonThread.start();
+
+    userThread.join();
+    daemonThread.join();
+
+    waitAndAssertMetrics(
+        metric ->
+            metric
+                .hasName("process.runtime.jvm.threads.count")
+                .hasUnit(UNIT_THREADS)
+                .satisfies(
+                    metricData -> {
+                      SumData<?> sumData = metricData.getLongSumData();
+                      assertThat(sumData.getPoints())
+                          .map(LongPointData.class::cast)
+                          .anyMatch(p -> p.getValue() > 0 && isDaemon(p))
+                          .anyMatch(p -> p.getValue() > 0 && !isDaemon(p));
+                    }));
+  }
+}