Merge branch 'master' into UnrecoverableException

CorfuDB · Dec 5, 2017 · e7d886c · e7d886c
2 parents 76835aa + 95f853e
commit e7d886c
Show file tree

Hide file tree

Showing 28 changed files with 689 additions and 180 deletions.
diff --git a/.github/PULL_REQUEST_TEMPLATE.md b/.github/PULL_REQUEST_TEMPLATE.md
@@ -4,7 +4,7 @@ Description:
 
 Why should this be merged: 
 
-Related issue(s) (if applicable): #1023
+Related issue(s) (if applicable): #<number>
 
 
 ## Checklist (Definition of Done):

diff --git a/generator/src/main/java/org/corfudb/generator/LongevityApp.java b/generator/src/main/java/org/corfudb/generator/LongevityApp.java
@@ -38,13 +38,14 @@ public class LongevityApp {
 
     // How much time we live the application hangs once the duration is finished
     // and the application is hanged
-    static final int APPLICATION_TIMEOUT_IN_MS = 20000;
+    static final int APPLICATION_TIMEOUT_IN_MS = 10000;
 
     static final int QUEUE_CAPACITY = 1000;
 
     long startTime;
     int numberThreads;
 
+
     public LongevityApp(long durationMs, int numberThreads, String configurationString, boolean checkPoint) {
         this.durationMs = durationMs;
         this.checkPoint = checkPoint;
@@ -60,6 +61,30 @@ public LongevityApp(long durationMs, int numberThreads, String configurationStri
         checkpointer = Executors.newScheduledThreadPool(1);
     }
 
+    /**
+     * Assess liveness of the application
+     *
+     * If the client was not able to do any operation during the last APPLICATION_TIMEOUT_IN_MS,
+     * we declare liveness of the client as failed. Also, if the client was not able to finish
+     * in time, it is marked as liveness failure.
+     *
+     * @param finishedInTime
+     * @return
+     */
+    private boolean livenessSuccess(boolean finishedInTime) {
+        if (!finishedInTime) {
+            return false;
+        }
+
+        long timeSinceSuccessfulReadOperation = System.currentTimeMillis()
+                - state.getLastSuccessfulReadOperationTimestamp();
+        long timeSinceSuccessfulWriteOperation = System.currentTimeMillis()
+                - state.getLastSuccessfulWriteOperationTimestamp();
+
+        return (timeSinceSuccessfulReadOperation < APPLICATION_TIMEOUT_IN_MS
+                && timeSinceSuccessfulWriteOperation < APPLICATION_TIMEOUT_IN_MS);
+    }
+
     /**
      * Give a chance to the workers to finish naturally (thanks to the timer) and then kill
      * the producer and the checkpointer.
@@ -76,7 +101,9 @@ private void waitForAppToFinish() {
         try {
             boolean finishedInTime = workers.
                     awaitTermination(durationMs + APPLICATION_TIMEOUT_IN_MS, TimeUnit.MILLISECONDS);
-            String livenessState = finishedInTime ? "Success" : "Fail";
+
+            String livenessState = livenessSuccess(finishedInTime) ? "Success" : "Fail";
+
             Correctness.recordOperation("Liveness, " + livenessState, false);
             if (!finishedInTime) {
                 System.exit(1);
@@ -86,6 +113,20 @@ private void waitForAppToFinish() {
         } finally {
             taskProducer.shutdownNow();
             checkpointer.shutdownNow();
+
+            boolean checkpointHasFinished = false;
+            int exitStatus;
+            try {
+                checkpointHasFinished = checkpointer.awaitTermination(APPLICATION_TIMEOUT_IN_MS, TimeUnit.MILLISECONDS);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                exitStatus = 1;
+            }
+
+            exitStatus = checkpointHasFinished ? 0 : 1;
+            System.exit(exitStatus);
+
+
         }
     }
 

diff --git a/generator/src/main/java/org/corfudb/generator/State.java b/generator/src/main/java/org/corfudb/generator/State.java
@@ -61,6 +61,14 @@ public enum StringIndexer implements
     @Getter
     final Map<UUID, CorfuTable> maps;
 
+    @Getter
+    @Setter
+    volatile long lastSuccessfulReadOperationTimestamp = -1;
+
+    @Getter
+    @Setter
+    volatile long lastSuccessfulWriteOperationTimestamp = -1;
+
     @Getter
     @Setter
     volatile long trimMark = -1;

diff --git a/generator/src/main/java/org/corfudb/generator/operations/OptimisticTxOperation.java b/generator/src/main/java/org/corfudb/generator/operations/OptimisticTxOperation.java
@@ -4,6 +4,7 @@
 import org.corfudb.generator.Correctness;
 import org.corfudb.generator.State;
 import org.corfudb.runtime.exceptions.TransactionAbortedException;
+import org.corfudb.runtime.view.Address;
 
 import java.util.List;
 
@@ -43,6 +44,11 @@ public void execute() {
 
             Correctness.recordTransactionMarkers(true, shortName, Correctness.TX_END,
                     Long.toString(timestamp));
+
+            if (Address.isAddress(timestamp)) {
+                state.setLastSuccessfulWriteOperationTimestamp(System.currentTimeMillis());
+            }
+
         } catch (TransactionAbortedException tae) {
             Correctness.recordTransactionMarkers(false, shortName, Correctness.TX_ABORTED);
         }

diff --git a/generator/src/main/java/org/corfudb/generator/operations/ReadOperation.java b/generator/src/main/java/org/corfudb/generator/operations/ReadOperation.java
@@ -35,5 +35,9 @@ public void execute() {
         ((CorfuTable)state.getMap((CorfuRuntime.getStreamID(streamId)))).
                 getByIndex(State.StringIndexer.BY_VALUE, val);
 
+        if (!TransactionalContext.isInTransaction()) {
+            state.setLastSuccessfulReadOperationTimestamp(System.currentTimeMillis());
+        }
+
     }
 }
diff --git a/generator/src/main/java/org/corfudb/generator/operations/RemoveOperation.java b/generator/src/main/java/org/corfudb/generator/operations/RemoveOperation.java
@@ -26,6 +26,10 @@ public void execute() {
 
             String correctnessRecord = String.format("%s, %s:%s", shortName, streamId, key);
             Correctness.recordOperation(correctnessRecord, TransactionalContext.isInTransaction());
+
+            if (!TransactionalContext.isInTransaction()) {
+                state.setLastSuccessfulWriteOperationTimestamp(System.currentTimeMillis());
+            }
         }
     }
 }
diff --git a/generator/src/main/java/org/corfudb/generator/operations/SnapshotTxOperation.java b/generator/src/main/java/org/corfudb/generator/operations/SnapshotTxOperation.java
@@ -57,6 +57,7 @@ public void execute() {
 
             state.stopTx();
             Correctness.recordTransactionMarkers(false, shortName, Correctness.TX_END);
+            state.setLastSuccessfulReadOperationTimestamp(System.currentTimeMillis());
         } catch (TransactionAbortedException tae) {
             Correctness.recordTransactionMarkers(false, shortName, Correctness.TX_ABORTED);
         }

diff --git a/generator/src/main/java/org/corfudb/generator/operations/WriteOperation.java b/generator/src/main/java/org/corfudb/generator/operations/WriteOperation.java
@@ -31,6 +31,10 @@ public void execute() {
 
             String correctnessRecord = String.format("%s, %s:%s=%s", shortName, streamId, key, val);
             Correctness.recordOperation(correctnessRecord, TransactionalContext.isInTransaction());
+
+            if (!TransactionalContext.isInTransaction()) {
+                state.setLastSuccessfulWriteOperationTimestamp(System.currentTimeMillis());
+            }
         }
     }
 }
diff --git a/infrastructure/src/main/java/org/corfudb/infrastructure/CorfuServer.java b/infrastructure/src/main/java/org/corfudb/infrastructure/CorfuServer.java
@@ -34,12 +34,14 @@
 
 import javax.net.ssl.SSLEngine;
 
+import javax.net.ssl.SSLException;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 
 import org.corfudb.protocols.wireprotocol.NettyCorfuMessageDecoder;
 import org.corfudb.protocols.wireprotocol.NettyCorfuMessageEncoder;
 import org.corfudb.security.sasl.plaintext.PlainTextSaslNettyServer;
+import org.corfudb.security.tls.SslContextConstructor;
 import org.corfudb.security.tls.TlsUtils;
 import org.corfudb.util.GitRepositoryState;
 import org.corfudb.util.Version;
@@ -304,26 +306,11 @@ public static void main(String[] args) {
             }
 
             try {
-                sslContext =
-                        TlsUtils.enableTls(TlsUtils.SslContextType.SERVER_CONTEXT,
-                                (String) opts.get("--keystore"), e -> {
-                                    log.error("Could not load keys from the key store.");
-                                    System.exit(1);
-                                },
-                                (String) opts.get("--keystore-password-file"), e -> {
-                                    log.error("Could not read the key store password file.");
-                                    System.exit(1);
-                                },
-                                (String) opts.get("--truststore"), e -> {
-                                    log.error("Could not load keys from the trust store.");
-                                    System.exit(1);
-                                },
-                                (String) opts.get("--truststore-password-file"), e -> {
-                                    log.error("Could not read the trust store password file.");
-                                    System.exit(1);
-                                });
-            } catch (Exception ex) {
-                log.error("Could not build the SSL context");
+                sslContext = SslContextConstructor.constructSslContext(true, (String) opts.get("--keystore"),
+                        (String) opts.get("--keystore-password-file"),
+                        (String) opts.get("--truststore"), (String) opts.get("--truststore-password-file"));
+            } catch (SSLException e) {
+                log.error("Could not build the SSL context", e);
                 System.exit(1);
             }
         }

diff --git a/runtime/src/main/java/org/corfudb/runtime/clients/NettyClientRouter.java b/runtime/src/main/java/org/corfudb/runtime/clients/NettyClientRouter.java
@@ -36,6 +36,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 
+import javax.net.ssl.SSLException;
 import lombok.Getter;
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
@@ -48,6 +49,7 @@
 import org.corfudb.runtime.exceptions.WrongEpochException;
 import org.corfudb.security.sasl.SaslUtils;
 import org.corfudb.security.sasl.plaintext.PlainTextSaslNettyClient;
+import org.corfudb.security.tls.SslContextConstructor;
 import org.corfudb.security.tls.TlsUtils;
 import org.corfudb.util.CFUtils;
 import org.corfudb.util.MetricsUtils;
@@ -270,24 +272,12 @@ public NettyClientRouter(String host, Integer port, Boolean tls,
         counterAsyncOpSent = metrics.counter(pfx + "async-op-sent");
 
         if (tls) {
-            sslContext =
-                    TlsUtils.enableTls(TlsUtils.SslContextType.CLIENT_CONTEXT,
-                            keyStore, e -> {
-                                throw new RuntimeException("Could not read the key store "
-                                        + "password file: " + e.getClass().getSimpleName(), e);
-                            },
-                            ksPasswordFile, e -> {
-                                throw new RuntimeException("Could not load keys from the key "
-                                        + "store: " + e.getClass().getSimpleName(), e);
-                            },
-                            trustStore, e -> {
-                                throw new RuntimeException("Could not read the trust store "
-                                        + "password file: " + e.getClass().getSimpleName(), e);
-                            },
-                            tsPasswordFile, e -> {
-                                throw new RuntimeException("Could not load keys from the trust "
-                                        + "store: " + e.getClass().getSimpleName(), e);
-                            });
+            try {
+                sslContext = SslContextConstructor.constructSslContext(false,
+                        keyStore, ksPasswordFile, trustStore, tsPasswordFile);
+            } catch (SSLException e) {
+                throw new RuntimeException(e);
+            }
             this.tlsEnabled = true;
         }
 

diff --git a/runtime/src/main/java/org/corfudb/security/tls/ReloadableTrustManager.java b/runtime/src/main/java/org/corfudb/security/tls/ReloadableTrustManager.java
@@ -0,0 +1,108 @@
+package org.corfudb.security.tls;
+
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.security.cert.CertificateException;
+import java.security.cert.X509Certificate;
+import javax.net.ssl.SSLException;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.X509TrustManager;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * This trust manager reloads the trust store whenever a checkClientTrusted
+ * or checkServerTrusted is called.
+ *
+ * Created by zjohnny on 9/18/17.
+ */
+@Slf4j
+public class ReloadableTrustManager implements X509TrustManager {
+    private String trustStorePath, trustPasswordPath;
+    private X509TrustManager trustManager;
+
+    /**
+     * Constructor.
+     *
+     * @param trustStorePath
+     *          Location of trust store.
+     * @param trustPasswordPath
+     *          Location of trust store password.
+     * @throws SSLException
+     *          Thrown when there's an issue with loading the trust store.
+     */
+    public ReloadableTrustManager(String trustStorePath, String trustPasswordPath) throws SSLException {
+        this.trustStorePath = trustStorePath;
+        this.trustPasswordPath = trustPasswordPath;
+        reloadTrustStore();
+    }
+
+    @Override
+    public void checkClientTrusted(X509Certificate[] chain, String authType) throws CertificateException {
+        reloadTrustStoreWrapper();
+        trustManager.checkClientTrusted(chain, authType);
+    }
+
+    @Override
+    public void checkServerTrusted(X509Certificate[] chain, String authType) throws CertificateException {
+        reloadTrustStoreWrapper();
+        trustManager.checkServerTrusted(chain, authType);
+    }
+
+    @Override
+    public X509Certificate[] getAcceptedIssuers() {
+        return new X509Certificate[0];
+    }
+
+    /**
+     * Just a wrapper due to IDE pointing out duplicate code.
+     *
+     * @throws CertificateException
+     *          Wrapper for any exception from reloading the trust store.
+     */
+    private void reloadTrustStoreWrapper() throws CertificateException {
+        try {
+            reloadTrustStore();
+        } catch (SSLException e) {
+            String message = "Unable to reload trust store " + trustStorePath + ".";
+            log.error(message, e);
+            throw new CertificateException(message, e);
+        }
+    }
+
+    /**
+     * Reload the trust manager.
+     *
+     * @throws SSLException
+     *          Thrown when there's an issue with loading the trust store.
+     */
+    private void reloadTrustStore() throws SSLException {
+        String trustPassword = TlsUtils.getKeyStorePassword(trustPasswordPath);
+        KeyStore trustStore = TlsUtils.openKeyStore(trustStorePath, trustPassword);
+
+        TrustManagerFactory tmf;
+        try {
+            tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+            tmf.init(trustStore);
+        } catch (NoSuchAlgorithmException e) {
+            String errorMessage = "No support for TrustManagerFactory default algorithm "
+                    + TrustManagerFactory.getDefaultAlgorithm() + ".";
+            log.error(errorMessage, e);
+            throw new SSLException(errorMessage, e);
+        } catch (KeyStoreException e) {
+            String errorMessage = "Unable to load trust store " + trustStorePath + ".";
+            log.error(errorMessage, e);
+            throw new SSLException(errorMessage, e);
+        }
+
+        for (TrustManager tm: tmf.getTrustManagers()) {
+            if (tm instanceof X509TrustManager) {
+                trustManager = (X509TrustManager)tm;
+                return;
+            }
+        }
+
+        throw new SSLException("No X509TrustManager in TrustManagerFactory.");
+    }
+}