#154 Log redis metrics to Eventbus

README.md

@@ -75,6 +75,9 @@ The following configuration values are available:
 | httpRequestHandlerPort                  | 7070                            | The port of the HTTP API                                                                                                                                                                        |
 | httpRequestHandlerUserHeader            | x-rp-usr                        | The name of the header property where the user information is provided. Used for the HTTP API                                                                                                   |
 | queueConfigurations                     |                                 | Configure retry intervals and enqueue delaying for queue patterns                                                                                                                               |
+| publish-metrics-address                 |                                 | The EventBus address to send collected redis metrics to                                                                                                                                         |
+| metric-storage-name                     | queue                           | The name of the storage used in the published metrics                                                                                                                                           |
+| metric-refresh-period                   | 10                              | The frequency [s] of collecting metrics from redis database                                                                                                                                     |
 
 ### Configuration util
 

pom.xml

@@ -70,6 +70,11 @@
             <artifactId>slf4j-api</artifactId>
             <version>${slf4j.version}</version>
         </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <version>${guava.version}</version>
+        </dependency>
 
         <!-- TEST dependencies -->
         <dependency>
@@ -388,6 +393,7 @@
         <slf4j.version>2.0.10</slf4j.version>
         <mockito.version>5.8.0</mockito.version>
         <junit.version>4.13.2</junit.version>
+        <guava.version>33.1.0-jre</guava.version>
         <awaitility.version>4.2.0</awaitility.version>
         <jedis.version>3.7.0</jedis.version>
         <rest-assured.version>5.4.0</rest-assured.version>

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

@@ -1,5 +1,6 @@
 package org.swisspush.redisques;
 
+import com.google.common.base.Strings;
 import io.vertx.core.*;
 import io.vertx.core.eventbus.DeliveryOptions;
 import io.vertx.core.eventbus.EventBus;
@@ -91,6 +92,7 @@ private enum QueueState {
     private MemoryUsageProvider memoryUsageProvider;
     private QueueActionFactory queueActionFactory;
     private RedisquesConfigurationProvider configurationProvider;
+    private RedisMonitor redisMonitor;
 
     private Map<QueueOperation, QueueAction> queueActions = new HashMap<>();
 
@@ -257,6 +259,19 @@ private void initialize() {
 
         registerActiveQueueRegistrationRefresh();
         registerQueueCheck();
+        registerMetricsGathering(configuration);
+    }
+
+    private void registerMetricsGathering(RedisquesConfiguration configuration){
+        String metricsAddress = configuration.getPublishMetricsAddress();
+        if(Strings.isNullOrEmpty(metricsAddress)) {
+            return;
+        }
+        String metricStorageName = configuration.getMetricStorageName();
+        int metricRefreshPeriod = configuration.getMetricRefreshPeriod();
+
+        redisMonitor = new RedisMonitor(vertx, redisProvider, metricsAddress, metricStorageName, metricRefreshPeriod);
+        redisMonitor.start();
     }
 
     private void registerActiveQueueRegistrationRefresh() {
@@ -379,6 +394,9 @@ private void unsupportedOperation(String operation, Message<JsonObject> event) {
     @Override
     public void stop() {
         unregisterConsumers(true);
+        if(redisMonitor != null) {
+            redisMonitor.stop();
+        }
     }
 
     private void gracefulStop(final Handler<Void> doneHandler) {

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

@@ -0,0 +1,23 @@
+package org.swisspush.redisques.util;
+
+import io.vertx.core.Vertx;
+import io.vertx.core.json.JsonObject;
+
+public class EventBusMetricsPublisher implements MetricsPublisher {
+
+    private final Vertx vertx;
+    private final String monitoringAddress;
+    private final String prefix;
+
+    public EventBusMetricsPublisher(Vertx vertx, String monitoringAddress, String prefix) {
+        this.vertx = vertx;
+        this.monitoringAddress = monitoringAddress;
+        this.prefix = prefix;
+    }
+
+    @Override
+    public void publishMetric(String name, long value) {
+        vertx.eventBus().publish(monitoringAddress,
+                new JsonObject().put("name", prefix + name).put("action", "set").put("n", value));
+    }
+}

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

@@ -0,0 +1,6 @@
+package org.swisspush.redisques.util;
+
+public interface MetricsPublisher {
+
+    void publishMetric(String name, long value);
+}

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

@@ -0,0 +1,107 @@
+package org.swisspush.redisques.util;
+
+import com.google.common.base.Splitter;
+import io.vertx.core.Vertx;
+import io.vertx.core.buffer.Buffer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class RedisMonitor {
+    private final Vertx vertx;
+    private final RedisProvider redisProvider;
+    private final int period;
+    private long timer;
+    private final Logger log = LoggerFactory.getLogger(RedisMonitor.class);
+
+    private static final String DELIMITER = ":";
+
+    private final MetricsPublisher publisher;
+
+    /**
+     * @param vertx         vertx
+     * @param redisProvider RedisProvider
+     * @param name          name
+     * @param period        in seconds.
+     */
+    public RedisMonitor(Vertx vertx, RedisProvider redisProvider, String monitoringAddress, String name, int period) {
+        this(vertx, redisProvider, name, period,
+                new EventBusMetricsPublisher(vertx, monitoringAddress, "redis." + name + ".")
+        );
+    }
+
+    public RedisMonitor(Vertx vertx, RedisProvider redisProvider, String name, int period, MetricsPublisher publisher) {
+        this.vertx = vertx;
+        this.redisProvider = redisProvider;
+        this.period = period * 1000;
+        this.publisher = publisher;
+    }
+
+    public void start() {
+        timer = vertx.setPeriodic(period, timer -> redisProvider.redis().onSuccess(redisAPI -> {
+            redisAPI.info(new ArrayList<>()).onComplete(event -> {
+                if (event.succeeded()) {
+                    collectMetrics(event.result().toBuffer());
+                } else {
+                    log.warn("Cannot collect INFO from redis");
+                }
+            });
+        }).onFailure(throwable -> log.warn("Cannot collect INFO from redis", throwable)));
+    }
+
+    public void stop() {
+        if (timer != 0) {
+            vertx.cancelTimer(timer);
+        }
+    }
+
+    private void collectMetrics(Buffer buffer) {
+        Map<String, String> map = new HashMap<>();
+
+        Splitter.on(System.lineSeparator()).omitEmptyStrings()
+                .trimResults().splitToList(buffer.toString()).stream()
+                .filter(input -> input != null && input.contains(DELIMITER)
+                        && !input.contains("executable")
+                        && !input.contains("config_file")).forEach(entry -> {
+                    List<String> keyValue = Splitter.on(DELIMITER).omitEmptyStrings().trimResults().splitToList(entry);
+                    if (keyValue.size() == 2) {
+                        map.put(keyValue.get(0), keyValue.get(1));
+                    }
+                });
+
+        log.debug("got redis metrics {}", map);
+
+        map.forEach((key, valueStr) -> {
+            long value;
+            try {
+                if (key.startsWith("db")) {
+                    String[] pairs = valueStr.split(",");
+                    for (String pair : pairs) {
+                        String[] tokens = pair.split("=");
+                        if (tokens.length == 2) {
+                            value = Long.parseLong(tokens[1]);
+                            publisher.publishMetric("keyspace." + key + "." + tokens[0], value);
+                        } else {
+                            log.warn("Invalid keyspace property. Will be ignored");
+                        }
+                    }
+                } else if (key.contains("_cpu_")) {
+                    value = (long) (Double.parseDouble(valueStr) * 1000.0);
+                    publisher.publishMetric(key, value);
+                } else if (key.contains("fragmentation_ratio")) {
+                    value = (long) (Double.parseDouble(valueStr));
+                    publisher.publishMetric(key, value);
+                } else {
+                    value = Long.parseLong(valueStr);
+                    publisher.publishMetric(key, value);
+                }
+            } catch (NumberFormatException e) {
+                // ignore this field
+            }
+        });
+    }
+}