From 625dbe96248770a076366a89527f1377da9f4741 Mon Sep 17 00:00:00 2001 From: P_Peaceful Date: Sun, 30 Nov 2025 19:39:40 +0800 Subject: [PATCH] [bugfix] redis cluster monitor master-slave relationship is inconsistent (#3874) Co-authored-by: Tomsun28 --- .../collect/redis/RedisCommonCollectImpl.java | 21 +++++++--- .../redis/RedisClusterCollectImplTest.java | 42 +++++++++++++------ 2 files changed, 45 insertions(+), 18 deletions(-) diff --git a/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/redis/RedisCommonCollectImpl.java b/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/redis/RedisCommonCollectImpl.java index 90b258545d..ac520cf72b 100644 --- a/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/redis/RedisCommonCollectImpl.java +++ b/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/redis/RedisCommonCollectImpl.java @@ -67,6 +67,8 @@ public class RedisCommonCollectImpl extends AbstractCollect { private static final String CLUSTER = "3"; + private static final String SINGLE = "1"; + private static final String CLUSTER_INFO = "cluster"; private static final String UNIQUE_IDENTITY = "identity"; @@ -140,7 +142,7 @@ public class RedisCommonCollectImpl extends AbstractCollect { * @return data */ private List> getClusterRedisInfo(Metrics metrics) throws GeneralSecurityException, IOException { - Map> connectionMap = getConnectionList(metrics.getRedis()); + Map> connectionMap = getConnectionList(metrics.getRedis()); List> list = new ArrayList<>(connectionMap.size()); connectionMap.forEach((identity, connection) ->{ String info = connection.sync().info(metrics.getName()); @@ -214,16 +216,23 @@ public class RedisCommonCollectImpl extends AbstractCollect { * @param redisProtocol protocol * @return connection map */ - private Map> getConnectionList(RedisProtocol redisProtocol) throws GeneralSecurityException, IOException { + private Map> getConnectionList(RedisProtocol redisProtocol) throws GeneralSecurityException, IOException { // first connection StatefulRedisClusterConnection connection = getClusterConnection(redisProtocol); Partitions partitions = connection.getPartitions(); - Map> clusterConnectionMap = new HashMap<>(partitions.size()); + Map> clusterConnectionMap = new HashMap<>(partitions.size()); for (RedisClusterNode partition : partitions) { RedisURI uri = partition.getUri(); - redisProtocol.setHost(uri.getHost()); - redisProtocol.setPort(String.valueOf(uri.getPort())); - StatefulRedisClusterConnection clusterConnection = getClusterConnection(redisProtocol); + RedisProtocol singleRedisProtocol = RedisProtocol.builder() + .host(uri.getHost()) + .port(String.valueOf(uri.getPort())) + .username(redisProtocol.getUsername()) + .password(redisProtocol.getPassword()) + .pattern(SINGLE) + .timeout(redisProtocol.getTimeout()) + .sshTunnel(redisProtocol.getSshTunnel()) + .build(); + StatefulRedisConnection clusterConnection = getSingleConnection(singleRedisProtocol); clusterConnectionMap.put(doUri(uri.getHost(), uri.getPort()), clusterConnection); } return clusterConnectionMap; diff --git a/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/redis/RedisClusterCollectImplTest.java b/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/redis/RedisClusterCollectImplTest.java index 96aa4db48f..5d74e92f2b 100644 --- a/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/redis/RedisClusterCollectImplTest.java +++ b/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/redis/RedisClusterCollectImplTest.java @@ -19,10 +19,13 @@ package org.apache.hertzbeat.collector.collect.redis; import static org.apache.hertzbeat.common.constants.CommonConstants.TYPE_STRING; import static org.junit.jupiter.api.Assertions.assertEquals; + +import io.lettuce.core.RedisClient; import io.lettuce.core.RedisURI; +import io.lettuce.core.api.StatefulRedisConnection; +import io.lettuce.core.api.sync.RedisCommands; import io.lettuce.core.cluster.RedisClusterClient; import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; -import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands; import io.lettuce.core.cluster.models.partitions.Partitions; import io.lettuce.core.cluster.models.partitions.RedisClusterNode; import io.lettuce.core.resource.ClientResources; @@ -37,6 +40,7 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.mockito.InjectMocks; import org.mockito.Mock; +import org.mockito.MockedStatic; import org.mockito.Mockito; import org.mockito.junit.jupiter.MockitoExtension; @@ -51,13 +55,19 @@ public class RedisClusterCollectImplTest { @Mock - private StatefulRedisClusterConnection connection; + private StatefulRedisClusterConnection clusterConnection; @Mock - private RedisAdvancedClusterCommands cmd; + private StatefulRedisConnection singleConnection; @Mock - private RedisClusterClient client; + private RedisCommands cmd; + + @Mock + private RedisClusterClient clusterClient; + + @Mock + private RedisClient singleClient; @BeforeEach void setUp() { @@ -65,8 +75,10 @@ public class RedisClusterCollectImplTest { @AfterEach void setDown() { - connection.close(); - client.shutdown(); + clusterConnection.close(); + singleConnection.close(); + clusterClient.shutdown(); + singleClient.shutdown(); } @Test @@ -110,10 +122,14 @@ public class RedisClusterCollectImplTest { metrics.setAliasFields(aliasField); metrics.setFields(fields); - - Mockito.mockStatic(RedisClusterClient.class).when(() -> RedisClusterClient.create(Mockito.any(ClientResources.class), - Mockito.any(RedisURI.class))).thenReturn(client); - Mockito.when(client.connect()).thenReturn(connection); + MockedStatic redisClusterClientMockedStatic = Mockito.mockStatic(RedisClusterClient.class); + redisClusterClientMockedStatic.when(() -> RedisClusterClient.create(Mockito.any(ClientResources.class), + Mockito.any(RedisURI.class))).thenReturn(clusterClient); + Mockito.when(clusterClient.connect()).thenReturn(clusterConnection); + MockedStatic redisClientMockedStatic = Mockito.mockStatic(RedisClient.class); + redisClientMockedStatic.when(() -> RedisClient.create(Mockito.any(ClientResources.class), + Mockito.any(RedisURI.class))).thenReturn(singleClient); + Mockito.when(singleClient.connect()).thenReturn(singleConnection); Partitions partitions = new Partitions(); RedisClusterNode node = new RedisClusterNode(); @@ -125,9 +141,9 @@ public class RedisClusterCollectImplTest { node2.setUri(RedisURI.create("redis://" + uri2)); partitions.add(node2); - Mockito.when(connection.getPartitions()).thenReturn(partitions); + Mockito.when(clusterConnection.getPartitions()).thenReturn(partitions); - Mockito.when(connection.sync()).thenReturn(cmd); + Mockito.when(singleConnection.sync()).thenReturn(cmd); Mockito.when(cmd.info(metrics.getName())).thenReturn(info); Mockito.when(cmd.clusterInfo()).thenReturn(clusterInfo); @@ -147,6 +163,8 @@ public class RedisClusterCollectImplTest { assertEquals(row.getColumns(2), uri2); } } + redisClusterClientMockedStatic.close(); + redisClientMockedStatic.close(); } }