Add --memory-report option to performance tool

This gives the Xmx and max direct memory allocation at the beginning
of the run and outputs Netty heap and direct allocation during the run.

This can help to diagnose any allocation that can occur, because of
some memory leak or more likely in an under-provisioned environment.

Author: acogoluegnes

Dockerfile

@@ -23,7 +23,7 @@ RUN set -eux; \
     if [ "$(uname -m)" = "x86_64" ] ; then JAVA_PATH="/usr/lib/jdk-$JAVA_VERSION"; \
     mkdir $JAVA_PATH && \
     tar --extract  --file jdk.tar.gz --directory "$JAVA_PATH" --strip-components 1; \
-	  $JAVA_PATH/bin/jlink --compress=2 --output /jre --add-modules java.base,java.naming,java.xml,jdk.unsupported,jdk.crypto.cryptoki; \
+	  $JAVA_PATH/bin/jlink --compress=2 --output /jre --add-modules java.base,jdk.management,java.naming,java.xml,jdk.unsupported,jdk.crypto.cryptoki; \
 	  /jre/bin/java -version; \
 	  fi
 

src/main/java/com/rabbitmq/stream/perf/DefaultPerformanceMetrics.java

@@ -35,6 +35,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
+import java.util.function.Supplier;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -47,16 +48,19 @@ class DefaultPerformanceMetrics implements PerformanceMetrics {
   private final boolean summaryFile;
   private final PrintWriter out;
   private final boolean includeByteRates;
+  private final Supplier<String> memoryReportSupplier;
   private volatile Closeable closingSequence = () -> {};
 
   DefaultPerformanceMetrics(
       CompositeMeterRegistry meterRegistry,
       String metricsPrefix,
       boolean summaryFile,
       boolean includeByteRates,
+      Supplier<String> memoryReportSupplier,
       PrintWriter out) {
     this.summaryFile = summaryFile;
     this.includeByteRates = includeByteRates;
+    this.memoryReportSupplier = memoryReportSupplier;
     this.out = out;
     DropwizardConfig dropwizardConfig =
         new DropwizardConfig() {
@@ -205,6 +209,10 @@ public void start(String description) throws Exception {
                   builder.append(formatLatency.apply(latency)).append(", ");
                   builder.append(formatChunkSize.apply(chunkSize));
                   this.out.println(builder);
+                  String memoryReport = this.memoryReportSupplier.get();
+                  if (!memoryReport.isEmpty()) {
+                    this.out.println(memoryReport);
+                  }
                 }
                 reportCount.incrementAndGet();
               } catch (Exception e) {
@@ -259,7 +267,7 @@ static String formatByteRate(String label, double bytes) {
     // based on
     // https://stackoverflow.com/questions/3758606/how-can-i-convert-byte-size-into-a-human-readable-format-in-java
     if (-1000 < bytes && bytes < 1000) {
-      return bytes + " B";
+      return bytes + " B/s";
     }
     CharacterIterator ci = new StringCharacterIterator("kMGTPE");
     while (bytes <= -999_950 || bytes >= 999_950) {

src/main/java/com/rabbitmq/stream/perf/StreamPerfTest.java

@@ -42,7 +42,11 @@
 import com.rabbitmq.stream.perf.Utils.NamedThreadFactory;
 import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.composite.CompositeMeterRegistry;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.ByteBufAllocatorMetric;
+import io.netty.buffer.ByteBufAllocatorMetricProvider;
 import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.util.internal.PlatformDependent;
 import java.io.PrintStream;
 import java.io.PrintWriter;
 import java.net.URI;
@@ -65,6 +69,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.BiFunction;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.slf4j.Logger;
@@ -286,6 +291,12 @@ public class StreamPerfTest implements Callable<Integer> {
       defaultValue = "false")
   private boolean includeByteRates;
 
+  @CommandLine.Option(
+      names = {"--memory-report", "-mr"},
+      description = "report information on memory settings and usage",
+      defaultValue = "false")
+  private boolean memoryReport;
+
   private MetricsCollector metricsCollector;
   private PerformanceMetrics performanceMetrics;
 
@@ -370,15 +381,61 @@ public Integer call() throws Exception {
     // FIXME assign codec
     this.codec = createCodec(this.codecClass);
 
+    ByteBufAllocator byteBufAllocator = ByteBufAllocator.DEFAULT;
+
     CompositeMeterRegistry meterRegistry = new CompositeMeterRegistry();
     String metricsPrefix = "rabbitmq.stream";
     this.metricsCollector = new MicrometerMetricsCollector(meterRegistry, metricsPrefix);
 
     Counter producerConfirm = meterRegistry.counter(metricsPrefix + ".producer_confirmed");
 
+    Supplier<String> memoryReportSupplier;
+    if (this.memoryReport) {
+      long physicalMemory = Utils.physicalMemory();
+      String physicalMemoryReport =
+          physicalMemory == 0
+              ? ""
+              : format(
+                  ", physical memory %s (%d bytes)",
+                  Utils.formatByte(physicalMemory), physicalMemory);
+      this.out.println(
+          format(
+              "Max memory %s (%d bytes), max direct memory %s (%d bytes)%s",
+              Utils.formatByte(Runtime.getRuntime().maxMemory()),
+              Runtime.getRuntime().maxMemory(),
+              Utils.formatByte(PlatformDependent.maxDirectMemory()),
+              PlatformDependent.maxDirectMemory(),
+              physicalMemoryReport));
+
+      if (byteBufAllocator instanceof ByteBufAllocatorMetricProvider) {
+        ByteBufAllocatorMetric allocatorMetric =
+            ((ByteBufAllocatorMetricProvider) byteBufAllocator).metric();
+        memoryReportSupplier =
+            () -> {
+              long usedHeapMemory = allocatorMetric.usedHeapMemory();
+              long usedDirectMemory = allocatorMetric.usedDirectMemory();
+              return format(
+                  "Used heap memory %s (%d bytes), used direct memory %s (%d bytes)",
+                  Utils.formatByte(usedHeapMemory),
+                  usedHeapMemory,
+                  Utils.formatByte(usedDirectMemory),
+                  usedDirectMemory);
+            };
+      } else {
+        memoryReportSupplier = () -> "";
+      }
+    } else {
+      memoryReportSupplier = () -> "";
+    }
+
     this.performanceMetrics =
         new DefaultPerformanceMetrics(
-            meterRegistry, metricsPrefix, this.summaryFile, this.includeByteRates, this.out);
+            meterRegistry,
+            metricsPrefix,
+            this.summaryFile,
+            this.includeByteRates,
+            memoryReportSupplier,
+            this.out);
 
     this.messageSize = this.messageSize < 8 ? 8 : this.messageSize; // we need to store a long in it
 
@@ -430,6 +487,7 @@ public Integer call() throws Exception {
             .addressResolver(addressResolver)
             .scheduledExecutorService(envExecutor)
             .metricsCollector(metricsCollector)
+            .byteBufAllocator(byteBufAllocator)
             .maxProducersByConnection(this.producersByConnection)
             .maxTrackingConsumersByConnection(this.trackingConsumersByConnection)
             .maxConsumersByConnection(this.consumersByConnection);

src/main/java/com/rabbitmq/stream/perf/Utils.java

@@ -18,6 +18,8 @@
 import com.rabbitmq.stream.StreamCreator.LeaderLocator;
 import com.rabbitmq.stream.compression.Compression;
 import java.security.cert.X509Certificate;
+import java.text.CharacterIterator;
+import java.text.StringCharacterIterator;
 import java.time.Duration;
 import java.time.Instant;
 import java.time.format.DateTimeFormatter;
@@ -102,6 +104,33 @@ static List<String> streams(String range, List<String> streams) {
     }
   }
 
+  static String formatByte(double bytes) {
+    // based on
+    // https://stackoverflow.com/questions/3758606/how-can-i-convert-byte-size-into-a-human-readable-format-in-java
+    if (-1000 < bytes && bytes < 1000) {
+      return String.valueOf(bytes);
+    }
+    CharacterIterator ci = new StringCharacterIterator("kMGTPE");
+    while (bytes <= -999_950 || bytes >= 999_950) {
+      bytes /= 1000;
+      ci.next();
+    }
+    return String.format("%.1f %cB", bytes / 1000.0, ci.current());
+  }
+
+  static long physicalMemory() {
+    try {
+      com.sun.management.OperatingSystemMXBean os =
+          (com.sun.management.OperatingSystemMXBean)
+              java.lang.management.ManagementFactory.getOperatingSystemMXBean();
+      return os.getTotalPhysicalMemorySize();
+    } catch (Throwable e) {
+      // we can get NoClassDefFoundError, so we catch from Throwable and below
+      LOGGER.warn("Could not get physical memory", e);
+      return 0;
+    }
+  }
+
   private static void throwConversionException(String format, String... arguments) {
     throw new CommandLine.TypeConversionException(String.format(format, (Object[]) arguments));
   }

src/test/java/com/rabbitmq/stream/perf/StreamPerfTestTest.java

@@ -250,6 +250,16 @@ void byteRatesShouldBeIncludedWhenOptionIsEnabled() throws Exception {
     assertThat(consoleOutput()).contains("written bytes").contains("read bytes");
   }
 
+  @Test
+  void memoryReportShouldBeIncludedWhenOptionIsEnabled() throws Exception {
+    Future<?> run = run(builder().memoryReport());
+    waitUntilStreamExists(s);
+    waitOneSecond();
+    run.cancel(true);
+    waitRunEnds();
+    assertThat(consoleOutput()).contains("Max memory").contains("max direct memory");
+  }
+
   @Test
   void subEntriesWithCompressionShouldRun() throws Exception {
     Future<?> run = run(builder().subEntrySize(10).compression(Compression.GZIP));
@@ -311,6 +321,11 @@ ArgumentsBuilder help() {
       return this;
     }
 
+    ArgumentsBuilder memoryReport() {
+      arguments.put("memory-report", "");
+      return this;
+    }
+
     ArgumentsBuilder maxLengthBytes(ByteCapacity capacity) {
       arguments.put("max-length-bytes", capacity.toString());
       return this;