fix(job-server): fix for 916,917: logging for manager_start.sh (spark-jobserver#918)

fix for 916,917: logging for manager_start.sh

* Add logging when starting context through manager_start.sh
* manager_start.sh process is now fully asynchronous
* spark-submit logs routed to main application logs
* removed wait-for-manager-start

Author: marcoboi

bin/manager_start.sh

@@ -48,5 +48,5 @@ else
   $appdir/spark-job-server.jar $2 $3 $4 $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 &
+eval $cmd
+

job-server/config/docker.conf

@@ -65,7 +65,6 @@ spark {
 
 deploy {
   manager-start-cmd = "app/manager_start.sh"
-  wait-for-manager-start = false
 }
 
 # Note that you can use this file to define settings not only for job server,

job-server/src/main/resources/application.conf

@@ -243,7 +243,4 @@ shiro {
 
 deploy {
   manager-start-cmd = "./manager_start.sh"
-  # If true, wait for the process which starts contexts in a forked JVM to exit. Set to false
-  # if this process needs to remain in the foreground, e.g. when running in Docker.
-  wait-for-manager-start = true
 }

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

@@ -1,6 +1,5 @@
 package spark.jobserver
 
-import java.io.IOException
 import java.nio.file.{Files, Paths}
 import java.nio.charset.Charset
 import java.util.concurrent.TimeUnit
@@ -20,6 +19,7 @@ import spark.jobserver.common.akka.InstrumentedActor
 import scala.concurrent.Await
 import akka.pattern.gracefulStop
 import org.joda.time.DateTime
+import org.slf4j.LoggerFactory
 import spark.jobserver.io.JobDAOActor.CleanContextJobInfos
 
 /**
@@ -37,7 +37,6 @@ import spark.jobserver.io.JobDAOActor.CleanContextJobInfos
  * {{{
  *   deploy {
  *     manager-start-cmd = "./manager_start.sh"
- *     wait-for-manager-start = true
  *   }
  * }}}
  */
@@ -54,7 +53,6 @@ class AkkaClusterSupervisorActor(daoActor: ActorRef, dataManagerActor: ActorRef)
                                                 TimeUnit.SECONDS)
   val contextDeletionTimeout = SparkJobUtils.getContextDeletionTimeout(config)
   val managerStartCommand = config.getString("deploy.manager-start-cmd")
-  val waitForManagerStart = config.getBoolean("deploy.wait-for-manager-start")
   import context.dispatcher
 
   //actor name -> (context isadhoc, success callback, failure callback)
@@ -238,28 +236,11 @@ class AkkaClusterSupervisorActor(daoActor: ActorRef, dataManagerActor: ActorRef)
       cmdString = cmdString + s" ${contextConfig.getString(SparkJobUtils.SPARK_PROXY_USER_PARAM)}"
     }
 
-    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)
-      if (waitForManagerStart) {
-        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)
-    }
+    val contextLogger = LoggerFactory.getLogger("manager_start")
+    val process = Process(cmdString.split(" "))
+    process.run(ProcessLogger(out => contextLogger.info(out), err => contextLogger.warn(err)))
 
+    contextInitInfos(contextActorName) = (isAdHoc, successFunc, failureFunc)
   }
 
   private def createContextDir(name: String,