Integration test activemq

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/JmxConnectorBuilderTest.java

@@ -31,23 +31,25 @@ static void afterAll() {
 
   @Test
   void noAuth() {
-    try (TestAppContainer app = new TestAppContainer().withNetwork(network).withJmxPort(9990)) {
+    int port = PortSelector.getAvailableRandomPort();
+    try (TestAppContainer app = new TestAppContainer().withHostAccessFixedJmxPort(port)) {
       app.start();
       testConnector(
-          () -> JmxConnectorBuilder.createNew(app.getHost(), app.getMappedPort(9990)).build());
+          () -> JmxConnectorBuilder.createNew(app.getHost(), app.getMappedPort(port)).build());
     }
   }
 
   @Test
   void loginPwdAuth() {
+    int port = PortSelector.getAvailableRandomPort();
     String login = "user";
     String pwd = "t0p!Secret";
     try (TestAppContainer app =
-        new TestAppContainer().withNetwork(network).withJmxPort(9999).withUserAuth(login, pwd)) {
+        new TestAppContainer().withHostAccessFixedJmxPort(port).withUserAuth(login, pwd)) {
       app.start();
       testConnector(
           () ->
-              JmxConnectorBuilder.createNew(app.getHost(), app.getMappedPort(9999))
+              JmxConnectorBuilder.createNew(app.getHost(), app.getMappedPort(port))
                   .userCredentials(login, pwd)
                   .build());
     }

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/PortSelector.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.jmxscraper;
+
+import java.io.IOException;
+import java.net.Socket;
+import java.util.Random;
+
+/** Class used for finding random free network port from range 1024-65535 */
+public class PortSelector {
+  private static final Random random = new Random(System.currentTimeMillis());
+
+  private static final int MIN_PORT = 1024;
+  private static final int MAX_PORT = 65535;
+
+  private PortSelector() {}
+
+  /**
+   * @return random available TCP port
+   */
+  public static synchronized int getAvailableRandomPort() {
+    int port;
+
+    do {
+      port = random.nextInt(MAX_PORT - MIN_PORT + 1) + MIN_PORT;
+    } while (!isPortAvailable(port));
+
+    return port;
+  }
+
+  private static boolean isPortAvailable(int port) {
+    // see https://stackoverflow.com/a/13826145 for the chosen implementation
+    try (Socket s = new Socket("localhost", port)) {
+      return false;
+    } catch (IOException e) {
+      return true;
+    }
+  }
+}

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

@@ -25,7 +25,6 @@
 public class TestAppContainer extends GenericContainer<TestAppContainer> {
 
   private final Map<String, String> properties;
-  private int port;
   private String login;
   private String pwd;
 
@@ -44,11 +43,16 @@ public TestAppContainer() {
         .withCommand("java", "-jar", "/app.jar");
   }
 
+  /**
+   * Configures app container for container-to-container access
+   *
+   * @param port mapped port to use
+   * @return this
+   */
   @CanIgnoreReturnValue
   public TestAppContainer withJmxPort(int port) {
-    this.port = port;
     properties.put("com.sun.management.jmxremote.port", Integer.toString(port));
-    return this.withExposedPorts(port);
+    return this;
   }
 
   @CanIgnoreReturnValue
@@ -58,9 +62,33 @@ public TestAppContainer withUserAuth(String login, String pwd) {
     return this;
   }
 
+  /**
+   * Configures app container for host-to-container access, port will be used as-is from host to
+   * work-around JMX in docker. This is optional on Linux as there is a network route and the
+   * container is accessible, but not on Mac where the container runs in an isolated VM.
+   *
+   * @param port port to use, must be available on host.
+   * @return this
+   */
+  @CanIgnoreReturnValue
+  public TestAppContainer withHostAccessFixedJmxPort(int port) {
+    // To get host->container JMX connection working docker must expose JMX/RMI port under the same
+    // port number. Because of this testcontainers' standard exposed port randomization approach
+    // can't be used.
+    // Explanation:
+    // https://forums.docker.com/t/exposing-mapped-jmx-ports-from-multiple-containers/5287/6
+    properties.put("com.sun.management.jmxremote.port", Integer.toString(port));
+    properties.put("com.sun.management.jmxremote.rmi.port", Integer.toString(port));
+    properties.put("java.rmi.server.hostname", getHost());
+    addFixedExposedPort(port, port);
+    return this;
+  }
+
   @Override
   public void start() {
-
+    //    properties.put("com.sun.management.jmxremote.local.only", "false");
+    //    properties.put("java.rmi.server.logCalls", "true");
+    //
     // TODO: add support for ssl
     properties.put("com.sun.management.jmxremote.ssl", "false");
 
@@ -92,11 +120,9 @@ public void start() {
 
     this.withEnv("JAVA_TOOL_OPTIONS", confArgs);
 
-    logger().info("Test application JAVA_TOOL_OPTIONS = " + confArgs);
+    logger().info("Test application JAVA_TOOL_OPTIONS = {}", confArgs);
 
     super.start();
-
-    logger().info("Test application JMX port mapped to {}:{}", getHost(), getMappedPort(port));
   }
 
   private static Path createPwdFile(String login, String pwd) {

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

@@ -0,0 +1,156 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.jmxscraper.target_systems;
+
+import static io.opentelemetry.contrib.jmxscraper.target_systems.MetricAssertions.assertGaugeWithAttributes;
+import static io.opentelemetry.contrib.jmxscraper.target_systems.MetricAssertions.assertSumWithAttributes;
+import static org.assertj.core.api.Assertions.entry;
+
+import io.opentelemetry.contrib.jmxscraper.JmxScraperContainer;
+import java.time.Duration;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.images.builder.ImageFromDockerfile;
+import org.testcontainers.utility.MountableFile;
+
+public class ActiveMqIntegrationTest extends TargetSystemIntegrationTest {
+
+  @Override
+  protected GenericContainer<?> createTargetContainer(int jmxPort) {
+    return new GenericContainer<>(
+            new ImageFromDockerfile()
+                .withDockerfileFromBuilder(
+                    builder -> builder.from("apache/activemq-classic:5.18.6").build()))
+        .withEnv("LOCAL_JMX", "no")
+        .withCopyFileToContainer(
+            MountableFile.forClasspathResource("activemq/env"), "/opt/apache-activemq/bin/env")
+        .withEnv(
+            "ACTIVEMQ_JMX_OPTS",
+            "-Dcom.sun.management.jmxremote.port="
+                + jmxPort
+                + " -Dcom.sun.management.jmxremote.rmi.port="
+                + jmxPort
+                + " -Dcom.sun.management.jmxremote.ssl=false"
+                + " -Dcom.sun.management.jmxremote.authenticate=false")
+        .withStartupTimeout(Duration.ofMinutes(2))
+        .waitingFor(Wait.forListeningPort());
+  }
+
+  @Override
+  protected JmxScraperContainer customizeScraperContainer(JmxScraperContainer scraper) {
+    return scraper.withTargetSystem("activemq");
+  }
+
+  @Override
+  protected void verifyMetrics() {
+    waitAndAssertMetrics(
+        metric ->
+            assertSumWithAttributes(
+                metric,
+                "activemq.consumer.count",
+                "The number of consumers currently reading from the broker.",
+                "consumers",
+                /* isMonotonic= */ false,
+                attrs ->
+                    attrs.containsOnly(
+                        entry("destination", "ActiveMQ.Advisory.MasterBroker"),
+                        entry("broker", "localhost"))),
+        metric ->
+            assertSumWithAttributes(
+                metric,
+                "activemq.producer.count",
+                "The number of producers currently attached to the broker.",
+                "producers",
+                /* isMonotonic= */ false,
+                attrs ->
+                    attrs.containsOnly(
+                        entry("destination", "ActiveMQ.Advisory.MasterBroker"),
+                        entry("broker", "localhost"))),
+        metric ->
+            assertSumWithAttributes(
+                metric,
+                "activemq.connection.count",
+                "The total number of current connections.",
+                "connections",
+                /* isMonotonic= */ false,
+                attrs -> attrs.containsOnly(entry("broker", "localhost"))),
+        metric ->
+            assertGaugeWithAttributes(
+                metric,
+                "activemq.memory.usage",
+                "The percentage of configured memory used.",
+                "%",
+                attrs ->
+                    attrs.containsOnly(
+                        entry("destination", "ActiveMQ.Advisory.MasterBroker"),
+                        entry("broker", "localhost"))),
+        metric ->
+            assertGaugeWithAttributes(
+                metric,
+                "activemq.disk.store_usage",
+                "The percentage of configured disk used for persistent messages.",
+                "%",
+                attrs -> attrs.containsOnly(entry("broker", "localhost"))),
+        metric ->
+            assertGaugeWithAttributes(
+                metric,
+                "activemq.disk.temp_usage",
+                "The percentage of configured disk used for non-persistent messages.",
+                "%",
+                attrs -> attrs.containsOnly(entry("broker", "localhost"))),
+        metric ->
+            assertSumWithAttributes(
+                metric,
+                "activemq.message.current",
+                "The current number of messages waiting to be consumed.",
+                "messages",
+                /* isMonotonic= */ false,
+                attrs ->
+                    attrs.containsOnly(
+                        entry("destination", "ActiveMQ.Advisory.MasterBroker"),
+                        entry("broker", "localhost"))),
+        metric ->
+            assertSumWithAttributes(
+                metric,
+                "activemq.message.expired",
+                "The total number of messages not delivered because they expired.",
+                "messages",
+                attrs ->
+                    attrs.containsOnly(
+                        entry("destination", "ActiveMQ.Advisory.MasterBroker"),
+                        entry("broker", "localhost"))),
+        metric ->
+            assertSumWithAttributes(
+                metric,
+                "activemq.message.enqueued",
+                "The total number of messages received by the broker.",
+                "messages",
+                attrs ->
+                    attrs.containsOnly(
+                        entry("destination", "ActiveMQ.Advisory.MasterBroker"),
+                        entry("broker", "localhost"))),
+        metric ->
+            assertSumWithAttributes(
+                metric,
+                "activemq.message.dequeued",
+                "The total number of messages delivered to consumers.",
+                "messages",
+                attrs ->
+                    attrs.containsOnly(
+                        entry("destination", "ActiveMQ.Advisory.MasterBroker"),
+                        entry("broker", "localhost"))),
+        metric ->
+            assertGaugeWithAttributes(
+                metric,
+                "activemq.message.wait_time.avg",
+                "The average time a message was held on a destination.",
+                "ms",
+                attrs ->
+                    attrs.containsOnly(
+                        entry("destination", "ActiveMQ.Advisory.MasterBroker"),
+                        entry("broker", "localhost"))));
+  }
+}