Add --metrics-command-line-arguments option

Fixes #225

Author: acogoluegnes

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

@@ -55,6 +55,7 @@ class DefaultPerformanceMetrics implements PerformanceMetrics {
   private volatile long lastPublishedCount = 0;
   private volatile long lastConsumedCount = 0;
   private volatile long offset;
+  private final String metricsSuffix;
 
   DefaultPerformanceMetrics(
       CompositeMeterRegistry meterRegistry,
@@ -111,28 +112,36 @@ protected Double nullGaugeValue() {
     } else {
       this.confirmLatency = null;
     }
+    // the metrics name contains the tags, if any,
+    // so we extract the suffix to use it later when looking up the metrics
+    String key = metricRegistry.getMeters().keySet().iterator().next();
+    int index = key.indexOf(".");
+    this.metricsSuffix = index == -1 ? "" : key.substring(index);
   }
 
   private long getPublishedCount() {
-    return this.metricRegistry.getMeters().get("rabbitmqStreamPublished").getCount();
+    return this.metricRegistry
+        .getMeters()
+        .get("rabbitmqStreamPublished" + metricsSuffix)
+        .getCount();
   }
 
   private long getConsumedCount() {
-    return this.metricRegistry.getMeters().get("rabbitmqStreamConsumed").getCount();
+    return this.metricRegistry.getMeters().get("rabbitmqStreamConsumed" + metricsSuffix).getCount();
   }
 
   @Override
   public void start(String description) throws Exception {
     long startTime = System.nanoTime();
 
-    String metricPublished = "rabbitmqStreamPublished";
-    String metricProducerConfirmed = "rabbitmqStreamProducer_confirmed";
-    String metricConsumed = "rabbitmqStreamConsumed";
-    String metricChunkSize = "rabbitmqStreamChunk_size";
-    String metricLatency = "rabbitmqStreamLatency";
-    String metricConfirmLatency = "rabbitmqStreamConfirm_latency";
-    String metricWrittenBytes = "rabbitmqStreamWritten_bytes";
-    String metricReadBytes = "rabbitmqStreamRead_bytes";
+    String metricPublished = "rabbitmqStreamPublished" + metricsSuffix;
+    String metricProducerConfirmed = "rabbitmqStreamProducer_confirmed" + metricsSuffix;
+    String metricConsumed = "rabbitmqStreamConsumed" + metricsSuffix;
+    String metricChunkSize = "rabbitmqStreamChunk_size" + metricsSuffix;
+    String metricLatency = "rabbitmqStreamLatency" + metricsSuffix;
+    String metricConfirmLatency = "rabbitmqStreamConfirm_latency" + metricsSuffix;
+    String metricWrittenBytes = "rabbitmqStreamWritten_bytes" + metricsSuffix;
+    String metricReadBytes = "rabbitmqStreamRead_bytes" + metricsSuffix;
 
     Set<String> allMetrics =
         new HashSet<>(

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

@@ -49,6 +49,7 @@
 import com.rabbitmq.stream.perf.ShutdownService.CloseCallback;
 import com.rabbitmq.stream.perf.Utils.NamedThreadFactory;
 import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.Tag;
 import io.micrometer.core.instrument.composite.CompositeMeterRegistry;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.ByteBufAllocatorMetric;
@@ -401,6 +402,13 @@ public void setMaxSegmentSize(ByteCapacity in) {
       converter = Utils.GreaterThanOrEqualToZeroIntegerTypeConverter.class)
   private int time;
 
+  @CommandLine.Option(
+      names = {"--metrics-tags", "-mt"},
+      description = "metrics tags as key-value pairs separated by commas",
+      defaultValue = "",
+      converter = Utils.MetricsTagsTypeConverter.class)
+  private Collection<Tag> metricsTags;
+
   private MetricsCollector metricsCollector;
   private PerformanceMetrics performanceMetrics;
   private List<Monitoring> monitorings;
@@ -503,6 +511,7 @@ public Integer call() throws Exception {
     ByteBufAllocator byteBufAllocator = ByteBufAllocator.DEFAULT;
 
     CompositeMeterRegistry meterRegistry = new CompositeMeterRegistry();
+    meterRegistry.config().commonTags(this.metricsTags);
     String metricsPrefix = "rabbitmq.stream";
     this.metricsCollector = new MicrometerMetricsCollector(meterRegistry, metricsPrefix);
 

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

@@ -24,6 +24,7 @@
 import com.rabbitmq.stream.StreamCreator.LeaderLocator;
 import com.rabbitmq.stream.compression.Compression;
 import com.sun.management.OperatingSystemMXBean;
+import io.micrometer.core.instrument.Tag;
 import java.lang.reflect.Constructor;
 import java.lang.reflect.Field;
 import java.lang.reflect.Method;
@@ -71,6 +72,7 @@
 import picocli.CommandLine.Model.CommandSpec;
 import picocli.CommandLine.Model.OptionSpec;
 import picocli.CommandLine.Option;
+import picocli.CommandLine.TypeConversionException;
 
 class Utils {
 
@@ -315,6 +317,28 @@ public ByteCapacity convert(String value) {
     }
   }
 
+  static class MetricsTagsTypeConverter implements CommandLine.ITypeConverter<Collection<Tag>> {
+
+    @Override
+    public Collection<Tag> convert(String value) throws Exception {
+      if (value == null || value.trim().isEmpty()) {
+        return Collections.emptyList();
+      } else {
+        try {
+          Collection<Tag> tags = new ArrayList<>();
+          for (String tag : value.split(",")) {
+            String[] keyValue = tag.split("=", 2);
+            tags.add(Tag.of(keyValue[0], keyValue[1]));
+          }
+          return tags;
+        } catch (Exception e) {
+          throw new TypeConversionException(
+              String.format("'%s' is not valid, use key/value pairs separated by commas"));
+        }
+      }
+    }
+  }
+
   static class NameStrategyConverter
       implements CommandLine.ITypeConverter<BiFunction<String, Integer, String>> {
 

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

@@ -387,6 +387,31 @@ void monitoringShouldReturnValidEndpoint() throws Exception {
     waitRunEnds();
   }
 
+  @Test
+  void metricsTagsShouldShowUpInHttpEndpoint() throws Exception {
+    int monitoringPort = randomNetworkPort();
+    Future<?> run =
+        run(
+            builder()
+                .deleteStreams()
+                .monitoring()
+                .monitoringPort(monitoringPort)
+                .prometheus()
+                .metricsTags("env=performance,datacenter=eu"));
+    waitUntilStreamExists(s);
+    waitOneSecond();
+
+    waitAtMost(
+        10,
+        () -> {
+          HttpResponse response = httpRequest("http://localhost:" + monitoringPort + "/metrics");
+          return response.responseCode == 200
+              && response.body.contains("{datacenter=\"eu\",env=\"performance\",}");
+        });
+    run.cancel(true);
+    waitRunEnds();
+  }
+
   @Test
   void publishConfirmLatencyShouldBeIncludedWhenOptionIsEnabled() throws Exception {
     Future<?> run = run(builder().confirmLatency());
@@ -640,6 +665,11 @@ ArgumentsBuilder prometheus() {
       return this;
     }
 
+    ArgumentsBuilder metricsTags(String tags) {
+      arguments.put("metrics-tags", tags);
+      return this;
+    }
+
     ArgumentsBuilder superStreams() {
       arguments.put("super-streams", "");
       return this;

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

@@ -23,10 +23,12 @@
 import com.rabbitmq.stream.OffsetSpecification;
 import com.rabbitmq.stream.compression.Compression;
 import com.rabbitmq.stream.perf.Utils.CompressionTypeConverter;
+import com.rabbitmq.stream.perf.Utils.MetricsTagsTypeConverter;
 import com.rabbitmq.stream.perf.Utils.NameStrategyConverter;
 import com.rabbitmq.stream.perf.Utils.PatternNameStrategy;
 import com.rabbitmq.stream.perf.Utils.RangeTypeConverter;
 import com.rabbitmq.stream.perf.Utils.SniServerNamesConverter;
+import io.micrometer.core.instrument.Tag;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
@@ -78,7 +80,7 @@ static Stream<Arguments> offsetSpecificationTypeConverterOkArguments() {
   static Stream<Arguments> streams() {
     Stream<Arguments> arguments =
         Stream.of(
-            of("1", "stream", Arrays.asList("stream")),
+            of("1", "stream", Collections.singletonList("stream")),
             of("5", "stream", IntStream.range(1, 6).mapToObj(i -> "stream-" + i).collect(toList())),
             of(
                 "10",
@@ -111,6 +113,10 @@ static Stream<Arguments> streams() {
         });
   }
 
+  private static Tag tag(String key, String value) {
+    return Tag.of(key, value);
+  }
+
   @ParameterizedTest
   @MethodSource("offsetSpecificationTypeConverterOkArguments")
   void offsetSpecificationTypeConverterOk(String value, OffsetSpecification expected)
@@ -191,6 +197,21 @@ void sniServerNamesConverter() throws Exception {
         .contains(new SNIHostName("dummy"));
   }
 
+  @Test
+  void metricsTagsConverter() throws Exception {
+    MetricsTagsTypeConverter converter = new MetricsTagsTypeConverter();
+    assertThat(converter.convert(null)).isNotNull().isEmpty();
+    assertThat(converter.convert("")).isNotNull().isEmpty();
+    assertThat(converter.convert("  ")).isNotNull().isEmpty();
+    assertThat(converter.convert("env=performance,datacenter=eu"))
+        .hasSize(2)
+        .contains(tag("env", "performance"))
+        .contains(tag("datacenter", "eu"));
+    assertThat(converter.convert("args=--queue-args \"x-max-length=100000\""))
+        .hasSize(1)
+        .contains(tag("args", "--queue-args \"x-max-length=100000\""));
+  }
+
   @Test
   void writeReadLongInByteArray() {
     byte[] array = new byte[8];
@@ -327,22 +348,23 @@ void superStreamPartitionsTest() {
 
   @Command(name = "test-command")
   static class TestCommand {
+
     @Option(
         names = {"aaa", "a"},
         defaultValue = "10")
-    private int a = 10;
+    private final int a = 10;
 
     @Option(names = "b", defaultValue = "false")
-    private boolean b = false;
+    private final boolean b = false;
 
     @Option(names = "c", defaultValue = "false")
-    private boolean c = false;
+    private final boolean c = false;
 
     @CommandLine.Option(
         names = {"offset"},
         defaultValue = "next",
         converter = Utils.OffsetSpecificationTypeConverter.class)
-    private OffsetSpecification offsetSpecification = OffsetSpecification.next();
+    private final OffsetSpecification offsetSpecification = OffsetSpecification.next();
 
     public TestCommand() {}
   }