Add publish confirm latency metrics to performance tool

acogoluegnes · acogoluegnes · commit ed1d30d18778 · 2022-06-02T16:56:52.000+02:00
Fixes #129
diff --git a/src/main/java/com/rabbitmq/stream/perf/DefaultPerformanceMetrics.java b/src/main/java/com/rabbitmq/stream/perf/DefaultPerformanceMetrics.java
@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2021 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2022 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -35,6 +35,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import org.slf4j.Logger;
@@ -45,7 +46,7 @@ class DefaultPerformanceMetrics implements PerformanceMetrics {
   private static final Logger LOGGER = LoggerFactory.getLogger(DefaultPerformanceMetrics.class);
 
   private final MetricRegistry metricRegistry;
-  private final Timer latency;
+  private final Timer latency, confirmLatency;
   private final boolean summaryFile;
   private final PrintWriter out;
   private final boolean includeByteRates;
@@ -60,6 +61,7 @@ class DefaultPerformanceMetrics implements PerformanceMetrics {
       String metricsPrefix,
       boolean summaryFile,
       boolean includeByteRates,
+      boolean confirmLatency,
       Supplier<String> memoryReportSupplier,
       PrintWriter out) {
     this.summaryFile = summaryFile;
@@ -98,6 +100,17 @@ protected Double nullGaugeValue() {
             .distributionStatisticExpiry(Duration.ofSeconds(1))
             .serviceLevelObjectives()
             .register(meterRegistry);
+    if (confirmLatency) {
+      this.confirmLatency =
+          Timer.builder(metricsPrefix + ".confirm_latency")
+              .description("publish confirm latency")
+              .publishPercentiles(0.5, 0.75, 0.95, 0.99)
+              .distributionStatisticExpiry(Duration.ofSeconds(1))
+              .serviceLevelObjectives()
+              .register(meterRegistry);
+    } else {
+      this.confirmLatency = null;
+    }
   }
 
   private long getPublishedCount() {
@@ -117,6 +130,7 @@ public void start(String description) throws Exception {
     String metricConsumed = "rabbitmqStreamConsumed";
     String metricChunkSize = "rabbitmqStreamChunk_size";
     String metricLatency = "rabbitmqStreamLatency";
+    String metricConfirmLatency = "rabbitmqStreamConfirm_latency";
     String metricWrittenBytes = "rabbitmqStreamWritten_bytes";
     String metricReadBytes = "rabbitmqStreamRead_bytes";
 
@@ -129,6 +143,10 @@ public void start(String description) throws Exception {
                 metricChunkSize,
                 metricLatency));
 
+    if (confirmLatency()) {
+      allMetrics.add(metricConfirmLatency);
+    }
+
     Map<String, String> metersNamesAndLabels = new LinkedHashMap<>();
     metersNamesAndLabels.put(metricPublished, "published");
     metersNamesAndLabels.put(metricProducerConfirmed, "confirmed");
@@ -184,14 +202,16 @@ public void start(String description) throws Exception {
         histogram -> String.format("chunk size %.0f", histogram.getSnapshot().getMean());
 
     com.codahale.metrics.Timer latency = metricRegistry.getTimers().get(metricLatency);
+    com.codahale.metrics.Timer confirmLatency =
+        confirmLatency() ? metricRegistry.getTimers().get(metricConfirmLatency) : null;
 
     Function<Number, Number> convertDuration =
         in -> in instanceof Long ? in.longValue() / 1_000_000 : in.doubleValue() / 1_000_000;
-    Function<com.codahale.metrics.Timer, String> formatLatency =
-        timer -> {
+    BiFunction<String, com.codahale.metrics.Timer, String> formatLatency =
+        (name, timer) -> {
           Snapshot snapshot = timer.getSnapshot();
           return String.format(
-              "latency min/median/75th/95th/99th %.0f/%.0f/%.0f/%.0f/%.0f ms",
+              name + " min/median/75th/95th/99th %.0f/%.0f/%.0f/%.0f/%.0f ms",
               convertDuration.apply(snapshot.getMin()),
               convertDuration.apply(snapshot.getMedian()),
               convertDuration.apply(snapshot.get75thPercentile()),
@@ -229,7 +249,12 @@ public void start(String description) throws Exception {
                                     .compute(lastValue, currentValue, duration));
                             lastMetersValues.put(meterName, currentValue);
                           });
-                  builder.append(formatLatency.apply(latency)).append(", ");
+                  if (confirmLatency()) {
+                    builder
+                        .append(formatLatency.apply("confirm latency", confirmLatency))
+                        .append(", ");
+                  }
+                  builder.append(formatLatency.apply("latency", latency)).append(", ");
                   builder.append(formatChunkSize.apply(chunkSize));
                   this.out.println(builder);
                   String memoryReport = this.memoryReportSupplier.get();
@@ -270,15 +295,20 @@ public void start(String description) throws Exception {
                 }
               };
 
-          Function<com.codahale.metrics.Timer, String> formatLatencySummary =
-              histogram ->
+          BiFunction<String, com.codahale.metrics.Timer, String> formatLatencySummary =
+              (name, histogram) ->
                   String.format(
-                      "latency 95th %.0f ms",
+                      name + " 95th %.0f ms",
                       convertDuration.apply(latency.getSnapshot().get95thPercentile()));
 
           StringBuilder builder = new StringBuilder("Summary: ");
           meters.entrySet().forEach(entry -> builder.append(formatMeterSummary.apply(entry)));
-          builder.append(formatLatencySummary.apply(latency)).append(", ");
+          if (confirmLatency()) {
+            builder
+                .append(formatLatencySummary.apply("confirm latency", confirmLatency))
+                .append(", ");
+          }
+          builder.append(formatLatencySummary.apply("latency", latency)).append(", ");
           builder.append(formatChunkSize.apply(chunkSize));
           this.out.println();
           this.out.println(builder);
@@ -369,6 +399,11 @@ public void latency(long latency, TimeUnit unit) {
     this.latency.record(latency, unit);
   }
 
+  @Override
+  public void confirmLatency(long latency, TimeUnit unit) {
+    this.confirmLatency.record(latency, unit);
+  }
+
   @Override
   public void offset(long offset) {
     this.offset = offset;
@@ -379,6 +414,10 @@ public void close() throws Exception {
     this.closingSequence.close();
   }
 
+  private boolean confirmLatency() {
+    return this.confirmLatency != null;
+  }
+
   private interface FormatCallback {
 
     String compute(long lastValue, long currentValue, Duration duration);
diff --git a/src/main/java/com/rabbitmq/stream/perf/PerformanceMetrics.java b/src/main/java/com/rabbitmq/stream/perf/PerformanceMetrics.java
@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2021 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2022 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -21,5 +21,7 @@ interface PerformanceMetrics extends AutoCloseable {
 
   void latency(long latency, TimeUnit unit);
 
+  void confirmLatency(long latency, TimeUnit unit);
+
   void offset(long offset);
 }
diff --git a/src/main/java/com/rabbitmq/stream/perf/StreamPerfTest.java b/src/main/java/com/rabbitmq/stream/perf/StreamPerfTest.java
@@ -347,6 +347,12 @@ public class StreamPerfTest implements Callable<Integer> {
       converter = Utils.PositiveIntegerTypeConverter.class)
   private int rpcTimeout;
 
+  @CommandLine.Option(
+      names = {"--confirm-latency", "-cl"},
+      description = "evaluate confirm latency",
+      defaultValue = "false")
+  private boolean confirmLatency;
+
   private MetricsCollector metricsCollector;
   private PerformanceMetrics performanceMetrics;
   private List<Monitoring> monitorings;
@@ -499,6 +505,7 @@ public Integer call() throws Exception {
             metricsPrefix,
             this.summaryFile,
             this.includeByteRates,
+            this.confirmLatency,
             memoryReportSupplier,
             this.out);
 
@@ -649,11 +656,6 @@ public Integer call() throws Exception {
                 }));
       }
 
-      // FIXME handle metadata update for consumers and publishers
-      // they should at least issue a warning that their stream has been deleted and that they're
-      // now
-      // useless
-
       List<Producer> producers = Collections.synchronizedList(new ArrayList<>(this.producers));
       List<Runnable> producerRunnables =
           IntStream.range(0, this.producers)
@@ -686,18 +688,48 @@ public Integer call() throws Exception {
                             .stream(stream)
                             .build();
 
+                    AtomicLong messageCount = new AtomicLong(0);
+                    ConfirmationHandler confirmationHandler;
+                    if (this.confirmLatency) {
+                      final PerformanceMetrics metrics = this.performanceMetrics;
+                      final int divisor = Utils.downSamplingDivisor(this.rate);
+                      confirmationHandler =
+                          confirmationStatus -> {
+                            if (confirmationStatus.isConfirmed()) {
+                              producerConfirm.increment();
+                              // at very high throughput ( > 1 M / s), the histogram can
+                              // become a bottleneck,
+                              // so we downsample and calculate latency for every x message
+                              // this should not affect the metric much
+                              if (messageCount.incrementAndGet() % divisor == 0) {
+                                try {
+                                  long time =
+                                      Utils.readLong(
+                                          confirmationStatus.getMessage().getBodyAsBinary());
+                                  // see below why we use current time to measure latency
+                                  metrics.confirmLatency(
+                                      System.currentTimeMillis() - time, TimeUnit.MILLISECONDS);
+                                } catch (Exception e) {
+                                  // not able to read the body, something wrong?
+                                }
+                              }
+                            }
+                          };
+                    } else {
+                      confirmationHandler =
+                          confirmationStatus -> {
+                            if (confirmationStatus.isConfirmed()) {
+                              producerConfirm.increment();
+                            }
+                          };
+                    }
+
                     producers.add(producer);
 
                     return (Runnable)
                         () -> {
                           final int msgSize = this.messageSize;
 
-                          ConfirmationHandler confirmationHandler =
-                              confirmationStatus -> {
-                                if (confirmationStatus.isConfirmed()) {
-                                  producerConfirm.increment();
-                                }
-                              };
                           try {
                             while (true && !Thread.currentThread().isInterrupted()) {
                               rateLimiterCallback.run();
@@ -747,14 +779,17 @@ public Integer call() throws Exception {
                                   .builder();
                         }
 
+                        // we assume the publishing rate is the same order as the consuming rate
+                        // we actually don't want to downsample for low rates
+                        final int divisor = Utils.downSamplingDivisor(this.rate);
                         consumerBuilder =
                             consumerBuilder.messageHandler(
                                 (context, message) -> {
                                   // at very high throughput ( > 1 M / s), the histogram can
                                   // become a bottleneck,
                                   // so we downsample and calculate latency for every x message
                                   // this should not affect the metric much
-                                  if (messageCount.incrementAndGet() % 100 == 0) {
+                                  if (messageCount.incrementAndGet() % 100 == divisor) {
                                     try {
                                       long time = Utils.readLong(message.getBodyAsBinary());
                                       // see above why we use current time to measure latency
diff --git a/src/main/java/com/rabbitmq/stream/perf/Utils.java b/src/main/java/com/rabbitmq/stream/perf/Utils.java
@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2021 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2022 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -278,6 +278,17 @@ static CommandSpec buildCommandSpec(Object... commands) {
     return spec;
   }
 
+  static int downSamplingDivisor(int rate) {
+    int divisor;
+    if (rate > 0) {
+      divisor = rate > 100 ? 100 : 1; // no downsampling for small rates
+    } else {
+      // no rate limitation, downsampling
+      divisor = 100;
+    }
+    return divisor;
+  }
+
   static class ByteCapacityTypeConverter implements CommandLine.ITypeConverter<ByteCapacity> {
 
     @Override
diff --git a/src/test/java/com/rabbitmq/stream/perf/StreamPerfTestTest.java b/src/test/java/com/rabbitmq/stream/perf/StreamPerfTestTest.java
@@ -1,4 +1,4 @@
-// Copyright (c) 2021 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2021-2022 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -341,6 +341,26 @@ void monitoringShouldReturnValidEndpoint() throws Exception {
     waitRunEnds();
   }
 
+  @Test
+  void publishConfirmLatencyShouldBeIncludedWhenOptionIsEnabled() throws Exception {
+    Future<?> run = run(builder().confirmLatency());
+    waitUntilStreamExists(s);
+    waitOneSecond();
+    run.cancel(true);
+    waitRunEnds();
+    assertThat(consoleOutput()).contains("confirm latency");
+  }
+
+  @Test
+  void publishConfirmLatencyShouldNotBeIncludedWhenOptionIsDisabled() throws Exception {
+    Future<?> run = run(builder());
+    waitUntilStreamExists(s);
+    waitOneSecond();
+    run.cancel(true);
+    waitRunEnds();
+    assertThat(consoleOutput()).doesNotContain("confirm latency");
+  }
+
   private static HttpResponse httpRequest(String urlString) throws Exception {
     URL url = new URL(urlString);
     HttpURLConnection con = (HttpURLConnection) url.openConnection();
@@ -439,6 +459,11 @@ ArgumentsBuilder memoryReport() {
       return this;
     }
 
+    ArgumentsBuilder confirmLatency() {
+      arguments.put("confirm-latency", "");
+      return this;
+    }
+
     ArgumentsBuilder maxLengthBytes(ByteCapacity capacity) {
       arguments.put("max-length-bytes", capacity.toString());
       return this;
diff --git a/src/test/java/com/rabbitmq/stream/perf/UtilsTest.java b/src/test/java/com/rabbitmq/stream/perf/UtilsTest.java
@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2021 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2022 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -289,6 +289,16 @@ void buildCommandSpec() {
     assertThat(spec.optionsMap()).hasSize(4).containsKeys("AAA", "B", "C", "OFFSET");
   }
 
+  @ParameterizedTest
+  @CsvSource({
+    "0,100",
+    "50,1",
+    "1000,100",
+  })
+  void testDownSamplingDivisor(int rate, int expected) {
+    assertThat(Utils.downSamplingDivisor(rate)).isEqualTo(expected);
+  }
+
   @ParameterizedTest
   @CsvSource({
     "--uris,URIS",
