pip 264 java runtime metrics

conf/pulsar_env.sh

@@ -94,3 +94,7 @@ PULSAR_EXTRA_OPTS="${PULSAR_EXTRA_OPTS:-" -Dpulsar.allocator.exit_on_oom=true -D
 #Wait time before forcefully kill the pulsar server instance, if the stop is not successful
 #PULSAR_STOP_TIMEOUT=
 
+# Enable semantically stable telemetry for JVM metrics, unless otherwise overridden by the user.
+if [ -z "$OTEL_SEMCONV_STABILITY_OPT_IN" ]; then
+  export OTEL_SEMCONV_STABILITY_OPT_IN=jvm
+fi

distribution/server/src/assemble/LICENSE.bin.txt

@@ -542,6 +542,8 @@ The Apache Software License, Version 2.0
     - io.opentelemetry.instrumentation-opentelemetry-instrumentation-api-1.33.2.jar
     - io.opentelemetry.instrumentation-opentelemetry-instrumentation-api-semconv-1.33.2-alpha.jar
     - io.opentelemetry.instrumentation-opentelemetry-resources-1.33.2-alpha.jar
+    - io.opentelemetry.instrumentation-opentelemetry-runtime-telemetry-java17-1.33.2-alpha.jar
+    - io.opentelemetry.instrumentation-opentelemetry-runtime-telemetry-java8-1.33.2-alpha.jar
     - io.opentelemetry.semconv-opentelemetry-semconv-1.25.0-alpha.jar
 
 BSD 3-clause "New" or "Revised" License

pulsar-broker-common/src/test/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsClient.java

@@ -59,7 +59,7 @@ public static Multimap<String, Metric> parseMetrics(String metrics) {
         // or
         // pulsar_subscriptions_count{cluster="standalone", namespace="public/default",
         // topic="persistent://public/default/test-2"} 0.0
-        Pattern pattern = Pattern.compile("^(\\w+)\\{([^}]+)}\\s([+-]?[\\d\\w.-]+)$");
+        Pattern pattern = Pattern.compile("^(\\w+)\\{([^}]+)}\\s([+-]?[\\d\\w.+-]+)$");
         Pattern tagsPattern = Pattern.compile("(\\w+)=\"([^\"]+)\"(,\\s?)?");
 
         Splitter.on("\n").split(metrics).forEach(line -> {

pulsar-opentelemetry/pom.xml

@@ -58,6 +58,10 @@
       <groupId>io.opentelemetry.semconv</groupId>
       <artifactId>opentelemetry-semconv</artifactId>
     </dependency>
+    <dependency>
+      <groupId>io.opentelemetry.instrumentation</groupId>
+      <artifactId>opentelemetry-runtime-telemetry-java17</artifactId>
+    </dependency>
 
     <dependency>
       <groupId>com.google.guava</groupId>
@@ -130,6 +134,16 @@
           </execution>
         </executions>
       </plugin>
+
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-surefire-plugin</artifactId>
+        <configuration>
+          <systemPropertyVariables>
+            <otel.semconv-stability.opt-in>jvm</otel.semconv-stability.opt-in>
+          </systemPropertyVariables>
+        </configuration>
+      </plugin>
     </plugins>
   </build>
 </project>

pulsar-opentelemetry/src/main/java/org/apache/pulsar/opentelemetry/OpenTelemetryService.java

@@ -21,6 +21,7 @@
 import static com.google.common.base.Preconditions.checkArgument;
 import com.google.common.annotations.VisibleForTesting;
 import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.instrumentation.runtimemetrics.java17.RuntimeMetrics;
 import io.opentelemetry.sdk.OpenTelemetrySdk;
 import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdk;
 import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdkBuilder;
@@ -29,6 +30,7 @@
 import java.io.Closeable;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import lombok.Builder;
 import org.apache.commons.lang3.StringUtils;
@@ -42,7 +44,9 @@ public class OpenTelemetryService implements Closeable {
     public static final String OTEL_SDK_DISABLED_KEY = "otel.sdk.disabled";
     static final int MAX_CARDINALITY_LIMIT = 10000;
 
-    private final OpenTelemetrySdk openTelemetrySdk;
+    private final AtomicReference<OpenTelemetrySdk> openTelemetrySdkReference = new AtomicReference<>();
+
+    private final AtomicReference<RuntimeMetrics> runtimeMetricsReference = new AtomicReference<>();
 
     /**
      * Instantiates the OpenTelemetry SDK. All attributes are overridden by system properties or environment
@@ -94,15 +98,28 @@ public OpenTelemetryService(String clusterName,
             builderCustomizer.accept(sdkBuilder);
         }
 
-        openTelemetrySdk = sdkBuilder.build().getOpenTelemetrySdk();
+        openTelemetrySdkReference.set(sdkBuilder.build().getOpenTelemetrySdk());
+
+        // For a list of exposed metrics, see https://opentelemetry.io/docs/specs/semconv/runtime/jvm-metrics/
+        runtimeMetricsReference.set(RuntimeMetrics.builder(openTelemetrySdkReference.get())
+                .enableAllFeatures()
+                .enableExperimentalJmxTelemetry()
+                .build());
     }
 
     public OpenTelemetry getOpenTelemetry() {
-        return openTelemetrySdk;
+        return openTelemetrySdkReference.get();
     }
 
     @Override
     public void close() {
-        openTelemetrySdk.close();
+        RuntimeMetrics runtimeMetrics = runtimeMetricsReference.getAndSet(null);
+        if (runtimeMetrics != null) {
+            runtimeMetrics.close();
+        }
+        OpenTelemetrySdk openTelemetrySdk = openTelemetrySdkReference.getAndSet(null);
+        if (openTelemetrySdk != null) {
+            openTelemetrySdk.close();
+        }
     }
 }

pulsar-opentelemetry/src/test/java/org/apache/pulsar/opentelemetry/OpenTelemetryServiceTest.java

@@ -198,4 +198,52 @@ public void testServiceIsDisabledByDefault() throws Exception {
         // Validate that the callback has not being called.
         assertThat(callback).isFalse();
     }
+
+    @Test
+    public void testJvmRuntimeMetrics() {
+        // Attempt collection of GC metrics. The metrics should be populated regardless if GC is triggered or not.
+        Runtime.getRuntime().gc();
+
+        var metrics = reader.collectAllMetrics();
+
+        // Process Metrics
+        // Replaces process_cpu_seconds_total
+        assertThat(metrics).anySatisfy(metric -> assertThat(metric).hasName("jvm.cpu.time"));
+
+        // Memory Metrics
+        // Replaces jvm_memory_bytes_used
+        assertThat(metrics).anySatisfy(metric -> assertThat(metric).hasName("jvm.memory.used"));
+        // Replaces jvm_memory_bytes_committed
+        assertThat(metrics).anySatisfy(metric -> assertThat(metric).hasName("jvm.memory.committed"));
+        // Replaces jvm_memory_bytes_max
+        assertThat(metrics).anySatisfy(metric -> assertThat(metric).hasName("jvm.memory.limit"));
+        // Replaces jvm_memory_bytes_init
+        assertThat(metrics).anySatisfy(metric -> assertThat(metric).hasName("jvm.memory.init"));
+        // Replaces jvm_memory_pool_allocated_bytes_total
+        assertThat(metrics).anySatisfy(metric -> assertThat(metric).hasName("jvm.memory.used_after_last_gc"));
+
+        // Buffer Pool Metrics
+        // Replaces jvm_buffer_pool_used_bytes
+        assertThat(metrics).anySatisfy(metric -> assertThat(metric).hasName("jvm.buffer.memory.usage"));
+        // Replaces jvm_buffer_pool_capacity_bytes
+        assertThat(metrics).anySatisfy(metric -> assertThat(metric).hasName("jvm.buffer.memory.limit"));
+        // Replaces jvm_buffer_pool_used_buffers
+        assertThat(metrics).anySatisfy(metric -> assertThat(metric).hasName("jvm.buffer.count"));
+
+        // Garbage Collector Metrics
+        // Replaces jvm_gc_collection_seconds
+        assertThat(metrics).anySatisfy(metric -> assertThat(metric).hasName("jvm.gc.duration"));
+
+        // Thread Metrics
+        // Replaces jvm_threads_state, jvm_threads_current and jvm_threads_daemon
+        assertThat(metrics).anySatisfy(metric -> assertThat(metric).hasName("jvm.thread.count"));
+
+        // Class Loading Metrics
+        // Replaces jvm_classes_currently_loaded
+        assertThat(metrics).anySatisfy(metric -> assertThat(metric).hasName("jvm.class.count"));
+        // Replaces jvm_classes_loaded_total
+        assertThat(metrics).anySatisfy(metric -> assertThat(metric).hasName("jvm.class.loaded"));
+        // Replaces jvm_classes_unloaded_total
+        assertThat(metrics).anySatisfy(metric -> assertThat(metric).hasName("jvm.class.unloaded"));
+    }
 }

tests/docker-images/latest-version-image/conf/bookie.conf

@@ -22,7 +22,7 @@ autostart=false
 redirect_stderr=true
 stdout_logfile=/var/log/pulsar/bookie.log
 directory=/pulsar
-environment=PULSAR_MEM="-Xmx128M -XX:MaxDirectMemorySize=512M",PULSAR_GC="-XX:+UseZGC"
+environment=PULSAR_MEM="-Xmx128M -XX:MaxDirectMemorySize=512M -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/var/log/pulsar",PULSAR_GC="-XX:+UseZGC"
 command=/pulsar/bin/pulsar bookie
 user=pulsar
 stopwaitsecs=15

tests/docker-images/latest-version-image/conf/broker.conf

@@ -22,7 +22,7 @@ autostart=false
 redirect_stderr=true
 stdout_logfile=/var/log/pulsar/broker.log
 directory=/pulsar
-environment=PULSAR_MEM="-Xmx128M",PULSAR_GC="-XX:+UseZGC"
+environment=PULSAR_MEM="-Xmx128M -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/var/log/pulsar",PULSAR_GC="-XX:+UseZGC"
 command=/pulsar/bin/pulsar broker
 user=pulsar
 stopwaitsecs=15

tests/docker-images/latest-version-image/conf/proxy.conf

@@ -22,7 +22,7 @@ autostart=false
 redirect_stderr=true
 stdout_logfile=/var/log/pulsar/proxy.log
 directory=/pulsar
-environment=PULSAR_MEM="-Xmx128M",PULSAR_GC="-XX:+UseZGC"
+environment=PULSAR_MEM="-Xmx128M -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/var/log/pulsar",PULSAR_GC="-XX:+UseZGC"
 command=/pulsar/bin/pulsar proxy
 user=pulsar
 stopwaitsecs=15

tests/docker-images/latest-version-image/conf/websocket.conf

@@ -22,7 +22,7 @@ autostart=false
 redirect_stderr=true
 stdout_logfile=/var/log/pulsar/pulsar-websocket.log
 directory=/pulsar
-environment=PULSAR_MEM="-Xmx128M",PULSAR_GC="-XX:+UseZGC"
+environment=PULSAR_MEM="-Xmx128M -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/var/log/pulsar",PULSAR_GC="-XX:+UseZGC"
 command=/pulsar/bin/pulsar websocket
 user=pulsar
 stopwaitsecs=15