Merge pull request #227 from swisspost/develop

PR for new release
swisspost · Nov 29, 2024 · a1b9023 · a1b9023
2 parents 04dee96 + bc92c6c
commit a1b9023
Show file tree

Hide file tree

Showing 5 changed files with 56 additions and 16 deletions.
diff --git a/pom.xml b/pom.xml
@@ -2,7 +2,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>org.swisspush</groupId>
     <artifactId>redisques</artifactId>
-    <version>4.1.7-SNAPSHOT</version>
+    <version>4.1.8-SNAPSHOT</version>
     <name>redisques</name>
     <description>
         A highly scalable redis-persistent queuing system for vertx

diff --git a/src/main/java/org/swisspush/redisques/RedisQues.java b/src/main/java/org/swisspush/redisques/RedisQues.java
@@ -393,7 +393,8 @@ private void initMicrometerMetrics(RedisquesConfiguration modConfig) {
 
         String address = modConfig.getAddress();
         int metricRefreshPeriod = modConfig.getMetricRefreshPeriod();
-        new PeriodicMetricsCollector(vertx, periodicSkipScheduler, address, meterRegistry, metricRefreshPeriod);
+        String identifier = modConfig.getMicrometerMetricsIdentifier();
+        new PeriodicMetricsCollector(vertx, periodicSkipScheduler, address, identifier, meterRegistry, metricRefreshPeriod);
     }
 
     private void initialize() {

diff --git a/src/main/java/org/swisspush/redisques/metrics/PeriodicMetricsCollector.java b/src/main/java/org/swisspush/redisques/metrics/PeriodicMetricsCollector.java
@@ -2,6 +2,7 @@
 
 import io.micrometer.core.instrument.Gauge;
 import io.micrometer.core.instrument.MeterRegistry;
+import io.netty.util.internal.StringUtil;
 import io.vertx.core.AsyncResult;
 import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
@@ -27,15 +28,23 @@ public class PeriodicMetricsCollector {
     private final Vertx vertx;
     private final String redisquesAddress;
 
+    private static final String DEFAULT_IDENTIFIER = "default";
+
     private final AtomicLong activeQueuesCount = new AtomicLong(0);
 
-    public PeriodicMetricsCollector(Vertx vertx, PeriodicSkipScheduler periodicSkipScheduler, String redisquesAddress, MeterRegistry meterRegistry, long metricCollectIntervalSec) {
+    public PeriodicMetricsCollector(Vertx vertx, PeriodicSkipScheduler periodicSkipScheduler, String redisquesAddress,
+                                    String identifier, MeterRegistry meterRegistry, long metricCollectIntervalSec) {
         this.vertx = vertx;
         this.redisquesAddress = redisquesAddress;
 
-        Gauge.builder(MetricMeter.ACTIVE_QUEUES.getId(), activeQueuesCount, AtomicLong::get).
-                description(MetricMeter.ACTIVE_QUEUES.getDescription()).
-                register(meterRegistry);
+        String id = identifier;
+
+        if(StringUtil.isNullOrEmpty(id)) {
+            id = DEFAULT_IDENTIFIER;
+        }
+
+        Gauge.builder(MetricMeter.ACTIVE_QUEUES.getId(), activeQueuesCount, AtomicLong::get).tag("identifier", id)
+                .description(MetricMeter.ACTIVE_QUEUES.getDescription()).register(meterRegistry);
 
         periodicSkipScheduler.setPeriodic(metricCollectIntervalSec * 1000, "metricCollectRefresh",
                 this::updateActiveQueuesCount);

diff --git a/src/main/java/org/swisspush/redisques/util/RedisquesConfiguration.java b/src/main/java/org/swisspush/redisques/util/RedisquesConfiguration.java
@@ -40,6 +40,7 @@ public class RedisquesConfiguration {
     private final boolean httpRequestHandlerEnabled;
     private final boolean httpRequestHandlerAuthenticationEnabled;
     private final boolean micrometerMetricsEnabled;
+    private final String micrometerMetricsIdentifier;
     private final String httpRequestHandlerPrefix;
     private final String httpRequestHandlerUsername;
     private final String httpRequestHandlerPassword;
@@ -114,6 +115,7 @@ public class RedisquesConfiguration {
     public static final String PROP_HTTP_REQUEST_HANDLER_ENABLED = "httpRequestHandlerEnabled";
     public static final String PROP_HTTP_REQUEST_HANDLER_AUTH_ENABLED = "httpRequestHandlerAuthenticationEnabled";
     public static final String PROP_MICROMETER_METRICS_ENABLED = "micrometerMetricsEnabled";
+    public static final String PROP_MICROMETER_METRICS_IDENTIFIER = "micrometerMetricsIdentifier";
     public static final String PROP_HTTP_REQUEST_HANDLER_PREFIX = "httpRequestHandlerPrefix";
     public static final String PROP_HTTP_REQUEST_HANDLER_USERNAME = "httpRequestHandlerUsername";
     public static final String PROP_HTTP_REQUEST_HANDLER_PASSWORD = "httpRequestHandlerPassword";
@@ -148,15 +150,16 @@ public RedisquesConfiguration(String address, String configurationUpdatedAddress
                                   String publishMetricsAddress, String metricStorageName, int metricRefreshPeriod, int refreshPeriod,
                                   String redisHost, int redisPort, String redisAuth, int checkInterval,
                                   int processorTimeout, long processorDelayMax, boolean httpRequestHandlerEnabled,
-                                  boolean httpRequestHandlerAuthenticationEnabled, boolean micrometerMetricsEnabled, String httpRequestHandlerPrefix,
+                                  boolean httpRequestHandlerAuthenticationEnabled, boolean micrometerMetricsEnabled,
+                                  String micrometerMetricsIdentifier, String httpRequestHandlerPrefix,
                                   String httpRequestHandlerUsername, String httpRequestHandlerPassword,
                                   Integer httpRequestHandlerPort, String httpRequestHandlerUserHeader,
                                   List<QueueConfiguration> queueConfigurations, boolean enableQueueNameDecoding) {
         this(address, configurationUpdatedAddress, redisPrefix, processorAddress, publishMetricsAddress, metricStorageName,
                 metricRefreshPeriod, refreshPeriod, DEFAULT_CONSUMER_LOCK_MULTIPLIER, Collections.singletonList(redisHost), Collections.singletonList(redisPort),
                 RedisClientType.STANDALONE, redisAuth, null, null, false, checkInterval,
                 processorTimeout, processorDelayMax, httpRequestHandlerEnabled,
-                httpRequestHandlerAuthenticationEnabled, micrometerMetricsEnabled, httpRequestHandlerPrefix, httpRequestHandlerUsername,
+                httpRequestHandlerAuthenticationEnabled, micrometerMetricsEnabled, micrometerMetricsIdentifier, httpRequestHandlerPrefix, httpRequestHandlerUsername,
                 httpRequestHandlerPassword, httpRequestHandlerPort, httpRequestHandlerUserHeader, queueConfigurations,
                 enableQueueNameDecoding, DEFAULT_REDIS_MAX_POOL_SIZE, DEFAULT_REDIS_MAX_POOL_WAIT_SIZE,
                 DEFAULT_REDIS_MAX_PIPELINE_WAIT_SIZE, DEFAULT_QUEUE_SPEED_INTERVAL_SEC, DEFAULT_MEMORY_USAGE_LIMIT_PCT,
@@ -173,15 +176,16 @@ public RedisquesConfiguration(String address, String configurationUpdatedAddress
                                   String publishMetricsAddress, String metricStorageName, int metricRefreshPeriod, int refreshPeriod,
                                   String redisHost, int redisPort, String redisPassword, String redisUser, boolean redisEnableTls,
                                   int checkInterval, int processorTimeout, long processorDelayMax, boolean httpRequestHandlerEnabled,
-                                  boolean httpRequestHandlerAuthenticationEnabled, boolean micrometerMetricsEnabled, String httpRequestHandlerPrefix,
+                                  boolean httpRequestHandlerAuthenticationEnabled, boolean micrometerMetricsEnabled,
+                                  String micrometerMetricsIdentifier, String httpRequestHandlerPrefix,
                                   String httpRequestHandlerUsername, String httpRequestHandlerPassword,
                                   Integer httpRequestHandlerPort, String httpRequestHandlerUserHeader,
                                   List<QueueConfiguration> queueConfigurations, boolean enableQueueNameDecoding) {
         this(address, configurationUpdatedAddress, redisPrefix, processorAddress, publishMetricsAddress, metricStorageName,
                 metricRefreshPeriod, refreshPeriod, DEFAULT_CONSUMER_LOCK_MULTIPLIER, Collections.singletonList(redisHost), Collections.singletonList(redisPort),
                 RedisClientType.STANDALONE, null, redisPassword, redisUser, redisEnableTls, checkInterval,
                 processorTimeout, processorDelayMax, httpRequestHandlerEnabled,
-                httpRequestHandlerAuthenticationEnabled, micrometerMetricsEnabled, httpRequestHandlerPrefix, httpRequestHandlerUsername,
+                httpRequestHandlerAuthenticationEnabled, micrometerMetricsEnabled, micrometerMetricsIdentifier, httpRequestHandlerPrefix, httpRequestHandlerUsername,
                 httpRequestHandlerPassword, httpRequestHandlerPort, httpRequestHandlerUserHeader, queueConfigurations,
                 enableQueueNameDecoding, DEFAULT_REDIS_MAX_POOL_SIZE, DEFAULT_REDIS_MAX_POOL_WAIT_SIZE,
                 DEFAULT_REDIS_MAX_PIPELINE_WAIT_SIZE, DEFAULT_QUEUE_SPEED_INTERVAL_SEC, DEFAULT_MEMORY_USAGE_LIMIT_PCT,
@@ -198,14 +202,15 @@ public RedisquesConfiguration(String address, String configurationUpdatedAddress
                                   String redisHost, int redisPort, RedisClientType redisClientType, String redisPassword,
                                   String redisUser, boolean redisEnableTls, int checkInterval,
                                   int processorTimeout, long processorDelayMax, boolean httpRequestHandlerEnabled,
-                                  boolean httpRequestHandlerAuthenticationEnabled, boolean micrometerMetricsEnabled, String httpRequestHandlerPrefix,
+                                  boolean httpRequestHandlerAuthenticationEnabled, boolean micrometerMetricsEnabled,
+                                  String micrometerMetricsIdentifier, String httpRequestHandlerPrefix,
                                   String httpRequestHandlerUsername, String httpRequestHandlerPassword,
                                   Integer httpRequestHandlerPort, String httpRequestHandlerUserHeader,
                                   List<QueueConfiguration> queueConfigurations, boolean enableQueueNameDecoding) {
         this(address, configurationUpdatedAddress, redisPrefix, processorAddress, publishMetricsAddress, metricStorageName,
                 metricRefreshPeriod, refreshPeriod, DEFAULT_CONSUMER_LOCK_MULTIPLIER, Collections.singletonList(redisHost), Collections.singletonList(redisPort),
                 redisClientType, null, redisPassword, redisUser, redisEnableTls, checkInterval, processorTimeout,
-                processorDelayMax, httpRequestHandlerEnabled, httpRequestHandlerAuthenticationEnabled, micrometerMetricsEnabled, httpRequestHandlerPrefix, httpRequestHandlerUsername,
+                processorDelayMax, httpRequestHandlerEnabled, httpRequestHandlerAuthenticationEnabled, micrometerMetricsEnabled, micrometerMetricsIdentifier, httpRequestHandlerPrefix, httpRequestHandlerUsername,
                 httpRequestHandlerPassword, httpRequestHandlerPort, httpRequestHandlerUserHeader, queueConfigurations,
                 enableQueueNameDecoding, DEFAULT_REDIS_MAX_POOL_SIZE, DEFAULT_REDIS_MAX_POOL_WAIT_SIZE,
                 DEFAULT_REDIS_MAX_PIPELINE_WAIT_SIZE, DEFAULT_QUEUE_SPEED_INTERVAL_SEC, DEFAULT_MEMORY_USAGE_LIMIT_PCT,
@@ -222,15 +227,16 @@ public RedisquesConfiguration(String address, String configurationUpdatedAddress
                                   int consumerLockMultiplier, String redisHost, int redisPort, RedisClientType redisClientType, String redisPassword,
                                   String redisUser, boolean redisEnableTls, int checkInterval,
                                   int processorTimeout, long processorDelayMax, boolean httpRequestHandlerEnabled,
-                                  boolean httpRequestHandlerAuthenticationEnabled, boolean micrometerMetricsEnabled, String httpRequestHandlerPrefix,
+                                  boolean httpRequestHandlerAuthenticationEnabled, boolean micrometerMetricsEnabled,
+                                  String micrometerMetricsIdentifier, String httpRequestHandlerPrefix,
                                   String httpRequestHandlerUsername, String httpRequestHandlerPassword,
                                   Integer httpRequestHandlerPort, String httpRequestHandlerUserHeader,
                                   List<QueueConfiguration> queueConfigurations, boolean enableQueueNameDecoding) {
         this(address, configurationUpdatedAddress, redisPrefix, processorAddress, publishMetricsAddress, metricStorageName,
                 metricRefreshPeriod, refreshPeriod, consumerLockMultiplier, Collections.singletonList(redisHost), Collections.singletonList(redisPort),
                 redisClientType, null, redisPassword, redisUser, redisEnableTls, checkInterval, processorTimeout,
                 processorDelayMax, httpRequestHandlerEnabled,
-                httpRequestHandlerAuthenticationEnabled, micrometerMetricsEnabled, httpRequestHandlerPrefix, httpRequestHandlerUsername,
+                httpRequestHandlerAuthenticationEnabled, micrometerMetricsEnabled, micrometerMetricsIdentifier, httpRequestHandlerPrefix, httpRequestHandlerUsername,
                 httpRequestHandlerPassword, httpRequestHandlerPort, httpRequestHandlerUserHeader, queueConfigurations,
                 enableQueueNameDecoding, DEFAULT_REDIS_MAX_POOL_SIZE, DEFAULT_REDIS_MAX_POOL_WAIT_SIZE,
                 DEFAULT_REDIS_MAX_PIPELINE_WAIT_SIZE, DEFAULT_QUEUE_SPEED_INTERVAL_SEC, DEFAULT_MEMORY_USAGE_LIMIT_PCT,
@@ -244,7 +250,7 @@ private RedisquesConfiguration(String address, String configurationUpdatedAddres
                                    int consumerLockMultiplier, List<String> redisHosts, List<Integer> redisPorts, RedisClientType redisClientType,
                                    String redisAuth, String redisPassword, String redisUser, boolean redisEnableTls, int checkInterval,
                                    int processorTimeout, long processorDelayMax, boolean httpRequestHandlerEnabled,
-                                   boolean httpRequestHandlerAuthenticationEnabled, boolean micrometerMetricsEnabled,
+                                   boolean httpRequestHandlerAuthenticationEnabled, boolean micrometerMetricsEnabled, String micrometerMetricsIdentifier,
                                    String httpRequestHandlerPrefix, String httpRequestHandlerUsername, String httpRequestHandlerPassword,
                                    Integer httpRequestHandlerPort, String httpRequestHandlerUserHeader,
                                    List<QueueConfiguration> queueConfigurations, boolean enableQueueNameDecoding,
@@ -313,6 +319,7 @@ 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;
@@ -361,7 +368,7 @@ private RedisquesConfiguration(RedisquesConfigurationBuilder builder) {
                 builder.redisPassword, builder.redisUser, builder.redisEnableTls, builder.checkInterval,
                 builder.processorTimeout, builder.processorDelayMax, builder.httpRequestHandlerEnabled,
                 builder.httpRequestHandlerAuthenticationEnabled,
-                builder.micrometerMetricsEnabled, builder.httpRequestHandlerPrefix,
+                builder.micrometerMetricsEnabled, builder.micrometerMetricsIdentifier, builder.httpRequestHandlerPrefix,
                 builder.httpRequestHandlerUsername, builder.httpRequestHandlerPassword, builder.httpRequestHandlerPort,
                 builder.httpRequestHandlerUserHeader, builder.queueConfigurations,
                 builder.enableQueueNameDecoding,
@@ -405,6 +412,7 @@ public JsonObject asJsonObject() {
         obj.put(PROP_HTTP_REQUEST_HANDLER_ENABLED, getHttpRequestHandlerEnabled());
         obj.put(PROP_HTTP_REQUEST_HANDLER_AUTH_ENABLED, getHttpRequestHandlerAuthenticationEnabled());
         obj.put(PROP_MICROMETER_METRICS_ENABLED, getMicrometerMetricsEnabled());
+        obj.put(PROP_MICROMETER_METRICS_IDENTIFIER, getMicrometerMetricsIdentifier());
         obj.put(PROP_HTTP_REQUEST_HANDLER_PREFIX, getHttpRequestHandlerPrefix());
         obj.put(PROP_HTTP_REQUEST_HANDLER_USERNAME, getHttpRequestHandlerUsername());
         obj.put(PROP_HTTP_REQUEST_HANDLER_PASSWORD, getHttpRequestHandlerPassword());
@@ -506,6 +514,9 @@ public static RedisquesConfiguration fromJsonObject(JsonObject json) {
         if (json.containsKey(PROP_MICROMETER_METRICS_ENABLED)) {
             builder.micrometerMetricsEnabled(json.getBoolean(PROP_MICROMETER_METRICS_ENABLED));
         }
+        if (json.containsKey(PROP_MICROMETER_METRICS_IDENTIFIER)) {
+            builder.micrometerMetricsIdentifier(json.getString(PROP_MICROMETER_METRICS_IDENTIFIER));
+        }
         if (json.containsKey(PROP_HTTP_REQUEST_HANDLER_PREFIX)) {
             builder.httpRequestHandlerPrefix(json.getString(PROP_HTTP_REQUEST_HANDLER_PREFIX));
         }
@@ -665,6 +676,10 @@ public boolean getHttpRequestHandlerAuthenticationEnabled() {
 
     public boolean getMicrometerMetricsEnabled() { return micrometerMetricsEnabled; }
 
+    public String getMicrometerMetricsIdentifier() {
+        return micrometerMetricsIdentifier;
+    }
+
     public String getHttpRequestHandlerPrefix() {
         return httpRequestHandlerPrefix;
     }
@@ -790,6 +805,7 @@ public static class RedisquesConfigurationBuilder {
         private int processorTimeout;
         private long processorDelayMax;
         private boolean micrometerMetricsEnabled;
+        private String micrometerMetricsIdentifier;
         private boolean httpRequestHandlerEnabled;
         private boolean httpRequestHandlerAuthenticationEnabled;
         private String httpRequestHandlerPrefix;
@@ -976,6 +992,11 @@ public RedisquesConfigurationBuilder micrometerMetricsEnabled(boolean micrometer
             return this;
         }
 
+        public RedisquesConfigurationBuilder micrometerMetricsIdentifier(String micrometerMetricsIdentifier) {
+            this.micrometerMetricsIdentifier = micrometerMetricsIdentifier;
+            return this;
+        }
+
         public RedisquesConfigurationBuilder httpRequestHandlerAuthenticationEnabled(boolean httpRequestHandlerAuthenticationEnabled) {
             this.httpRequestHandlerAuthenticationEnabled = httpRequestHandlerAuthenticationEnabled;
             return this;