PR for release

README.md

@@ -67,6 +67,7 @@ The following configuration values are available:
 | redisReconnectDelaySec                  | 30                              | The interval [s] to attempt to reconnect when redis connection is lost.                                                                                                                                          |
 | redisPoolRecycleTimeoutMs               | 180000                          | The timeout [ms] when the connection pool is recycled. Use **-1** when having reconnect feature enabled.                                                                                                         |
 | micrometerMetricsEnabled                | false                           | Enable / disable collection of metrics using micrometer                                                                                                                                                          |
+| micrometerMetricsIdentifier             | default                         | Identifier to track values from multiple redisques instances                                                                                                                                                     |
 | httpRequestHandlerEnabled               | false                           | Enable / disable the HTTP API                                                                                                                                                                                    |
 | httpRequestHandlerAuthenticationEnabled | false                           | Enable / disable authentication for the HTTP API                                                                                                                                                                 |
 | httpRequestHandlerUsername              |                                 | The username for the HTTP API authentication                                                                                                                                                                     |

pom.xml

@@ -2,7 +2,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>org.swisspush</groupId>
     <artifactId>redisques</artifactId>
-    <version>4.1.8-SNAPSHOT</version>
+    <version>4.1.9-SNAPSHOT</version>
     <name>redisques</name>
     <description>
         A highly scalable redis-persistent queuing system for vertx

src/main/java/org/swisspush/redisques/QueueStatsService.java

@@ -8,10 +8,7 @@
 import io.vertx.core.json.JsonObject;
 import org.slf4j.Logger;
 import org.swisspush.redisques.exception.RedisQuesExceptionFactory;
-import org.swisspush.redisques.util.DequeueStatistic;
-import org.swisspush.redisques.util.DequeueStatisticCollector;
-import org.swisspush.redisques.util.MetricMeter;
-import org.swisspush.redisques.util.QueueStatisticsCollector;
+import org.swisspush.redisques.util.*;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -55,14 +52,15 @@ public class QueueStatsService {
     private final AtomicLong maxQueueSize = new AtomicLong(0);
 
     public QueueStatsService(
-        Vertx vertx,
-        EventBus eventBus,
-        String redisquesAddress,
-        QueueStatisticsCollector queueStatisticsCollector,
-        DequeueStatisticCollector dequeueStatisticCollector,
-        RedisQuesExceptionFactory exceptionFactory,
-        Semaphore incomingRequestQuota,
-        MeterRegistry meterRegistry
+            Vertx vertx,
+            EventBus eventBus,
+            String redisquesAddress,
+            QueueStatisticsCollector queueStatisticsCollector,
+            DequeueStatisticCollector dequeueStatisticCollector,
+            RedisQuesExceptionFactory exceptionFactory,
+            Semaphore incomingRequestQuota,
+            MeterRegistry meterRegistry,
+            String metricsIdentifier
     ) {
         this.vertx = vertx;
         this.eventBus = eventBus;
@@ -72,9 +70,10 @@ public QueueStatsService(
         this.exceptionFactory = exceptionFactory;
         this.incomingRequestQuota = incomingRequestQuota;
 
-        if(meterRegistry != null) {
+        if (meterRegistry != null) {
             Gauge.builder(MetricMeter.MAX_QUEUE_SIZE.getId(), maxQueueSize, AtomicLong::get).
-                    description(MetricMeter.MAX_QUEUE_SIZE.getDescription()).
+                    description(MetricMeter.MAX_QUEUE_SIZE.getDescription())
+                    .tag(MetricTags.IDENTIFIER.getId(), metricsIdentifier).
                     register(meterRegistry);
         }
     }
@@ -101,14 +100,23 @@ public <CTX> void getQueueStats(CTX mCtx, GetQueueStatsMentor<CTX> mentor) {
             req0.mCtx = mCtx;
             req0.mentor = mentor;
             fetchQueueNamesAndSize(req0, (ex1, req1) -> {
-                if (ex1 != null) { onDone.accept(ex1, null); return; }
+                if (ex1 != null) {
+                    onDone.accept(ex1, null);
+                    return;
+                }
                 // Prepare a list of queue names as it is needed to fetch retryDetails.
                 req1.queueNames = new ArrayList<>(req1.queues.size());
                 for (Queue q : req1.queues) req1.queueNames.add(q.name);
                 fetchRetryDetails(req1, (ex2, req2) -> {
-                    if (ex2 != null) { onDone.accept(ex2, null); return; }
+                    if (ex2 != null) {
+                        onDone.accept(ex2, null);
+                        return;
+                    }
                     attachDequeueStats(req2, (ex3, req3) -> {
-                        if (ex3 != null) { onDone.accept(ex3, null); return; }
+                        if (ex3 != null) {
+                            onDone.accept(ex3, null);
+                            return;
+                        }
                         onDone.accept(null, req3.queues);
                     });
                 });
@@ -172,7 +180,7 @@ private <CTX> void fetchQueueNamesAndSize(GetQueueStatsRequest<CTX> req, BiConsu
     }
 
     private void collectMaxQueueSize(List<Queue> queues) {
-        if(queues.isEmpty()) {
+        if (queues.isEmpty()) {
             maxQueueSize.set(0);
         } else {
             maxQueueSize.set(queues.get(0).getSize());
@@ -182,7 +190,7 @@ private void collectMaxQueueSize(List<Queue> queues) {
     private <CTX> void fetchRetryDetails(GetQueueStatsRequest<CTX> req, BiConsumer<Throwable, GetQueueStatsRequest<CTX>> onDone) {
         long begGetQueueStatsMs = currentTimeMillis();
         assert req.queueNames != null;
-        queueStatisticsCollector.getQueueStatistics(req.queueNames).onComplete( ev -> {
+        queueStatisticsCollector.getQueueStatistics(req.queueNames).onComplete(ev -> {
             req.queueNames = null; // <- no longer needed
             long durGetQueueStatsMs = currentTimeMillis() - begGetQueueStatsMs;
             log.debug("queueStatisticsCollector.getQueueStatistics() took {}ms", durGetQueueStatsMs);
@@ -244,44 +252,62 @@ public static class Queue {
         private Long lastDequeueAttemptEpochMs;
         private Long lastDequeueSuccessEpochMs;
         private Long nextDequeueDueTimestampEpochMs;
-        private Queue(String name, long size){
+
+        private Queue(String name, long size) {
             assert name != null;
             this.name = name;
             this.size = size;
         }
 
-        public String getName() { return name; }
-        public long getSize() { return size; }
-        public Long getLastDequeueAttemptEpochMs() { return lastDequeueAttemptEpochMs; }
-        public Long getLastDequeueSuccessEpochMs() { return lastDequeueSuccessEpochMs; }
-        public Long getNextDequeueDueTimestampEpochMs() { return nextDequeueDueTimestampEpochMs; }
+        public String getName() {
+            return name;
+        }
+
+        public long getSize() {
+            return size;
+        }
+
+        public Long getLastDequeueAttemptEpochMs() {
+            return lastDequeueAttemptEpochMs;
+        }
+
+        public Long getLastDequeueSuccessEpochMs() {
+            return lastDequeueSuccessEpochMs;
+        }
+
+        public Long getNextDequeueDueTimestampEpochMs() {
+            return nextDequeueDueTimestampEpochMs;
+        }
     }
 
 
     /**
      * <p>Mentors fetching operations and so provides the fetcher the required
      * information. Finally it also receives the operations result.</p>
      *
-     * @param <CTX>
-     *     The context object of choice handled back to each callback so the mentor
-     *     knows about what request the fetcher is talking.
+     * @param <CTX> The context object of choice handled back to each callback so the mentor
+     *              knows about what request the fetcher is talking.
      */
     public static interface GetQueueStatsMentor<CTX> {
 
         /**
          * <p>Returning true means that all queues will be present in the result. If
          * false, empty queues won't show up the result.</p>
          *
-         * @param ctx  See {@link GetQueueStatsMentor}.
+         * @param ctx See {@link GetQueueStatsMentor}.
          */
-        public boolean includeEmptyQueues( CTX ctx );
+        public boolean includeEmptyQueues(CTX ctx);
 
-        /** <p>Limits the result to the largest N queues.</p> */
-        public int limit( CTX ctx );
+        /**
+         * <p>Limits the result to the largest N queues.</p>
+         */
+        public int limit(CTX ctx);
 
-        public String filter( CTX ctx);
+        public String filter(CTX ctx);
 
-        /** <p>Called ONCE with the final result.</p> */
+        /**
+         * <p>Called ONCE with the final result.</p>
+         */
         public void onQueueStatistics(List<Queue> queues, CTX ctx);
 
         /**

src/main/java/org/swisspush/redisques/RedisQues.java

@@ -388,8 +388,9 @@ private void initMicrometerMetrics(RedisquesConfiguration modConfig) {
         if(meterRegistry == null) {
             meterRegistry = BackendRegistries.getDefaultNow();
         }
+        String metricsIdentifier = modConfig.getMicrometerMetricsIdentifier();
         dequeueCounter =  Counter.builder(MetricMeter.DEQUEUE.getId())
-                .description(MetricMeter.DEQUEUE.getDescription()).register(meterRegistry);
+                .description(MetricMeter.DEQUEUE.getDescription()).tag(MetricTags.IDENTIFIER.getId(), metricsIdentifier).register(meterRegistry);
 
         String address = modConfig.getAddress();
         int metricRefreshPeriod = modConfig.getMetricRefreshPeriod();

src/main/java/org/swisspush/redisques/action/EnqueueAction.java

@@ -25,16 +25,18 @@ public EnqueueAction(
             Vertx vertx, RedisProvider redisProvider, String address, String queuesKey, String queuesPrefix,
             String consumersPrefix, String locksKey, List<QueueConfiguration> queueConfigurations,
             RedisQuesExceptionFactory exceptionFactory, QueueStatisticsCollector queueStatisticsCollector, Logger log,
-            MemoryUsageProvider memoryUsageProvider, int memoryUsageLimitPercent, MeterRegistry meterRegistry
+            MemoryUsageProvider memoryUsageProvider, int memoryUsageLimitPercent, MeterRegistry meterRegistry, String metricsIdentifier
     ) {
         super(vertx, redisProvider, address, queuesKey, queuesPrefix, consumersPrefix, locksKey,
                 queueConfigurations, exceptionFactory, queueStatisticsCollector, log);
         this.memoryUsageProvider = memoryUsageProvider;
         this.memoryUsageLimitPercent = memoryUsageLimitPercent;
 
         if(meterRegistry != null) {
-            enqueueCounterSuccess =  Counter.builder(MetricMeter.ENQUEUE_SUCCESS.getId()).description(MetricMeter.ENQUEUE_SUCCESS.getDescription()).register(meterRegistry);
-            enqueueCounterFail =  Counter.builder(MetricMeter.ENQUEUE_FAIL.getId()).description(MetricMeter.ENQUEUE_FAIL.getDescription()).register(meterRegistry);
+            enqueueCounterSuccess =  Counter.builder(MetricMeter.ENQUEUE_SUCCESS.getId()).description(MetricMeter.ENQUEUE_SUCCESS.getDescription())
+                    .tag(MetricTags.IDENTIFIER.getId(), metricsIdentifier).register(meterRegistry);
+            enqueueCounterFail =  Counter.builder(MetricMeter.ENQUEUE_FAIL.getId()).description(MetricMeter.ENQUEUE_FAIL.getDescription())
+                    .tag(MetricTags.IDENTIFIER.getId(), metricsIdentifier).register(meterRegistry);
         }
     }
 

src/main/java/org/swisspush/redisques/action/LockedEnqueueAction.java

@@ -23,10 +23,11 @@ public LockedEnqueueAction(Vertx vertx, RedisProvider redisProvider,
                                String consumersPrefix, String locksKey, List<QueueConfiguration> queueConfigurations,
                                RedisQuesExceptionFactory exceptionFactory,
                                QueueStatisticsCollector queueStatisticsCollector, Logger log,
-                               MemoryUsageProvider memoryUsageProvider, int memoryUsageLimitPercent, MeterRegistry meterRegistry) {
+                               MemoryUsageProvider memoryUsageProvider, int memoryUsageLimitPercent, MeterRegistry meterRegistry,
+                               String metricsIdentifier) {
         super(vertx, redisProvider, address, queuesKey, queuesPrefix, consumersPrefix,
                 locksKey, queueConfigurations, exceptionFactory, queueStatisticsCollector, log, memoryUsageProvider,
-                memoryUsageLimitPercent, meterRegistry);
+                memoryUsageLimitPercent, meterRegistry, metricsIdentifier);
     }
 
     @Override

src/main/java/org/swisspush/redisques/handler/RedisquesHttpRequestHandler.java

@@ -140,7 +140,7 @@ private RedisquesHttpRequestHandler(
         this.exceptionFactory = exceptionFactory;
         this.queueStatsService = new QueueStatsService(
                 vertx, eventBus, redisquesAddress, queueStatisticsCollector, dequeueStatisticCollector,
-                exceptionFactory, queueStatsRequestQuota, meterRegistry);
+                exceptionFactory, queueStatsRequestQuota, meterRegistry, modConfig.getMicrometerMetricsIdentifier());
 
         final String prefix = modConfig.getHttpRequestHandlerPrefix();
 

src/main/java/org/swisspush/redisques/util/MetricTags.java

@@ -0,0 +1,16 @@
+package org.swisspush.redisques.util;
+
+public enum MetricTags {
+
+    IDENTIFIER("identifier");
+
+    private final String id;
+
+    MetricTags(String id) {
+        this.id = id;
+    }
+
+    public String getId() {
+        return id;
+    }
+}

src/main/java/org/swisspush/redisques/util/QueueActionFactory.java

@@ -23,6 +23,7 @@ public class QueueActionFactory {
     private final QueueStatisticsCollector queueStatisticsCollector;
     private final int memoryUsageLimitPercent;
     private final MeterRegistry meterRegistry;
+    private final String metricsIdentifier;
     private final MemoryUsageProvider memoryUsageProvider;
     private final RedisQuesExceptionFactory exceptionFactory;
     private final Semaphore getQueuesItemsCountRedisRequestQuota;
@@ -60,6 +61,8 @@ public QueueActionFactory(
         this.memoryUsageLimitPercent = configurationProvider.configuration().getMemoryUsageLimitPercent();
         this.getQueuesItemsCountRedisRequestQuota = getQueuesItemsCountRedisRequestQuota;
         this.meterRegistry = meterRegistry;
+
+        metricsIdentifier = configurationProvider.configuration().getMicrometerMetricsIdentifier();
     }
 
     public QueueAction buildQueueAction(RedisquesAPI.QueueOperation queueOperation){
@@ -101,11 +104,11 @@ public QueueAction buildQueueAction(RedisquesAPI.QueueOperation queueOperation){
             case enqueue:
                 return new EnqueueAction(vertx, redisProvider, address, queuesKey, queuesPrefix,
                         consumersPrefix, locksKey, queueConfigurations, exceptionFactory, queueStatisticsCollector, log,
-                        memoryUsageProvider, memoryUsageLimitPercent, meterRegistry);
+                        memoryUsageProvider, memoryUsageLimitPercent, meterRegistry, metricsIdentifier);
             case lockedEnqueue:
                 return new LockedEnqueueAction(vertx, redisProvider, address, queuesKey, queuesPrefix,
                         consumersPrefix, locksKey, queueConfigurations, exceptionFactory, queueStatisticsCollector, log,
-                        memoryUsageProvider, memoryUsageLimitPercent, meterRegistry);
+                        memoryUsageProvider, memoryUsageLimitPercent, meterRegistry, metricsIdentifier);
             case getLock:
                 return new GetLockAction(vertx, redisProvider, address, queuesKey, queuesPrefix,
                         consumersPrefix, locksKey, queueConfigurations, exceptionFactory, queueStatisticsCollector, log);

src/main/java/org/swisspush/redisques/util/RedisquesConfiguration.java

@@ -64,6 +64,7 @@ public class RedisquesConfiguration {
     private static final int DEFAULT_PROCESSOR_TIMEOUT_MS = 240000; // 240s
     private static final int DEFAULT_METRIC_REFRESH_PERIOD_S = 10; // 10s
     private static final String DEFAULT_METRIC_STORAGE_NAME = "queue";
+    private static final String DEFAULT_MICROMETER_METRICS_IDENTIFIER = "default";
     private static final long DEFAULT_PROCESSOR_DELAY_MAX = 0;
     private static final int DEFAULT_REDIS_MAX_POOL_SIZE = 200;
     private static final int DEFAULT_REDIS_RECONNECT_ATTEMPTS = 0;
@@ -277,6 +278,13 @@ private RedisquesConfiguration(String address, String configurationUpdatedAddres
         this.maxPipelineWaitSize = maxPipelineWaitSize;
         Logger log = LoggerFactory.getLogger(RedisquesConfiguration.class);
 
+        String maybeEmptyMicrometerMetricsIdentifier = Strings.nullToEmpty(micrometerMetricsIdentifier).trim();
+        if(Strings.isNullOrEmpty(maybeEmptyMicrometerMetricsIdentifier)) {
+            this.micrometerMetricsIdentifier = DEFAULT_MICROMETER_METRICS_IDENTIFIER;
+        } else {
+            this.micrometerMetricsIdentifier = maybeEmptyMicrometerMetricsIdentifier;
+        }
+
         String maybeEmptyMetricStorageName = Strings.nullToEmpty(metricStorageName).trim();
         if(Strings.isNullOrEmpty(maybeEmptyMetricStorageName)) {
             this.metricStorageName = DEFAULT_METRIC_STORAGE_NAME;
@@ -319,7 +327,6 @@ private RedisquesConfiguration(String address, String configurationUpdatedAddres
         this.httpRequestHandlerEnabled = httpRequestHandlerEnabled;
         this.httpRequestHandlerAuthenticationEnabled = httpRequestHandlerAuthenticationEnabled;
         this.micrometerMetricsEnabled = micrometerMetricsEnabled;
-        this.micrometerMetricsIdentifier = micrometerMetricsIdentifier;
         this.httpRequestHandlerPrefix = httpRequestHandlerPrefix;
         this.httpRequestHandlerUsername = httpRequestHandlerUsername;
         this.httpRequestHandlerPassword = httpRequestHandlerPassword;

src/test/java/org/swisspush/redisques/RedisQuesTest.java

@@ -37,6 +37,8 @@ public class RedisQuesTest extends AbstractTestCase {
     private Counter enqueueCounterFail;
     private Counter dequeueCounter;
 
+    private final String metricsIdentifier = "foo";
+
     @Rule
     public Timeout rule = Timeout.seconds(50);
 
@@ -47,6 +49,7 @@ public void deployRedisques(TestContext context) {
         JsonObject config = RedisquesConfiguration.with()
                 .processorAddress(PROCESSOR_ADDRESS)
                 .micrometerMetricsEnabled(true)
+                .micrometerMetricsIdentifier(metricsIdentifier)
                 .refreshPeriod(2)
                 .publishMetricsAddress("my-metrics-eb-address")
                 .metricStorageName("foobar")
@@ -61,9 +64,9 @@ public void deployRedisques(TestContext context) {
                 .asJsonObject();
 
         MeterRegistry meterRegistry = new SimpleMeterRegistry();
-        enqueueCounterSuccess = meterRegistry.counter(MetricMeter.ENQUEUE_SUCCESS.getId());
-        enqueueCounterFail = meterRegistry.counter(MetricMeter.ENQUEUE_FAIL.getId());
-        dequeueCounter =  meterRegistry.counter(MetricMeter.DEQUEUE.getId());
+        enqueueCounterSuccess = meterRegistry.counter(MetricMeter.ENQUEUE_SUCCESS.getId(), MetricTags.IDENTIFIER.getId(), metricsIdentifier);
+        enqueueCounterFail = meterRegistry.counter(MetricMeter.ENQUEUE_FAIL.getId(), MetricTags.IDENTIFIER.getId(), metricsIdentifier);
+        dequeueCounter =  meterRegistry.counter(MetricMeter.DEQUEUE.getId(), MetricTags.IDENTIFIER.getId(), metricsIdentifier);
 
         memoryUsageProvider = new TestMemoryUsageProvider(Optional.of(50));
         redisQues = RedisQues.builder()