Add option for metrics prefix

[#158209647]

References #42

Author: acogoluegnes

Makefile

@@ -21,6 +21,9 @@ clean: 	## Clean all build artefacts
 compile: ## Compile the source code
 	@mvnw compile
 
+install: clean ## Create and copy the binaries into the local Maven repository
+	@mvnw install -Dmaven.test.skip
+
 jar: clean ## Build the JAR file
 	@mvnw package
 

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

@@ -24,8 +24,8 @@ protected void delay(long now, AgentState state) {
 
         long elapsed = now - state.getLastStatsTime();
         //example: rateLimit is 5000 msg/s,
-        //10 ms have elapsed, we have sent 200 messages
-        //the 200 msgs we have actually sent should have taken us
+        //10 ms have elapsed, we have published 200 messages
+        //the 200 msgs we have actually published should have taken us
         //200 * 1000 / 5000 = 40 ms. So we pause for 40ms - 10ms
         long pause = (long) (state.getRateLimit() == 0.0f ?
             0.0f : (state.getMsgCount() * 1000.0 / state.getRateLimit() - elapsed));

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

@@ -41,6 +41,7 @@ public class BaseMetrics implements Metrics {
     public Options options() {
         Options options = new Options();
         options.addOption(new Option("mt", "metrics-tags", true, "metrics tags as key-value pairs separated by commas"));
+        options.addOption(new Option("mpx", "metrics-prefix", true, "prefix for PerfTest metrics, default is perftest_"));
         options.addOption(new Option("mc", "metrics-client", false, "enable client metrics"));
         options.addOption(new Option("mcl", "metrics-class-loader", false, "enable JVM class loader metrics"));
         options.addOption(new Option("mjm", "metrics-jvm-memory", false, "enable JVM memory metrics"));
@@ -63,7 +64,7 @@ public void configure(CommandLineProxy cmd, CompositeMeterRegistry meterRegistry
         }
         meterRegistry.config().commonTags(tags);
         if (cmd.hasOption("mc")) {
-            factory.setMetricsCollector(new MicrometerMetricsCollector(meterRegistry));
+            factory.setMetricsCollector(new MicrometerMetricsCollector(meterRegistry, "client"));
         }
 
         if (cmd.hasOption("mcl")) {

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

@@ -179,13 +179,14 @@ public static void main(String [] args, PerfTestOptions perfTestOptions) {
                 uris = singletonList(uri);
             }
 
+            String metricsPrefix = strArg(cmd, "mpx", "perftest_");
             //setup
             PrintlnStats stats = new PrintlnStats(testID,
                 1000L * samplingInterval,
                 producerCount > 0,
                 consumerCount > 0,
                 (flags.contains("mandatory") || flags.contains("immediate")),
-                confirm != -1, legacyMetrics, useMillis, output, registry);
+                confirm != -1, legacyMetrics, useMillis, output, registry, metricsPrefix);
 
             SSLContext sslContext = perfTestOptions.skipSslContextConfiguration ? null :
                 getSslContextIfNecessary(cmd, System.getProperties());
@@ -571,8 +572,8 @@ public PrintlnStats(String testID, long interval,
             boolean sendStatsEnabled, boolean recvStatsEnabled,
             boolean returnStatsEnabled, boolean confirmStatsEnabled,
             boolean legacyMetrics, boolean useMillis,
-            PrintWriter out, MeterRegistry registry) {
-            super(interval, useMillis, registry);
+            PrintWriter out, MeterRegistry registry, String metricsPrefix) {
+            super(interval, useMillis, registry, metricsPrefix);
             this.sendStatsEnabled = sendStatsEnabled;
             this.recvStatsEnabled = recvStatsEnabled;
             this.returnStatsEnabled = returnStatsEnabled;
@@ -582,7 +583,7 @@ public PrintlnStats(String testID, long interval,
             this.useMillis = useMillis;
             this.out = out;
             if (out != null) {
-                out.printf("id,time (s),sent (msg/s),returned (msg/s)," +
+                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",
@@ -594,15 +595,15 @@ public PrintlnStats(String testID, long interval,
         protected void report(long now) {
             String output = "id: " + testID + ", ";
 
-            double rateSent = 0.0;
+            double ratePublished = 0.0;
             double rateReturned = 0.0;
             double rateConfirmed = 0.0;
             double rateNacked = 0.0;
-            double rateReceived = 0.0;
+            double rateConsumed = 0.0;
 
             if (sendStatsEnabled) {
-                rateSent = rate(sendCountInterval, elapsedInterval);
-                sent(rateSent);
+                ratePublished = rate(sendCountInterval, elapsedInterval);
+                published(ratePublished);
             }
             if (sendStatsEnabled && returnStatsEnabled) {
                 rateReturned = rate(returnCountInterval, elapsedInterval);
@@ -617,17 +618,17 @@ protected void report(long now) {
                 nacked(rateNacked);
             }
             if (recvStatsEnabled) {
-                rateReceived = rate(recvCountInterval, elapsedInterval);
-                received(rateReceived);
+                rateConsumed = rate(recvCountInterval, elapsedInterval);
+                received(rateConsumed);
             }
 
             output += "time: " + format("%.3f", (now - startTime)/1000.0) + "s";
             output +=
-                getRate("sent",      rateSent,    sendStatsEnabled) +
+                getRate("sent",      ratePublished,    sendStatsEnabled) +
                     getRate("returned",  rateReturned,  sendStatsEnabled && returnStatsEnabled) +
                     getRate("confirmed", rateConfirmed, sendStatsEnabled && confirmStatsEnabled) +
                     getRate("nacked",    rateNacked,    sendStatsEnabled && confirmStatsEnabled) +
-                    getRate("received",  rateReceived,    recvStatsEnabled);
+                    getRate("received",  rateConsumed,    recvStatsEnabled);
 
             if (legacyMetrics) {
                 output += (latencyCountInterval > 0 ?
@@ -649,11 +650,11 @@ protected void report(long now) {
             System.out.println(output);
             if (out != null) {
                 out.println(testID + "," + format("%.3f", (now - startTime)/1000.0) + "," +
-                    rate(rateSent, sendStatsEnabled)+ "," +
+                    rate(ratePublished, sendStatsEnabled)+ "," +
                     rate(rateReturned, sendStatsEnabled && returnStatsEnabled)+ "," +
                     rate(rateConfirmed, sendStatsEnabled && confirmStatsEnabled)+ "," +
                     rate(rateNacked, sendStatsEnabled && confirmStatsEnabled)+ "," +
-                    rate(rateReceived, recvStatsEnabled) + "," +
+                    rate(rateConsumed, recvStatsEnabled) + "," +
                     (latencyCountInterval > 0 ?
                         div(latency.getSnapshot().getMin()) + "," +
                         div(latency.getSnapshot().getMedian()) + "," +

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

@@ -33,7 +33,7 @@ public abstract class Stats {
 
     protected final long startTime;
     private final Consumer<Long> updateLatency;
-    private final DoubleAccumulator sent, returned, confirmed, nacked, received;
+    private final DoubleAccumulator published, returned, confirmed, nacked, consumed;
     protected long lastStatsTime;
     protected int sendCountInterval;
     protected int returnCountInterval;
@@ -53,27 +53,29 @@ public abstract class Stats {
     protected Histogram latency = new MetricRegistry().histogram("latency");
 
     public Stats(long interval) {
-        this(interval, false, new SimpleMeterRegistry());
+        this(interval, false, new SimpleMeterRegistry(), null);
     }
 
-    public Stats(long interval, boolean useMs, MeterRegistry registry) {
+    public Stats(long interval, boolean useMs, MeterRegistry registry, String metricsPrefix) {
         this.interval = interval;
         startTime = System.currentTimeMillis();
 
+        metricsPrefix = metricsPrefix == null ? "" : metricsPrefix;
+
         Timer latencyTimer = Timer
-            .builder("latency")
+            .builder(metricsPrefix + "latency")
             .description("message latency")
             .publishPercentiles(0.5, 0.75, 0.95, 0.99)
             .distributionStatisticExpiry(Duration.ofMillis(this.interval))
             .sla()
             .register(registry);
 
         DoubleBinaryOperator accumulatorFunction = (x, y) -> y;
-        sent = registry.gauge("sent", new DoubleAccumulator(accumulatorFunction, 0.0));
-        returned = registry.gauge("returned", new DoubleAccumulator(accumulatorFunction, 0.0));
-        confirmed = registry.gauge("confirmed", new DoubleAccumulator(accumulatorFunction, 0.0));
-        nacked = registry.gauge("nacked", new DoubleAccumulator(accumulatorFunction, 0.0));
-        received = registry.gauge("received", new DoubleAccumulator(accumulatorFunction, 0.0));
+        published = registry.gauge(metricsPrefix + "published", new DoubleAccumulator(accumulatorFunction, 0.0));
+        returned = registry.gauge(metricsPrefix + "returned", new DoubleAccumulator(accumulatorFunction, 0.0));
+        confirmed = registry.gauge(metricsPrefix + "confirmed", new DoubleAccumulator(accumulatorFunction, 0.0));
+        nacked = registry.gauge(metricsPrefix + "nacked", new DoubleAccumulator(accumulatorFunction, 0.0));
+        consumed = registry.gauge(metricsPrefix + "consumed", new DoubleAccumulator(accumulatorFunction, 0.0));
 
         updateLatency = useMs ? latency -> latencyTimer.record(latency, TimeUnit.MILLISECONDS) :
             latency -> latencyTimer.record(latency, TimeUnit.NANOSECONDS);
@@ -147,8 +149,8 @@ public synchronized void handleRecv(long latency) {
         report();
     }
 
-    protected void sent(double rate) {
-        this.sent.accumulate(rate);
+    protected void published(double rate) {
+        this.published.accumulate(rate);
     }
 
     protected void returned(double rate) {
@@ -164,6 +166,6 @@ protected void nacked(double rate) {
     }
 
     protected void received(double rate) {
-        this.received.accumulate(rate);
+        this.consumed.accumulate(rate);
     }
 }

src/test/java/com/rabbitmq/perf/TopologyTest.java

@@ -582,7 +582,7 @@ public void sequenceMoreQueuesThanProducers() throws Exception {
         when(ch.queueDeclare(queueNameCaptor.capture(), anyBoolean(), anyBoolean(), anyBoolean(), isNull()))
             .then(invocation -> new AMQImpl.Queue.DeclareOk(invocation.getArgument(0), 0, 0));
 
-        // once all producers have sent messages (producerCount routing keys in the set),
+        // once all producers have published messages (producerCount routing keys in the set),
         // we open the latch so MulticastSet.run can end
         Set<String> routingKeys = new HashSet<>();
         CountDownLatch latchPublishing = new CountDownLatch(1);