[CROSSDATA-577] [SERVER] Service Discovery (#643)

* [CROSSDATA] [CONFIG] Service discovery for Hazelcast & Akka members
Stratio · Sep 9, 2016 · 275e886 · 275e886
1 parent f9305b8
commit 275e886
Show file tree

Hide file tree

Showing 12 changed files with 405 additions and 31 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -10,7 +10,8 @@ Only listing significant user-visible, not internal code cleanups and minor bug
     * Adds support for sub documents at Native side.
 
 * Major Akka version upgrade to 2.4 (2.4.9), thus enabling putting the server behind HAProxy. 
-* Resolve queries using fully qualified identifiers. 
+* Resolve queries using fully qualified identifiers.
+* Service discovery of Akka Cluster and Hazelcast members for servers through Zookeeper.
 
 ## 1.5.0 (August 2016)
 

diff --git a/core/src/main/scala/org/apache/spark/sql/crossdata/XDContext.scala b/core/src/main/scala/org/apache/spark/sql/crossdata/XDContext.scala
@@ -92,7 +92,6 @@ class XDContext protected (@transient val sc: SparkContext,
 
   catalogConfig = Try(xdConfig.getConfig(CoreConfig.CatalogConfigKey)).getOrElse(ConfigFactory.empty())
 
-
   override protected[sql] def executeSql(sql: String): org.apache.spark.sql.execution.QueryExecution = executePlan(parseSql(sql))
 
   override protected[sql] def executePlan(plan: LogicalPlan): sparkexecution.QueryExecution =

diff --git a/core/src/main/scala/org/apache/spark/sql/crossdata/catalyst/streaming/streamingDdl.scala b/core/src/main/scala/org/apache/spark/sql/crossdata/catalyst/streaming/streamingDdl.scala
@@ -100,9 +100,6 @@ private[crossdata] case class CreateEphemeralTable(tableIdent: TableIdentifier,
 
 }
 
-
-
-
 private[crossdata] case class DropEphemeralTable(tableIdent: TableIdentifier)
   extends LogicalPlan with RunnableCommand {
 

diff --git a/dist/src/main/resources/server/server-application.conf b/dist/src/main/resources/server/server-application.conf
@@ -77,6 +77,21 @@ crossdata-server.config.spark.metrics.conf="/etc/sds/crossdata/server/xdmetrics.
 ################################
 #crossdata-server.akka.cluster.auto-down-unreachable-after = 10s
 
+################################
+#                              #
+#      Service Discovery       #
+#                              #
+################################
+
+#crossdata-server.service-discovery.activated = false
+#crossdata-server.service-discovery.url = "127.0.0.1:2181"
+#crossdata-server.service-discovery.cluster-subscription.path = "/stratio/crossdata/discovery/subscription/leader"
+#crossdata-server.service-discovery.cluster-subscription.timeout = 10
+#crossdata-server.service-discovery.cluster-leader.path = "/stratio/crossdata/discovery/cluster/leader"
+#crossdata-server.service-discovery.cluster-leader.scheduler.delay = 10
+#crossdata-server.service-discovery.seeds.path = "/stratio/crossdata/discovery/seeds"
+#crossdata-server.service-discovery.provider.path = "/stratio/crossdata/discovery/provider/members"
+
 ################################
 #                              #
 #      Security                #

diff --git a/pom.xml b/pom.xml
@@ -116,9 +116,9 @@
         <derby.version>10.10.1.1</derby.version>
         <jackson4s.version>3.2.10</jackson4s.version>
         <mockito.version>1.10.19</mockito.version>
-        <curator.version>2.9.0</curator.version>
         <common.utils.version>0.5.0</common.utils.version>
         <guava.version>18.0</guava.version>
+        <curator.version>2.11.0</curator.version>
         <!-- Scala version and cross build properties -->
         <scala.binary.version>2.11</scala.binary.version>
         <scala.version>2.11.8</scala.version>

diff --git a/server/pom.xml b/server/pom.xml
@@ -118,6 +118,16 @@
             <artifactId>hazelcast</artifactId>
             <version>${hazelcast.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.curator</groupId>
+            <artifactId>curator-client</artifactId>
+            <version>${curator.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.curator</groupId>
+            <artifactId>curator-x-discovery</artifactId>
+            <version>${curator.version}</version>
+        </dependency>
         <!-- TESTS -->
         <dependency>
             <groupId>com.stratio.crossdata</groupId>

diff --git a/server/src/main/resources/server-reference.conf b/server/src/main/resources/server-reference.conf
@@ -134,6 +134,31 @@ crossdata-server.akka.cluster.acceptable-heartbeat-pause = 20s
 crossdata-server.akka.cluster.heartbeat-interval = 6s
 
 crossdata-server.akka.http.server.parsing.max-content-length=1024m
+
+################################
+#                              #
+#      Service Discovery       #
+#                              #
+################################
+
+crossdata-server.service-discovery.activated = false
+crossdata-server.service-discovery.activated = ${?CROSSDATA_SERVER_SERVICE_DISCOVERY_ACTIVATED}
+#crossdata-server.service-discovery.url = "127.0.0.1:2181"
+crossdata-server.service-discovery.url = ${?CROSSDATA_SERVER_SERVICE_DISCOVERY_URL}
+#crossdata-server.service-discovery.cluster-subscription.path = "/stratio/crossdata/discovery/subscription/leader"
+crossdata-server.service-discovery.cluster-subscription.path = ${?CROSSDATA_SERVER_SERVICE_DISCOVERY_CLUSTER_SUBSCRIPTION_PATH}
+#crossdata-server.service-discovery.cluster-subscription.timeout = 10
+crossdata-server.service-discovery.cluster-subscription.timeout = ${?CROSSDATA_SERVER_SERVICE_DISCOVERY_CLUSTER_SUBSCRIPTION_TIMEOUT}
+#crossdata-server.service-discovery.cluster-leader.path = "/stratio/crossdata/discovery/cluster/leader"
+crossdata-server.service-discovery.cluster-leader.path = ${?CROSSDATA_SERVER_SERVICE_DISCOVERY_CLUSTER_LEADER_PATH}
+#crossdata-server.service-discovery.cluster-leader.scheduler.delay = 10
+crossdata-server.service-discovery.cluster-leader.scheduler.delay = ${?CROSSDATA_SERVER_SERVICE_DISCOVERY_CLUSTER_LEADER_DELAY}
+#crossdata-server.service-discovery.seeds.path = "/stratio/crossdata/discovery/seeds"
+crossdata-server.service-discovery.seeds.path = ${?CROSSDATA_SERVER_SERVICE_DISCOVERY_SEEDS_PATH}
+#crossdata-server.service-discovery.provider.path = "/stratio/crossdata/discovery/provider/members"
+crossdata-server.service-discovery.provider.path = ${?CROSSDATA_SERVER_SERVICE_DISCOVERY_PROVIDER_PATH}
+
+
 ################################
 #                              #
 #      Security                #

diff --git a/server/src/main/scala/com/stratio/crossdata/server/CrossdataApplication.scala b/server/src/main/scala/com/stratio/crossdata/server/CrossdataApplication.scala
@@ -15,10 +15,25 @@
  */
 package com.stratio.crossdata.server
 
+import com.typesafe.config.{Config, ConfigFactory}
+
+import scala.collection.JavaConversions._
+
 
 object CrossdataApplication extends App {
 
-  val crossdataServer = new CrossdataServer
+  val mapConfig = {
+    val (keys, values) = args.toList.partition(_.startsWith("--"))
+    val argsConfig = keys.map(_.replace("--", "")).zip(values).toMap
+    ConfigFactory.parseMap(argsConfig)
+  }
+
+  val params = mapConfig match {
+    case m: Config if !m.isEmpty => Some(m)
+    case _ => None
+  }
+
+  val crossdataServer = new CrossdataServer(params)
   crossdataServer.start()
 
 }