Issue 140 - add SSL support

doc/configuration.md

@@ -10,6 +10,7 @@ topology from the initial node, so there is no need to provide the rest of the c
 * `spark.redis.timeout` - connection timeout in ms, 2000 ms by default
 * `spark.redis.max.pipeline.size` - the maximum number of commands per pipeline (used to batch commands). The default value is 100.
 * `spark.redis.scan.count` - count option of SCAN command (used to iterate over keys). The default value is 100.
+* `spark.redis.ssl` - set to true to use tls
 
 
 

src/main/scala/com/redislabs/provider/redis/ConnectionPool.scala

@@ -11,6 +11,7 @@ import scala.collection.JavaConversions._
 object ConnectionPool {
   @transient private lazy val pools: ConcurrentHashMap[RedisEndpoint, JedisPool] =
     new ConcurrentHashMap[RedisEndpoint, JedisPool]()
+
   def connect(re: RedisEndpoint): Jedis = {
     val pool = pools.getOrElseUpdate(re,
       {
@@ -23,7 +24,8 @@ object ConnectionPool {
         poolConfig.setMinEvictableIdleTimeMillis(60000)
         poolConfig.setTimeBetweenEvictionRunsMillis(30000)
         poolConfig.setNumTestsPerEvictionRun(-1)
-        new JedisPool(poolConfig, re.host, re.port, re.timeout, re.auth, re.dbNum)
+
+        new JedisPool(poolConfig, re.host, re.port, re.timeout, re.auth, re.dbNum, re.ssl)
       }
     )
     var sleepTime: Int = 4

src/main/scala/com/redislabs/provider/redis/RedisConfig.scala

@@ -11,23 +11,24 @@ import scala.collection.JavaConversions._
 
 /**
   * RedisEndpoint represents a redis connection endpoint info: host, port, auth password
-  * db number, and timeout
+  * db number, timeout and ssl mode
   *
   * @param host  the redis host or ip
   * @param port  the redis port
   * @param auth  the authentication password
   * @param dbNum database number (should be avoided in general)
+  * @param ssl true to enable SSL connection. Defaults to false
   */
 case class RedisEndpoint(host: String = Protocol.DEFAULT_HOST,
                          port: Int = Protocol.DEFAULT_PORT,
                          auth: String = null,
                          dbNum: Int = Protocol.DEFAULT_DATABASE,
-                         timeout: Int = Protocol.DEFAULT_TIMEOUT)
+                         timeout: Int = Protocol.DEFAULT_TIMEOUT,
+                         ssl: Boolean = false)
   extends Serializable {
 
   /**
-    * Constructor from spark config. set params with spark.redis.host, spark.redis.port, spark.redis.auth and
-    * spark.redis.db
+    * Constructor from spark config. set params with spark.redis.host, spark.redis.port, spark.redis.auth, spark.redis.db and spark.redis.ssl
     *
     * @param conf spark context config
     */
@@ -37,23 +38,25 @@ case class RedisEndpoint(host: String = Protocol.DEFAULT_HOST,
       conf.getInt("spark.redis.port", Protocol.DEFAULT_PORT),
       conf.get("spark.redis.auth", null),
       conf.getInt("spark.redis.db", Protocol.DEFAULT_DATABASE),
-      conf.getInt("spark.redis.timeout", Protocol.DEFAULT_TIMEOUT)
+      conf.getInt("spark.redis.timeout", Protocol.DEFAULT_TIMEOUT),
+      conf.getBoolean("spark.redis.ssl", false)
     )
   }
 
   /**
     * Constructor with Jedis URI
     *
-    * @param uri connection URI in the form of redis://:$password@$host:$port/[dbnum]
+    * @param uri connection URI in the form of redis://:$password@$host:$port/[dbnum]. Use "rediss://" scheme for redis SSL
     */
   def this(uri: URI) {
-    this(uri.getHost, uri.getPort, JedisURIHelper.getPassword(uri), JedisURIHelper.getDBIndex(uri))
+    this(uri.getHost, uri.getPort, JedisURIHelper.getPassword(uri), JedisURIHelper.getDBIndex(uri),
+      Protocol.DEFAULT_TIMEOUT, uri.getScheme == RedisSslScheme)
   }
 
   /**
     * Constructor with Jedis URI from String
     *
-    * @param uri connection URI in the form of redis://:$password@$host:$port/[dbnum]
+    * @param uri connection URI in the form of redis://:$password@$host:$port/[dbnum]. Use "rediss://" scheme for redis SSL
     */
   def this(uri: String) {
     this(URI.create(uri))
@@ -251,7 +254,7 @@ class RedisConfig(val initialHost: RedisEndpoint) extends Serializable {
 
       //simply re-enter this function witht he master host/port
       getNonClusterNodes(initialHost = new RedisEndpoint(host, port,
-        initialHost.auth, initialHost.dbNum))
+        initialHost.auth, initialHost.dbNum, ssl = initialHost.ssl))
 
     } else {
       //this is a master - take its slaves
@@ -267,7 +270,7 @@ class RedisConfig(val initialHost: RedisEndpoint) extends Serializable {
       val range = nodes.length
       (0 until range).map(i =>
         RedisNode(RedisEndpoint(nodes(i)._1, nodes(i)._2, initialHost.auth, initialHost.dbNum,
-          initialHost.timeout),
+          initialHost.timeout, initialHost.ssl),
           0, 16383, i, range)).toArray
     }
   }
@@ -297,7 +300,7 @@ class RedisConfig(val initialHost: RedisEndpoint) extends Serializable {
           val host = SafeEncoder.encode(node.get(0).asInstanceOf[Array[scala.Byte]])
           val port = node.get(1).toString.toInt
           RedisNode(RedisEndpoint(host, port, initialHost.auth, initialHost.dbNum,
-            initialHost.timeout),
+            initialHost.timeout, initialHost.ssl),
             sPos,
             ePos,
             i,

src/main/scala/com/redislabs/provider/redis/package.scala

@@ -1,7 +1,7 @@
 package com.redislabs.provider
 
 package object redis extends RedisFunctions {
-
+  val RedisSslScheme: String = "rediss"
   val RedisDataTypeHash: String = "hash"
   val RedisDataTypeString: String = "string"
 }

src/main/scala/org/apache/spark/sql/redis/RedisSourceRelation.scala

@@ -40,7 +40,8 @@ class RedisSourceRelation(override val sqlContext: SQLContext,
         val auth = parameters.getOrElse("auth", null)
         val dbNum = parameters.get("dbNum").map(_.toInt).getOrElse(Protocol.DEFAULT_DATABASE)
         val timeout = parameters.get("timeout").map(_.toInt).getOrElse(Protocol.DEFAULT_TIMEOUT)
-        RedisEndpoint(host, port, auth, dbNum, timeout)
+        val ssl = parameters.get("ssl").map(_.toBoolean).getOrElse(false)
+        RedisEndpoint(host, port, auth, dbNum, timeout, ssl)
       }
     )
   }