Browse Source

[livy] ProcessSession should read output to determine the repl url

This allows us to add a ProcessSessionSpec test
Erick Tryzelaar 10 years ago
parent
commit
7c75d278fa

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

@@ -60,9 +60,11 @@ object Main extends Logging {
 
 
     try {
     try {
       val replUrl = s"http://${server.host}:${server.port}"
       val replUrl = s"http://${server.host}:${server.port}"
-      println(s"Starting livy-repl on $replUrl")
       System.setProperty("livy.repl.url", replUrl)
       System.setProperty("livy.repl.url", replUrl)
 
 
+      println(s"Starting livy-repl on $replUrl")
+      Console.flush()
+
       server.join()
       server.join()
       server.stop()
       server.stop()
     } finally {
     } finally {

+ 1 - 1
apps/spark/java/livy-server/pom.xml

@@ -167,7 +167,7 @@
                         <spark.driver.allowMultipleContexts>true</spark.driver.allowMultipleContexts>
                         <spark.driver.allowMultipleContexts>true</spark.driver.allowMultipleContexts>
                         <spark.ui.enabled>false</spark.ui.enabled>
                         <spark.ui.enabled>false</spark.ui.enabled>
                         <settings.usejavacp.value>true</settings.usejavacp.value>
                         <settings.usejavacp.value>true</settings.usejavacp.value>
-                        <livy.assembly.jar>../livy-repl/target/livy-repl_${scala.binary.version}-${project.version}.jar</livy.assembly.jar>
+                        <livy.repl.jar>../livy-repl/target/livy-repl_${scala.binary.version}-${project.version}.jar</livy.repl.jar>
                     </systemProperties>
                     </systemProperties>
                 </configuration>
                 </configuration>
             </plugin>
             </plugin>

+ 42 - 2
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/sessions/ProcessSession.scala

@@ -1,12 +1,16 @@
 package com.cloudera.hue.livy.server.sessions
 package com.cloudera.hue.livy.server.sessions
 
 
 import java.lang.ProcessBuilder.Redirect
 import java.lang.ProcessBuilder.Redirect
+import java.net.URL
 
 
 import com.cloudera.hue.livy.{LivyConf, Logging, Utils}
 import com.cloudera.hue.livy.{LivyConf, Logging, Utils}
 
 
+import scala.annotation.tailrec
 import scala.collection.JavaConversions._
 import scala.collection.JavaConversions._
 import scala.collection.mutable.ArrayBuffer
 import scala.collection.mutable.ArrayBuffer
 import scala.concurrent.Future
 import scala.concurrent.Future
+import scala.io.Source
+import scala.util.control.Breaks._
 
 
 object ProcessSession extends Logging {
 object ProcessSession extends Logging {
 
 
@@ -48,8 +52,7 @@ object ProcessSession extends Logging {
 
 
     pb.environment().put("LIVY_PORT", "0")
     pb.environment().put("LIVY_PORT", "0")
 
 
-
-    pb.redirectOutput(Redirect.INHERIT)
+    pb.redirectOutput(Redirect.PIPE)
     pb.redirectError(Redirect.INHERIT)
     pb.redirectError(Redirect.INHERIT)
 
 
     pb.start()
     pb.start()
@@ -64,10 +67,47 @@ object ProcessSession extends Logging {
 
 
 private class ProcessSession(id: String, process: Process) extends WebSession(id) {
 private class ProcessSession(id: String, process: Process) extends WebSession(id) {
 
 
+  val stdoutThread = new Thread {
+    override def run() = {
+      val regex = """Starting livy-repl on (https?://.*)""".r
+
+      val lines = Source.fromInputStream(process.getInputStream).getLines()
+
+      // Loop until we find the ip address to talk to livy-repl.
+      @tailrec
+      def readUntilURL(): Boolean = {
+        if (lines.hasNext) {
+          val line = lines.next()
+          println(line)
+
+          line match {
+            case regex(url_) =>
+              url = new URL(url_)
+              true
+            case _ => readUntilURL()
+          }
+        } else {
+          false
+        }
+      }
+
+      if (readUntilURL()) {
+        for (line <- lines) {
+          println(line)
+        }
+      }
+    }
+  }
+
+  stdoutThread.setName("process session stdout reader")
+  stdoutThread.setDaemon(true)
+  stdoutThread.start()
+
   override def stop(): Future[Unit] = {
   override def stop(): Future[Unit] = {
     super.stop() andThen { case r =>
     super.stop() andThen { case r =>
       // Make sure the process is reaped.
       // Make sure the process is reaped.
       process.waitFor()
       process.waitFor()
+      stdoutThread.join()
 
 
       r
       r
     }
     }

+ 14 - 0
apps/spark/java/livy-server/src/test/scala/com/cloudera/hue/livy/server/ProcessSessionSpec.scala

@@ -0,0 +1,14 @@
+package com.cloudera.hue.livy.server
+
+import com.cloudera.hue.livy.LivyConf
+import com.cloudera.hue.livy.server.sessions.{ProcessSession, Session}
+import org.scalatest.matchers.ShouldMatchers
+import org.scalatest.{Matchers, FunSpecLike, BeforeAndAfter, FunSpec}
+
+class ProcessSessionSpec extends BaseSessionSpec with FunSpecLike with Matchers with BeforeAndAfter {
+
+  val livyConf = new LivyConf()
+  livyConf.set("livy.repl.driverClassPath", sys.props("java.class.path"))
+
+  def createSession() = ProcessSession.create(livyConf, "0", "spark")
+}