Procházet zdrojové kódy

HUE-2832 [livy] Make session timeout configurable

Erick Tryzelaar před 10 roky
rodič
revize
15b1d8c

+ 10 - 6
apps/spark/java/conf/livy-defaults.conf.tmpl

@@ -1,20 +1,24 @@
 # What host address to start the server on. Defaults to 0.0.0.0. If using the
 # `yarn` factory mode, this address must be accessible from the YARN nodes.
-# livy.server.host = 0.0.0.0
+## livy.server.host = 0.0.0.0
 
 # What port to start the server on. Defaults to 8998.
-# livy.server.port = 8998
+## livy.server.port = 8998
 
 # What session factory to use. The options are `process` and `yarn`.
-# livy.server.session.factory = process
+## livy.server.session.factory = process
 
 # What spark-submit executable path to use to submit spark applications. Defaults to
 # `spark-submit`.
-# livy.server.spark-submit = spark-submit
+## livy.server.spark-submit = spark-submit
+
+# Time in milliseconds on how long Livy will wait before timing out an idle session.
+# Default is one hour.
+## livy.server.session.timeout = 3600000
 
 # Location to find the livy assembly. If not specified, livy will determine the
 # assembly from the local jarfile. If using `yarn` sessions, this may be on HDFS.
-# livy.yarn.jar = hdfs://localhost:8020/user/hue/share/lib/livy-assembly.jar
+## livy.yarn.jar = hdfs://localhost:8020/user/hue/share/lib/livy-assembly.jar
 
 # If livy should use proxy users when submitting a job.
-# livy.impersonation.enabled = true
+## livy.impersonation.enabled = true

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

@@ -140,8 +140,8 @@ class ScalatraBootstrap extends LifeCycle with Logging {
           (new InteractiveSessionYarnFactory(livyConf), new BatchSessionYarnFactory(livyConf))
       }
 
-      sessionManager = new SessionManager(sessionFactory)
-      batchManager = new SessionManager(batchFactory)
+      sessionManager = new SessionManager(livyConf, sessionFactory)
+      batchManager = new SessionManager(livyConf, batchFactory)
 
       context.mount(new InteractiveSessionServlet(sessionManager), "/sessions/*")
       context.mount(new BatchSessionServlet(batchManager), "/batches/*")

+ 10 - 18
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/SessionManager.scala

@@ -20,31 +20,23 @@ package com.cloudera.hue.livy.server
 
 import java.util.concurrent.atomic.AtomicInteger
 
-import com.cloudera.hue.livy.Logging
+import com.cloudera.hue.livy.{LivyConf, Logging}
 import org.json4s.JValue
 
 import scala.collection.mutable
 import scala.concurrent.{ExecutionContext, Future}
 
-object SessionManager {
-  // Time in milliseconds; TODO: make configurable
-  val TIMEOUT = 60000
-
-  // Time in milliseconds; TODO: make configurable
-  val GC_PERIOD = 1000 * 60 * 60
-}
-
-class SessionManager[S <: Session](factory: SessionFactory[S])
+class SessionManager[S <: Session](livyConf: LivyConf, factory: SessionFactory[S])
   extends Logging {
 
-  import SessionManager._
-
   private implicit def executor: ExecutionContext = ExecutionContext.global
 
-  protected[this] val _idCounter = new AtomicInteger()
-  protected[this] val _sessions = mutable.Map[Int, S]()
+  private[this] val _idCounter = new AtomicInteger()
+  private[this] val _sessions = mutable.Map[Int, S]()
 
-  private val garbageCollector = new GarbageCollector
+  private[this] val sessionTimeout = livyConf.getInt("livy.server.session.timeout", 1000 * 60 * 60)
+  private[this] val garbageCollector = new GarbageCollector
+  garbageCollector.setDaemon(true)
   garbageCollector.start()
 
   def create(createRequest: JValue): S = {
@@ -84,7 +76,7 @@ class SessionManager[S <: Session](factory: SessionFactory[S])
   def collectGarbage() = {
     def expired(session: Session): Boolean = {
       session.lastActivity match {
-        case Some(lastActivity) => System.currentTimeMillis() - lastActivity > TIMEOUT
+        case Some(lastActivity) => System.currentTimeMillis() - lastActivity > sessionTimeout
         case None => false
       }
     }
@@ -92,14 +84,14 @@ class SessionManager[S <: Session](factory: SessionFactory[S])
     all().filter(expired).foreach(delete)
   }
 
-  private class GarbageCollector extends Thread {
+  private class GarbageCollector extends Thread("session gc thread") {
 
     private var finished = false
 
     override def run(): Unit = {
       while (!finished) {
         collectGarbage()
-        Thread.sleep(SessionManager.GC_PERIOD)
+        Thread.sleep(60 * 1000)
       }
     }
 

+ 3 - 2
apps/spark/java/livy-server/src/test/scala/com/cloudera/hue/livy/server/batches/BatchServletSpec.scala

@@ -54,8 +54,9 @@ class BatchServletSpec extends ScalatraSuite with FunSpecLike with BeforeAndAfte
     script
   }
 
-  val batchFactory = new BatchSessionProcessFactory(new LivyConf())
-  val batchManager = new SessionManager(batchFactory)
+  val livyConf = new LivyConf()
+  val batchFactory = new BatchSessionProcessFactory(livyConf)
+  val batchManager = new SessionManager(livyConf, batchFactory)
   val servlet = new BatchSessionServlet(batchManager)
 
   addServlet(servlet, "/*")

+ 3 - 1
apps/spark/java/livy-server/src/test/scala/com/cloudera/hue/livy/server/interactive/InteractiveSessionServletSpec.scala

@@ -21,6 +21,7 @@ package com.cloudera.hue.livy.server.interactive
 import java.net.URL
 import java.util.concurrent.atomic.AtomicInteger
 
+import com.cloudera.hue.livy.LivyConf
 import com.cloudera.hue.livy.msgs.ExecuteRequest
 import com.cloudera.hue.livy.server.SessionManager
 import com.cloudera.hue.livy.sessions._
@@ -80,7 +81,8 @@ class InteractiveSessionServletSpec extends ScalatraSuite with FunSpecLike {
     }
   }
 
-  val sessionManager = new SessionManager(new MockInteractiveSessionFactory())
+  val livyConf = new LivyConf()
+  val sessionManager = new SessionManager(livyConf, new MockInteractiveSessionFactory())
   val servlet = new InteractiveSessionServlet(sessionManager)
 
   addServlet(servlet, "/*")