Merge pull request spark-jobserver#427 from Atigeo/xPatterns_6.0

Support user impersonation for an already Kerberos authenticated user

Author: velvia

README.md

@@ -592,6 +592,10 @@ or in the job config when using POST /jobs,
         spark.cores.max = 10
     }
 
+User impersonation for an already Kerberos authenticated user is supported via `spark.proxy.user` query param:
+  
+  POST /contexts/my-new-context?spark.proxy.user=<user-to-impersonate>
+
 To pass settings directly to the sparkConf that do not use the "spark." prefix "as-is", use the "passthrough" section.
 
     spark.context-settings {

bin/manager_start.sh

@@ -1,6 +1,6 @@
 #!/bin/bash
 # Script to start the job manager
-# args: <work dir for context> <cluster address>
+# args: <work dir for context> <cluster address> [proxy_user]
 set -e
 
 get_abs_script_path() {
@@ -27,10 +27,18 @@ JAVA_OPTS="-XX:MaxDirectMemorySize=$MAX_DIRECT_MEMORY
 
 MAIN="spark.jobserver.JobManager"
 
-cmd='$SPARK_HOME/bin/spark-submit --class $MAIN --driver-memory $JOBSERVER_MEMORY
+if [ ! -z $3 ]; then
+  cmd='$SPARK_HOME/bin/spark-submit --class $MAIN --driver-memory $JOBSERVER_MEMORY
   --conf "spark.executor.extraJavaOptions=$LOGGING_OPTS"
+  --proxy-user $3
   --driver-java-options "$GC_OPTS $JAVA_OPTS $LOGGING_OPTS $CONFIG_OVERRIDES"
   $appdir/spark-job-server.jar $1 $2 $conffile'
+else
+  cmd='$SPARK_HOME/bin/spark-submit --class $MAIN --driver-memory $JOBSERVER_MEMORY
+  --conf "spark.executor.extraJavaOptions=$LOGGING_OPTS"
+  --driver-java-options "$GC_OPTS $JAVA_OPTS $LOGGING_OPTS $CONFIG_OVERRIDES"
+  $appdir/spark-job-server.jar $1 $2 $conffile'
+fi
 
 eval $cmd > /dev/null 2>&1 &
 # exec java -cp $CLASSPATH $GC_OPTS $JAVA_OPTS $LOGGING_OPTS $CONFIG_OVERRIDES $MAIN $1 $2 $conffile 2>&1 &

job-server/src/spark.jobserver/AkkaClusterSupervisorActor.scala

@@ -204,7 +204,15 @@ class AkkaClusterSupervisorActor(daoActor: ActorRef) extends InstrumentedActor {
           failureFunc(e)
           return
       }
-    val pb = Process(s"$managerStartCommand $contextDir ${selfAddress.toString}")
+
+    //extract spark.proxy.user from contextConfig, if available and pass it to $managerStartCommand
+    var cmdString = s"$managerStartCommand $contextDir ${selfAddress.toString}"
+
+    if (contextConfig.hasPath("spark.proxy.user")) {
+      cmdString = cmdString + s" ${contextConfig.getString("spark.proxy.user")}"
+    }
+
+    val pb = Process(cmdString)
     val pio = new ProcessIO(_ => (),
                         stdout => scala.io.Source.fromInputStream(stdout)
                           .getLines.foreach(println),