diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/configuration/LocalFaultTolerantRedisClusterFactory.java b/service/src/test/java/org/whispersystems/textsecuregcm/configuration/LocalFaultTolerantRedisClusterFactory.java index eec0c86d6..866ee76f5 100644 --- a/service/src/test/java/org/whispersystems/textsecuregcm/configuration/LocalFaultTolerantRedisClusterFactory.java +++ b/service/src/test/java/org/whispersystems/textsecuregcm/configuration/LocalFaultTolerantRedisClusterFactory.java @@ -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())); } } diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/redis/FaultTolerantRedisClusterClientTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/redis/FaultTolerantRedisClusterClientTest.java index dafc93de1..6eaf3cc88 100644 --- a/service/src/test/java/org/whispersystems/textsecuregcm/redis/FaultTolerantRedisClusterClientTest.java +++ b/service/src/test/java/org/whispersystems/textsecuregcm/redis/FaultTolerantRedisClusterClientTest.java @@ -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()); } } diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/redis/RedisClusterExtension.java b/service/src/test/java/org/whispersystems/textsecuregcm/redis/RedisClusterExtension.java index bfdac19a6..264730567 100644 --- a/service/src/test/java/org/whispersystems/textsecuregcm/redis/RedisClusterExtension.java +++ b/service/src/test/java/org/whispersystems/textsecuregcm/redis/RedisClusterExtension.java @@ -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 exposedAddressesByInternalAddress; + private static List 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 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 connection = meetClient.connect(); - final RedisCommands 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 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 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 { diff --git a/service/src/test/resources/org/whispersystems/textsecuregcm/redis/redis-cluster.yml b/service/src/test/resources/org/whispersystems/textsecuregcm/redis/redis-cluster.yml new file mode 100644 index 000000000..f158dba09 --- /dev/null +++ b/service/src/test/resources/org/whispersystems/textsecuregcm/redis/redis-cluster.yml @@ -0,0 +1,26 @@ +# Image is redis-cluster:7.4; please see +# https://hub.docker.com/layers/bitnami/redis-cluster/7.4/images/sha256-c11efe6a53692829b6e031ea8b5b4caa380df3c84ad4242549851d345592708d + +services: + redis-0: + image: docker.io/bitnami/redis-cluster@sha256:a53d023fdfaf8a8d7ddc58da040d3494e4cb45772644618ffa44c42dcd32b9af + environment: + - 'ALLOW_EMPTY_PASSWORD=yes' + - 'REDIS_NODES=redis-0 redis-1 redis-2' + + redis-1: + image: docker.io/bitnami/redis-cluster@sha256:a53d023fdfaf8a8d7ddc58da040d3494e4cb45772644618ffa44c42dcd32b9af + environment: + - 'ALLOW_EMPTY_PASSWORD=yes' + - 'REDIS_NODES=redis-0 redis-1 redis-2' + + redis-2: + image: docker.io/bitnami/redis-cluster@sha256:a53d023fdfaf8a8d7ddc58da040d3494e4cb45772644618ffa44c42dcd32b9af + depends_on: + - redis-0 + - redis-1 + environment: + - 'ALLOW_EMPTY_PASSWORD=yes' + - 'REDIS_CLUSTER_REPLICAS=0' + - 'REDIS_NODES=redis-0 redis-1 redis-2' + - 'REDIS_CLUSTER_CREATOR=yes'