mirror of
https://github.com/signalapp/Signal-Server
synced 2026-04-21 10:28:04 +01:00
Use a Redis testcontainer instead of embedded-redis in RedisClusterExtension
This commit is contained in:
Jon Chambers
committed by
Jon Chambers
Jon Chambers
parent
a3356d0188
commit
803e73bd1e
@@ -44,7 +44,7 @@ public class LocalFaultTolerantRedisClusterFactory implements FaultTolerantRedis
|
||||
final RedisClusterConfiguration config = new RedisClusterConfiguration();
|
||||
config.setConfigurationUri(RedisClusterExtension.getRedisURIs().getFirst().toString());
|
||||
|
||||
return new FaultTolerantRedisClusterClient(name, config, clientResourcesBuilder);
|
||||
return new FaultTolerantRedisClusterClient(name, config, clientResourcesBuilder.socketAddressResolver(redisClusterExtension.getSocketAddressResolver()));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -91,7 +91,8 @@ class FaultTolerantRedisClusterClientTest {
|
||||
@Nullable final CircuitBreakerConfiguration circuitBreakerConfiguration,
|
||||
final ClientResources.Builder clientResourcesBuilder) {
|
||||
|
||||
return new FaultTolerantRedisClusterClient("test", clientResourcesBuilder,
|
||||
return new FaultTolerantRedisClusterClient("test",
|
||||
clientResourcesBuilder.socketAddressResolver(REDIS_CLUSTER_EXTENSION.getSocketAddressResolver()),
|
||||
RedisClusterExtension.getRedisURIs(), TIMEOUT,
|
||||
Optional.ofNullable(circuitBreakerConfiguration).orElseGet(CircuitBreakerConfiguration::new),
|
||||
RETRY_CONFIGURATION);
|
||||
@@ -338,11 +339,13 @@ class FaultTolerantRedisClusterClientTest {
|
||||
}
|
||||
|
||||
void openBreaker(final RedisURI redisURI) {
|
||||
urisToChannelBreakers.get(redisURI).forEach(handler -> handler.breaker.transitionToOpenState());
|
||||
urisToChannelBreakers.get(REDIS_CLUSTER_EXTENSION.getExposedRedisURI(redisURI))
|
||||
.forEach(handler -> handler.breaker.transitionToOpenState());
|
||||
}
|
||||
|
||||
void closeBreaker(final RedisURI redisURI) {
|
||||
urisToChannelBreakers.get(redisURI).forEach(handler -> handler.breaker.transitionToClosedState());
|
||||
urisToChannelBreakers.get(REDIS_CLUSTER_EXTENSION.getExposedRedisURI(redisURI))
|
||||
.forEach(handler -> handler.breaker.transitionToClosedState());
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -5,45 +5,53 @@
|
||||
|
||||
package org.whispersystems.textsecuregcm.redis;
|
||||
|
||||
import static org.junit.jupiter.api.Assumptions.assumeFalse;
|
||||
|
||||
import io.github.resilience4j.circuitbreaker.CallNotPermittedException;
|
||||
import io.lettuce.core.FlushMode;
|
||||
import io.lettuce.core.RedisClient;
|
||||
import io.lettuce.core.RedisException;
|
||||
import io.lettuce.core.RedisURI;
|
||||
import io.lettuce.core.api.StatefulRedisConnection;
|
||||
import io.lettuce.core.api.sync.RedisCommands;
|
||||
import io.lettuce.core.cluster.SlotHash;
|
||||
import io.lettuce.core.internal.HostAndPort;
|
||||
import io.lettuce.core.resource.ClientResources;
|
||||
import io.lettuce.core.resource.DnsResolvers;
|
||||
import io.lettuce.core.resource.MappingSocketAddressResolver;
|
||||
import io.lettuce.core.resource.SocketAddressResolver;
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.net.ServerSocket;
|
||||
import java.net.URL;
|
||||
import java.nio.file.Paths;
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.stream.Collectors;
|
||||
import org.junit.jupiter.api.extension.AfterEachCallback;
|
||||
import org.junit.jupiter.api.extension.BeforeAllCallback;
|
||||
import org.junit.jupiter.api.extension.BeforeEachCallback;
|
||||
import org.junit.jupiter.api.extension.ExtensionContext;
|
||||
import org.testcontainers.containers.ComposeContainer;
|
||||
import org.testcontainers.containers.wait.strategy.Wait;
|
||||
import org.testcontainers.containers.wait.strategy.WaitStrategy;
|
||||
import org.whispersystems.textsecuregcm.configuration.CircuitBreakerConfiguration;
|
||||
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, AfterEachCallback,
|
||||
ExtensionContext.Store.CloseableResource {
|
||||
|
||||
private static final Duration DEFAULT_TIMEOUT = Duration.ofSeconds(2);
|
||||
private static final int NODE_COUNT = 2;
|
||||
|
||||
private static final RedisServer[] CLUSTER_NODES = new RedisServer[NODE_COUNT];
|
||||
private static ComposeContainer composeContainer;
|
||||
private static ClientResources redisClientResources;
|
||||
private static Map<HostAndPort, HostAndPort> exposedAddressesByInternalAddress;
|
||||
private static List<RedisURI> redisUris;
|
||||
|
||||
private final Duration timeout;
|
||||
private final RetryConfiguration retryConfiguration;
|
||||
private FaultTolerantRedisClusterClient redisCluster;
|
||||
|
||||
private FaultTolerantRedisClusterClient redisClusterClient;
|
||||
|
||||
private static final Duration DEFAULT_TIMEOUT = Duration.ofSeconds(2);
|
||||
|
||||
private static final int REDIS_PORT = 6379;
|
||||
private static final WaitStrategy WAIT_STRATEGY = Wait.forListeningPort().withStartupTimeout(Duration.ofMinutes(1));
|
||||
private static final Duration CLUSTER_UP_DEADLINE = Duration.ofSeconds(15);
|
||||
|
||||
private static final String[] REDIS_SERVICE_NAMES = new String[] { "redis-0-1", "redis-1-1", "redis-2-1" };
|
||||
|
||||
public RedisClusterExtension(final Duration timeout, final RetryConfiguration retryConfiguration) {
|
||||
this.timeout = timeout;
|
||||
@@ -59,34 +67,88 @@ public class RedisClusterExtension implements BeforeAllCallback, BeforeEachCallb
|
||||
public void close() throws Throwable {
|
||||
if (redisClientResources != null) {
|
||||
redisClientResources.shutdown().get();
|
||||
|
||||
for (final RedisServer node : CLUSTER_NODES) {
|
||||
node.stop();
|
||||
}
|
||||
composeContainer.stop();
|
||||
}
|
||||
|
||||
redisClientResources = null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void afterEach(final ExtensionContext context) throws Exception {
|
||||
redisCluster.shutdown();
|
||||
public void afterEach(final ExtensionContext context) {
|
||||
redisClusterClient.shutdown();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void beforeAll(final ExtensionContext context) throws Exception {
|
||||
assumeFalse(System.getProperty("os.name").equalsIgnoreCase("windows"));
|
||||
|
||||
if (redisClientResources == null) {
|
||||
redisClientResources = ClientResources.builder().build();
|
||||
final URL clusterComposeFileUrl = getClass().getResource("redis-cluster.yml");
|
||||
|
||||
for (int i = 0; i < NODE_COUNT; i++) {
|
||||
// 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);
|
||||
if (clusterComposeFileUrl == null) {
|
||||
throw new IllegalStateException("Cluster compose file not found");
|
||||
}
|
||||
|
||||
assembleCluster(CLUSTER_NODES);
|
||||
final File clusterComposeFile = Paths.get(clusterComposeFileUrl.toURI()).toFile();
|
||||
|
||||
composeContainer = new ComposeContainer(clusterComposeFile);
|
||||
|
||||
for (final String serviceName : REDIS_SERVICE_NAMES) {
|
||||
composeContainer = composeContainer.withExposedService(serviceName, REDIS_PORT, WAIT_STRATEGY);
|
||||
}
|
||||
|
||||
composeContainer.start();
|
||||
|
||||
exposedAddressesByInternalAddress = Arrays.stream(REDIS_SERVICE_NAMES)
|
||||
.collect(Collectors.toMap(serviceName -> {
|
||||
final String internalIp = composeContainer.getContainerByServiceName(serviceName).orElseThrow()
|
||||
.getContainerInfo()
|
||||
.getNetworkSettings()
|
||||
.getNetworks().values().stream().findFirst().orElseThrow()
|
||||
.getIpAddress();
|
||||
|
||||
if (internalIp == null) {
|
||||
throw new IllegalStateException("Could not determine internal IP address of service container: " + serviceName);
|
||||
}
|
||||
|
||||
return HostAndPort.of(internalIp, REDIS_PORT);
|
||||
},
|
||||
serviceName -> HostAndPort.of(
|
||||
composeContainer.getServiceHost(serviceName, REDIS_PORT),
|
||||
composeContainer.getServicePort(serviceName, REDIS_PORT))));
|
||||
|
||||
redisUris = Arrays.stream(REDIS_SERVICE_NAMES)
|
||||
.map(serviceName -> RedisURI.create(
|
||||
composeContainer.getServiceHost("redis-0-1", REDIS_PORT),
|
||||
composeContainer.getServicePort("redis-0-1", REDIS_PORT)))
|
||||
.toList();
|
||||
|
||||
redisClientResources = ClientResources.builder()
|
||||
.socketAddressResolver(getSocketAddressResolver())
|
||||
.build();
|
||||
|
||||
// Wait for the cluster to be fully up; just having the containers running isn't enough since they still need to do
|
||||
// some post-launch cluster setup work.
|
||||
boolean allNodesUp;
|
||||
final Instant deadline = Instant.now().plus(CLUSTER_UP_DEADLINE);
|
||||
|
||||
do {
|
||||
allNodesUp = redisUris.stream()
|
||||
.allMatch(redisUri -> {
|
||||
try (final RedisClient redisClient = RedisClient.create(redisClientResources, redisUri)) {
|
||||
final String clusterInfo = redisClient.connect().sync().clusterInfo();
|
||||
return clusterInfo.contains("cluster_state:ok") && clusterInfo.contains("cluster_slots_ok:16384");
|
||||
} catch (final Exception e) {
|
||||
return false;
|
||||
}
|
||||
});
|
||||
|
||||
if (Instant.now().isAfter(deadline)) {
|
||||
throw new RuntimeException("Cluster did not start before deadline");
|
||||
}
|
||||
|
||||
if (!allNodesUp) {
|
||||
Thread.sleep(100);
|
||||
}
|
||||
} while (!allNodesUp);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -94,142 +156,34 @@ public class RedisClusterExtension implements BeforeAllCallback, BeforeEachCallb
|
||||
public void beforeEach(final ExtensionContext context) throws Exception {
|
||||
final CircuitBreakerConfiguration circuitBreakerConfig = new CircuitBreakerConfiguration();
|
||||
circuitBreakerConfig.setWaitDurationInOpenState(Duration.ofMillis(500));
|
||||
redisCluster = new FaultTolerantRedisClusterClient("test-cluster",
|
||||
redisClusterClient = new FaultTolerantRedisClusterClient("test-cluster",
|
||||
redisClientResources.mutate(),
|
||||
getRedisURIs(),
|
||||
timeout,
|
||||
circuitBreakerConfig,
|
||||
retryConfiguration);
|
||||
|
||||
redisCluster.useCluster(connection -> {
|
||||
boolean setAll = false;
|
||||
|
||||
final String[] keys = new String[NODE_COUNT];
|
||||
|
||||
for (int i = 0; i < keys.length; i++) {
|
||||
keys[i] = RedisClusterUtil.getMinimalHashTag(i * SlotHash.SLOT_COUNT / keys.length);
|
||||
}
|
||||
|
||||
while (!setAll) {
|
||||
try {
|
||||
for (final String key : keys) {
|
||||
connection.sync().set(key, "warmup");
|
||||
}
|
||||
|
||||
setAll = true;
|
||||
} catch (final RedisException | CallNotPermittedException ignored) {
|
||||
// Cluster isn't ready; wait and retry.
|
||||
try {
|
||||
Thread.sleep(500);
|
||||
} catch (final InterruptedException ignored2) {
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
redisCluster.useCluster(connection -> connection.sync().flushall(FlushMode.SYNC));
|
||||
redisClusterClient.useCluster(connection -> connection.sync().flushall(FlushMode.SYNC));
|
||||
}
|
||||
|
||||
public static List<RedisURI> getRedisURIs() {
|
||||
return Arrays.stream(CLUSTER_NODES)
|
||||
.map(node -> "redis://127.0.0.1:%d".formatted(node.ports().getFirst()))
|
||||
.map(RedisURI::create)
|
||||
.toList();
|
||||
return redisUris;
|
||||
}
|
||||
|
||||
public RedisURI getExposedRedisURI(final RedisURI internalRedisURI) {
|
||||
final HostAndPort internalHostAndPort = HostAndPort.of(internalRedisURI.getHost(), internalRedisURI.getPort());
|
||||
final HostAndPort exposedHostAndPort = exposedAddressesByInternalAddress.getOrDefault(internalHostAndPort, internalHostAndPort);
|
||||
|
||||
return RedisURI.create(exposedHostAndPort.getHostText(), exposedHostAndPort.getPort());
|
||||
}
|
||||
|
||||
public SocketAddressResolver getSocketAddressResolver() {
|
||||
return MappingSocketAddressResolver.create(DnsResolvers.UNRESOLVED,
|
||||
hostAndPort -> exposedAddressesByInternalAddress.getOrDefault(hostAndPort, hostAndPort));
|
||||
}
|
||||
|
||||
public FaultTolerantRedisClusterClient getRedisCluster() {
|
||||
return redisCluster;
|
||||
}
|
||||
|
||||
private static RedisServer buildClusterNode(final int port) throws IOException {
|
||||
final File clusterConfigFile = File.createTempFile("redis", ".conf");
|
||||
clusterConfigFile.deleteOnExit();
|
||||
|
||||
return RedisServer.builder()
|
||||
.setting("cluster-enabled yes")
|
||||
.setting("cluster-config-file " + clusterConfigFile.getAbsolutePath())
|
||||
.setting("cluster-node-timeout 5000")
|
||||
.setting("appendonly no")
|
||||
.setting("save \"\"")
|
||||
.setting("dir " + System.getProperty("java.io.tmpdir"))
|
||||
.port(port)
|
||||
.build();
|
||||
}
|
||||
|
||||
private static void assembleCluster(final RedisServer... nodes) throws InterruptedException {
|
||||
try (final RedisClient meetClient = RedisClient.create(RedisURI.create("127.0.0.1", nodes[0].ports().getFirst()))) {
|
||||
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().getFirst());
|
||||
}
|
||||
}
|
||||
|
||||
final int slotsPerNode = SlotHash.SLOT_COUNT / nodes.length;
|
||||
|
||||
for (int i = 0; i < nodes.length; i++) {
|
||||
final int startInclusive = i * slotsPerNode;
|
||||
final int endExclusive = i == nodes.length - 1 ? SlotHash.SLOT_COUNT : (i + 1) * slotsPerNode;
|
||||
|
||||
try (final RedisClient assignSlotClient = RedisClient.create(
|
||||
RedisURI.create("127.0.0.1", nodes[i].ports().getFirst()));
|
||||
final StatefulRedisConnection<String, String> assignSlotConnection = assignSlotClient.connect()) {
|
||||
final int[] slots = new int[endExclusive - startInclusive];
|
||||
|
||||
for (int s = startInclusive; s < endExclusive; s++) {
|
||||
slots[s - startInclusive] = s;
|
||||
}
|
||||
|
||||
assignSlotConnection.sync().clusterAddSlots(slots);
|
||||
}
|
||||
}
|
||||
|
||||
try (final RedisClient waitClient = RedisClient.create(RedisURI.create("127.0.0.1", nodes[0].ports().getFirst()));
|
||||
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 leader nodes are reachable.
|
||||
|
||||
final int sleepMillis = 500;
|
||||
int tries = 0;
|
||||
while (!connection.sync().clusterInfo().contains("cluster_state:ok")) {
|
||||
Thread.sleep(sleepMillis);
|
||||
tries++;
|
||||
|
||||
if (tries == 20) {
|
||||
throw new RuntimeException(
|
||||
"Timeout: Redis not ready after waiting %d milliseconds".formatted(tries * sleepMillis));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public static int getNextRedisClusterPort() throws IOException {
|
||||
final int maxIterations = 11_000;
|
||||
for (int i = 0; i < maxIterations; i++) {
|
||||
try (final ServerSocket socket = new ServerSocket(0)) {
|
||||
final int port = socket.getLocalPort();
|
||||
if (port < 55535) {
|
||||
return port;
|
||||
}
|
||||
}
|
||||
}
|
||||
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);
|
||||
}
|
||||
return redisClusterClient;
|
||||
}
|
||||
|
||||
public static class Builder {
|
||||
|
||||
Reference in New Issue
Block a user