Feature/issue224 micrometer metrics

README.md

@@ -66,6 +66,7 @@ The following configuration values are available:
 | redisReconnectAttempts                  | 0                               | The amount of attempts to reconnect when redis connection is lost. Use **0** to not reconnect at all or **-1** to reconnect indefinitely.                                                                        |
 | 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                                                                                                                                                          |
 | httpRequestHandlerEnabled               | false                           | Enable / disable the HTTP API                                                                                                                                                                                    |
 | httpRequestHandlerAuthenticationEnabled | false                           | Enable / disable authentication for the HTTP API                                                                                                                                                                 |
 | httpRequestHandlerUsername              |                                 | The username for the HTTP API authentication                                                                                                                                                                     |
@@ -630,7 +631,6 @@ Response Data
 }
 ```
 
-
 #### getQueuesSpeed
 
 Request Data
@@ -651,7 +651,6 @@ Response Data
 }
 ```
 
-
 ## RedisQues HTTP API
 RedisQues provides a HTTP API to modify queues, queue items and get information about queue counts and queue item counts.
 
@@ -1021,7 +1020,7 @@ Available url parameters are:
 * _filter=<regex pattern>_: Filter the queues for which the cumulated speed is evaluated
 
 The result will be a json object with the speed of the last measurement period calculated
-over all queues matching the given filter regex. Additionally the used measurement time in seconds
+over all queues matching the given filter regex. Additionally, the used measurement time in seconds
 is returned (eg. 60 seconds by default)
 
 ```json
@@ -1040,6 +1039,45 @@ SemaphoreConfig semaphoreConfig = new SemaphoreConfig().setInitialPermits(1).set
 hazelcastConfig.getCPSubsystemConfig().addSemaphoreConfig(semaphoreConfig);
 ```
 
+## Metric collection
+Besides the API, redisques provides some key metrics collected by [micrometer.io](https://micrometer.io/).
+
+The collected metrics include:
+
+| Metric name                     | Description                                                 |
+|:--------------------------------|:------------------------------------------------------------|
+| redisques_enqueue_success_total | Overall count of queue items to be enqueued                 |
+| redisques_enqueue_fail_total    | Overall count of failing enqueues                           |
+| redisques_dequeue_total         | Overall count of queue items to be dequeued from the queues |
+| redisques_active_queues         | Overall count of active queues                              |
+| redisques_max_queue_size        | Amount of queue items of the biggest queue                  |
+
+### Testing locally
+When you include redisques in you project, you probably already have the configuration for publishing the metrics.
+
+To export the metrics locally you have to add this dependency to the `pom.xml`
+
+```
+<dependency>
+    <groupId>io.micrometer</groupId>
+    <artifactId>micrometer-registry-prometheus</artifactId>
+    <version>${micrometer.version}</version>
+</dependency>
+```
+Also add the micrometer configuration to `RedisQuesRunner` class like this:
+
+```java
+MicrometerMetricsOptions options = new MicrometerMetricsOptions()
+        .setPrometheusOptions(new VertxPrometheusOptions()
+                .setStartEmbeddedServer(true)
+                .setEmbeddedServerOptions(new HttpServerOptions().setPort(9101))
+                .setEnabled(true))
+        .setEnabled(true);
+Vertx vertx = Vertx.vertx(new VertxOptions().setMetricsOptions(options));
+```
+Using the configuration above, the metrics can be accessed with
+
+> GET http://localhost:9101/metrics
 
 
 ## Dependencies

pom.xml

@@ -65,6 +65,16 @@
             <artifactId>vertx-web</artifactId>
             <version>${vertx.version}</version>
         </dependency>
+        <dependency>
+            <groupId>io.vertx</groupId>
+            <artifactId>vertx-micrometer-metrics</artifactId>
+            <version>${vertx.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.micrometer</groupId>
+            <artifactId>micrometer-core</artifactId>
+            <version>${micrometer.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>
@@ -390,6 +400,7 @@
     </profiles>
     <properties>
         <vertx.version>4.5.2</vertx.version>
+        <micrometer.version>1.12.13</micrometer.version>
         <slf4j.version>2.0.10</slf4j.version>
         <mockito.version>5.8.0</mockito.version>
         <junit.version>4.13.2</junit.version>

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

@@ -1,5 +1,7 @@
 package org.swisspush.redisques;
 
+import io.micrometer.core.instrument.Gauge;
+import io.micrometer.core.instrument.MeterRegistry;
 import io.vertx.core.Vertx;
 import io.vertx.core.eventbus.EventBus;
 import io.vertx.core.json.JsonArray;
@@ -8,15 +10,14 @@
 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 java.util.ArrayList;
-import java.util.HashMap;
-import java.util.Iterator;
 import java.util.List;
-import java.util.Map;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.BiConsumer;
 
 import static java.lang.Long.compare;
@@ -51,14 +52,17 @@ public class QueueStatsService {
     private final RedisQuesExceptionFactory exceptionFactory;
     private final Semaphore incomingRequestQuota;
 
+    private final AtomicLong maxQueueSize = new AtomicLong(0);
+
     public QueueStatsService(
         Vertx vertx,
         EventBus eventBus,
         String redisquesAddress,
         QueueStatisticsCollector queueStatisticsCollector,
         DequeueStatisticCollector dequeueStatisticCollector,
         RedisQuesExceptionFactory exceptionFactory,
-        Semaphore incomingRequestQuota
+        Semaphore incomingRequestQuota,
+        MeterRegistry meterRegistry
     ) {
         this.vertx = vertx;
         this.eventBus = eventBus;
@@ -67,6 +71,12 @@ public QueueStatsService(
         this.dequeueStatisticCollector = dequeueStatisticCollector;
         this.exceptionFactory = exceptionFactory;
         this.incomingRequestQuota = incomingRequestQuota;
+
+        if(meterRegistry != null) {
+            Gauge.builder(MetricMeter.MAX_QUEUE_SIZE.getId(), maxQueueSize, AtomicLong::get).
+                    description(MetricMeter.MAX_QUEUE_SIZE.getDescription()).
+                    register(meterRegistry);
+        }
     }
 
     public <CTX> void getQueueStats(CTX mCtx, GetQueueStatsMentor<CTX> mentor) {
@@ -156,10 +166,19 @@ private <CTX> void fetchQueueNamesAndSize(GetQueueStatsRequest<CTX> req, BiConsu
             int limit = req.mentor.limit(req.mCtx);
             if (limit != 0 && queues.size() > limit) queues = queues.subList(0, limit);
             req.queues = queues;
+            collectMaxQueueSize(queues);
             onDone.accept(null, req);
         });
     }
 
+    private void collectMaxQueueSize(List<Queue> queues) {
+        if(queues.isEmpty()) {
+            maxQueueSize.set(0);
+        } else {
+            maxQueueSize.set(queues.get(0).getSize());
+        }
+    }
+
     private <CTX> void fetchRetryDetails(GetQueueStatsRequest<CTX> req, BiConsumer<Throwable, GetQueueStatsRequest<CTX>> onDone) {
         long begGetQueueStatsMs = currentTimeMillis();
         assert req.queueNames != null;

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

@@ -1,6 +1,8 @@
 package org.swisspush.redisques;
 
 import com.google.common.base.Strings;
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.MeterRegistry;
 import io.vertx.core.AbstractVerticle;
 import io.vertx.core.AsyncResult;
 import io.vertx.core.CompositeFuture;
@@ -13,6 +15,7 @@
 import io.vertx.core.eventbus.MessageConsumer;
 import io.vertx.core.json.JsonArray;
 import io.vertx.core.json.JsonObject;
+import io.vertx.micrometer.backends.BackendRegistries;
 import io.vertx.redis.client.Command;
 import io.vertx.redis.client.RedisAPI;
 import io.vertx.redis.client.Response;
@@ -21,6 +24,7 @@
 import org.swisspush.redisques.action.QueueAction;
 import org.swisspush.redisques.exception.RedisQuesExceptionFactory;
 import org.swisspush.redisques.handler.RedisquesHttpRequestHandler;
+import org.swisspush.redisques.metrics.PeriodicMetricsCollector;
 import org.swisspush.redisques.performance.UpperBoundParallel;
 import org.swisspush.redisques.scheduling.PeriodicSkipScheduler;
 import org.swisspush.redisques.util.*;
@@ -84,6 +88,7 @@ public static class RedisQuesBuilder {
         private RedisquesConfigurationProvider configurationProvider;
         private RedisProvider redisProvider;
         private RedisQuesExceptionFactory exceptionFactory;
+        private MeterRegistry meterRegistry;
         private Semaphore redisMonitoringReqQuota;
         private Semaphore checkQueueRequestsQuota;
         private Semaphore queueStatsRequestQuota;
@@ -113,6 +118,11 @@ public RedisQuesBuilder withExceptionFactory(RedisQuesExceptionFactory exception
             return this;
         }
 
+        public RedisQuesBuilder withMeterRegistry(MeterRegistry meterRegistry) {
+            this.meterRegistry = meterRegistry;
+            return this;
+        }
+
         /**
          * How many redis requests monitoring related component will trigger
          * simultaneously. One of those components for example is
@@ -172,7 +182,7 @@ public RedisQues build() {
             }
             return new RedisQues(memoryUsageProvider, configurationProvider, redisProvider, exceptionFactory,
                     redisMonitoringReqQuota, checkQueueRequestsQuota, queueStatsRequestQuota,
-                    getQueuesItemsCountRedisRequestQuota);
+                    getQueuesItemsCountRedisRequestQuota, meterRegistry);
         }
     }
 
@@ -218,6 +228,9 @@ private enum QueueState {
     private RedisquesConfigurationProvider configurationProvider;
     private RedisMonitor redisMonitor;
 
+    private MeterRegistry meterRegistry;
+    private Counter dequeueCounter;
+
     private Map<QueueOperation, QueueAction> queueActions = new HashMap<>();
 
     private Map<String, DequeueStatistic> dequeueStatistic = new ConcurrentHashMap<>();
@@ -235,6 +248,20 @@ public RedisQues() {
         log.warn("Fallback to legacy behavior and allow up to {} simultaneous requests to redis", Integer.MAX_VALUE);
     }
 
+    public RedisQues(
+            MemoryUsageProvider memoryUsageProvider,
+            RedisquesConfigurationProvider configurationProvider,
+            RedisProvider redisProvider,
+            RedisQuesExceptionFactory exceptionFactory,
+            Semaphore redisMonitoringReqQuota,
+            Semaphore checkQueueRequestsQuota,
+            Semaphore queueStatsRequestQuota,
+            Semaphore getQueuesItemsCountRedisRequestQuota
+    ) {
+        this(memoryUsageProvider, configurationProvider, redisProvider, exceptionFactory, redisMonitoringReqQuota,
+                checkQueueRequestsQuota, queueStatsRequestQuota, getQueuesItemsCountRedisRequestQuota, null);
+    }
+
     public RedisQues(
         MemoryUsageProvider memoryUsageProvider,
         RedisquesConfigurationProvider configurationProvider,
@@ -243,7 +270,8 @@ public RedisQues(
         Semaphore redisMonitoringReqQuota,
         Semaphore checkQueueRequestsQuota,
         Semaphore queueStatsRequestQuota,
-        Semaphore getQueuesItemsCountRedisRequestQuota
+        Semaphore getQueuesItemsCountRedisRequestQuota,
+        MeterRegistry meterRegistry
     ) {
         this.memoryUsageProvider = memoryUsageProvider;
         this.configurationProvider = configurationProvider;
@@ -253,6 +281,7 @@ public RedisQues(
         this.checkQueueRequestsQuota = checkQueueRequestsQuota;
         this.queueStatsRequestQuota = queueStatsRequestQuota;
         this.getQueuesItemsCountRedisRequestQuota = getQueuesItemsCountRedisRequestQuota;
+        this.meterRegistry = meterRegistry;
     }
 
     public static RedisQuesBuilder builder() {
@@ -318,6 +347,10 @@ public void start(Promise<Void> promise) {
         RedisquesConfiguration modConfig = configurationProvider.configuration();
         log.info("Starting Redisques module with configuration: {}", configurationProvider.configuration());
 
+        if(configurationProvider.configuration().getMicrometerMetricsEnabled()) {
+            initMicrometerMetrics(modConfig);
+        }
+
         int dequeueStatisticReportIntervalSec = modConfig.getDequeueStatisticReportIntervalSec();
         if (modConfig.isDequeueStatsEnabled()) {
             dequeueStatisticEnabled = true;
@@ -351,6 +384,18 @@ public void start(Promise<Void> promise) {
         });
     }
 
+    private void initMicrometerMetrics(RedisquesConfiguration modConfig) {
+        if(meterRegistry == null) {
+            meterRegistry = BackendRegistries.getDefaultNow();
+        }
+        dequeueCounter =  Counter.builder(MetricMeter.DEQUEUE.getId())
+                .description(MetricMeter.DEQUEUE.getDescription()).register(meterRegistry);
+
+        String address = modConfig.getAddress();
+        int metricRefreshPeriod = modConfig.getMetricRefreshPeriod();
+        new PeriodicMetricsCollector(vertx, periodicSkipScheduler, address, meterRegistry, metricRefreshPeriod);
+    }
+
     private void initialize() {
         RedisquesConfiguration configuration = configurationProvider.configuration();
         this.queueStatisticsCollector = new QueueStatisticsCollector(
@@ -359,7 +404,7 @@ private void initialize() {
 
         RedisquesHttpRequestHandler.init(
             vertx, configuration, queueStatisticsCollector, dequeueStatisticCollector,
-            exceptionFactory, queueStatsRequestQuota);
+            exceptionFactory, queueStatsRequestQuota, meterRegistry);
 
         // only initialize memoryUsageProvider when not provided in the constructor
         if (memoryUsageProvider == null) {
@@ -371,7 +416,7 @@ private void initialize() {
         queueActionFactory = new QueueActionFactory(
                 redisProvider, vertx, log, queuesKey, queuesPrefix, consumersPrefix, locksKey,
                 memoryUsageProvider, queueStatisticsCollector, exceptionFactory,
-                configurationProvider, getQueuesItemsCountRedisRequestQuota);
+                configurationProvider, getQueuesItemsCountRedisRequestQuota, meterRegistry);
 
         queueActions.put(addQueueItem, queueActionFactory.buildQueueAction(addQueueItem));
         queueActions.put(deleteQueueItem, queueActionFactory.buildQueueAction(deleteQueueItem));
@@ -950,6 +995,9 @@ private Future<Void> readQueue(final String queueName) {
                                         log.error("Failed to pop from queue '{}'", queueName, jsonAnswer.cause());
                                         // We should return here. See: "https://softwareengineering.stackexchange.com/a/190535"
                                     }
+                                    if(dequeueCounter != null) {
+                                        dequeueCounter.increment();
+                                    }
                                     log.debug("RedisQues Message removed, queue {} is ready again", queueName);
                                     myQueues.put(queueName, QueueState.READY);