Merge pull request #122 from rabbitmq/rabbitmq-perf-test-121-publish-confirm-latencies

Calculate publisher confirm latency

Author: michaelklishin

pom.xml

@@ -63,6 +63,7 @@
     <junit.jupiter.version>5.3.0</junit.jupiter.version>
     <mockito.version>2.23.0</mockito.version>
     <hamcrest.version>2.0.0.0</hamcrest.version>
+    <commons-lang3.version>3.8.1</commons-lang3.version>
     <jmh.version>1.20</jmh.version>
 
     <!-- to sign artifacts when releasing -->
@@ -205,6 +206,15 @@
       <scope>test</scope>
     </dependency>
 
+    <!-- https://mvnrepository.com/artifact/org.apache.commons/commons-lang3 -->
+    <dependency>
+      <groupId>org.apache.commons</groupId>
+      <artifactId>commons-lang3</artifactId>
+      <version>${commons-lang3.version}</version>
+      <scope>test</scope>
+    </dependency>
+
+
     <dependency>
       <groupId>org.openjdk.jmh</groupId>
       <artifactId>jmh-core</artifactId>

src/docs/asciidoc/monitoring.adoc

@@ -20,7 +20,10 @@ PerfTest supports.
 Here are the metrics PerfTest can gather:
 
 * default metrics: number of published, returned, confirmed, nacked, and consumed messages, message
-latency. Latency is a major concern in many types of workload, it can be easily monitored here.
+latency, publisher confirm latency. Message latency is a major concern in many types of workload, it can be easily monitored here.
+https://www.rabbitmq.com/confirms.html#publisher-confirms[Publisher confirm]
+latency reflects the time a message can be considered unsafe. It is
+calculated as soon as the `--confirm`/`-c` option is used.
 Default metrics are available as long as PerfTest support for a monitoring system
 is enabled.
 * client metrics: these are the https://www.rabbitmq.com/api-guide.html#metrics[Java Client metrics].

src/docs/asciidoc/usage.adoc

@@ -62,7 +62,7 @@ https://www.rabbitmq.com/confirms.html[Consumer prefetch (QoS)] can be configure
  bin/runjava com.rabbitmq.perf.PerfTest -x 1 -y 2 -u "throughput-test-7" --id "test-7" \
    -f persistent --multi-ack-every 200 -q 500
 
-Publisher confirms can be used with maximum of N outstanding publishes:
+Publisher confirms can be used with a maximum of N outstanding publishes:
 
  bin/runjava com.rabbitmq.perf.PerfTest -x 1 -y 2 -u "throughput-test-8" --id "test-8" \
    -f persistent -q 500 -c 500

src/main/java/com/rabbitmq/perf/LocalFilesMessageBodySource.java

@@ -24,6 +24,8 @@
  */
 public class LocalFilesMessageBodySource implements MessageBodySource {
 
+    private final long ZERO = 0L;
+
     private final List<byte[]> bodies;
 
     private final String contentType;
@@ -56,9 +58,9 @@ public LocalFilesMessageBodySource(List<String> filesNames) throws IOException {
     }
 
     @Override
-    public MessageBodyAndContentType create(int sequenceNumber) {
-        return new MessageBodyAndContentType(
-            bodies.get(sequenceNumber % bodies.size()), contentType
+    public MessageEnvelope create(int sequenceNumber) {
+        return new MessageEnvelope(
+            bodies.get(sequenceNumber % bodies.size()), contentType, ZERO
         );
     }
 }

src/main/java/com/rabbitmq/perf/MessageBodySource.java

@@ -23,15 +23,17 @@
  */
 public interface MessageBodySource {
 
-    MessageBodyAndContentType create(int sequenceNumber) throws IOException;
+    MessageEnvelope create(int sequenceNumber) throws IOException;
 
-    class MessageBodyAndContentType {
+    class MessageEnvelope {
         private final byte [] body;
         private final String contentType;
+        private final long time;
 
-        public MessageBodyAndContentType(byte[] body, String contentType) {
+        public MessageEnvelope(byte[] body, String contentType, long time) {
             this.body = body;
             this.contentType = contentType;
+            this.time = time;
         }
 
         public byte[] getBody() {
@@ -41,6 +43,10 @@ public byte[] getBody() {
         public String getContentType() {
             return contentType;
         }
+
+        public long getTime() {
+            return time;
+        }
     }
 
 }

src/main/java/com/rabbitmq/perf/PerfTest.java

@@ -15,40 +15,26 @@
 
 package com.rabbitmq.perf;
 
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.DefaultSaslConfig;
+import com.rabbitmq.client.impl.ClientVersion;
+import com.rabbitmq.client.impl.nio.NioParams;
+import io.micrometer.core.instrument.composite.CompositeMeterRegistry;
+import org.apache.commons.cli.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.SSLContext;
 import java.io.*;
 import java.nio.charset.Charset;
 import java.security.NoSuchAlgorithmException;
 import java.text.SimpleDateFormat;
-import java.util.Calendar;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Locale;
-import java.util.Map;
-import java.util.Properties;
+import java.util.*;
 import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 
-import com.rabbitmq.client.impl.ClientVersion;
-import com.rabbitmq.client.impl.nio.NioParams;
-import io.micrometer.core.instrument.MeterRegistry;
-import io.micrometer.core.instrument.composite.CompositeMeterRegistry;
-import org.apache.commons.cli.CommandLine;
-import org.apache.commons.cli.CommandLineParser;
-import org.apache.commons.cli.GnuParser;
-import org.apache.commons.cli.HelpFormatter;
-import org.apache.commons.cli.Option;
-import org.apache.commons.cli.Options;
-import org.apache.commons.cli.ParseException;
-
-import com.rabbitmq.client.ConnectionFactory;
-import com.rabbitmq.client.DefaultSaslConfig;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.net.ssl.SSLContext;
-
 import static com.rabbitmq.perf.OptionsUtils.forEach;
 import static java.lang.String.format;
 import static java.util.Arrays.asList;
@@ -562,176 +548,6 @@ private static String getExchangeName(CommandLineProxy cmd, String def) {
         return exchangeName;
     }
 
-    private static class PrintlnStats extends Stats {
-        private final boolean sendStatsEnabled;
-        private final boolean recvStatsEnabled;
-        private final boolean returnStatsEnabled;
-        private final boolean confirmStatsEnabled;
-        private final boolean legacyMetrics;
-        private final boolean useMillis;
-
-        private final String testID;
-        private final PrintWriter out;
-
-        public PrintlnStats(String testID, long interval,
-            boolean sendStatsEnabled, boolean recvStatsEnabled,
-            boolean returnStatsEnabled, boolean confirmStatsEnabled,
-            boolean legacyMetrics, boolean useMillis,
-            PrintWriter out, MeterRegistry registry, String metricsPrefix) {
-            super(interval, useMillis, registry, metricsPrefix);
-            this.sendStatsEnabled = sendStatsEnabled;
-            this.recvStatsEnabled = recvStatsEnabled;
-            this.returnStatsEnabled = returnStatsEnabled;
-            this.confirmStatsEnabled = confirmStatsEnabled;
-            this.testID = testID;
-            this.legacyMetrics = legacyMetrics;
-            this.useMillis = useMillis;
-            this.out = out;
-            if (out != null) {
-                out.printf("id,time (s),published (msg/s),returned (msg/s)," +
-                        "confirmed (msg/s),nacked (msg/s)," +
-                        "received (msg/s),min latency (%s),median latency (%s)," +
-                        "75th p. latency (%s),95th p. latency (%s),99th p. latency (%s)%n",
-                        units(), units(), units(), units(), units());
-            }
-        }
-
-        @Override
-        protected void report(long now) {
-            String output = "id: " + testID + ", ";
-
-            double ratePublished = 0.0;
-            double rateReturned = 0.0;
-            double rateConfirmed = 0.0;
-            double rateNacked = 0.0;
-            double rateConsumed = 0.0;
-
-            if (sendStatsEnabled) {
-                ratePublished = rate(sendCountInterval, elapsedInterval);
-                published(ratePublished);
-            }
-            if (sendStatsEnabled && returnStatsEnabled) {
-                rateReturned = rate(returnCountInterval, elapsedInterval);
-                returned(rateReturned);
-            }
-            if (sendStatsEnabled && confirmStatsEnabled) {
-                rateConfirmed = rate(confirmCountInterval, elapsedInterval);
-                confirmed(rateConfirmed);
-            }
-            if (sendStatsEnabled && confirmStatsEnabled) {
-                rateNacked = rate(nackCountInterval, elapsedInterval);
-                nacked(rateNacked);
-            }
-            if (recvStatsEnabled) {
-                rateConsumed = rate(recvCountInterval, elapsedInterval);
-                received(rateConsumed);
-            }
-
-            output += "time: " + format("%.3f", (now - startTime)/1000.0) + "s";
-            output +=
-                getRate("sent",      ratePublished,    sendStatsEnabled) +
-                    getRate("returned",  rateReturned,  sendStatsEnabled && returnStatsEnabled) +
-                    getRate("confirmed", rateConfirmed, sendStatsEnabled && confirmStatsEnabled) +
-                    getRate("nacked",    rateNacked,    sendStatsEnabled && confirmStatsEnabled) +
-                    getRate("received",  rateConsumed,    recvStatsEnabled);
-
-            if (legacyMetrics) {
-                output += (latencyCountInterval > 0 ?
-                    ", min/avg/max latency: " +
-                        minLatency/1000L + "/" +
-                        cumulativeLatencyInterval / (1000L * latencyCountInterval) + "/" +
-                        maxLatency/1000L + " µs " :
-                    "");
-            } else {
-                output += (latencyCountInterval > 0 ?
-                    ", min/median/75th/95th/99th latency: "
-                        + div(latency.getSnapshot().getMin()) + "/"
-                        + div(latency.getSnapshot().getMedian()) + "/"
-                        + div(latency.getSnapshot().get75thPercentile()) + "/"
-                        + div(latency.getSnapshot().get95thPercentile()) + "/"
-                        + div(latency.getSnapshot().get99thPercentile()) + " " + units() :
-                    "");
-            }
-
-            System.out.println(output);
-            if (out != null) {
-                out.println(testID + "," + format("%.3f", (now - startTime)/1000.0) + "," +
-                    rate(ratePublished, sendStatsEnabled)+ "," +
-                    rate(rateReturned, sendStatsEnabled && returnStatsEnabled)+ "," +
-                    rate(rateConfirmed, sendStatsEnabled && confirmStatsEnabled)+ "," +
-                    rate(rateNacked, sendStatsEnabled && confirmStatsEnabled)+ "," +
-                    rate(rateConsumed, recvStatsEnabled) + "," +
-                    (latencyCountInterval > 0 ?
-                        div(latency.getSnapshot().getMin()) + "," +
-                        div(latency.getSnapshot().getMedian()) + "," +
-                        div(latency.getSnapshot().get75thPercentile()) + "," +
-                        div(latency.getSnapshot().get95thPercentile()) + "," +
-                        div(latency.getSnapshot().get99thPercentile())
-                        : ",,,,")
-                );
-            }
-
-        }
-
-        private String units() {
-            if (useMillis) {
-                return "ms";
-            } else {
-                return "µs";
-            }
-        }
-
-        private long div(double p) {
-            if (useMillis) {
-                return (long)p;
-            } else {
-                return (long)(p / 1000L);
-            }
-        }
-
-        private String getRate(String descr, double rate, boolean display) {
-            if (display) {
-                return ", " + descr + ": " + formatRate(rate) + " msg/s";
-            } else {
-                return "";
-            }
-        }
-
-        public void printFinal() {
-            long now = System.currentTimeMillis();
-
-            System.out.println("id: " + testID + ", sending rate avg: " +
-                formatRate(sendCountTotal * 1000.0 / (now - startTime)) +
-                " msg/s");
-
-            long elapsed = now - startTime;
-            if (elapsed > 0) {
-                System.out.println("id: " + testID + ", receiving rate avg: " +
-                    formatRate(recvCountTotal * 1000.0 / elapsed) +
-                    " msg/s");
-            }
-        }
-
-        private static String formatRate(double rate) {
-            if (rate == 0.0)    return format("%d", (long)rate);
-            else if (rate < 1)  return format("%1.2f", rate);
-            else if (rate < 10) return format("%1.1f", rate);
-            else                return format("%d", (long)rate);
-        }
-
-        private static String rate(double rate, boolean display) {
-            if (display) {
-                return formatRate(rate);
-            } else {
-                return "";
-            }
-        }
-
-        private static double rate(long count, long elapsed) {
-            return 1000.0 * count / elapsed;
-        }
-    }
-
     private static void versionInformation() {
         String lineSeparator = System.getProperty("line.separator");
         String version = format(