Adds ssl capability for cassandra connectivity

ndbench-cass-plugins/README.md

@@ -26,6 +26,8 @@ Every config has to be prefixed with configuration prefix mentioned above
 |cass.cfname| 	emp | Destination CF Name|
 |cass.readConsistencyLevel| 	LOCAL_ONE | Read CL|
 |cass.writeConsistencyLevel| 	LOCAL_ONE | Write CL|
+|cass.truststorePath| null | Absolute path of truststore
+|cass.truststorePass| null | Password of truststore
 
 
 #### Cassandra *CassJavaDriverGeneric* Plugin configurations

ndbench-cass-plugins/src/main/java/com/netflix/ndbench/plugin/cass/CJavaDriverBasePlugin.java

@@ -46,6 +46,8 @@ public abstract class CJavaDriverBasePlugin<C extends CassandraConfigurationBase
     protected volatile Session session;
     protected volatile PreparedStatement readPstmt;
     protected volatile PreparedStatement writePstmt;
+    protected volatile String truststorePath;
+    protected volatile String truststorePass;
 
     /**
      * Creates an instance of the abstract CJavaDriverBasePlugin class. Subclasses calling this method should use
@@ -70,6 +72,8 @@ public void init(DataGenerator dataGenerator) throws Exception {
         this.connections = config.getConnections();
         this.username = config.getUsername();
         this.password = config.getPassword();
+        this.truststorePath = config.getTruststorePath();
+        this.truststorePass = config.getTruststorePass();
 
         // we do not set ReadConsistencyLevel and WriteConsistencyLevel and MaxColCount here because the
         // enum classes corresponding to the consistency levels differ among the concrete subclasses and because
@@ -100,7 +104,7 @@ private void initDriver() {
         logger.info("Cassandra  Cluster: " + clusterName);
 
         this.cluster = cassJavaDriverManager.registerCluster(clusterName, clusterContactPoint, connections, port,
-                                                             username, password);
+                                                             username, password, truststorePath, truststorePass);
         this.session = cassJavaDriverManager.getSession(cluster);
         if(config.getCreateSchema())
         {

ndbench-cass-plugins/src/main/java/com/netflix/ndbench/plugin/cass/CassJavaDriverManager.java

@@ -12,7 +12,7 @@
  */
 @ImplementedBy(CassJavaDriverManagerImpl.class)
 public interface CassJavaDriverManager {
-    Cluster registerCluster(String clName, String contactPoint, int connections, int port, String username, String password);
+    Cluster registerCluster(String clName, String contactPoint, int connections, int port, String username, String password, String truststorePath, String truststorePass);
     Cluster registerCluster(String clName, String contactPoint, int connections, int port);
     Session getSession(Cluster cluster);
 

ndbench-cass-plugins/src/main/java/com/netflix/ndbench/plugin/cass/CassJavaDriverManagerImpl.java

@@ -3,11 +3,16 @@
  */
 package com.netflix.ndbench.plugin.cass;
 
-import com.datastax.driver.core.Cluster;
-import com.datastax.driver.core.Session;
-import com.datastax.driver.core.PoolingOptions;
-import com.datastax.driver.core.HostDistance;
+import com.datastax.driver.core.*;
 import com.datastax.driver.core.policies.*;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslContext;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManagerFactory;
+import java.io.InputStream;
+import java.security.KeyStore;
+import java.security.KeyStoreException;
 
 
 /**
@@ -20,15 +25,31 @@ public class CassJavaDriverManagerImpl implements CassJavaDriverManager {
 
     @Override
     public Cluster registerCluster(String clName, String contactPoint, int connections, int port) {
-        return registerCluster(clName,contactPoint,connections,port,null,null);
+        return registerCluster(clName,contactPoint,connections,port,null,null,null,null);
     }
         @Override
-    public Cluster registerCluster(String clName, String contactPoint, int connections, int port, String username, String password) {
+    public Cluster registerCluster(String clName, String contactPoint, int connections, int port, String username, String password, String truststorePath, String truststorePass) {
 
         PoolingOptions poolingOpts = new PoolingOptions()
                                      .setConnectionsPerHost(HostDistance.LOCAL, connections, connections)
                                      .setMaxRequestsPerConnection(HostDistance.LOCAL, 32768);
+        KeyStore ks = null;
+        SSLContext sslContext = null ;
+
+        try {
+            ks = KeyStore.getInstance("JKS");
+            InputStream trustStore = new java.io.FileInputStream(truststorePath);
+            ks.load(trustStore, truststorePass.toCharArray());
+            TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+            tmf.init(ks);
 
+            sslContext = SSLContext.getInstance("TLS");
+            sslContext.init(null,tmf.getTrustManagers(),null);
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+
+        SSLOptions sslOptions = RemoteEndpointAwareJdkSSLOptions.builder().withSSLContext(sslContext).build();
 
         Cluster.Builder clusterBuilder = Cluster.builder()
                 .withClusterName(clName)
@@ -40,6 +61,10 @@ public Cluster registerCluster(String clName, String contactPoint, int connectio
             clusterBuilder = clusterBuilder.withCredentials(username, password);
         }
 
+        if ((truststorePath !=null)) {
+            clusterBuilder = clusterBuilder.withSSL(sslOptions);
+        }
+
         cluster = clusterBuilder.build();
         return cluster;
     }

ndbench-cass-plugins/src/main/java/com/netflix/ndbench/plugin/configs/CassandraConfigurationBase.java

@@ -75,4 +75,10 @@ public interface CassandraConfigurationBase {
 
     @DefaultValue("true")
     Boolean getCreateSchema();
+
+    @PropertyName(name = "truststorePath")
+    String getTruststorePath();
+
+    @PropertyName(name = "truststorePass")
+    String getTruststorePass();
 }