Merge pull request #16 from riptano/DSP-15163-dse

DSP-15163 use dse script to start context in separate jvm

bin/manager_start.sh

@@ -14,7 +14,7 @@ get_abs_script_path
 . $appdir/setenv.sh
 
 # Override logging options to provide per-context logging
-LOGGING_OPTS="-Dlog4j.configuration=file:$appdir/log4j-server.properties
+LOGGING_OPTS="$LOGGING_OPTS_FILE
               -DLOG_DIR=$1"
 
 GC_OPTS="-XX:+UseConcMarkSweepGC
@@ -40,5 +40,5 @@ else
   $appdir/spark-job-server.jar $1 $2 $conffile'
 fi
 
-eval $cmd > /dev/null 2>&1 &
+eval $cmd > /dev/null 2>&1
 # exec java -cp $CLASSPATH $GC_OPTS $JAVA_OPTS $LOGGING_OPTS $CONFIG_OVERRIDES $MAIN $1 $2 $conffile 2>&1 &

bin/setenv.sh

@@ -43,8 +43,10 @@ if [ -z "$LOG_DIR" ]; then
 fi
 mkdir -p $LOG_DIR
 
-LOGGING_OPTS="-Dlogback.configurationFile=file:$appdir/logback-server.xml
-              -DLOG_DIR=$LOG_DIR"
+# used in server_start and in manager_start
+LOGGING_OPTS_FILE="-Dlogback.configurationFile=file:$appdir/logback-server.xml"
+
+LOGGING_OPTS="$LOGGING_OPTS_FILE -DLOG_DIR=$LOG_DIR"
 
 # For Mesos
 CONFIG_OVERRIDES="-Dspark.executor.uri=$SPARK_EXECUTOR_URI "

job-server/config/dse.conf

@@ -56,3 +56,8 @@ spark {
 
 # Note that you can use this file to define settings not only for job server,
 # but for your Spark jobs as well.  Spark job configuration merges with this configuration file as defaults.
+
+
+deploy {
+  manager-start-cmd = "dse spark-jobserver context-per-jvm-managed-start"
+}

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

@@ -3,19 +3,23 @@ package spark.jobserver
 import java.io.IOException
 import java.nio.file.{Files, Paths}
 import java.nio.charset.Charset
-import java.util.concurrent.TimeUnit
+import java.util.concurrent.{ExecutorService, Executors, TimeUnit}
 
 import akka.actor._
 import akka.cluster.Cluster
 import akka.cluster.ClusterEvent.{MemberUp, MemberEvent, InitialStateAsEvents}
 import akka.util.Timeout
+import com.google.common.util.concurrent.ThreadFactoryBuilder
 import com.typesafe.config.{Config, ConfigFactory, ConfigRenderOptions}
 import ooyala.common.akka.InstrumentedActor
 import spark.jobserver.util.SparkJobUtils
+
 import scala.collection.mutable
 import scala.util.{Try, Success, Failure}
 import scala.sys.process._
 
+import scala.collection.concurrent.TrieMap
+
 /**
  * The AkkaClusterSupervisorActor launches Spark Contexts as external processes
  * that connect back with the master node via Akka Cluster.
@@ -50,8 +54,10 @@ class AkkaClusterSupervisorActor(daoActor: ActorRef) extends InstrumentedActor {
   //TODO: try to pass this state to the jobManager at start instead of having to track
   //extra state.  What happens if the WebApi process dies before the forked process
   //starts up?  Then it never gets initialized, and this state disappears.
-  private val contextInitInfos = mutable.HashMap.empty[String, (Boolean, ActorRef => Unit, Throwable => Unit)]
-
+  private val contextInitInfos = TrieMap.empty[String, (Boolean, ActorRef => Unit, Throwable => Unit)]
+  private val contextInitExecutorService = Executors.newCachedThreadPool(
+    new ThreadFactoryBuilder().setDaemon(true).setNameFormat("job-server-context-init-thread -% d").build
+  )
   // actor name -> (JobManagerActor ref, ResultActor ref)
   private val contexts = mutable.HashMap.empty[String, (ActorRef, ActorRef)]
 
@@ -212,26 +218,31 @@ class AkkaClusterSupervisorActor(daoActor: ActorRef) extends InstrumentedActor {
       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),
-                        stderr => scala.io.Source.fromInputStream(stderr).getLines().foreach(println))
-    logger.info("Starting to execute sub process {}", pb)
-    val processStart = Try {
-      val process = pb.run(pio)
-      val exitVal = process.exitValue()
-      if (exitVal != 0) {
-        throw new IOException("Failed to launch context process, got exit code " + exitVal)
-      }
-    }
-
-    if (processStart.isSuccess) {
-      contextInitInfos(contextActorName) = (isAdHoc, successFunc, failureFunc)
-    } else {
-      failureFunc(processStart.failed.get)
-    }
+    contextInitInfos(contextActorName) = (isAdHoc, successFunc, failureFunc)
+
+    contextInitExecutorService.submit(new Runnable {
+      override def run(): Unit = {
+        val pb = Process(cmdString)
+        val pio = new ProcessIO(_ => (),
+          stdout => scala.io.Source.fromInputStream(stdout)
+            .getLines.foreach(println),
+          stderr => scala.io.Source.fromInputStream(stderr).getLines().foreach(println))
+
+        logger.info("Starting to execute sub process {}", pb)
+        val processStart = Try {
+          val process = pb.run(pio)
+          val exitVal = process.exitValue()
+          if (exitVal != 0) {
+            throw new IOException("Failed to launch context process, got exit code " + exitVal)
+          }
+        }
 
+        if (processStart.isFailure) {
+          failureFunc(processStart.failed.get)
+          contextInitInfos.remove(contextActorName)
+        }
+      }
+    })
   }
 
   private def createContextDir(name: String,