mirror/Signal-Server
0
0
Fork 0
mirror of https://github.com/signalapp/Signal-Server.git synced 2024-09-20 03:52:16 +02:00
Code

fixing embedded redis based tests

Browse Source
This commit is contained in:
Sergey Skrobotov 2023-03-15 08:56:06 -07:00
parent 7c52be2ac1
commit ebf8aa7b15
2 changed files with 44 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
pom.xml
View File

@ -56,7 +56,7 @@
<lettuce.version>6.2.1.RELEASE</lettuce.version>
<libphonenumber.version>8.12.54</libphonenumber.version>
<logstash.logback.version>7.2</logstash.logback.version>
<luajava.version>3.3.0</luajava.version>
<luajava.version>3.4.0</luajava.version>
<micrometer.version>1.10.3</micrometer.version>
<mockito.version>4.11.0</mockito.version>
<netty.version>4.1.82.Final</netty.version>

84
service/src/test/java/org/whispersystems/textsecuregcm/redis/RedisClusterExtension.java
View File

@ -1,5 +1,5 @@
/*
* Copyright 2013-2022 Signal Messenger, LLC
* Copyright 2013 Signal Messenger, LLC
* SPDX-License-Identifier: AGPL-3.0-only
*/
@ -30,17 +30,24 @@ import org.whispersystems.textsecuregcm.configuration.CircuitBreakerConfiguratio
import org.whispersystems.textsecuregcm.configuration.RetryConfiguration;
import org.whispersystems.textsecuregcm.util.RedisClusterUtil;
import redis.embedded.RedisServer;
import redis.embedded.exceptions.EmbeddedRedisException;
public class RedisClusterExtension implements BeforeAllCallback, BeforeEachCallback, AfterAllCallback, AfterEachCallback {
private static final int NODE_COUNT = 2;
private static final RedisServer[] CLUSTER_NODES = new RedisServer[NODE_COUNT];
private FaultTolerantRedisCluster redisCluster;
public static RedisClusterExtensionBuilder builder() {
return new RedisClusterExtensionBuilder();
}
@Override
public void afterAll(final ExtensionContext context) throws Exception {
for (final RedisServer node : clusterNodes) {
for (final RedisServer node : CLUSTER_NODES) {
node.stop();
}
}
@ -54,21 +61,20 @@ public class RedisClusterExtension implements BeforeAllCallback, BeforeEachCallb
public void beforeAll(final ExtensionContext context) throws Exception {
assumeFalse(System.getProperty("os.name").equalsIgnoreCase("windows"));
clusterNodes = new RedisServer[NODE_COUNT];
for (int i = 0; i < NODE_COUNT; i++) {
clusterNodes[i] = buildClusterNode(getNextRedisClusterPort());
clusterNodes[i].start();
// We're occasionally seeing redis server startup failing due to the bind address being already in use.
// To mitigate that, we're going to just retry a couple of times before failing the test.
CLUSTER_NODES[i] = startWithRetries(3);
}
assembleCluster(clusterNodes);
assembleCluster(CLUSTER_NODES);
}
@Override
public void beforeEach(final ExtensionContext context) throws Exception {
final List<String> urls = Arrays.stream(clusterNodes)
final List<String> urls = Arrays.stream(CLUSTER_NODES)
.map(node -> String.format("redis://127.0.0.1:%d", node.ports().get(0)))
.collect(Collectors.toList());
.toList();
redisCluster = new FaultTolerantRedisCluster("test-cluster",
RedisClusterClient.create(urls.stream().map(RedisURI::create).collect(Collectors.toList())),
@ -105,12 +111,6 @@ public class RedisClusterExtension implements BeforeAllCallback, BeforeEachCallb
redisCluster.useCluster(connection -> connection.sync().flushall());
}
private static final int NODE_COUNT = 2;
private static RedisServer[] clusterNodes;
private FaultTolerantRedisCluster redisCluster;
public FaultTolerantRedisCluster getRedisCluster() {
return redisCluster;
}
@ -131,17 +131,13 @@ public class RedisClusterExtension implements BeforeAllCallback, BeforeEachCallb
}
private static void assembleCluster(final RedisServer... nodes) throws InterruptedException {
final RedisClient meetClient = RedisClient.create(RedisURI.create("127.0.0.1", nodes[0].ports().get(0)));
try {
try (final RedisClient meetClient = RedisClient.create(RedisURI.create("127.0.0.1", nodes[0].ports().get(0)))) {
final StatefulRedisConnection<String, String> connection = meetClient.connect();
final RedisCommands<String, String> commands = connection.sync();
for (int i = 1; i < nodes.length; i++) {
commands.clusterMeet("127.0.0.1", nodes[i].ports().get(0));
}
} finally {
meetClient.shutdown();
}
final int slotsPerNode = SlotHash.SLOT_COUNT / nodes.length;
@ -150,9 +146,8 @@ public class RedisClusterExtension implements BeforeAllCallback, BeforeEachCallb
final int startInclusive = i * slotsPerNode;
final int endExclusive = i == nodes.length - 1 ? SlotHash.SLOT_COUNT : (i + 1) * slotsPerNode;
final RedisClient assignSlotClient = RedisClient.create(RedisURI.create("127.0.0.1", nodes[i].ports().get(0)));
try (final StatefulRedisConnection<String, String> assignSlotConnection = assignSlotClient.connect()) {
try (final RedisClient assignSlotClient = RedisClient.create(RedisURI.create("127.0.0.1", nodes[i].ports().get(0)));
final StatefulRedisConnection<String, String> assignSlotConnection = assignSlotClient.connect()) {
final int[] slots = new int[endExclusive - startInclusive];
for (int s = startInclusive; s < endExclusive; s++) {
@ -160,14 +155,11 @@ public class RedisClusterExtension implements BeforeAllCallback, BeforeEachCallb
}
assignSlotConnection.sync().clusterAddSlots(slots);
} finally {
assignSlotClient.shutdown();
}
}
final RedisClient waitClient = RedisClient.create(RedisURI.create("127.0.0.1", nodes[0].ports().get(0)));
try (final StatefulRedisConnection<String, String> connection = waitClient.connect()) {
try (final RedisClient waitClient = RedisClient.create(RedisURI.create("127.0.0.1", nodes[0].ports().get(0)));
final StatefulRedisConnection<String, String> connection = waitClient.connect()) {
// CLUSTER INFO gives us a big blob of key-value pairs, but the one we're interested in is `cluster_state`.
// According to https://redis.io/commands/cluster-info, `cluster_state:ok` means that the node is ready to
// receive queries, all slots are assigned, and a majority of master nodes are reachable.
@ -183,30 +175,40 @@ public class RedisClusterExtension implements BeforeAllCallback, BeforeEachCallb
String.format("Timeout: Redis not ready after waiting %d milliseconds", tries * sleepMillis));
}
}
} finally {
waitClient.shutdown();
}
}
public static int getNextRedisClusterPort() throws IOException {
final int MAX_ITERATIONS = 11_000;
int port;
for (int i = 0; i < MAX_ITERATIONS; i++) {
try (ServerSocket socket = new ServerSocket(0)) {
final int maxIterations = 11_000;
for (int i = 0; i < maxIterations; i++) {
try (final ServerSocket socket = new ServerSocket(0)) {
socket.setReuseAddress(false);
port = socket.getLocalPort();
}
if (port < 55535) {
return port;
final int port = socket.getLocalPort();
if (port < 55535) {
return port;
}
}
}
throw new IOException("Couldn't find an open port below 55,535 in " + MAX_ITERATIONS + " tries");
throw new IOException("Couldn't find an unused open port below 55,535 in " + maxIterations + " tries");
}
private static RedisServer startWithRetries(final int attemptsLeft) throws Exception {
try {
final RedisServer redisServer = buildClusterNode(getNextRedisClusterPort());
redisServer.start();
return redisServer;
} catch (final IOException | EmbeddedRedisException e) {
if (attemptsLeft == 0) {
throw e;
}
Thread.sleep(500);
return startWithRetries(attemptsLeft - 1);
}
}
public static class RedisClusterExtensionBuilder {
private RedisClusterExtensionBuilder() {
}
public RedisClusterExtension build() {