Add Prometheus metrics on HTTP monitoring endpoint

Fixes #29

Author: acogoluegnes

pom.xml

@@ -168,6 +168,13 @@
       <optional>true</optional>
     </dependency>
 
+    <dependency>
+      <groupId>io.micrometer</groupId>
+      <artifactId>micrometer-registry-prometheus</artifactId>
+      <version>${micrometer.version}</version>
+      <optional>true</optional>
+    </dependency>
+
     <!-- for rate limiter -->
     <dependency>
       <groupId>com.google.guava</groupId>
@@ -562,6 +569,12 @@
           <version>${micrometer.version}</version>
         </dependency>
 
+        <dependency>
+          <groupId>io.micrometer</groupId>
+          <artifactId>micrometer-registry-prometheus</artifactId>
+          <version>${micrometer.version}</version>
+        </dependency>
+
         <!-- for rate limiter -->
         <dependency>
           <groupId>com.google.guava</groupId>

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

@@ -13,6 +13,7 @@
 // [email protected].
 package com.rabbitmq.stream.perf;
 
+import io.micrometer.core.instrument.composite.CompositeMeterRegistry;
 import java.util.ArrayList;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -29,13 +30,15 @@
 class MonitoringContext {
 
   private final int monitoringPort;
+  private final CompositeMeterRegistry meterRegistry;
 
   private final Map<String, Handler> handlers = new LinkedHashMap<>();
 
   private volatile Server server;
 
-  MonitoringContext(int monitoringPort) {
+  MonitoringContext(int monitoringPort, CompositeMeterRegistry meterRegistry) {
     this.monitoringPort = monitoringPort;
+    this.meterRegistry = meterRegistry;
   }
 
   void addHttpEndpoint(String path, Handler handler) {
@@ -77,4 +80,8 @@ void close() throws Exception {
       server.stop();
     }
   }
+
+  CompositeMeterRegistry meterRegistry() {
+    return meterRegistry;
+  }
 }

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

@@ -0,0 +1,64 @@
+// Copyright (c) 2021 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").
+// For the MPL, please see LICENSE-MPL-RabbitMQ. For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// [email protected].
+package com.rabbitmq.stream.perf;
+
+import io.micrometer.prometheus.PrometheusConfig;
+import io.micrometer.prometheus.PrometheusMeterRegistry;
+import java.io.IOException;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import org.eclipse.jetty.server.Request;
+import org.eclipse.jetty.server.handler.AbstractHandler;
+import picocli.CommandLine.Option;
+
+class PrometheusEndpointMonitoring implements Monitoring {
+
+  @Option(
+      names = {"--prometheus"},
+      description = "Enable HTTP Prometheus metrics endpoint",
+      defaultValue = "false")
+  private boolean enabled;
+
+  private volatile PrometheusMeterRegistry registry;
+
+  @Override
+  public void configure(MonitoringContext context) {
+    if (enabled) {
+      registry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
+      context.meterRegistry().add(registry);
+      context.addHttpEndpoint(
+          "metrics",
+          new AbstractHandler() {
+            @Override
+            public void handle(
+                String target,
+                Request baseRequest,
+                HttpServletRequest request,
+                HttpServletResponse response)
+                throws IOException {
+
+              String scraped = registry.scrape();
+
+              response.setStatus(HttpServletResponse.SC_OK);
+              response.setContentLength(scraped.length());
+              response.setContentType("text/plain");
+
+              response.getWriter().print(scraped);
+
+              baseRequest.setHandled(true);
+            }
+          });
+    }
+  }
+}

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

@@ -360,7 +360,8 @@ static int run(String[] args, PrintStream consoleOut, PrintStream consoleErr) {
     CommandLine commandLine =
         new CommandLine(streamPerfTest).setOut(streamPerfTest.out).setErr(streamPerfTest.err);
 
-    List<Monitoring> monitorings = Arrays.asList(new DebugEndpointMonitoring());
+    List<Monitoring> monitorings =
+        Arrays.asList(new DebugEndpointMonitoring(), new PrometheusEndpointMonitoring());
 
     monitorings.forEach(m -> commandLine.addMixin(m.getClass().getSimpleName(), m));
 
@@ -477,7 +478,7 @@ public Integer call() throws Exception {
 
     ShutdownService shutdownService = new ShutdownService();
 
-    MonitoringContext monitoringContext = new MonitoringContext(this.monitoringPort);
+    MonitoringContext monitoringContext = new MonitoringContext(this.monitoringPort, meterRegistry);
     this.monitorings.forEach(m -> m.configure(monitoringContext));
     monitoringContext.start();
 

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

@@ -315,7 +315,13 @@ void compressionWithoutSubEntriesShouldNotStart() throws Exception {
   @Test
   void monitoringShouldReturnValidEndpoint() throws Exception {
     int monitoringPort = randomNetworkPort();
-    Future<?> run = run(builder().deleteStreams().monitoring(true).monitoringPort(monitoringPort));
+    Future<?> run =
+        run(
+            builder()
+                .deleteStreams()
+                .monitoring(true)
+                .monitoringPort(monitoringPort)
+                .prometheus(true));
     waitUntilStreamExists(s);
     waitOneSecond();
 
@@ -327,6 +333,13 @@ void monitoringShouldReturnValidEndpoint() throws Exception {
               && response.body.contains("stream-perf-test-publishers-");
         });
 
+    waitAtMost(
+        10,
+        () -> {
+          HttpResponse response = httpRequest("http://localhost:" + monitoringPort + "/metrics");
+          return response.responseCode == 200
+              && response.body.contains("# HELP rabbitmq_stream_published_total");
+        });
     run.cancel(true);
     waitRunEnds();
   }
@@ -509,6 +522,11 @@ ArgumentsBuilder monitoringPort(int port) {
       return this;
     }
 
+    ArgumentsBuilder prometheus(boolean prometheus) {
+      arguments.put("prometheus", String.valueOf(prometheus));
+      return this;
+    }
+
     String build() {
       return this.arguments.entrySet().stream()
           .map(e -> "--" + e.getKey() + (e.getValue().isEmpty() ? "" : (" " + e.getValue())))