fix according to comments

apache · Sep 21, 2014 · c90d84e · c90d84e
1 parent 8f7cc96
commit c90d84e
Show file tree

Hide file tree

Showing 9 changed files with 31 additions and 33 deletions.
diff --git a/core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala b/core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala
@@ -39,8 +39,7 @@ private[deploy] object DeployMessages {
       port: Int,
       cores: Int,
       memory: Int,
-      workerWebUiUrl: String,
-      publicAddress: String)
+      workerWebUiUrl: String)
     extends DeployMessage {
     Utils.checkHost(host, "Required hostname")
     assert (port > 0)

diff --git a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala
@@ -122,7 +122,11 @@ private[spark] class Master(
     // Listen for remote client disconnection events, since they don't go through Akka's watch()
     context.system.eventStream.subscribe(self, classOf[RemotingLifecycleEvent])
     webUi.bind()
-    val masterWebUiUrlPrefix = conf.get("spark.http.policy") + "://"
+    val masterWebUiUrlPrefix = if( conf.get("spark.ui.https.enabled", "false").toBoolean) {
+      "https://"
+    } else{
+      "http://"
+    }
     masterWebUiUrl = masterWebUiUrlPrefix + masterPublicAddress + ":" + webUi.boundPort
     context.system.scheduler.schedule(0 millis, WORKER_TIMEOUT millis, self, CheckForWorkerTimeOut)
 
@@ -191,7 +195,7 @@ private[spark] class Master(
       System.exit(0)
     }
 
-    case RegisterWorker(id, workerHost, workerPort, cores, memory, workerWebUiUrl, publicAddress) =>
+    case RegisterWorker(id, workerHost, workerPort, cores, memory, workerWebUiUrl) =>
     {
       logInfo("Registering worker %s:%d with %d cores, %s RAM".format(
         workerHost, workerPort, cores, Utils.megabytesToString(memory)))
@@ -201,7 +205,7 @@ private[spark] class Master(
         sender ! RegisterWorkerFailed("Duplicate worker ID")
       } else {
         val worker = new WorkerInfo(id, workerHost, workerPort, cores, memory,
-          sender, workerWebUiUrl, publicAddress)
+          sender, workerWebUiUrl)
         if (registerWorker(worker)) {
           persistenceEngine.addWorker(worker)
           sender ! RegisteredWorker(masterUrl, masterWebUiUrl)

diff --git a/core/src/main/scala/org/apache/spark/deploy/master/WorkerInfo.scala b/core/src/main/scala/org/apache/spark/deploy/master/WorkerInfo.scala
@@ -30,8 +30,7 @@ private[spark] class WorkerInfo(
     val cores: Int,
     val memory: Int,
     val actor: ActorRef,
-    val webUiAddress: String,
-    val publicAddress: String)
+    val webUiAddress: String)
   extends Serializable {
 
   Utils.checkHost(host, "Expected hostname")

diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala b/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
@@ -78,7 +78,7 @@ private[spark] class Worker(
   var activeMasterUrl: String = ""
   var activeMasterWebUiUrl : String = ""
   val akkaUrl = "akka.tcp://%s@%s:%s/user/%s".format(actorSystemName, host, port, actorName)
-  var workerWebUiUrl: String = _
+  var workerWebUiUrl: String = ""
   @volatile var registered = false
   @volatile var connected = false
   val workerId = generateWorkerId()
@@ -139,7 +139,12 @@ private[spark] class Worker(
     context.system.eventStream.subscribe(self, classOf[RemotingLifecycleEvent])
     webUi = new WorkerWebUI(this, workDir, webUiPort)
     webUi.bind()
-    workerWebUiUrl = conf.get("spark.http.policy") + "://" + publicAddress + ":" + webUi.boundPort
+    val workerWebUiUrlPrefix = if( conf.get("spark.ui.https.enabled", "false").toBoolean) {
+      "https://"
+    } else{
+      "http://"
+    }
+    workerWebUiUrl = workerWebUiUrlPrefix + publicAddress + ":" + webUi.boundPort
     registerWithMaster()
 
     metricsSystem.registerSource(workerSource)
@@ -165,7 +170,7 @@ private[spark] class Worker(
     for (masterUrl <- masterUrls) {
       logInfo("Connecting to master " + masterUrl + "...")
       val actor = context.actorSelection(Master.toAkkaUrl(masterUrl))
-      actor ! RegisterWorker(workerId, host, port, cores, memory, workerWebUiUrl, publicAddress)
+      actor ! RegisterWorker(workerId, host, port, cores, memory, workerWebUiUrl)
     }
   }
 

diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/WorkerArguments.scala b/core/src/main/scala/org/apache/spark/deploy/worker/WorkerArguments.scala
@@ -53,9 +53,6 @@ private[spark] class WorkerArguments(args: Array[String], conf: SparkConf) {
   if (System.getenv("SPARK_WORKER_DIR") != null) {
     workDir = System.getenv("SPARK_WORKER_DIR")
   }
-  if (conf.contains("worker.ui.port")) {
-    webUiPort = conf.get("worker.ui.port").toInt
-  }
 
   parse(args.toList)
 

diff --git a/core/src/main/scala/org/apache/spark/ui/JettyUtils.scala b/core/src/main/scala/org/apache/spark/ui/JettyUtils.scala
@@ -30,13 +30,13 @@ import org.eclipse.jetty.server.{Connector, Server}
 import org.eclipse.jetty.server.handler._
 import org.eclipse.jetty.servlet._
 import org.eclipse.jetty.util.thread.QueuedThreadPool
+import org.eclipse.jetty.server.nio.SelectChannelConnector
+import org.eclipse.jetty.server.ssl.SslSelectChannelConnector
 import org.json4s.JValue
 import org.json4s.jackson.JsonMethods.{pretty, render}
 
 import org.apache.spark.{Logging, SecurityManager, SparkConf}
 import org.apache.spark.util.Utils
-import org.eclipse.jetty.server.nio.SelectChannelConnector
-import org.eclipse.jetty.server.ssl.SslSelectChannelConnector
 
 /**
  * Utilities for launching a web server using Jetty's HTTP Server class
@@ -212,11 +212,10 @@ private[spark] object JettyUtils extends Logging {
   }
 
   private def getConnector(port: Int, conf: SparkConf): Connector = {
-    val https = getHttpPolicy(conf)
+    val https = conf.get("spark.ui.https.enabled", "false").toBoolean
     if (https) {
       buildSslSelectChannelConnector(port, conf)
     } else {
-      conf.set("spark.http.policy", "http")
       val connector = new SelectChannelConnector
       connector.setPort(port)
       connector
@@ -245,13 +244,6 @@ private[spark] object JettyUtils extends Logging {
     connector
   }
 
-  def getHttpPolicy(conf: SparkConf): Boolean = {
-    if (conf.contains("spark.http.policy") && conf.get("spark.http.policy").equals("https")) {
-      true
-    } else {
-      false
-    }
-  }
 }
 
 private[spark] case class ServerInfo(

diff --git a/core/src/main/scala/org/apache/spark/ui/SparkUI.scala b/core/src/main/scala/org/apache/spark/ui/SparkUI.scala
@@ -97,8 +97,6 @@ private[spark] class SparkUI(
    */
   private[spark] def appUIHostPort = publicHostName + ":" + boundPort
 
-  private def appUiAddressPrefix = conf.get("spark.http.policy")
-
   private[spark] def appUIAddress = s"$appUiAddressPrefix://$appUIHostPort"
 }
 

diff --git a/core/src/main/scala/org/apache/spark/ui/WebUI.scala b/core/src/main/scala/org/apache/spark/ui/WebUI.scala
@@ -91,6 +91,12 @@ private[spark] abstract class WebUI(
     }
   }
 
+  def appUiAddressPrefix = if(conf.get("spark.ui.https.enabled", "false").toBoolean) {
+    "https"
+  } else {
+    "http"
+  }
+
   /** Initialize all components of the server. */
   def initialize()
 
@@ -99,11 +105,7 @@ private[spark] abstract class WebUI(
     assert(!serverInfo.isDefined, "Attempted to bind %s more than once!".format(className))
     try {
       serverInfo = Some(startJettyServer("0.0.0.0", port, handlers, conf, name))
-      if (conf.get("spark.http.policy").equals("https")) {
-        logInfo("Started %s at https://%s:%d".format(className, publicHostName, boundPort))
-      } else {
-        logInfo("Started %s at http://%s:%d".format(className, publicHostName, boundPort))
-      }
+      logInfo(s"Started %s at $appUiAddressPrefix://%s:%d".format(className, publicHostName, boundPort))
     } catch {
       case e: Exception =>
         logError("Failed to bind %s".format(className), e)

diff --git a/core/src/test/scala/org/apache/spark/ui/UISuite.scala b/core/src/test/scala/org/apache/spark/ui/UISuite.scala
@@ -124,8 +124,10 @@ class UISuite extends FunSuite {
       case Failure(e) =>
       // Either case server port is busy hence setup for test complete
     }
+    //keytool -export -keystore core/src/test/resources/spark.keystore -alias spark -file /home/wf/code/spark2/core/src/test/resources/spark.cer -storepass 123456
+
     val sparkConf = new SparkConf()
-      .set("spark.http.policy", "https")
+      .set("spark.ui.https.enabled", "true")
       .set("spark.ssl.server.keystore.location", "./src/test/resources/spark.keystore")
     val serverInfo1 = JettyUtils.startJettyServer(
       "0.0.0.0", startPort, Seq[ServletContextHandler](), sparkConf)
@@ -154,7 +156,7 @@ class UISuite extends FunSuite {
 
   test("jetty with https binds to port 0 correctly") {
     val sparkConf = new SparkConf()
-      .set("spark.http.policy", "https")
+      .set("spark.ui.https.enabled", "true")
       .set("spark.ssl.server.keystore.location", "./src/test/resources/spark.keystore")
     val serverInfo = JettyUtils.startJettyServer(
       "0.0.0.0", 0, Seq[ServletContextHandler](), sparkConf)