Initial hystrix support

pom.xml

@@ -14,6 +14,7 @@
     <properties>
         <dropwizard.version>1.3.7</dropwizard.version>
         <jackson.api.version>2.9.6</jackson.api.version>
+        <hystrix.version>1.5.12</hystrix.version>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     </properties>
 
@@ -110,6 +111,17 @@
             <version>0.5.0</version>
         </dependency>
 
+        <dependency>
+            <groupId>com.netflix.hystrix</groupId>
+            <artifactId>hystrix-core</artifactId>
+            <version>${hystrix.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.netflix.hystrix</groupId>
+            <artifactId>hystrix-codahale-metrics-publisher</artifactId>
+            <version>${hystrix.version}</version>
+        </dependency>
+
         <dependency>
             <groupId>com.relayrides</groupId>
             <artifactId>pushy</artifactId>

src/main/java/org/whispersystems/textsecuregcm/WhisperServerConfiguration.java

@@ -131,6 +131,10 @@ public class WhisperServerConfiguration extends Configuration {
   @JsonProperty
   private UnidentifiedDeliveryConfiguration unidentifiedDelivery;
 
+  @NotNull
+  @JsonProperty
+  private Map<String, Object> hystrix = new HashMap<>();
+
   public WebSocketConfiguration getWebSocketConfiguration() {
     return webSocket;
   }
@@ -225,4 +229,7 @@ public class WhisperServerConfiguration extends Configuration {
     return results;
   }
 
+  public Map<String, Object> getHystrixConfiguration() {
+    return hystrix;
+  }
 }

src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java

@@ -20,6 +20,10 @@ import com.codahale.metrics.SharedMetricRegistries;
 import com.fasterxml.jackson.annotation.JsonAutoDetect;
 import com.fasterxml.jackson.annotation.PropertyAccessor;
 import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.netflix.config.ConfigurationManager;
+import com.netflix.hystrix.contrib.codahalemetricspublisher.HystrixCodaHaleMetricsPublisher;
+import com.netflix.hystrix.strategy.HystrixPlugins;
+import org.apache.commons.configuration.MapConfiguration;
 import org.bouncycastle.jce.provider.BouncyCastleProvider;
 import org.eclipse.jetty.servlets.CrossOriginFilter;
 import org.skife.jdbi.v2.DBI;
@@ -102,6 +106,9 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
 
   @Override
   public void initialize(Bootstrap<WhisperServerConfiguration> bootstrap) {
+    HystrixCodaHaleMetricsPublisher hystrixMetricsPublisher = new HystrixCodaHaleMetricsPublisher(bootstrap.getMetricRegistry());
+    HystrixPlugins.getInstance().registerMetricsPublisher(hystrixMetricsPublisher);
+
     bootstrap.addCommand(new DirectoryCommand());
     bootstrap.addCommand(new VacuumCommand());
     bootstrap.addCommand(new TrimMessagesCommand());
@@ -137,6 +144,8 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     environment.getObjectMapper().setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.NONE);
     environment.getObjectMapper().setVisibility(PropertyAccessor.FIELD, JsonAutoDetect.Visibility.ANY);
 
+    ConfigurationManager.install(new MapConfiguration(config.getHystrixConfiguration()));
+
     DBIFactory dbiFactory = new DBIFactory();
     DBI        database   = dbiFactory.build(environment, config.getDataSourceFactory(), "accountdb");
     DBI        messagedb  = dbiFactory.build(environment, config.getMessageStoreConfiguration(), "messagedb");

src/main/java/org/whispersystems/textsecuregcm/hystrix/GroupKeys.java

@@ -0,0 +1,9 @@
+package org.whispersystems.textsecuregcm.hystrix;
+
+public class GroupKeys {
+
+  public static final String REDIS_CACHE       = "redis_cache";
+  public static final String DATABASE_ACCOUNTS = "database_accounts";
+  public static final String DIRECTORY_SERVICE = "directory_service";
+
+}

src/main/java/org/whispersystems/textsecuregcm/storage/Accounts.java

@@ -63,7 +63,7 @@ public abstract class Accounts {
 
   @Mapper(AccountMapper.class)
   @SqlQuery("SELECT * FROM accounts WHERE " + NUMBER + " = :number")
-  abstract Account get(@Bind("number") String number);
+  public abstract Account get(@Bind("number") String number);
 
   @SqlQuery("SELECT COUNT(DISTINCT " + NUMBER + ") from accounts")
   public abstract long getCount();

src/main/java/org/whispersystems/textsecuregcm/storage/AccountsManager.java

@@ -1,5 +1,5 @@
-/**
+/*
- * Copyright (C) 2013 Open WhisperSystems
+ * Copyright (C) 2013-2018 Signal
  *
  * This program is free software: you can redistribute it and/or modify
  * it under the terms of the GNU Affero General Public License as published by
@@ -19,9 +19,13 @@ package org.whispersystems.textsecuregcm.storage;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.netflix.hystrix.HystrixCommand;
+import com.netflix.hystrix.HystrixCommandGroupKey;
+import com.netflix.hystrix.exception.HystrixBadRequestException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.whispersystems.textsecuregcm.entities.ClientContact;
+import org.whispersystems.textsecuregcm.hystrix.GroupKeys;
 import org.whispersystems.textsecuregcm.redis.ReplicatedJedisPool;
 import org.whispersystems.textsecuregcm.util.SystemMapper;
 import org.whispersystems.textsecuregcm.util.Util;
@@ -62,72 +66,119 @@ public class AccountsManager {
   }
 
   public boolean create(Account account) {
-    boolean freshUser = accounts.create(account);
+    boolean freshUser = databaseCreate(account);
-    memcacheSet(account.getNumber(), account);
+    redisSet(account.getNumber(), account, false);
     updateDirectory(account);
 
     return freshUser;
   }
 
   public void update(Account account) {
-    memcacheSet(account.getNumber(), account);
+    redisSet(account.getNumber(), account, false);
-    accounts.update(account);
+    databaseUpdate(account);
     updateDirectory(account);
   }
 
   public Optional<Account> get(String number) {
-    Optional<Account> account = memcacheGet(number);
+    Optional<Account> account = redisGet(number);
 
     if (!account.isPresent()) {
-      account = Optional.ofNullable(accounts.get(number));
+      account = databaseGet(number);
-
+      account.ifPresent(value -> redisSet(number, value, true));
-      if (account.isPresent()) {
-        memcacheSet(number, account.get());
-      }
     }
 
     return account;
   }
 
-  public boolean isRelayListed(String number) {
-    byte[]                  token   = Util.getContactToken(number);
-    Optional<ClientContact> contact = directory.get(token);
-
-    return contact.isPresent() && !Util.isEmpty(contact.get().getRelay());
-  }
-
   private void updateDirectory(Account account) {
-    if (account.isActive()) {
+    new HystrixCommand<Void>(HystrixCommandGroupKey.Factory.asKey(GroupKeys.DIRECTORY_SERVICE)) {
-      byte[]        token         = Util.getContactToken(account.getNumber());
+      @Override
-      ClientContact clientContact = new ClientContact(token, null, true, true);
+      protected Void run() {
-      directory.add(clientContact);
+        if (account.isActive()) {
-    } else {
+          byte[]        token         = Util.getContactToken(account.getNumber());
-      directory.remove(account.getNumber());
+          ClientContact clientContact = new ClientContact(token, null, true, true);
-    }
+          directory.add(clientContact);
+        } else {
+          directory.remove(account.getNumber());
+        }
+
+        return null;
+      }
+    }.execute();
   }
 
   private String getKey(String number) {
     return Account.class.getSimpleName() + Account.MEMCACHE_VERION + number;
   }
 
-  private void memcacheSet(String number, Account account) {
+  private void redisSet(String number, Account account, boolean optional) {
-    try (Jedis jedis = cacheClient.getWriteResource()) {
+    new HystrixCommand<Boolean>(HystrixCommandGroupKey.Factory.asKey(GroupKeys.REDIS_CACHE)) {
-      jedis.set(getKey(number), mapper.writeValueAsString(account));
+      @Override
-    } catch (JsonProcessingException e) {
+      protected Boolean run() {
-      throw new IllegalArgumentException(e);
+        try (Jedis jedis = cacheClient.getWriteResource()) {
-    }
+          jedis.set(getKey(number), mapper.writeValueAsString(account));
+        } catch (JsonProcessingException e) {
+          throw new HystrixBadRequestException("Json processing error", e);
+        }
+
+        return true;
+      }
+
+      @Override
+      protected Boolean getFallback() {
+        if (optional) return true;
+        else          return super.getFallback();
+      }
+    }.execute();
   }
 
-  private Optional<Account> memcacheGet(String number) {
+  private Optional<Account> redisGet(String number) {
-    try (Jedis jedis = cacheClient.getReadResource()) {
+    return new HystrixCommand<Optional<Account>>(HystrixCommandGroupKey.Factory.asKey(GroupKeys.REDIS_CACHE)) {
-      String json = jedis.get(getKey(number));
+      @Override
+      protected Optional<Account> run() {
+        try (Jedis jedis = cacheClient.getReadResource()) {
+          String json = jedis.get(getKey(number));
 
-      if (json != null) return Optional.of(mapper.readValue(json, Account.class));
+          if (json != null) return Optional.of(mapper.readValue(json, Account.class));
-      else              return Optional.empty();
+          else              return Optional.empty();
-    } catch (IOException e) {
+        } catch (IOException e) {
-      logger.warn("AccountsManager", "Deserialization error", e);
+          logger.warn("AccountsManager", "Deserialization error", e);
-      return Optional.empty();
+          return Optional.empty();
-    }
+        }
+      }
+
+      @Override
+      protected Optional<Account> getFallback() {
+        return Optional.empty();
+      }
+    }.execute();
   }
 
+  private Optional<Account> databaseGet(String number) {
+    return new HystrixCommand<Optional<Account>>(HystrixCommandGroupKey.Factory.asKey(GroupKeys.DATABASE_ACCOUNTS)) {
+      @Override
+      protected Optional<Account> run() {
+        return Optional.ofNullable(accounts.get(number));
+      }
+    }.execute();
+  }
+
+  private boolean databaseCreate(Account account) {
+    return new HystrixCommand<Boolean>(HystrixCommandGroupKey.Factory.asKey(GroupKeys.DATABASE_ACCOUNTS)) {
+      @Override
+      protected Boolean run() {
+        return accounts.create(account);
+      }
+    }.execute();
+  }
+
+  private void databaseUpdate(Account account) {
+    new HystrixCommand<Void>(HystrixCommandGroupKey.Factory.asKey(GroupKeys.DATABASE_ACCOUNTS)) {
+      @Override
+      protected Void run() {
+        accounts.update(account);
+        return null;
+      }
+    }.execute();
+  }
 }

src/test/java/org/whispersystems/textsecuregcm/tests/storage/AccountsManagerTest.java

@@ -0,0 +1,134 @@
+package org.whispersystems.textsecuregcm.tests.storage;
+
+import com.netflix.hystrix.exception.HystrixRuntimeException;
+import org.junit.Test;
+import org.skife.jdbi.v2.exceptions.UnableToObtainConnectionException;
+import org.whispersystems.textsecuregcm.redis.ReplicatedJedisPool;
+import org.whispersystems.textsecuregcm.storage.Account;
+import org.whispersystems.textsecuregcm.storage.Accounts;
+import org.whispersystems.textsecuregcm.storage.AccountsManager;
+import org.whispersystems.textsecuregcm.storage.DirectoryManager;
+
+import java.util.HashSet;
+import java.util.Optional;
+
+import static junit.framework.TestCase.assertSame;
+import static junit.framework.TestCase.assertTrue;
+import static org.junit.Assert.assertEquals;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.*;
+import redis.clients.jedis.Jedis;
+import redis.clients.jedis.exceptions.JedisException;
+
+public class AccountsManagerTest {
+
+  @Test
+  public void testGetAccountInCache() {
+    ReplicatedJedisPool cacheClient      = mock(ReplicatedJedisPool.class);
+    Jedis               jedis            = mock(Jedis.class              );
+    Accounts            accounts         = mock(Accounts.class           );
+    DirectoryManager    directoryManager = mock(DirectoryManager.class   );
+
+    when(cacheClient.getReadResource()).thenReturn(jedis);
+    when(jedis.get(eq("Account5+14152222222"))).thenReturn("{\"number\": \"+14152222222\", \"name\": \"test\"}");
+
+    AccountsManager   accountsManager = new AccountsManager(accounts, directoryManager, cacheClient);
+    Optional<Account> account         = accountsManager.get("+14152222222");
+
+    assertTrue(account.isPresent());
+    assertEquals(account.get().getNumber(), "+14152222222");
+    assertEquals(account.get().getProfileName(), "test");
+
+    verify(jedis, times(1)).get(eq("Account5+14152222222"));
+    verify(jedis, times(1)).close();
+    verifyNoMoreInteractions(jedis);
+    verifyNoMoreInteractions(accounts);
+  }
+
+  @Test
+  public void testGetAccountNotInCache() {
+    ReplicatedJedisPool cacheClient      = mock(ReplicatedJedisPool.class);
+    Jedis               jedis            = mock(Jedis.class              );
+    Accounts            accounts         = mock(Accounts.class           );
+    DirectoryManager    directoryManager = mock(DirectoryManager.class   );
+    Account             account          = new Account("+14152222222", new HashSet<>(), new byte[16]);
+
+    when(cacheClient.getReadResource()).thenReturn(jedis);
+    when(cacheClient.getWriteResource()).thenReturn(jedis);
+    when(jedis.get(eq("Account5+14152222222"))).thenReturn(null);
+    when(accounts.get(eq("+14152222222"))).thenReturn(account);
+
+    AccountsManager   accountsManager = new AccountsManager(accounts, directoryManager, cacheClient);
+    Optional<Account> retrieved       = accountsManager.get("+14152222222");
+
+    assertTrue(retrieved.isPresent());
+    assertSame(retrieved.get(), account);
+
+    verify(jedis, times(1)).get(eq("Account5+14152222222"));
+    verify(jedis, times(1)).set(eq("Account5+14152222222"), anyString());
+    verify(jedis, times(2)).close();
+    verifyNoMoreInteractions(jedis);
+
+    verify(accounts, times(1)).get(eq("+14152222222"));
+    verifyNoMoreInteractions(accounts);
+  }
+
+  @Test
+  public void testGetAccountBrokenCache() {
+    ReplicatedJedisPool cacheClient      = mock(ReplicatedJedisPool.class);
+    Jedis               jedis            = mock(Jedis.class              );
+    Accounts            accounts         = mock(Accounts.class           );
+    DirectoryManager    directoryManager = mock(DirectoryManager.class   );
+    Account             account          = new Account("+14152222222", new HashSet<>(), new byte[16]);
+
+    when(cacheClient.getReadResource()).thenReturn(jedis);
+    when(cacheClient.getWriteResource()).thenReturn(jedis);
+    when(jedis.get(eq("Account5+14152222222"))).thenThrow(new JedisException("Connection lost!"));
+    when(accounts.get(eq("+14152222222"))).thenReturn(account);
+
+    AccountsManager   accountsManager = new AccountsManager(accounts, directoryManager, cacheClient);
+    Optional<Account> retrieved       = accountsManager.get("+14152222222");
+
+    assertTrue(retrieved.isPresent());
+    assertSame(retrieved.get(), account);
+
+    verify(jedis, times(1)).get(eq("Account5+14152222222"));
+    verify(jedis, times(1)).set(eq("Account5+14152222222"), anyString());
+    verify(jedis, times(2)).close();
+    verifyNoMoreInteractions(jedis);
+
+    verify(accounts, times(1)).get(eq("+14152222222"));
+    verifyNoMoreInteractions(accounts);
+  }
+
+  @Test
+  public void testGetAccountEmptyCacheBrokenDatabase() {
+    ReplicatedJedisPool cacheClient      = mock(ReplicatedJedisPool.class);
+    Jedis               jedis            = mock(Jedis.class              );
+    Accounts            accounts         = mock(Accounts.class           );
+    DirectoryManager    directoryManager = mock(DirectoryManager.class   );
+    Account             account          = new Account("+14152222222", new HashSet<>(), new byte[16]);
+
+    when(cacheClient.getReadResource()).thenReturn(jedis);
+    when(cacheClient.getWriteResource()).thenReturn(jedis);
+    when(jedis.get(eq("Account5+14152222222"))).thenReturn(null);
+    when(accounts.get(eq("+14152222222"))).thenThrow(new UnableToObtainConnectionException(new Exception()));
+
+    AccountsManager   accountsManager = new AccountsManager(accounts, directoryManager, cacheClient);
+
+    try {
+      Optional<Account> retrieved = accountsManager.get("+14152222222");
+      throw new AssertionError("Should not have succeeded!");
+    } catch (HystrixRuntimeException e) {
+      // good
+      verify(jedis, times(1)).get(eq("Account5+14152222222"));
+      verify(jedis, times(1)).close();
+      verifyNoMoreInteractions(jedis);
+
+      verify(accounts, times(1)).get(eq("+14152222222"));
+      verifyNoMoreInteractions(accounts);
+    }
+  }
+
+
+}