fix(job-server-api,job-server,akka-app): fixed starting a new job does not determine proxy user issue spark-jobserver#827 (spark-jobserver#854)

Issues:

Starting a new job does not determine proxy user spark-jobserver#827
Fixes:

Added determine proxy user code to adHoc context creation and job starting
Pull Request checklist:

 The commit(s) message(s) follows the contribution guidelines ?
 Tests for the changes have been added (for bug fixes / features) ?
[No] Docs have been added / updated (for bug fixes / features) ?
Current behavior :

After context is created by proxy-user then starting a job for such a context fails or the context name has to be prepended with user name and separator (~)
New behavior :

After context is created by proxy-user then starting a job for such a context with just its name succeeds,
AdHoc context names are internally prepended with user name and separator (~)

* fix(job-server-api,job-server,akka-app): changes starting SparkJobs with adHoc and explicit context
* Move userNamePrefix, removeProxyUserPrefix, NameContextDelimiter and regRexPart2 from WebApi to SparkJobUtils
* Change jobRoutes implementation in WebApi, so it checks proxy user
* Change adHoc context starting in AkkaClusterSupervisorActor and LocalContextSupervisorActor, so it handles proxy user
* Add tests to LocalContextSupervisorSpec covering adHoc context starting
* Change WebApiWithAuthenticationSpec test as a consequence of moving class members from WebApi to SparkJobUtils
* Added shiro.basic.ini, WebApiSecuredSpec and WebApiSecuredRoutesSpec to job-server/tests

Author: zromi18

.gitignore

@@ -15,7 +15,6 @@ config/*.conf
 config/*.sh
 config/*.ini
 job-server/config/*.conf
-job-server/config/*.sh
 job-server/config/*.ini
 job-server-extras/spark-warehouse/
 metastore_db/

README.md

@@ -89,7 +89,7 @@ Spark Job Server is now included in Datastax Enterprise 4.8!
 
 - *"Spark as a Service"*: Simple REST interface (including HTTPS) for all aspects of job, context management
 - Support for Spark SQL, Hive, Streaming Contexts/jobs and custom job contexts!  See [Contexts](doc/contexts.md).
-- [Python](doc/python.md), Scala, and Java (see [TestJob.java](https://github.com/spark-jobserver/spark-jobserver/blob/master/job-server-api/src/main/java/spark/jobserver/api/TestJob.java)) support
+- [Python](doc/python.md), Scala, and [Java](doc/javaapi.md) (see [TestJob.java](https://github.com/spark-jobserver/spark-jobserver/blob/master/job-server-api/src/main/java/spark/jobserver/api/TestJob.java)) support
 - LDAP Auth support via Apache Shiro integration
 - Separate JVM per SparkContext for isolation (EXPERIMENTAL)
 - Supports sub-second low-latency jobs via long-running job contexts
@@ -602,7 +602,7 @@ jdbc {
         user = "secret"
         password = "secret"
       }
-```      
+```
 
 ### Chef
 
@@ -629,14 +629,14 @@ Flow diagrams are checked in in the doc/ subdirectory.  .diagram files are for w
     GET /binaries               - lists all current binaries
     POST /binaries/<appName>    - upload a new binary file
     DELETE /binaries/<appName>  - delete defined binary
-    
+
 When POSTing new binaries, the content-type header must be set to one of the types supported by the subclasses of the `BinaryType` trait. e.g. "application/java-archive" or application/python-archive"
 
 ### Jars (deprecated)
 
     GET /jars                   - lists all the jars and the last upload timestamp
     POST /jars/<appName>        - uploads a new jar under <appName>
-    
+
 These routes are kept for legacy purposes but are deprecated in favour of the /binaries routes
 
 ### Contexts
@@ -807,8 +807,8 @@ for instance: `sbt ++2.11.6 job-server/compile`
    - You may need to set `SPARK_LOCAL_IP` to `localhost` to ensure Akka port can bind successfully
    - Note for Windows users: very few tests fail on Windows. Thus, run `testOnly -- -l WindowsIgnore` from SBT shell to ignore them.
 - Logging for tests goes to "job-server-test.log"
-- Run `scoverage:test` to check the code coverage and improve it. 
-  - Windows users: run `; coverage ; testOnly -- -l WindowsIgnore ; coverageReport` from SBT shell. 
+- Run `scoverage:test` to check the code coverage and improve it.
+  - Windows users: run `; coverage ; testOnly -- -l WindowsIgnore ; coverageReport` from SBT shell.
 - Please run scalastyle to ensure your code changes don't break the style guide.
 - Do "reStart" from SBT for quick restarts of the job server process
 - Please update the g8 template if you change the SparkJob API

akka-app/src/test/scala/spark/jobserver/common/akka/ActorMetricsSpec.scala

@@ -1,14 +1,18 @@
 package spark.jobserver.common.akka
 
-import org.scalatest.{Matchers, FunSpec}
-import akka.testkit.TestActorRef
+import org.scalatest.{BeforeAndAfterAll, Matchers, FunSpec}
+import akka.testkit.{TestKit, TestActorRef}
 
 import akka.actor.{Actor, ActorSystem}
 
 
-class ActorMetricsSpec extends FunSpec with Matchers {
+class ActorMetricsSpec extends FunSpec with Matchers with BeforeAndAfterAll {
   implicit val system = ActorSystem("test")
 
+  override def afterAll(): Unit = {
+    TestKit.shutdownActorSystem(system)
+  }
+
   describe("actor metrics") {
     it("should increment receive count metric when a message is received") {
       val actorRef = TestActorRef(new DummyActor with ActorMetrics)

akka-app/src/test/scala/spark/jobserver/common/akka/ActorStackSpec.scala

@@ -1,7 +1,7 @@
 package spark.jobserver.common.akka
 
-import org.scalatest.{Matchers, FunSpec}
-import akka.testkit.TestActorRef
+import org.scalatest.{BeforeAndAfterAll, Matchers, FunSpec}
+import akka.testkit.{TestKit, TestActorRef}
 
 import akka.actor.{Actor, ActorSystem}
 
@@ -22,9 +22,13 @@ trait AddPrefix extends ActorStack {
   }
 }
 
-class ActorStackSpec extends FunSpec with Matchers {
+class ActorStackSpec extends FunSpec with Matchers with BeforeAndAfterAll {
   implicit val system = ActorSystem("test")
 
+  override def afterAll(): Unit = {
+    TestKit.shutdownActorSystem(system)
+  }
+
   describe("stacking traits") {
     it("should be able to stack traits and receive messages") {
       val actorRef = TestActorRef(new DummyActor with AddPrefix)

akka-app/src/test/scala/spark/jobserver/common/akka/actor/ReaperSpec.scala

@@ -3,6 +3,7 @@ package spark.jobserver.common.akka.actor
 import akka.actor.{ActorSystem, Props, ActorRef}
 import akka.testkit.{TestKit, ImplicitSender, TestProbe}
 import org.scalatest.{MustMatchers, FunSpecLike, BeforeAndAfterAll}
+import spark.jobserver.common.akka.AkkaTestUtils
 
 // Our test reaper.  Sends the snooper a message when all
 // the souls have been reaped
@@ -19,7 +20,7 @@ class ReaperSpec extends TestKit(ActorSystem("ReaperSpec")) with ImplicitSender
   import scala.concurrent.duration._
 
   override def afterAll() {
-    system.shutdown()
+    AkkaTestUtils.shutdownAndWait(system)
   }
 
   describe("Reaper") {

akka-app/src/test/scala/spark/jobserver/common/akka/web/CommonRoutesSpec.scala

@@ -2,6 +2,7 @@ package spark.jobserver.common.akka.web
 
 import java.util.concurrent.TimeUnit
 
+import akka.testkit.TestKit
 import org.scalatest.{Matchers, FunSpec}
 import spray.testkit.ScalatestRouteTest
 
@@ -32,6 +33,10 @@ class CommonRoutesSpec extends FunSpec with Matchers with ScalatestRouteTest wit
   val timerMap = Map("type" -> "timer", "rate" -> (meterMap - "type"),
     "duration" -> (histMap ++ Map("units" -> "milliseconds") - "type"))
 
+  override def afterAll():Unit = {
+    TestKit.shutdownActorSystem(system)
+  }
+
   describe("/metricz route") {
     it("should report all metrics") {
       Get("/metricz") ~> commonRoutes ~> check {

build.sbt

@@ -150,7 +150,7 @@ lazy val dockerSettings = Seq(
                 apt-get -y update && \
                 apt-get -y install mesos=${MESOS_VERSION} && \
                 apt-get clean
-             """)
+        """)
       env("MAVEN_VERSION","3.3.9")
       runRaw(
         """mkdir -p /usr/share/maven /usr/share/maven/ref \
@@ -160,7 +160,7 @@ lazy val dockerSettings = Seq(
         """)
       env("MAVEN_HOME","/usr/share/maven")
       env("MAVEN_CONFIG", "/.m2")
-      
+
       copy(artifact, artifactTargetPath)
       copy(baseDirectory(_ / "bin" / "server_start.sh").value, file("app/server_start.sh"))
       copy(baseDirectory(_ / "bin" / "server_stop.sh").value, file("app/server_stop.sh"))
@@ -192,14 +192,14 @@ lazy val dockerSettings = Seq(
   },
   imageNames in docker := Seq(
     sbtdocker.ImageName(namespace = Some("velvia"),
-                        repository = "spark-jobserver",
-                        tag = Some(
-                          s"${version.value}" +
-                          s".mesos-${Versions.mesos.split('-')(0)}" +
-                          s".spark-${Versions.spark}" +
-                          s".scala-${scalaBinaryVersion.value}" +
-                          s".jdk-${Versions.java}")
-                        )
+      repository = "spark-jobserver",
+      tag = Some(
+        s"${version.value}" +
+          s".mesos-${Versions.mesos.split('-')(0)}" +
+          s".spark-${Versions.spark}" +
+          s".scala-${scalaBinaryVersion.value}" +
+          s".jdk-${Versions.java}")
+    )
   )
 )
 

job-server-api/src/main/scala/spark/jobserver/context/SparkContextFactory.scala

@@ -20,24 +20,24 @@ case object JobWrongType extends LoadingError
 case class JobLoadError(ex: Exception) extends LoadingError
 
 /**
- * Factory trait for creating a SparkContext or any derived Contexts,
- * such as SQLContext, StreamingContext, HiveContext, etc.
- * My implementing classes can be dynamically loaded using classloaders to ensure that the entire
- * SparkContext has access to certain dynamically loaded classes, for example, job jars.
- * Also, this is capable of loading jobs which don't necessarily implement SparkJobBase, or even
- * Python jobs etc., so long as a wrapping SparkJobBase is returned.
- */
+  * Factory trait for creating a SparkContext or any derived Contexts,
+  * such as SQLContext, StreamingContext, HiveContext, etc.
+  * My implementing classes can be dynamically loaded using classloaders to ensure that the entire
+  * SparkContext has access to certain dynamically loaded classes, for example, job jars.
+  * Also, this is capable of loading jobs which don't necessarily implement SparkJobBase, or even
+  * Python jobs etc., so long as a wrapping SparkJobBase is returned.
+  */
 trait SparkContextFactory {
   import SparkJobUtils._
 
   type C <: ContextLike
   type J <: JobContainer
 
   /**
-   * Loads the job of the given appName, version, and class path, and validates that it is
-   * the right type of job given the current context type.  For example, it may load a JAR
-   * and validate the classpath exists and try to invoke its constructor.
-   */
+    * Loads the job of the given appName, version, and class path, and validates that it is
+    * the right type of job given the current context type.  For example, it may load a JAR
+    * and validate the classpath exists and try to invoke its constructor.
+    */
   def loadAndValidateJob(appName: String,
                          uploadTime: DateTime,
                          classPath: String,
@@ -53,12 +53,12 @@ trait SparkContextFactory {
   def makeContext(sparkConf: SparkConf, config: Config, contextName: String): C
 
   /**
-   * Creates a SparkContext or derived context.
-   * @param config the overall system / job server Typesafe Config
-   * @param contextConfig the config specific to this particular context
-   * @param contextName the name of the context to start
-   * @return the newly created context.
-   */
+    * Creates a SparkContext or derived context.
+    * @param config the overall system / job server Typesafe Config
+    * @param contextConfig the config specific to this particular context
+    * @param contextName the name of the context to start
+    * @return the newly created context.
+    */
   def makeContext(config: Config, contextConfig: Config, contextName: String): C = {
     val sparkConf = configToSparkConf(config, contextConfig, contextName)
     makeContext(sparkConf, contextConfig, contextName)
@@ -70,8 +70,8 @@ case class ScalaJobContainer(job: api.SparkJobBase) extends JobContainer {
 }
 
 /**
- * A SparkContextFactory designed for Scala and Java jobs that loads jars
- */
+  * A SparkContextFactory designed for Scala and Java jobs that loads jars
+  */
 trait ScalaContextFactory extends SparkContextFactory {
   type J = ScalaJobContainer
 
@@ -95,12 +95,12 @@ trait ScalaContextFactory extends SparkContextFactory {
   }
 
   /**
-   * Returns true if the job is valid for this context.
-   * At the minimum this should check for if the job can actually take a context of this type;
-   * for example, a SQLContext should only accept jobs that take a SQLContext.
-   * The recommendation is to define a trait for each type of context job;  the standard
-   * [[DefaultSparkContextFactory]] checks to see if the job is of type [[SparkJob]].
-   */
+    * Returns true if the job is valid for this context.
+    * At the minimum this should check for if the job can actually take a context of this type;
+    * for example, a SQLContext should only accept jobs that take a SQLContext.
+    * The recommendation is to define a trait for each type of context job;  the standard
+    * [[DefaultSparkContextFactory]] checks to see if the job is of type [[SparkJob]].
+    */
   def isValidJob(job: api.SparkJobBase): Boolean
 }
 
@@ -130,12 +130,12 @@ trait JavaContextFactory extends SparkContextFactory {
 }
 
 /**
- * The default factory creates a standard SparkContext.
- * In the future if we want to add additional methods, etc. then we can have additional factories.
- * For example a specialized SparkContext to manage RDDs in a user-defined way.
- *
- * If you create your own SparkContextFactory, please make sure it has zero constructor args.
- */
+  * The default factory creates a standard SparkContext.
+  * In the future if we want to add additional methods, etc. then we can have additional factories.
+  * For example a specialized SparkContext to manage RDDs in a user-defined way.
+  *
+  * If you create your own SparkContextFactory, please make sure it has zero constructor args.
+  */
 class DefaultSparkContextFactory extends ScalaContextFactory {
 
   type C = SparkContext with ContextLike

job-server-api/src/main/scala/spark/jobserver/util/SparkJobUtils.scala

@@ -13,12 +13,41 @@ import scala.util.Try
 object SparkJobUtils {
   import collection.JavaConverters._
 
+  val NameContextDelimiter = "~"
+
   /**
    * User impersonation for an already Kerberos authenticated user is supported via the
    * `spark.proxy.user` query param
    */
   val SPARK_PROXY_USER_PARAM = "spark.proxy.user"
 
+  private val regRexPart2 = "([^" + SparkJobUtils.NameContextDelimiter + "]+.*)"
+
+  /**
+    * appends the NameContextDelimiter to the user name and,
+    * if the user name contains the delimiter as well, then it doubles it so that we can be sure
+    * that our prefix is unique
+    */
+  def userNamePrefix(userName: String) : String = {
+    userName.replaceAll(NameContextDelimiter,
+      NameContextDelimiter + NameContextDelimiter) +
+      NameContextDelimiter
+  }
+
+  /**
+    * filter the given context names so that the user may only see his/her own contexts
+    */
+  def removeProxyUserPrefix(userName: => String, contextNames: Seq[String], filter: Boolean): Seq[String] = {
+    if (filter) {
+      val RegExPrefix = ("^" + userNamePrefix(userName) + regRexPart2).r
+      contextNames collect {
+        case RegExPrefix(cName) => cName
+      }
+    } else {
+      contextNames
+    }
+  }
+
   /**
    * Creates a SparkConf for initializing a SparkContext based on various configs.
    * Note that anything in contextConfig with keys beginning with spark. get