Ver Fonte

HUE-2894 [livy] Shut down and server repl if process dies or exception thrown in server.

Erick Tryzelaar há 10 anos atrás
pai
commit
2c07ba7

+ 16 - 10
apps/spark/java/livy-repl/src/main/scala/com/cloudera/hue/livy/repl/Main.scala

@@ -101,19 +101,25 @@ class ScalatraBootstrap extends LifeCycle with Logging {
   var session: Session = null
 
   override def init(context: ServletContext): Unit = {
-    session = context.getInitParameter(Main.SESSION_KIND) match {
-      case Main.PYSPARK_SESSION => PythonSession.create()
-      case Main.SPARK_SESSION => SparkSession.create()
-      case Main.SPARKR_SESSION => SparkRSession.create()
-    }
+    try {
+      session = context.getInitParameter(Main.SESSION_KIND) match {
+        case Main.PYSPARK_SESSION => PythonSession.create()
+        case Main.SPARK_SESSION => SparkSession.create()
+        case Main.SPARKR_SESSION => SparkRSession.create()
+      }
 
-    context.mount(new WebApp(session), "/*")
+      context.mount(new WebApp(session), "/*")
 
-    val callbackUrl = Option(System.getProperty("livy.repl.callback-url"))
-      .orElse(sys.env.get("LIVY_CALLBACK_URL"))
+      val callbackUrl = Option(System.getProperty("livy.repl.callback-url"))
+        .orElse(sys.env.get("LIVY_CALLBACK_URL"))
 
-    // See if we want to notify someone that we've started on a url
-    callbackUrl.foreach(notifyCallback)
+      // See if we want to notify someone that we've started on a url
+      callbackUrl.foreach(notifyCallback)
+    } catch {
+      case e: Throwable =>
+        println(f"Exception thrown when initializing server: $e")
+        sys.exit(1)
+    }
   }
 
   override def destroy(context: ServletContext): Unit = {

+ 18 - 0
apps/spark/java/livy-repl/src/main/scala/com/cloudera/hue/livy/repl/process/ProcessInterpreter.scala

@@ -167,4 +167,22 @@ abstract class ProcessInterpreter(process: Process)
         throw e
     }
   }
+
+
+  private[this] val processWatcherThread = new Thread("process watcher thread") {
+    override def run() = {
+      val exitCode = process.waitFor()
+      if (exitCode != 0) {
+        _state = Error()
+
+        // Give livy-server a moment to see that we've died.
+        Thread.sleep(1000)
+
+        System.exit(1)
+      }
+    }
+  }
+
+  processWatcherThread.setDaemon(true)
+  processWatcherThread.start()
 }

+ 19 - 20
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/Main.scala

@@ -127,30 +127,29 @@ class ScalatraBootstrap extends LifeCycle with Logging {
   var batchManager: SessionManager[BatchSession] = null
 
   override def init(context: ServletContext): Unit = {
-    val livyConf = new LivyConf()
-
-    val sessionFactoryKind = try {
-      livyConf.sessionKind()
-    } catch {
-      case e: IllegalStateException =>
-        println(f"Unknown session factory: $e}")
-        sys.exit(1)
-    }
+    try {
+      val livyConf = new LivyConf()
+      val sessionFactoryKind = livyConf.sessionKind()
 
-    info(f"Using $sessionFactoryKind sessions")
+      info(f"Using $sessionFactoryKind sessions")
 
-    val (sessionFactory, batchFactory) = sessionFactoryKind match {
-      case LivyConf.Process() =>
-        (new InteractiveSessionProcessFactory(livyConf), new BatchSessionProcessFactory(livyConf))
-      case LivyConf.Yarn() =>
-        (new InteractiveSessionYarnFactory(livyConf), new BatchSessionYarnFactory(livyConf))
-    }
+      val (sessionFactory, batchFactory) = sessionFactoryKind match {
+        case LivyConf.Process() =>
+          (new InteractiveSessionProcessFactory(livyConf), new BatchSessionProcessFactory(livyConf))
+        case LivyConf.Yarn() =>
+          (new InteractiveSessionYarnFactory(livyConf), new BatchSessionYarnFactory(livyConf))
+      }
 
-    sessionManager = new SessionManager(sessionFactory)
-    batchManager = new SessionManager(batchFactory)
+      sessionManager = new SessionManager(sessionFactory)
+      batchManager = new SessionManager(batchFactory)
 
-    context.mount(new InteractiveSessionServlet(sessionManager), "/sessions/*")
-    context.mount(new BatchSessionServlet(batchManager), "/batches/*")
+      context.mount(new InteractiveSessionServlet(sessionManager), "/sessions/*")
+      context.mount(new BatchSessionServlet(batchManager), "/batches/*")
+    } catch {
+      case e: Throwable =>
+        println(f"Exception thrown when initializing server: $e")
+        sys.exit(1)
+    }
   }
 
   override def destroy(context: ServletContext): Unit = {