MirrorMirror
/
Signal-Server
mirror of https://github.com/signalapp/Signal-Server.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add pub/sub affordances to FaultTolerantRedisCluster.

This commit is contained in:
Jon Chambers 2020-08-04 13:31:29 -04:00 committed by Jon Chambers
parent 189f8afcc9
commit 9457325119
2 changed files with 45 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
service/src/main/java/org/whispersystems/textsecuregcm/redis/FaultTolerantRedisCluster.java
View File

@ -6,6 +6,7 @@ import io.github.resilience4j.circuitbreaker.CircuitBreaker;
import io.lettuce.core.RedisURI; import io.lettuce.core.RedisURI;
import io.lettuce.core.cluster.RedisClusterClient; import io.lettuce.core.cluster.RedisClusterClient;
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
import io.lettuce.core.cluster.pubsub.StatefulRedisClusterPubSubConnection;
import io.lettuce.core.codec.ByteArrayCodec; import io.lettuce.core.codec.ByteArrayCodec;
import org.whispersystems.textsecuregcm.configuration.CircuitBreakerConfiguration; import org.whispersystems.textsecuregcm.configuration.CircuitBreakerConfiguration;
import org.whispersystems.textsecuregcm.util.CircuitBreakerUtil; import org.whispersystems.textsecuregcm.util.CircuitBreakerUtil;
@ -26,11 +27,13 @@ public class FaultTolerantRedisCluster {
private final RedisClusterClient clusterClient; private final RedisClusterClient clusterClient;
private final StatefulRedisClusterConnection<String, String> stringClusterConnection; private final StatefulRedisClusterConnection<String, String> stringClusterConnection;
private final StatefulRedisClusterConnection<byte[], byte[]> binaryClusterConnection; private final StatefulRedisClusterConnection<byte[], byte[]> binaryClusterConnection;
private final StatefulRedisClusterPubSubConnection<String, String> pubSubClusterConnection;
private final CircuitBreaker readCircuitBreaker; private final CircuitBreaker readCircuitBreaker;
private final CircuitBreaker writeCircuitBreaker; private final CircuitBreaker writeCircuitBreaker;
private final CircuitBreaker pubSubCircuitBreaker;
public FaultTolerantRedisCluster(final String name, final List<String> urls, final Duration timeout, final CircuitBreakerConfiguration circuitBreakerConfiguration) { public FaultTolerantRedisCluster(final String name, final List<String> urls, final Duration timeout, final CircuitBreakerConfiguration circuitBreakerConfiguration) {
this(name, RedisClusterClient.create(urls.stream().map(RedisURI::create).collect(Collectors.toList())), timeout, circuitBreakerConfiguration); this(name, RedisClusterClient.create(urls.stream().map(RedisURI::create).collect(Collectors.toList())), timeout, circuitBreakerConfiguration);
@ -43,8 +46,10 @@ public class FaultTolerantRedisCluster {
this.stringClusterConnection = clusterClient.connect(); this.stringClusterConnection = clusterClient.connect();
this.binaryClusterConnection = clusterClient.connect(ByteArrayCodec.INSTANCE); this.binaryClusterConnection = clusterClient.connect(ByteArrayCodec.INSTANCE);
this.pubSubClusterConnection = clusterClient.connectPubSub();
this.readCircuitBreaker = CircuitBreaker.of(name + "-read", circuitBreakerConfiguration.toCircuitBreakerConfig()); this.readCircuitBreaker = CircuitBreaker.of(name + "-read", circuitBreakerConfiguration.toCircuitBreakerConfig());
this.writeCircuitBreaker = CircuitBreaker.of(name + "-write", circuitBreakerConfiguration.toCircuitBreakerConfig()); this.writeCircuitBreaker = CircuitBreaker.of(name + "-write", circuitBreakerConfiguration.toCircuitBreakerConfig());
this.pubSubCircuitBreaker = CircuitBreaker.of(name + "-pubsub", circuitBreakerConfiguration.toCircuitBreakerConfig());
CircuitBreakerUtil.registerMetrics(SharedMetricRegistries.getOrCreate(Constants.METRICS_NAME), CircuitBreakerUtil.registerMetrics(SharedMetricRegistries.getOrCreate(Constants.METRICS_NAME),
readCircuitBreaker, readCircuitBreaker,
@ -53,11 +58,16 @@ public class FaultTolerantRedisCluster {
CircuitBreakerUtil.registerMetrics(SharedMetricRegistries.getOrCreate(Constants.METRICS_NAME), CircuitBreakerUtil.registerMetrics(SharedMetricRegistries.getOrCreate(Constants.METRICS_NAME),
writeCircuitBreaker, writeCircuitBreaker,
FaultTolerantRedisCluster.class); FaultTolerantRedisCluster.class);
CircuitBreakerUtil.registerMetrics(SharedMetricRegistries.getOrCreate(Constants.METRICS_NAME),
pubSubCircuitBreaker,
FaultTolerantRedisCluster.class);
} }
void shutdown() { void shutdown() {
stringClusterConnection.close(); stringClusterConnection.close();
binaryClusterConnection.close(); binaryClusterConnection.close();
pubSubClusterConnection.close();
clusterClient.shutdown(); clusterClient.shutdown();
} }
@ -93,4 +103,12 @@ public class FaultTolerantRedisCluster {
public <T> T withBinaryWriteCluster(final Function<StatefulRedisClusterConnection<byte[], byte[]>, T> consumer) { public <T> T withBinaryWriteCluster(final Function<StatefulRedisClusterConnection<byte[], byte[]>, T> consumer) {
return this.writeCircuitBreaker.executeSupplier(() -> consumer.apply(binaryClusterConnection)); return this.writeCircuitBreaker.executeSupplier(() -> consumer.apply(binaryClusterConnection));
} }
public void usePubSubConnection(final Consumer<StatefulRedisClusterPubSubConnection<String, String>> consumer) {
this.pubSubCircuitBreaker.executeRunnable(() -> consumer.accept(pubSubClusterConnection));
}
public <T> T withPubSubConnection(final Function<StatefulRedisClusterPubSubConnection<String, String>, T> consumer) {
return this.pubSubCircuitBreaker.executeSupplier(() -> consumer.apply(pubSubClusterConnection));
}
} }

28
service/src/test/java/org/whispersystems/textsecuregcm/redis/FaultTolerantRedisClusterTest.java
View File

@ -5,6 +5,8 @@ import io.lettuce.core.RedisException;
import io.lettuce.core.cluster.RedisClusterClient; import io.lettuce.core.cluster.RedisClusterClient;
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands; import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands;
import io.lettuce.core.cluster.pubsub.StatefulRedisClusterPubSubConnection;
import io.lettuce.core.cluster.pubsub.api.sync.RedisClusterPubSubCommands;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
import org.whispersystems.textsecuregcm.configuration.CircuitBreakerConfiguration; import org.whispersystems.textsecuregcm.configuration.CircuitBreakerConfiguration;
@ -18,20 +20,25 @@ import static org.mockito.Mockito.when;
public class FaultTolerantRedisClusterTest { public class FaultTolerantRedisClusterTest {
private RedisAdvancedClusterCommands<String, String> clusterCommands; private RedisAdvancedClusterCommands<String, String> clusterCommands;
private RedisClusterPubSubCommands<String, String> pubSubCommands;
private FaultTolerantRedisCluster faultTolerantCluster; private FaultTolerantRedisCluster faultTolerantCluster;
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
@Before @Before
public void setUp() { public void setUp() {
final RedisClusterClient clusterClient = mock(RedisClusterClient.class); final RedisClusterClient clusterClient = mock(RedisClusterClient.class);
final StatefulRedisClusterConnection<String, String> clusterConnection = mock(StatefulRedisClusterConnection.class); final StatefulRedisClusterConnection<String, String> clusterConnection = mock(StatefulRedisClusterConnection.class);
final StatefulRedisClusterPubSubConnection<String, String> pubSubConnection = mock(StatefulRedisClusterPubSubConnection.class);
clusterCommands = mock(RedisAdvancedClusterCommands.class); clusterCommands = mock(RedisAdvancedClusterCommands.class);
pubSubCommands = mock(RedisClusterPubSubCommands.class);
when(clusterClient.connect()).thenReturn(clusterConnection); when(clusterClient.connect()).thenReturn(clusterConnection);
when(clusterClient.connectPubSub()).thenReturn(pubSubConnection);
when(clusterConnection.sync()).thenReturn(clusterCommands); when(clusterConnection.sync()).thenReturn(clusterCommands);
when(pubSubConnection.sync()).thenReturn(pubSubCommands);
final CircuitBreakerConfiguration breakerConfiguration = new CircuitBreakerConfiguration(); final CircuitBreakerConfiguration breakerConfiguration = new CircuitBreakerConfiguration();
breakerConfiguration.setFailureRateThreshold(100); breakerConfiguration.setFailureRateThreshold(100);
@ -85,4 +92,19 @@ public class FaultTolerantRedisClusterTest {
assertThrows(CircuitBreakerOpenException.class, assertThrows(CircuitBreakerOpenException.class,
() -> faultTolerantCluster.withWriteCluster(connection -> connection.sync().get("OH NO"))); () -> faultTolerantCluster.withWriteCluster(connection -> connection.sync().get("OH NO")));
} }
@Test
public void testPubSubBreaker() {
when(pubSubCommands.publish(anyString(), anyString()))
.thenReturn(1L)
.thenThrow(new RedisException("Badness has ensued."));
assertEquals(1L, (long)faultTolerantCluster.withPubSubConnection(connection -> connection.sync().publish("channel", "message")));
assertThrows(RedisException.class,
() -> faultTolerantCluster.withPubSubConnection(connection -> connection.sync().publish("channel", "OH NO")));
assertThrows(CircuitBreakerOpenException.class,
() -> faultTolerantCluster.withPubSubConnection(connection -> connection.sync().publish("channel", "OH NO")));
}
} }