[collect]Add redis cluster auto-discovery display metrics information

collector/src/main/java/com/usthe/collector/collect/redis/RedisClusterCollectImpl.java

@@ -7,14 +7,21 @@
 import com.usthe.common.entity.job.protocol.RedisProtocol;
 import com.usthe.common.entity.message.CollectRep;
 import com.usthe.common.util.CommonConstants;
+import io.lettuce.core.RedisURI;
 import io.lettuce.core.cluster.RedisClusterClient;
 import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
+import io.lettuce.core.cluster.models.partitions.Partitions;
+import io.lettuce.core.cluster.models.partitions.RedisClusterNode;
 import io.lettuce.core.resource.ClientResources;
 import io.lettuce.core.resource.DefaultClientResources;
 import lombok.extern.slf4j.Slf4j;
 
-import java.util.Map;
-import java.util.Objects;
+import java.net.URI;
+import java.util.*;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import static com.usthe.common.util.SignConstants.DOUBLE_MARK;
 
 /**
  * @description: Redis 集群指标收集器
@@ -26,29 +33,63 @@ public class RedisClusterCollectImpl extends RedisCommonCollectImpl {
 
     private static final String CLUSTER_INFO = "cluster";
 
+    private static final String UNIQUE_IDENTITY = "identity";
+
     private final ClientResources defaultClientResources;
 
     public RedisClusterCollectImpl() {
         defaultClientResources = DefaultClientResources.create();
     }
 
 
-    public Map<String, String> getRedisInfo(Metrics metrics) {
-        StatefulRedisClusterConnection<String, String> connection = getConnection(metrics.getRedis());
-        String info = connection.sync().info(metrics.getName());
-        Map<String, String> valueMap = parseInfo(info);
-        if (Objects.equals(metrics.getName(), CLUSTER_INFO)) {
-            String clusterNodes = connection.sync().clusterInfo();
-            valueMap.putAll(parseInfo(clusterNodes));
-        }
-        if (log.isDebugEnabled()) {
-            log.debug("[RedisSingleCollectImpl] fetch redis info");
-            valueMap.forEach((k, v) -> log.debug("{} : {}", k, v));
+    public List<Map<String, String>> getRedisInfo(Metrics metrics) {
+        Map<String, StatefulRedisClusterConnection<String, String>> connectionMap = getConnectionList(metrics.getRedis());
+        List<Map<String, String>> list = new ArrayList<>(connectionMap.size());
+        connectionMap.forEach((identity, connection) ->{
+            String info = connection.sync().info(metrics.getName());
+            Map<String, String> valueMap = parseInfo(info);
+            valueMap.put(UNIQUE_IDENTITY, identity);
+            if (Objects.equals(metrics.getName(), CLUSTER_INFO)) {
+                String clusterNodes = connection.sync().clusterInfo();
+                valueMap.putAll(parseInfo(clusterNodes));
+            }
+            if (log.isDebugEnabled()) {
+                log.debug("[RedisSingleCollectImpl] fetch redis info");
+                valueMap.forEach((k, v) -> log.debug("{} : {}", k, v));
+            }
+            list.add(valueMap);
+        });
+        return list;
+    }
+
+
+
+    private Map<String, StatefulRedisClusterConnection<String, String>> getConnectionList(RedisProtocol redisProtocol) {
+
+        // first connection
+        StatefulRedisClusterConnection<String, String> connection = getConnection(redisProtocol);
+        Partitions partitions = connection.getPartitions();
+        Map<String, StatefulRedisClusterConnection<String, String>> clusterConnectionMap = new HashMap<>(partitions.size());
+        for (RedisClusterNode partition : partitions) {
+            RedisURI uri = partition.getUri();
+            StatefulRedisClusterConnection<String, String> clusterConnection = getConnection(uri, redisProtocol);
+            clusterConnectionMap.put(doUri(uri.getHost(), uri.getPort()), clusterConnection);
         }
-        return valueMap;
+        return clusterConnectionMap;
+    }
+
+
+
+
+
+    private StatefulRedisClusterConnection<String, String> getConnection(RedisURI uri, RedisProtocol redisProtocol) {
+        redisProtocol.setHost(uri.getHost());
+        redisProtocol.setPort(String.valueOf(uri.getPort()));
+        return getConnection(redisProtocol);
     }
 
 
+
     /**
      * obtain StatefulRedisClusterConnection
      *
@@ -78,4 +119,14 @@ private StatefulRedisClusterConnection<String, String> getConnection(RedisProtoc
     private RedisClusterClient buildClient(RedisProtocol redisProtocol) {
         return RedisClusterClient.create(defaultClientResources, redisUri(redisProtocol));
     }
+
+    /**
+     * build single identity
+     * @param ip
+     * @param port
+     * @return
+     */
+    private String doUri(String ip, Integer port) {
+        return ip + DOUBLE_MARK + port;
+    }
 }

collector/src/main/java/com/usthe/collector/collect/redis/RedisCommonCollectImpl.java

@@ -21,6 +21,9 @@
 import java.time.Duration;
 import java.util.*;
 
+import static com.usthe.common.util.MapCapUtil.calInitMap;
+import static com.usthe.common.util.SignConstants.*;
+
 /**
  * @description:
  * @author: hdd
@@ -42,15 +45,15 @@ public void collect(CollectRep.MetricsData.Builder builder, long appId, String a
             return;
         }
         try {
-            Map<String, String> redisInfo ;
             if (Objects.nonNull(metrics.getRedis().getPattern()) && Objects.equals(metrics.getRedis().getPattern(), CLUSTER)) {
                 RedisClusterCollectImpl redisClusterCollect = new RedisClusterCollectImpl();
-                redisInfo = redisClusterCollect.getRedisInfo(metrics);
+                List<Map<String, String>> redisInfoList = redisClusterCollect.getRedisInfo(metrics);
+                doMetricsDataList(builder, redisInfoList, metrics);
             } else {
                 RedisSingleCollectImpl redisSingleCollect = new RedisSingleCollectImpl();
-                redisInfo = redisSingleCollect.getRedisInfo(metrics);
+                Map<String, String> redisInfo = redisSingleCollect.getRedisInfo(metrics);
+                doMetricsData(builder, redisInfo, metrics);
             }
-            doMetricsData(builder, redisInfo, metrics);
         } catch (RedisConnectionException connectionException) {
             String errorMsg = CommonUtil.getMessageFromThrowable(connectionException);
             log.info("[redis connection] error: {}", errorMsg);
@@ -77,12 +80,12 @@ public String supportProtocol() {
      * @return parsed redis info
      */
     protected Map<String, String> parseInfo(String info) {
-        String[] lines = info.split("\n");
-        Map<String, String> result = new HashMap<>(128);
+        String[] lines = info.split(LINE_FEED);
+        Map<String, String> result = new HashMap<>(calInitMap(lines.length));
         Arrays.stream(lines)
-                .filter(it -> StringUtils.hasText(it) && !it.startsWith("#") && it.contains(":"))
+                .filter(it -> StringUtils.hasText(it) && !it.startsWith(WELL_NO) && it.contains(DOUBLE_MARK))
                 .map(this::removeCr)
-                .map(r -> r.split(":"))
+                .map(r -> r.split(DOUBLE_MARK))
                 .forEach(it -> {
                     if (it.length > 1) {
                         result.put(it[0], it[1]);
@@ -94,6 +97,7 @@ protected Map<String, String> parseInfo(String info) {
 
     /**
      * structure
+     *
      * @param redisProtocol
      * @return
      */
@@ -113,6 +117,7 @@ protected RedisURI redisUri(RedisProtocol redisProtocol) {
 
     /**
      * build redis cache key
+     *
      * @param redisProtocol
      * @return
      */
@@ -128,6 +133,7 @@ protected CacheIdentifier doIdentifier(RedisProtocol redisProtocol) {
 
     /**
      * get redis connection
+     *
      * @param identifier
      * @return
      */
@@ -154,6 +160,17 @@ protected StatefulConnection<String, String> getStatefulConnection(CacheIdentifi
     /**
      * Build monitoring parameters according to redis info
      * @param builder
+     * @param valueMapList
+     * @param metrics
+     */
+    private void doMetricsDataList(CollectRep.MetricsData.Builder builder, List<Map<String, String>> valueMapList, Metrics metrics) {
+        valueMapList.forEach(e -> doMetricsData(builder, e, metrics));
+    }
+
+    /**
+     * Build monitoring parameters according to redis info
+     *
+     * @param builder
      * @param valueMap
      * @param metrics
      */
@@ -189,7 +206,7 @@ private void preCheck(Metrics metrics) {
 
 
     private String removeCr(String value) {
-        return value.replace("\r", "");
+        return value.replace(CARRIAGE_RETURN, "");
     }
 
 

common/src/main/java/com/usthe/common/util/MapCapUtil.java

@@ -0,0 +1,21 @@
+package com.usthe.common.util;
+
+/**
+ * map initial capacity calculation
+ * @author: hdd
+ * @create: 2023/03/04
+ */
+public class MapCapUtil {
+
+
+    private static final float LOAD_FACTOR = 0.75f;
+
+    /**
+     * Prevent expansion
+     * @param size
+     * @return
+     */
+    public static int calInitMap(int size) {
+        return (int) Math.ceil (size / LOAD_FACTOR);
+    }
+}

common/src/main/java/com/usthe/common/util/SignConstants.java

@@ -0,0 +1,18 @@
+package com.usthe.common.util;
+
+/**
+ * 特殊服务常量
+ * @author: hdd
+ * @create: 2023/03/04
+ */
+public interface SignConstants {
+
+    String DOUBLE_MARK = ":";
+
+    String WELL_NO = "#";
+
+
+    String LINE_FEED = "\n";
+
+    String CARRIAGE_RETURN = "\r";
+}

manager/src/main/resources/define/app-redis_cluster.yml

@@ -66,6 +66,8 @@ metrics:
   - name: server
     priority: 0
     fields:
+      - field: identity
+        type: 1
       - field: redis_version
         type: 1
       - field: redis_git_sha1
@@ -124,6 +126,8 @@ metrics:
   - name: clients
     priority: 1
     fields:
+      - field: identity
+        type: 1
       - field: connected_clients
         type: 0
       - field: cluster_connections
@@ -152,6 +156,8 @@ metrics:
   - name: memory
     priority: 2
     fields:
+      - field: identity
+        type: 1
       - field: used_memory
         type: 0
       - field: used_memory_human
@@ -255,6 +261,8 @@ metrics:
   - name: persistence
     priority: 3
     fields:
+      - field: identity
+        type: 1
       - field: loading
         type: 0
       - field: current_cow_size
@@ -313,6 +321,8 @@ metrics:
   - name: stats
     priority: 4
     fields:
+      - field: identity
+        type: 1
       - field: total_connections_received
         type: 0
       - field: total_commands_processed
@@ -401,6 +411,8 @@ metrics:
   - name: replication
     priority: 5
     fields:
+      - field: identity
+        type: 1
       - field: role
         type: 1
       - field: connected_slaves
@@ -435,6 +447,8 @@ metrics:
   - name: cpu
     priority: 6
     fields:
+      - field: identity
+        type: 1
       - field: used_cpu_sys
         type: 0
       - field: used_cpu_user
@@ -459,6 +473,8 @@ metrics:
   - name: errorstats
     priority: 8
     fields:
+      - field: identity
+        type: 1
       - field: errorstat_ERR
         type: 1
       - field: errorstat_MISCONF
@@ -475,6 +491,8 @@ metrics:
   - name: cluster
     priority: 9
     fields:
+      - field: identity
+        type: 1
       - field: cluster_enabled
         type: 0
       - field: cluster_state
@@ -521,6 +539,8 @@ metrics:
   - name: commandstats
     priority: 9
     fields:
+      - field: identity
+        type: 1
       - field: cmdstat_set
         type: 1
       - field: cmdstat_get
@@ -554,6 +574,8 @@ metrics:
   - name: keyspace
     priority: 9
     fields:
+      - field: identity
+        type: 1
       - field: db0
         type: 1
       - field: db1