Adding Hysterix library for getting metrics on calls to Vault and KMS (…

…#63)

* Adding Hysterix library for getting metrics on calls to Vault and KMS
* Allowing environment.properties to override initial values loaded into Archaius
* Upgrading vault client
* Adding retry on create orphan token
* Additional configuration of the vault admin client

.travis.yml

@@ -11,4 +11,4 @@ deploy:
   file: build/libs/cms.jar
   on:
     repo: Nike-Inc/cerberus-management-service
-    tags: true
+    tags: true

gradle.properties

@@ -14,6 +14,6 @@
 # limitations under the License.
 #
 
-version=0.29.0
+version=0.30.0-hystrixRC5
 groupId=com.nike.cerberus
 artifactId=cms

gradle/dependencies.gradle

@@ -28,7 +28,7 @@ def groovyVersion = '2.3.9'
 
 dependencies {
     compile (
-            "com.nike:vault-client:1.4.0",
+            "com.nike:vault-client:1.6.0",
             "com.nike.riposte:riposte-spi:$riposteVersion",
             "com.nike.riposte:riposte-core:$riposteVersion",
             "com.nike.riposte:riposte-typesafe-config:$riposteVersion",
@@ -38,6 +38,7 @@ dependencies {
             "com.nike.riposte:riposte-metrics-codahale:$riposteVersion",
             "com.nike.riposte:riposte-metrics-codahale-signalfx:$riposteVersion",
 
+            "com.netflix.hystrix:hystrix-core:1.5.12",
             "javax:javaee-api:7.0",
             "org.codehaus.groovy:groovy-all:$groovyVersion", // For logback groovy config processing
             "ch.qos.logback:logback-classic:$logbackVersion",

gradle/develop.gradle

@@ -19,7 +19,7 @@ import groovyx.net.http.RESTClient
 import static groovyx.net.http.ContentType.*
 
 def dashboardRelease = 'v1.6.0'
-def vaultVersion = "0.7.3"
+def vaultVersion = "0.8.1"
 
 def reverseProxyPort = 9001
 def nodeServerPort = 8000

src/main/java/com/nike/cerberus/aws/KmsClientFactory.java

@@ -29,7 +29,6 @@
 /**
  * Factory for AWS KMS clients.  Caches clients by region as they are requested.
  */
-@Singleton
 public class KmsClientFactory {
 
     private final Map<Region, AWSKMSClient> kmsClientMap = Maps.newConcurrentMap();

src/main/java/com/nike/cerberus/hystrix/HystrixKmsClient.java

@@ -0,0 +1,102 @@
+package com.nike.cerberus.hystrix;
+
+import com.amazonaws.services.kms.AWSKMSClient;
+import com.amazonaws.services.kms.model.CreateAliasRequest;
+import com.amazonaws.services.kms.model.CreateAliasResult;
+import com.amazonaws.services.kms.model.CreateKeyRequest;
+import com.amazonaws.services.kms.model.CreateKeyResult;
+import com.amazonaws.services.kms.model.DescribeKeyRequest;
+import com.amazonaws.services.kms.model.DescribeKeyResult;
+import com.amazonaws.services.kms.model.EncryptRequest;
+import com.amazonaws.services.kms.model.EncryptResult;
+import com.amazonaws.services.kms.model.GetKeyPolicyRequest;
+import com.amazonaws.services.kms.model.GetKeyPolicyResult;
+import com.amazonaws.services.kms.model.PutKeyPolicyRequest;
+import com.amazonaws.services.kms.model.PutKeyPolicyResult;
+import com.amazonaws.services.kms.model.ScheduleKeyDeletionRequest;
+import com.amazonaws.services.kms.model.ScheduleKeyDeletionResult;
+import com.netflix.hystrix.HystrixCommand;
+import com.netflix.hystrix.HystrixCommandGroupKey;
+import com.netflix.hystrix.HystrixCommandKey;
+import com.netflix.hystrix.HystrixThreadPoolKey;
+
+import java.util.function.Supplier;
+
+/**
+ * Hystrix wrapper around AWSKMSClient
+ * <p>
+ * Most of these commands should execute in their own Thread Pools because they have unique limits.
+ * http://docs.aws.amazon.com/kms/latest/developerguide/limits.html#requests-per-second-table
+ */
+public class HystrixKmsClient extends AWSKMSClient {
+
+    private static final String KMS = "KMS";
+
+    private final AWSKMSClient client;
+
+    public HystrixKmsClient(AWSKMSClient client) {
+        this.client = client;
+    }
+
+    public EncryptResult encrypt(EncryptRequest request) {
+        // Default AWS limit was 1200 shared as of Aug 2017
+        return execute("KmsEncryptDecrypt", "KmsEncrypt", () -> client.encrypt(request));
+    }
+
+    public CreateKeyResult createKey(CreateKeyRequest request) {
+        // Default AWS limit was 5 as of Aug 2017
+        return execute("KmsCreateKey", () -> client.createKey(request));
+    }
+
+    public CreateAliasResult createAlias(CreateAliasRequest request) {
+        // Default AWS limit was 5 as of Aug 2017
+        return execute("KmsCreateAlias", () -> client.createAlias(request));
+    }
+
+    public DescribeKeyResult describeKey(DescribeKeyRequest request) {
+        // Default AWS limit was 30 as of Aug 2017
+        return execute("KmsDescribeKey", () -> client.describeKey(request));
+    }
+
+    public ScheduleKeyDeletionResult scheduleKeyDeletion(ScheduleKeyDeletionRequest request) {
+        // Default AWS limit was 5 as of Aug 2017
+        return execute("KmsScheduleKeyDeletion", () -> client.scheduleKeyDeletion(request));
+    }
+
+    public GetKeyPolicyResult getKeyPolicy(GetKeyPolicyRequest request) {
+        // Default AWS limit was 30 as of Aug 2017
+        return execute("KmsGetKeyPolicy", () -> client.getKeyPolicy(request));
+    }
+
+    public PutKeyPolicyResult putKeyPolicy(PutKeyPolicyRequest request) {
+        // Default AWS limit was 5 as of Aug 2017
+        return execute("KmsPutKeyPolicy", () -> client.putKeyPolicy(request));
+    }
+
+    /**
+     * Execute a function that returns a value in a ThreadPool unique to that command.
+     */
+    private static <T> T execute(String commandKey, Supplier<T> function) {
+        return execute(commandKey, commandKey, function);
+    }
+
+    /**
+     * Execute a function that returns a value in a specified ThreadPool
+     */
+    private static <T> T execute(String threadPoolName, String commandKey, Supplier<T> function) {
+        return new HystrixCommand<T>(buildSetter(threadPoolName, commandKey)) {
+
+            @Override
+            protected T run() throws Exception {
+                return function.get();
+            }
+        }.execute();
+    }
+
+    private static HystrixCommand.Setter buildSetter(String threadPoolName, String commandKey) {
+        return HystrixCommand.Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey(KMS))
+                .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey(threadPoolName))
+                .andCommandKey(HystrixCommandKey.Factory.asKey(commandKey));
+    }
+
+}

src/main/java/com/nike/cerberus/hystrix/HystrixKmsClientFactory.java

@@ -0,0 +1,23 @@
+package com.nike.cerberus.hystrix;
+
+import com.amazonaws.regions.Region;
+import com.nike.cerberus.aws.KmsClientFactory;
+
+public class HystrixKmsClientFactory extends KmsClientFactory {
+
+    private final KmsClientFactory kmsClientFactory;
+
+    public HystrixKmsClientFactory(KmsClientFactory kmsClientFactory) {
+        this.kmsClientFactory = kmsClientFactory;
+    }
+
+    @Override
+    public HystrixKmsClient getClient(Region region) {
+        return new HystrixKmsClient(kmsClientFactory.getClient(region));
+    }
+
+    @Override
+    public HystrixKmsClient getClient(String regionName) {
+        return new HystrixKmsClient(kmsClientFactory.getClient(regionName));
+    }
+}

src/main/java/com/nike/cerberus/hystrix/HystrixMetricsLogger.java

@@ -0,0 +1,75 @@
+package com.nike.cerberus.hystrix;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import com.google.inject.name.Named;
+import com.netflix.hystrix.HystrixCircuitBreaker;
+import com.netflix.hystrix.HystrixCommandMetrics;
+import com.netflix.hystrix.HystrixThreadPoolMetrics;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Periodically print Hystrix metrics to the log.
+ */
+@Singleton
+public class HystrixMetricsLogger implements Runnable {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(HystrixMetricsLogger.class);
+
+    private final ScheduledExecutorService executor;
+
+    @Inject
+    public HystrixMetricsLogger(@Named("hystrixExecutor") ScheduledExecutorService executor,
+                                @Named("cms.hystrix.metricsLoggingInitialIntervalSeconds") long initialIntervalSeconds,
+                                @Named("cms.hystrix.metricsLoggingIntervalSeconds") long intervalSeconds) {
+        this.executor = executor;
+        LOGGER.info("Hystrix metrics logging initialIntervalSeconds:{}, intervalSeconds:{}", initialIntervalSeconds, intervalSeconds);
+        this.executor.scheduleWithFixedDelay(this, initialIntervalSeconds, intervalSeconds, TimeUnit.SECONDS);
+    }
+
+    @Override
+    public void run() {
+        try {
+            printHystrixCommandMetrics();
+            printHystrixThreadPoolMetrics();
+        } catch (Exception e) {
+            LOGGER.warn("Error printing Hystrix metrics", e);
+        }
+    }
+
+    public void printHystrixCommandMetrics() {
+        for (HystrixCommandMetrics metrics : HystrixCommandMetrics.getInstances()) {
+            boolean isCircuitOpen = HystrixCircuitBreaker.Factory.getInstance(metrics.getCommandKey()).isOpen();
+
+            LOGGER.info("group:{}, commandKey:{}, CircuitOpen:{}, Mean:{}, 95%:{}, 99%:{}, 99.5%:{}, {}",
+                    metrics.getCommandGroup().name(),
+                    metrics.getCommandKey().name(),
+                    isCircuitOpen,
+                    metrics.getExecutionTimeMean(),
+                    metrics.getExecutionTimePercentile(95.0),
+                    metrics.getExecutionTimePercentile(99.5),
+                    metrics.getExecutionTimePercentile(99.5),
+                    metrics.getHealthCounts()
+            );
+        }
+    }
+
+    public void printHystrixThreadPoolMetrics() {
+        for (HystrixThreadPoolMetrics metrics : HystrixThreadPoolMetrics.getInstances()) {
+            LOGGER.info("threadPool:{}, rollingCounts[rejected:{}, executed:{}, maxActiveThreads:{}], cumulativeCounts[rejected:{}, executed:{}], {}",
+                    metrics.getThreadPoolKey().name(),
+                    metrics.getRollingCountThreadsRejected(),
+                    metrics.getRollingCountThreadsExecuted(),
+                    metrics.getRollingMaxActiveThreads(),
+                    metrics.getCumulativeCountThreadsRejected(),
+                    metrics.getCumulativeCountThreadsExecuted(),
+                    metrics.getThreadPool()
+            );
+        }
+    }
+
+}

src/main/java/com/nike/cerberus/hystrix/HystrixRequestAndResponseFilter.java

@@ -0,0 +1,46 @@
+package com.nike.cerberus.hystrix;
+
+import com.netflix.hystrix.strategy.concurrency.HystrixRequestContext;
+import com.nike.riposte.server.http.RequestInfo;
+import com.nike.riposte.server.http.ResponseInfo;
+import com.nike.riposte.server.http.filter.RequestAndResponseFilter;
+import io.netty.channel.ChannelHandlerContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Riposte filter that sets up and shuts down a HystrixRequestContext
+ */
+public class HystrixRequestAndResponseFilter implements RequestAndResponseFilter {
+
+    private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+    @Override
+    public <T> RequestInfo<T> filterRequestFirstChunkNoPayload(RequestInfo<T> currentRequestInfo,
+                                                               ChannelHandlerContext ctx) {
+        HystrixRequestContext context = HystrixRequestContext.initializeContext();
+        currentRequestInfo.addRequestAttribute("HystrixRequestContext", context);
+        return currentRequestInfo;
+    }
+
+    @Override
+    public <T> RequestInfo<T> filterRequestLastChunkWithFullPayload(RequestInfo<T> currentRequestInfo,
+                                                                    ChannelHandlerContext ctx) {
+        // Nothing to do - the other filterRequest method already handled Hystrix initialization.
+        // Returning null just means use the passed-in response, which is what we want.
+        return null;
+    }
+
+    @Override
+    public <T> ResponseInfo<T> filterResponse(ResponseInfo<T> currentResponseInfo, RequestInfo<?> requestInfo,
+                                              ChannelHandlerContext ctx) {
+        try {
+            ((HystrixRequestContext) requestInfo.getRequestAttributes().get("HystrixRequestContext")).shutdown();
+        } catch (Throwable t) {
+            logger.error("An unexpected error occurred trying to shutdown the HystrixRequestContext for this request.", t);
+        }
+
+        // Returning null just means use the passed-in response, which is what we want.
+        return null;
+    }
+}