Merge branch 'master' of https://github.com/apache/spark into jdbc-schema-rdd

Author: chutium

core/pom.xml

@@ -150,7 +150,7 @@
     <dependency>
       <groupId>org.json4s</groupId>
       <artifactId>json4s-jackson_${scala.binary.version}</artifactId>
-      <version>3.2.6</version>
+      <version>3.2.10</version>
     </dependency>
     <dependency>
       <groupId>colt</groupId>

core/src/main/scala/org/apache/spark/Accumulators.scala

@@ -36,15 +36,21 @@ import org.apache.spark.serializer.JavaSerializer
  *
  * @param initialValue initial value of accumulator
  * @param param helper object defining how to add elements of type `R` and `T`
+ * @param name human-readable name for use in Spark's web UI
  * @tparam R the full accumulated data (result type)
  * @tparam T partial data that can be added in
  */
 class Accumulable[R, T] (
     @transient initialValue: R,
-    param: AccumulableParam[R, T])
+    param: AccumulableParam[R, T],
+    val name: Option[String])
   extends Serializable {
 
-  val id = Accumulators.newId
+  def this(@transient initialValue: R, param: AccumulableParam[R, T]) =
+    this(initialValue, param, None)
+
+  val id: Long = Accumulators.newId
+
   @transient private var value_ = initialValue // Current value on master
   val zero = param.zero(initialValue)  // Zero value to be passed to workers
   private var deserialized = false
@@ -219,8 +225,10 @@ GrowableAccumulableParam[R <% Growable[T] with TraversableOnce[T] with Serializa
  * @param param helper object defining how to add elements of type `T`
  * @tparam T result type
  */
-class Accumulator[T](@transient initialValue: T, param: AccumulatorParam[T])
-  extends Accumulable[T,T](initialValue, param)
+class Accumulator[T](@transient initialValue: T, param: AccumulatorParam[T], name: Option[String])
+    extends Accumulable[T,T](initialValue, param, name) {
+  def this(initialValue: T, param: AccumulatorParam[T]) = this(initialValue, param, None)
+}
 
 /**
  * A simpler version of [[org.apache.spark.AccumulableParam]] where the only data type you can add
@@ -281,4 +289,7 @@ private object Accumulators {
       }
     }
   }
+
+  def stringifyPartialValue(partialValue: Any) = "%s".format(partialValue)
+  def stringifyValue(value: Any) = "%s".format(value)
 }

core/src/main/scala/org/apache/spark/HttpFileServer.scala

@@ -23,7 +23,10 @@ import com.google.common.io.Files
 
 import org.apache.spark.util.Utils
 
-private[spark] class HttpFileServer(securityManager: SecurityManager) extends Logging {
+private[spark] class HttpFileServer(
+    securityManager: SecurityManager,
+    requestedPort: Int = 0)
+  extends Logging {
 
   var baseDir : File = null
   var fileDir : File = null
@@ -38,7 +41,7 @@ private[spark] class HttpFileServer(securityManager: SecurityManager) extends Lo
     fileDir.mkdir()
     jarDir.mkdir()
     logInfo("HTTP File server directory is " + baseDir)
-    httpServer = new HttpServer(baseDir, securityManager)
+    httpServer = new HttpServer(baseDir, securityManager, requestedPort, "HTTP file server")
     httpServer.start()
     serverUri = httpServer.uri
     logDebug("HTTP file server started at: " + serverUri)

core/src/main/scala/org/apache/spark/HttpServer.scala

@@ -21,7 +21,7 @@ import java.io.File
 
 import org.eclipse.jetty.util.security.{Constraint, Password}
 import org.eclipse.jetty.security.authentication.DigestAuthenticator
-import org.eclipse.jetty.security.{ConstraintMapping, ConstraintSecurityHandler, HashLoginService, SecurityHandler}
+import org.eclipse.jetty.security.{ConstraintMapping, ConstraintSecurityHandler, HashLoginService}
 
 import org.eclipse.jetty.server.Server
 import org.eclipse.jetty.server.bio.SocketConnector
@@ -41,48 +41,68 @@ private[spark] class ServerStateException(message: String) extends Exception(mes
  * as well as classes created by the interpreter when the user types in code. This is just a wrapper
  * around a Jetty server.
  */
-private[spark] class HttpServer(resourceBase: File, securityManager: SecurityManager)
-    extends Logging {
+private[spark] class HttpServer(
+    resourceBase: File,
+    securityManager: SecurityManager,
+    requestedPort: Int = 0,
+    serverName: String = "HTTP server")
+  extends Logging {
+
   private var server: Server = null
-  private var port: Int = -1
+  private var port: Int = requestedPort
 
   def start() {
     if (server != null) {
       throw new ServerStateException("Server is already started")
     } else {
       logInfo("Starting HTTP Server")
-      server = new Server()
-      val connector = new SocketConnector
-      connector.setMaxIdleTime(60*1000)
-      connector.setSoLingerTime(-1)
-      connector.setPort(0)
-      server.addConnector(connector)
-
-      val threadPool = new QueuedThreadPool
-      threadPool.setDaemon(true)
-      server.setThreadPool(threadPool)
-      val resHandler = new ResourceHandler
-      resHandler.setResourceBase(resourceBase.getAbsolutePath)
-
-      val handlerList = new HandlerList
-      handlerList.setHandlers(Array(resHandler, new DefaultHandler))
-
-      if (securityManager.isAuthenticationEnabled()) {
-        logDebug("HttpServer is using security")
-        val sh = setupSecurityHandler(securityManager)
-        // make sure we go through security handler to get resources
-        sh.setHandler(handlerList)
-        server.setHandler(sh)
-      } else {
-        logDebug("HttpServer is not using security")
-        server.setHandler(handlerList)
-      }
-
-      server.start()
-      port = server.getConnectors()(0).getLocalPort()
+      val (actualServer, actualPort) =
+        Utils.startServiceOnPort[Server](requestedPort, doStart, serverName)
+      server = actualServer
+      port = actualPort
     }
   }
 
+  /**
+   * Actually start the HTTP server on the given port.
+   *
+   * Note that this is only best effort in the sense that we may end up binding to a nearby port
+   * in the event of port collision. Return the bound server and the actual port used.
+   */
+  private def doStart(startPort: Int): (Server, Int) = {
+    val server = new Server()
+    val connector = new SocketConnector
+    connector.setMaxIdleTime(60 * 1000)
+    connector.setSoLingerTime(-1)
+    connector.setPort(startPort)
+    server.addConnector(connector)
+
+    val threadPool = new QueuedThreadPool
+    threadPool.setDaemon(true)
+    server.setThreadPool(threadPool)
+    val resHandler = new ResourceHandler
+    resHandler.setResourceBase(resourceBase.getAbsolutePath)
+
+    val handlerList = new HandlerList
+    handlerList.setHandlers(Array(resHandler, new DefaultHandler))
+
+    if (securityManager.isAuthenticationEnabled()) {
+      logDebug("HttpServer is using security")
+      val sh = setupSecurityHandler(securityManager)
+      // make sure we go through security handler to get resources
+      sh.setHandler(handlerList)
+      server.setHandler(sh)
+    } else {
+      logDebug("HttpServer is not using security")
+      server.setHandler(handlerList)
+    }
+
+    server.start()
+    val actualPort = server.getConnectors()(0).getLocalPort
+
+    (server, actualPort)
+  }
+
   /**
    * Setup Jetty to the HashLoginService using a single user with our
    * shared secret. Configure it to use DIGEST-MD5 authentication so that the password
@@ -134,7 +154,7 @@ private[spark] class HttpServer(resourceBase: File, securityManager: SecurityMan
     if (server == null) {
       throw new ServerStateException("Server is not started")
     } else {
-      return "http://" + Utils.localIpAddress + ":" + port
+      "http://" + Utils.localIpAddress + ":" + port
     }
   }
 }

core/src/main/scala/org/apache/spark/SecurityManager.scala

@@ -41,10 +41,19 @@ import org.apache.spark.deploy.SparkHadoopUtil
  * secure the UI if it has data that other users should not be allowed to see. The javax
  * servlet filter specified by the user can authenticate the user and then once the user
  * is logged in, Spark can compare that user versus the view acls to make sure they are
- * authorized to view the UI. The configs 'spark.ui.acls.enable' and 'spark.ui.view.acls'
+ * authorized to view the UI. The configs 'spark.acls.enable' and 'spark.ui.view.acls'
  * control the behavior of the acls. Note that the person who started the application
  * always has view access to the UI.
  *
+ * Spark has a set of modify acls (`spark.modify.acls`) that controls which users have permission
+ * to  modify a single application. This would include things like killing the application. By
+ * default the person who started the application has modify access. For modify access through
+ * the UI, you must have a filter that does authentication in place for the modify acls to work
+ * properly.
+ *
+ * Spark also has a set of admin acls (`spark.admin.acls`) which is a set of users/administrators
+ * who always have permission to view or modify the Spark application.
+ *
  * Spark does not currently support encryption after authentication.
  *
  * At this point spark has multiple communication protocols that need to be secured and
@@ -137,18 +146,32 @@ private[spark] class SecurityManager(sparkConf: SparkConf) extends Logging {
   private val sparkSecretLookupKey = "sparkCookie"
 
   private val authOn = sparkConf.getBoolean("spark.authenticate", false)
-  private var uiAclsOn = sparkConf.getBoolean("spark.ui.acls.enable", false)
+  // keep spark.ui.acls.enable for backwards compatibility with 1.0
+  private var aclsOn = sparkConf.getOption("spark.acls.enable").getOrElse(
+    sparkConf.get("spark.ui.acls.enable", "false")).toBoolean
+
+  // admin acls should be set before view or modify acls
+  private var adminAcls: Set[String] =
+    stringToSet(sparkConf.get("spark.admin.acls", ""))
 
   private var viewAcls: Set[String] = _
+
+  // list of users who have permission to modify the application. This should
+  // apply to both UI and CLI for things like killing the application.
+  private var modifyAcls: Set[String] = _
+
   // always add the current user and SPARK_USER to the viewAcls
-  private val defaultAclUsers = Seq[String](System.getProperty("user.name", ""),
+  private val defaultAclUsers = Set[String](System.getProperty("user.name", ""),
     Option(System.getenv("SPARK_USER")).getOrElse(""))
+
   setViewAcls(defaultAclUsers, sparkConf.get("spark.ui.view.acls", ""))
+  setModifyAcls(defaultAclUsers, sparkConf.get("spark.modify.acls", ""))
 
   private val secretKey = generateSecretKey()
   logInfo("SecurityManager: authentication " + (if (authOn) "enabled" else "disabled") +
-    "; ui acls " + (if (uiAclsOn) "enabled" else "disabled") +
-    "; users with view permissions: " + viewAcls.toString())
+    "; ui acls " + (if (aclsOn) "enabled" else "disabled") +
+    "; users with view permissions: " + viewAcls.toString() +
+    "; users with modify permissions: " + modifyAcls.toString())
 
   // Set our own authenticator to properly negotiate user/password for HTTP connections.
   // This is needed by the HTTP client fetching from the HttpServer. Put here so its
@@ -169,18 +192,51 @@ private[spark] class SecurityManager(sparkConf: SparkConf) extends Logging {
     )
   }
 
-  private[spark] def setViewAcls(defaultUsers: Seq[String], allowedUsers: String) {
-    viewAcls = (defaultUsers ++ allowedUsers.split(',')).map(_.trim()).filter(!_.isEmpty).toSet 
+  /**
+   * Split a comma separated String, filter out any empty items, and return a Set of strings
+   */
+  private def stringToSet(list: String): Set[String] = {
+    list.split(',').map(_.trim).filter(!_.isEmpty).toSet
+  }
+
+  /**
+   * Admin acls should be set before the view or modify acls.  If you modify the admin
+   * acls you should also set the view and modify acls again to pick up the changes.
+   */
+  def setViewAcls(defaultUsers: Set[String], allowedUsers: String) {
+    viewAcls = (adminAcls ++ defaultUsers ++ stringToSet(allowedUsers))
     logInfo("Changing view acls to: " + viewAcls.mkString(","))
   }
 
-  private[spark] def setViewAcls(defaultUser: String, allowedUsers: String) {
-    setViewAcls(Seq[String](defaultUser), allowedUsers)
+  def setViewAcls(defaultUser: String, allowedUsers: String) {
+    setViewAcls(Set[String](defaultUser), allowedUsers)
+  }
+
+  def getViewAcls: String = viewAcls.mkString(",")
+
+  /**
+   * Admin acls should be set before the view or modify acls.  If you modify the admin
+   * acls you should also set the view and modify acls again to pick up the changes.
+   */
+  def setModifyAcls(defaultUsers: Set[String], allowedUsers: String) {
+    modifyAcls = (adminAcls ++ defaultUsers ++ stringToSet(allowedUsers))
+    logInfo("Changing modify acls to: " + modifyAcls.mkString(","))
+  }
+
+  def getModifyAcls: String = modifyAcls.mkString(",")
+
+  /**
+   * Admin acls should be set before the view or modify acls.  If you modify the admin
+   * acls you should also set the view and modify acls again to pick up the changes.
+   */
+  def setAdminAcls(adminUsers: String) {
+    adminAcls = stringToSet(adminUsers)
+    logInfo("Changing admin acls to: " + adminAcls.mkString(","))
   }
 
-  private[spark] def setUIAcls(aclSetting: Boolean) { 
-    uiAclsOn = aclSetting 
-    logInfo("Changing acls enabled to: " + uiAclsOn)
+  def setAcls(aclSetting: Boolean) {
+    aclsOn = aclSetting
+    logInfo("Changing acls enabled to: " + aclsOn)
   }
 
   /**
@@ -224,22 +280,39 @@ private[spark] class SecurityManager(sparkConf: SparkConf) extends Logging {
    * Check to see if Acls for the UI are enabled
    * @return true if UI authentication is enabled, otherwise false
    */
-  def uiAclsEnabled(): Boolean = uiAclsOn
+  def aclsEnabled(): Boolean = aclsOn
 
   /**
    * Checks the given user against the view acl list to see if they have
-   * authorization to view the UI. If the UI acls must are disabled
-   * via spark.ui.acls.enable, all users have view access.
+   * authorization to view the UI. If the UI acls are disabled
+   * via spark.acls.enable, all users have view access. If the user is null
+   * it is assumed authentication is off and all users have access.
    *
    * @param user to see if is authorized
    * @return true is the user has permission, otherwise false
    */
   def checkUIViewPermissions(user: String): Boolean = {
-    logDebug("user=" + user + " uiAclsEnabled=" + uiAclsEnabled() + " viewAcls=" + 
+    logDebug("user=" + user + " aclsEnabled=" + aclsEnabled() + " viewAcls=" +
       viewAcls.mkString(","))
-    if (uiAclsEnabled() && (user != null) && (!viewAcls.contains(user))) false else true
+    if (aclsEnabled() && (user != null) && (!viewAcls.contains(user))) false else true
   }
 
+  /**
+   * Checks the given user against the modify acl list to see if they have
+   * authorization to modify the application. If the UI acls are disabled
+   * via spark.acls.enable, all users have modify access. If the user is null
+   * it is assumed authentication isn't turned on and all users have access.
+   *
+   * @param user to see if is authorized
+   * @return true is the user has permission, otherwise false
+   */
+  def checkModifyPermissions(user: String): Boolean = {
+    logDebug("user=" + user + " aclsEnabled=" + aclsEnabled() + " modifyAcls=" +
+      modifyAcls.mkString(","))
+    if (aclsEnabled() && (user != null) && (!modifyAcls.contains(user))) false else true
+  }
+
+
   /**
    * Check to see if authentication for the Spark communication protocols is enabled
    * @return true if authentication is enabled, otherwise false

core/src/main/scala/org/apache/spark/SparkConf.scala

@@ -323,6 +323,14 @@ private[spark] object SparkConf {
    * the scheduler, while the rest of the spark configs can be inherited from the driver later.
    */
   def isExecutorStartupConf(name: String): Boolean = {
-    isAkkaConf(name) || name.startsWith("spark.akka") || name.startsWith("spark.auth")
+    isAkkaConf(name) ||
+    name.startsWith("spark.akka") ||
+    name.startsWith("spark.auth") ||
+    isSparkPortConf(name)
   }
+
+  /**
+   * Return whether the given config is a Spark port config.
+   */
+  def isSparkPortConf(name: String): Boolean = name.startsWith("spark.") && name.endsWith(".port")
 }