Start Prometheus server after instance sync

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2018-2022 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2018-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Java client library, is triple-licensed under the
 // Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
@@ -62,6 +62,11 @@ public void configure(ConfigurationContext context) throws Exception {
         }
     }
 
+    @Override
+    public void start() {
+        metrics.forEach(Metrics::start);
+    }
+
     @Override
     public boolean isEnabled(CommandLineProxy cmd) {
         for (Metrics metric : metrics) {

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

@@ -25,6 +25,9 @@
 import java.io.OutputStream;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -68,6 +71,11 @@ static String processConfigurationFile(InputStream configuration, String namespa
     return builder.toString().replace("${namespace}", namespace);
   }
 
+  @Override
+  public void addPostSyncListener(Runnable listener) {
+    this.delegate.addPostSyncListener(listener);
+  }
+
   @Override
   public void synchronize() throws Exception {
     this.delegate.synchronize();
@@ -88,6 +96,7 @@ private static class JGroupsInstanceSynchronization implements InstanceSynchroni
     private final int expectedInstances;
     private final Duration timeout;
     private final JChannel channel;
+    private Set<Runnable> listeners = Collections.synchronizedSet(new LinkedHashSet<>());
 
     private JGroupsInstanceSynchronization(String id, int expectedInstances, String namespace,
         Duration timeout) {
@@ -112,6 +121,11 @@ private JGroupsInstanceSynchronization(String id, int expectedInstances, String
       }
     }
 
+    @Override
+    public void addPostSyncListener(Runnable listener) {
+      listeners.add(listener);
+    }
+
     @Override
     public void synchronize() throws Exception {
       long start = System.nanoTime();
@@ -188,6 +202,14 @@ public void setState(InputStream input) throws Exception {
       } catch (Exception e) {
         LOGGER.info("Error while closing JGroups channel: {}", e.getMessage());
       }
+
+      for (Runnable listener : listeners) {
+        try {
+          listener.run();
+        } catch (Exception e) {
+          LOGGER.info("Error while running instance synchronization listener: {}", e.getMessage());
+        }
+      }
     }
 
   }

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

@@ -7,4 +7,8 @@ interface InstanceSynchronization {
 
   void synchronize() throws Exception;
 
+  default void addPostSyncListener(Runnable listener) {
+    listener.run();
+  }
+
 }

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2018-2022 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2018-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Java client library, is triple-licensed under the
 // Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
@@ -39,6 +39,9 @@ default boolean isEnabled(CommandLineProxy cmd) {
         return false;
     }
 
+    default void start() {
+    }
+
     default void close() throws Exception { }
 
     class ConfigurationContext {

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

@@ -550,13 +550,15 @@ private void shutdown(List<Connection> configurationConnections, Connection[] co
                 if (Thread.interrupted()) {
                     return;
                 }
-                boolean cancelled = producerState.task.cancel(true);
-                LOGGER.debug("Producer has been correctly cancelled: {}", cancelled);
+                if (producerState.task != null) {
+                    boolean cancelled = producerState.task.cancel(true);
+                    LOGGER.debug("Producer has been correctly cancelled: {}", cancelled);
+                }
             }
 
             // we do our best to stop producers before closing their connections
             for (AgentState producerState : producerStates) {
-                if (!producerState.task.isDone()) {
+                if (producerState.task != null && !producerState.task.isDone()) {
                     try {
                         if (Thread.interrupted()) {
                             return;

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

@@ -253,6 +253,7 @@ public static void main(String [] args, PerfTestOptions perfTestOptions) {
             InstanceSynchronization instanceSynchronization = new DefaultInstanceSynchronization(
                 testID, expectedInstances, instanceSyncNamespace, Duration.ofSeconds(instanceSyncTimeout)
             );
+            instanceSynchronization.addPostSyncListener(() -> metrics.start());
 
             MulticastSet set = new MulticastSet(performanceMetrics, factory, p, testID, uris, completionHandler,
                 shutdownService, expectedMetrics, instanceSynchronization);

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2018-2022 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2018-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Java client library, is triple-licensed under the
 // Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
@@ -19,9 +19,11 @@
 import io.micrometer.core.instrument.composite.CompositeMeterRegistry;
 import io.micrometer.prometheus.PrometheusConfig;
 import io.micrometer.prometheus.PrometheusMeterRegistry;
+import java.io.IOException;
 import java.io.OutputStream;
 import java.net.InetSocketAddress;
 import java.nio.charset.StandardCharsets;
+import java.util.concurrent.atomic.AtomicBoolean;
 import org.apache.commons.cli.Option;
 import org.apache.commons.cli.Options;
 
@@ -34,6 +36,8 @@
 public class PrometheusMetrics implements Metrics {
 
     private volatile HttpServer server;
+    private volatile Runnable serverStart = () -> { };
+    private volatile Runnable serverStop = () -> { };
 
     private volatile PrometheusMeterRegistry registry;
 
@@ -53,25 +57,44 @@ public void configure(ConfigurationContext context) throws Exception {
             registry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
             meterRegistry.add(registry);
             int prometheusHttpEndpointPort = intArg(cmd, "mpp", 8080);
-            String prometheusHttpEndpoint = strArg(cmd, "mpe", "metrics");
-            prometheusHttpEndpoint = prometheusHttpEndpoint.startsWith("/") ? prometheusHttpEndpoint : "/" + prometheusHttpEndpoint;
-            server = HttpServer.create(new InetSocketAddress(prometheusHttpEndpointPort), 0);
-            server.createContext(prometheusHttpEndpoint, exchange -> {
-                exchange.getResponseHeaders().set("Content-Type", "text/plain");
-                byte[] content = registry.scrape().getBytes(StandardCharsets.UTF_8);
-                exchange.sendResponseHeaders(200, content.length);
-                try (OutputStream out = exchange.getResponseBody()) {
-                    out.write(content);
+            String endpoint = strArg(cmd, "mpe", "metrics");
+            String prometheusHttpEndpoint = endpoint.startsWith("/") ? endpoint : "/" + endpoint;
+            AtomicBoolean serverStarted = new AtomicBoolean(false);
+            this.serverStart = () -> {
+                try {
+                    server = HttpServer.create(new InetSocketAddress(prometheusHttpEndpointPort), 0);
+                } catch (IOException e) {
+                    throw new PerfTestException("Error while starting Prometheus HTTP server", e);
                 }
-            });
-            server.start();
+                server.createContext(prometheusHttpEndpoint, exchange -> {
+                    exchange.getResponseHeaders().set("Content-Type", "text/plain");
+                    byte[] content = registry.scrape().getBytes(StandardCharsets.UTF_8);
+                    exchange.sendResponseHeaders(200, content.length);
+                    try (OutputStream out = exchange.getResponseBody()) {
+                        out.write(content);
+                    }
+                });
+                server.start();
+                serverStarted.set(true);
+            };
+            this.serverStop = () -> {
+                if (serverStarted.compareAndSet(true, false)) {
+                    server.stop(0);
+                }
+            };
+        } else {
+            this.serverStart = () -> { };
+            this.serverStop = () -> { };
         }
     }
 
+    @Override
+    public void start() {
+        this.serverStart.run();
+    }
+
     public void close() throws Exception {
-        if (server != null) {
-            server.stop(0);
-        }
+        this.serverStop.run();
         if (registry != null) {
             registry.close();
         }