Reuse JMX Insights in Scraper + Tomcat support

jmx-scraper/build.gradle.kts

@@ -13,19 +13,32 @@ otelJava.moduleName.set("io.opentelemetry.contrib.jmxscraper")
 
 application.mainClass.set("io.opentelemetry.contrib.jmxscraper.JmxScraper")
 
+repositories {
+  mavenCentral()
+  mavenLocal()
+  // TODO: remove snapshot repository once 2.9.0 is released
+  maven {
+    setUrl("https://oss.sonatype.org/content/repositories/snapshots")
+  }
+}
+
 dependencies {
   // TODO remove snapshot dependency on upstream once 2.9.0 is released
-  // api(enforcedPlatform("io.opentelemetry.instrumentation:opentelemetry-instrumentation-bom-alpha:2.9.0-SNAPSHOT-alpha",))
+  api(enforcedPlatform("io.opentelemetry.instrumentation:opentelemetry-instrumentation-bom-alpha:2.9.0-alpha-SNAPSHOT"))
 
   implementation("io.opentelemetry:opentelemetry-api")
   implementation("io.opentelemetry:opentelemetry-sdk")
   implementation("io.opentelemetry:opentelemetry-sdk-metrics")
   implementation("io.opentelemetry:opentelemetry-sdk-extension-autoconfigure")
 
+  runtimeOnly("io.opentelemetry:opentelemetry-exporter-otlp")
+  runtimeOnly("io.opentelemetry:opentelemetry-exporter-logging")
+
   implementation("io.opentelemetry.instrumentation:opentelemetry-jmx-metrics")
 
   testImplementation("org.junit-pioneer:junit-pioneer")
   testImplementation("io.opentelemetry:opentelemetry-sdk-testing")
+  testImplementation("org.awaitility:awaitility")
 }
 
 testing {

jmx-scraper/src/integrationTest/java/io/opentelemetry/contrib/jmxscraper/JmxScraperContainer.java

@@ -76,6 +76,7 @@ public void start() {
     // for now only configure through JVM args
     List<String> arguments = new ArrayList<>();
     arguments.add("java");
+    arguments.add("-Dotel.metrics.exporter=otlp");
     arguments.add("-Dotel.exporter.otlp.endpoint=" + endpoint);
 
     if (!targetSystems.isEmpty()) {

jmx-scraper/src/integrationTest/java/io/opentelemetry/contrib/jmxscraper/target_systems/JvmIntegrationTest.java

@@ -5,9 +5,13 @@
 
 package io.opentelemetry.contrib.jmxscraper.target_systems;
 
+import static io.opentelemetry.contrib.jmxscraper.target_systems.MetricAssertions.assertGauge;
+import static io.opentelemetry.contrib.jmxscraper.target_systems.MetricAssertions.assertTypedGauge;
+import static io.opentelemetry.contrib.jmxscraper.target_systems.MetricAssertions.assertTypedSum;
+
 import io.opentelemetry.contrib.jmxscraper.JmxScraperContainer;
 import io.opentelemetry.contrib.jmxscraper.TestAppContainer;
-import io.opentelemetry.proto.collector.metrics.v1.ExportMetricsServiceRequest;
+import java.util.Arrays;
 import java.util.List;
 import org.testcontainers.containers.GenericContainer;
 
@@ -25,7 +29,55 @@ protected JmxScraperContainer customizeScraperContainer(JmxScraperContainer scra
   }
 
   @Override
-  protected void verifyMetrics(List<ExportMetricsServiceRequest> metrics) {
-    // TODO: Verify gathered metrics
+  protected void verifyMetrics() {
+    // those values depend on the JVM GC configured
+    List<String> gcLabels =
+        Arrays.asList(
+            "Code Cache",
+            "PS Eden Space",
+            "PS Old Gen",
+            "Metaspace",
+            "Compressed Class Space",
+            "PS Survivor Space");
+    List<String> gcCollectionLabels = Arrays.asList("PS MarkSweep", "PS Scavenge");
+
+    waitAndAssertMetrics(
+        metric -> assertGauge(metric, "jvm.classes.loaded", "number of loaded classes", "1"),
+        metric ->
+            assertTypedSum(
+                metric,
+                "jvm.gc.collections.count",
+                "total number of collections that have occurred",
+                "1",
+                gcCollectionLabels),
+        metric ->
+            assertTypedSum(
+                metric,
+                "jvm.gc.collections.elapsed",
+                "the approximate accumulated collection elapsed time in milliseconds",
+                "ms",
+                gcCollectionLabels),
+        metric -> assertGauge(metric, "jvm.memory.heap.committed", "current heap usage", "by"),
+        metric -> assertGauge(metric, "jvm.memory.heap.init", "current heap usage", "by"),
+        metric -> assertGauge(metric, "jvm.memory.heap.max", "current heap usage", "by"),
+        metric -> assertGauge(metric, "jvm.memory.heap.used", "current heap usage", "by"),
+        metric ->
+            assertGauge(metric, "jvm.memory.nonheap.committed", "current non-heap usage", "by"),
+        metric -> assertGauge(metric, "jvm.memory.nonheap.init", "current non-heap usage", "by"),
+        metric -> assertGauge(metric, "jvm.memory.nonheap.max", "current non-heap usage", "by"),
+        metric -> assertGauge(metric, "jvm.memory.nonheap.used", "current non-heap usage", "by"),
+        metric ->
+            assertTypedGauge(
+                metric, "jvm.memory.pool.committed", "current memory pool usage", "by", gcLabels),
+        metric ->
+            assertTypedGauge(
+                metric, "jvm.memory.pool.init", "current memory pool usage", "by", gcLabels),
+        metric ->
+            assertTypedGauge(
+                metric, "jvm.memory.pool.max", "current memory pool usage", "by", gcLabels),
+        metric ->
+            assertTypedGauge(
+                metric, "jvm.memory.pool.used", "current memory pool usage", "by", gcLabels),
+        metric -> assertGauge(metric, "jvm.threads.count", "number of threads", "1"));
   }
 }

jmx-scraper/src/integrationTest/java/io/opentelemetry/contrib/jmxscraper/target_systems/MetricAssertions.java

@@ -0,0 +1,126 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.jmxscraper.target_systems;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.entry;
+
+import io.opentelemetry.proto.common.v1.KeyValue;
+import io.opentelemetry.proto.metrics.v1.Metric;
+import io.opentelemetry.proto.metrics.v1.NumberDataPoint;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+import org.assertj.core.api.MapAssert;
+
+/** Metrics assertions */
+class MetricAssertions {
+
+  private MetricAssertions() {}
+
+  static void assertGauge(Metric metric, String name, String description, String unit) {
+    assertThat(metric.getName()).isEqualTo(name);
+    assertThat(metric.getDescription()).isEqualTo(description);
+    assertThat(metric.getUnit()).isEqualTo(unit);
+    assertThat(metric.hasGauge()).isTrue();
+    assertThat(metric.getGauge().getDataPointsList())
+        .satisfiesExactly(point -> assertThat(point.getAttributesList()).isEmpty());
+  }
+
+  static void assertSum(Metric metric, String name, String description, String unit) {
+    assertSum(metric, name, description, unit, /* isMonotonic= */ true);
+  }
+
+  static void assertSum(
+      Metric metric, String name, String description, String unit, boolean isMonotonic) {
+    assertThat(metric.getName()).isEqualTo(name);
+    assertThat(metric.getDescription()).isEqualTo(description);
+    assertThat(metric.getUnit()).isEqualTo(unit);
+    assertThat(metric.hasSum()).isTrue();
+    assertThat(metric.getSum().getDataPointsList())
+        .satisfiesExactly(point -> assertThat(point.getAttributesList()).isEmpty());
+    assertThat(metric.getSum().getIsMonotonic()).isEqualTo(isMonotonic);
+  }
+
+  static void assertTypedGauge(
+      Metric metric, String name, String description, String unit, List<String> types) {
+    assertThat(metric.getName()).isEqualTo(name);
+    assertThat(metric.getDescription()).isEqualTo(description);
+    assertThat(metric.getUnit()).isEqualTo(unit);
+    assertThat(metric.hasGauge()).isTrue();
+    assertTypedPoints(metric.getGauge().getDataPointsList(), types);
+  }
+
+  static void assertTypedSum(
+      Metric metric, String name, String description, String unit, List<String> types) {
+    assertThat(metric.getName()).isEqualTo(name);
+    assertThat(metric.getDescription()).isEqualTo(description);
+    assertThat(metric.getUnit()).isEqualTo(unit);
+    assertThat(metric.hasSum()).isTrue();
+    assertTypedPoints(metric.getSum().getDataPointsList(), types);
+  }
+
+  @SafeVarargs
+  static void assertSumWithAttributes(
+      Metric metric,
+      String name,
+      String description,
+      String unit,
+      Consumer<MapAssert<String, String>>... attributeGroupAssertions) {
+    assertThat(metric.getName()).isEqualTo(name);
+    assertThat(metric.getDescription()).isEqualTo(description);
+    assertThat(metric.getUnit()).isEqualTo(unit);
+    assertThat(metric.hasSum()).isTrue();
+    assertAttributedPoints(metric.getSum().getDataPointsList(), attributeGroupAssertions);
+  }
+
+  @SafeVarargs
+  static void assertGaugeWithAttributes(
+      Metric metric,
+      String name,
+      String description,
+      String unit,
+      Consumer<MapAssert<String, String>>... attributeGroupAssertions) {
+    assertThat(metric.getName()).isEqualTo(name);
+    assertThat(metric.getDescription()).isEqualTo(description);
+    assertThat(metric.getUnit()).isEqualTo(unit);
+    assertThat(metric.hasGauge()).isTrue();
+    assertAttributedPoints(metric.getGauge().getDataPointsList(), attributeGroupAssertions);
+  }
+
+  @SuppressWarnings("unchecked")
+  private static void assertTypedPoints(List<NumberDataPoint> points, List<String> types) {
+    Consumer<MapAssert<String, String>>[] assertions =
+        types.stream()
+            .map(
+                type ->
+                    (Consumer<MapAssert<String, String>>)
+                        attrs -> attrs.containsOnly(entry("name", type)))
+            .toArray(Consumer[]::new);
+
+    assertAttributedPoints(points, assertions);
+  }
+
+  @SuppressWarnings("unchecked")
+  private static void assertAttributedPoints(
+      List<NumberDataPoint> points,
+      Consumer<MapAssert<String, String>>... attributeGroupAssertions) {
+    Consumer<Map<String, String>>[] assertions =
+        Arrays.stream(attributeGroupAssertions)
+            .map(assertion -> (Consumer<Map<String, String>>) m -> assertion.accept(assertThat(m)))
+            .toArray(Consumer[]::new);
+    assertThat(points)
+        .extracting(
+            numberDataPoint ->
+                numberDataPoint.getAttributesList().stream()
+                    .collect(
+                        Collectors.toMap(
+                            KeyValue::getKey, keyValue -> keyValue.getValue().getStringValue())))
+        .satisfiesExactlyInAnyOrder(assertions);
+  }
+}

jmx-scraper/src/integrationTest/java/io/opentelemetry/contrib/jmxscraper/target_systems/TargetSystemIntegrationTest.java

@@ -6,23 +6,28 @@
 package io.opentelemetry.contrib.jmxscraper.target_systems;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 
 import com.linecorp.armeria.server.ServerBuilder;
 import com.linecorp.armeria.server.grpc.GrpcService;
 import com.linecorp.armeria.testing.junit5.server.ServerExtension;
 import io.grpc.stub.StreamObserver;
-import io.opentelemetry.contrib.jmxscraper.JmxConnectorBuilder;
 import io.opentelemetry.contrib.jmxscraper.JmxScraperContainer;
 import io.opentelemetry.proto.collector.metrics.v1.ExportMetricsServiceRequest;
 import io.opentelemetry.proto.collector.metrics.v1.ExportMetricsServiceResponse;
 import io.opentelemetry.proto.collector.metrics.v1.MetricsServiceGrpc;
-import java.io.IOException;
+import io.opentelemetry.proto.metrics.v1.Metric;
+import io.opentelemetry.proto.metrics.v1.ResourceMetrics;
+import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.LinkedBlockingDeque;
-import javax.management.remote.JMXConnector;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeAll;
@@ -93,7 +98,7 @@ void endToEndTest() {
 
     target =
         createTargetContainer(JMX_PORT)
-            .withLogConsumer(new Slf4jLogConsumer(logger))
+            .withLogConsumer(new Slf4jLogConsumer(logger).withPrefix("target-system"))
             .withNetwork(network)
             .withExposedPorts(JMX_PORT)
             .withNetworkAliases(TARGET_SYSTEM_NETWORK_ALIAS);
@@ -104,26 +109,54 @@ void endToEndTest() {
     logger.info(
         "Target system started, JMX port: {} mapped to {}:{}", JMX_PORT, targetHost, targetPort);
 
-    // TODO : wait for metrics to be sent and add assertions on what is being captured
-    // for now we just test that we can connect to remote JMX using our client.
-    try (JMXConnector connector = JmxConnectorBuilder.createNew(targetHost, targetPort).build()) {
-      assertThat(connector.getMBeanServerConnection()).isNotNull();
-    } catch (IOException e) {
-      throw new RuntimeException(e);
-    }
-
     scraper =
         new JmxScraperContainer(otlpEndpoint)
+            .withLogConsumer(new Slf4jLogConsumer(logger).withPrefix("jmx-scraper"))
             .withNetwork(network)
             .withService(TARGET_SYSTEM_NETWORK_ALIAS, JMX_PORT);
 
     scraper = customizeScraperContainer(scraper);
     scraper.start();
 
-    verifyMetrics(otlpServer.getMetrics());
+    verifyMetrics();
   }
 
-  protected abstract void verifyMetrics(List<ExportMetricsServiceRequest> metrics);
+  protected void waitAndAssertMetrics(Iterable<Consumer<Metric>> assertions) {
+    await()
+        .atMost(Duration.ofSeconds(30))
+        .untilAsserted(
+            () -> {
+              List<ExportMetricsServiceRequest> receivedMetrics = otlpServer.getMetrics();
+              assertThat(receivedMetrics).describedAs("no metric received").isNotEmpty();
+
+              List<Metric> metrics =
+                  receivedMetrics.stream()
+                      .map(ExportMetricsServiceRequest::getResourceMetricsList)
+                      .flatMap(rm -> rm.stream().map(ResourceMetrics::getScopeMetricsList))
+                      .flatMap(Collection::stream)
+                      .filter(
+                          // TODO: disabling batch span exporter might help remove unwanted metrics
+                          sm -> sm.getScope().getName().equals("io.opentelemetry.jmx"))
+                      .flatMap(sm -> sm.getMetricsList().stream())
+                      .collect(Collectors.toList());
+
+              assertThat(metrics)
+                  .describedAs("metrics reported but none from JMX scraper")
+                  .isNotEmpty();
+
+              for (Consumer<Metric> assertion : assertions) {
+                assertThat(metrics).anySatisfy(assertion);
+              }
+            });
+  }
+
+  @SafeVarargs
+  @SuppressWarnings("varargs")
+  protected final void waitAndAssertMetrics(Consumer<Metric>... assertions) {
+    waitAndAssertMetrics(Arrays.asList(assertions));
+  }
+
+  protected abstract void verifyMetrics();
 
   protected JmxScraperContainer customizeScraperContainer(JmxScraperContainer scraper) {
     return scraper;
@@ -152,6 +185,10 @@ protected void configure(ServerBuilder sb) {
                     public void export(
                         ExportMetricsServiceRequest request,
                         StreamObserver<ExportMetricsServiceResponse> responseObserver) {
+
+                      // verbose but helpful to diagnose what is received
+                      logger.info("receiving metrics {}", request);
+
                       metricRequests.add(request);
                       responseObserver.onNext(ExportMetricsServiceResponse.getDefaultInstance());
                       responseObserver.onCompleted();