-
Notifications
You must be signed in to change notification settings - Fork 135
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #5 from spark-jobserver/sc-hang
Code & unit test for detecting hang Spark workers
- Loading branch information
Showing
7 changed files
with
285 additions
and
7 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,90 @@ | ||
| package spark.jobserver | ||
|
|
||
| import akka.actor.ActorSystem | ||
| import akka.io.IO | ||
| import akka.pattern.ask | ||
| import akka.util.Timeout | ||
| import ooyala.common.akka.InstrumentedActor | ||
| import scala.util.{Success, Failure} | ||
| import scala.concurrent.Future | ||
| import spark.jobserver.SparkWebUiActor.{SparkWorkersErrorInfo, SparkWorkersInfo, GetWorkerStatus} | ||
| import spray.can.Http | ||
| import spray.client.pipelining.{Get, sendReceive, SendReceive} | ||
| import spray.http.HttpResponse | ||
|
|
||
| object SparkWebUiActor { | ||
| // Requests | ||
| case class GetWorkerStatus() | ||
|
|
||
| // Responses | ||
| case class SparkWorkersInfo(alive: Int, dead: Int) | ||
| case class SparkWorkersErrorInfo(message :String) | ||
| } | ||
| /** | ||
| * This actor pulls Spark worker status info (ALIVE, DEAD etc) from Spark admin web ui | ||
| * Collecting worker info from HTML page is not ideal. | ||
| * But at this time Spark does not provide public API yet to expose worker status. | ||
| * Also, the current implementation only works for Spark standalone mode | ||
| */ | ||
| class SparkWebUiActor extends InstrumentedActor { | ||
| import actorSystem.dispatcher // execution context for futures | ||
| import scala.concurrent.duration._ | ||
|
|
||
| implicit val actorSystem: ActorSystem = context.system | ||
|
|
||
| // Used in the asks (?) below to request info from contextSupervisor and resultActor | ||
| implicit val shortTimeout = Timeout(3 seconds) | ||
|
|
||
| val config = context.system.settings.config | ||
|
|
||
| val sparkWebHostUrl = getSparkHostName() | ||
| val sparkWebHostPort = config.getInt("spark.webUrlPort") | ||
|
|
||
| val pipeline: Future[SendReceive] = | ||
| for ( | ||
| Http.HostConnectorInfo(connector, _) <- | ||
| IO(Http) ? Http.HostConnectorSetup(sparkWebHostUrl, port = sparkWebHostPort) | ||
| ) yield sendReceive(connector) | ||
|
|
||
| override def postStop() { | ||
| logger.info("Shutting down actor system for SparkWebUiActor") | ||
| } | ||
|
|
||
| override def wrappedReceive: Receive = { | ||
| case GetWorkerStatus() => | ||
| val request = Get("/") | ||
| logger.info("Get the request for spark web UI") | ||
|
|
||
| val theSender = sender | ||
| val responseFuture: Future[HttpResponse] = pipeline.flatMap(_(request)) | ||
| responseFuture onComplete { | ||
| case Success(httpResponse) => | ||
| val content = httpResponse.entity.asString; | ||
|
|
||
| val aliveWorkerNum = "<td>ALIVE</td>".r.findAllIn(content).length | ||
| val deadWorkerNum = "<td>DEAD</td>".r.findAllIn(content).length | ||
|
|
||
| theSender ! SparkWorkersInfo(aliveWorkerNum, deadWorkerNum) | ||
| case Failure(error) => | ||
| val msg = s"Failed to retrieve Spark web UI $sparkWebHostUrl:$sparkWebHostPort" | ||
| logger.error( msg ) | ||
| theSender ! SparkWorkersErrorInfo(msg) | ||
| } | ||
| } | ||
|
|
||
| def getSparkHostName(): String = { | ||
| val master = config.getString("spark.master") | ||
| // Regular expression used for local[N] and local[*] master formats | ||
| val LOCAL_N_REGEX = """local\[([0-9\*]+)\]""".r | ||
| // Regular expression for connecting to Spark deploy clusters | ||
| val SPARK_REGEX = """spark://(.*):.*""".r | ||
|
|
||
| master match { | ||
| case "localhost" | "local" | LOCAL_N_REGEX(_) => "localhost" | ||
| case SPARK_REGEX(sparkUrl) => sparkUrl | ||
| case _ => throw new RuntimeException("Could not parse Master URL: '" + master + "'") | ||
| } | ||
| } | ||
|
|
||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
133 changes: 133 additions & 0 deletions
133
job-server/test/spark.jobserver/SparkWebUiActorSpec.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,133 @@ | ||
| package spark.jobserver | ||
|
|
||
| import akka.actor.{Props, ActorRef, ActorSystem} | ||
| import akka.io.IO | ||
| import akka.pattern.ask | ||
| import akka.testkit.{TestKit, ImplicitSender} | ||
| import com.typesafe.config.ConfigFactory | ||
| import org.scalatest.{FunSpec, BeforeAndAfter, BeforeAndAfterAll} | ||
| import org.scalatest.matchers.ShouldMatchers | ||
| import spray.client.pipelining._ | ||
|
|
||
| import scala.concurrent.{Await, Future} | ||
| import scala.concurrent.duration._ | ||
| import akka.util.Timeout | ||
| import akka.actor._ | ||
| import spray.can.Http | ||
| import spray.http._ | ||
| import HttpMethods._ | ||
| import SparkWebUiActor._ | ||
|
|
||
| import scala.util.{Failure, Success} | ||
|
|
||
| object SparkWebUiActorSpec { | ||
| val sparkWebUrl = "localhost" | ||
| val sparkWebPort = 8098 | ||
| val config = ConfigFactory.parseString(s""" | ||
| spark { | ||
| master = "spark://localhost:7077" | ||
| webUrl = $sparkWebUrl | ||
| webUrlPort = $sparkWebPort | ||
| temp-contexts { | ||
| num-cpu-cores = 4 # Number of cores to allocate. Required. | ||
| memory-per-node = 512m # Executor memory per node, -Xmx style eg 512m, 1G, etc. | ||
| } | ||
| jobserver.job-result-cache-size = 100 | ||
| jobserver.context-creation-timeout = 5 s | ||
| jobserver.context-factory = spark.jobserver.util.DefaultSparkContextFactory | ||
| contexts { | ||
| olap-demo { | ||
| num-cpu-cores = 4 | ||
| memory-per-node = 512m | ||
| } | ||
| } | ||
| context-settings { | ||
| num-cpu-cores = 2 | ||
| memory-per-node = 512m | ||
| } | ||
| } | ||
| akka.log-dead-letters = 0 | ||
| """) | ||
|
|
||
| val system = ActorSystem("test", config) | ||
| } | ||
|
|
||
| /** | ||
| * Created by senqiang on 8/22/14. | ||
| */ | ||
|
|
||
| // simple http service | ||
| class SimpleHttpServer extends Actor with ActorLogging { | ||
| implicit val timeout: Timeout = 1.second // for the actor 'asks' | ||
| import context.dispatcher // ExecutionContext for the futures and scheduler | ||
|
|
||
| def receive = { | ||
| // when a new connection comes in we register ourselves as the connection handler | ||
| case _: Http.Connected => sender ! Http.Register(self) | ||
| case HttpRequest(GET, Uri.Path("/"), _, _, _) => { | ||
| sender ! HttpResponse(entity = | ||
| """<td>ALIVE</td> | ||
| |<td>DEAD</td> | ||
| """.stripMargin) | ||
| } | ||
| case HttpRequest(GET, Uri.Path("/stop"), _, _, _) => | ||
| sender ! HttpResponse(entity = "Shutting down in 1 second ...") | ||
| sender ! Http.Close | ||
| context.system.scheduler.scheduleOnce(1.second) { context.system.shutdown() } | ||
| } | ||
| } | ||
|
|
||
| class SparkWebUiActorSpec extends TestKit(SparkWebUiActorSpec.system) with ImplicitSender | ||
| with FunSpec with ShouldMatchers with BeforeAndAfter with BeforeAndAfterAll { | ||
|
|
||
| // Used in the asks (?) below to request info from contextSupervisor and resultActor | ||
| implicit val ShortTimeout = Timeout(3 seconds) | ||
|
|
||
| override def beforeAll(): Unit = { | ||
| import akka.actor.{ActorSystem, Props} | ||
| import akka.io.IO | ||
| import spray.can.Http | ||
|
|
||
| implicit val system = ActorSystem() | ||
| // the handler actor replies to incoming HttpRequests | ||
| val handler = system.actorOf(Props[SimpleHttpServer], name = "simpleHttpServer") | ||
| IO(Http) ! Http.Bind(handler, interface = SparkWebUiActorSpec.sparkWebUrl, port = SparkWebUiActorSpec.sparkWebPort) | ||
| } | ||
|
|
||
| override def afterAll() { | ||
| ooyala.common.akka.AkkaTestUtils.shutdownAndWait(SparkWebUiActorSpec.system) | ||
|
|
||
| // close the web service | ||
| implicit val system = ActorSystem("test") | ||
| import system.dispatcher // execution context for futures below | ||
|
|
||
| val pipeline: Future[SendReceive] = | ||
| for ( | ||
| Http.HostConnectorInfo(connector, _) <- | ||
| IO(Http) ? Http.HostConnectorSetup(SparkWebUiActorSpec.sparkWebUrl, port = SparkWebUiActorSpec.sparkWebPort) | ||
| ) yield sendReceive(connector) | ||
|
|
||
| val request = Get("/stop") | ||
| pipeline.flatMap(_(request)) // async call | ||
|
|
||
| } | ||
|
|
||
| var actor: ActorRef = _ | ||
|
|
||
| before { | ||
| actor = SparkWebUiActorSpec.system.actorOf(Props(classOf[SparkWebUiActor]), "spark-web-ui") | ||
| } | ||
|
|
||
| after { | ||
| ooyala.common.akka.AkkaTestUtils.shutdownAndWait(actor) | ||
| } | ||
|
|
||
| describe("SparkWebUiActor") { | ||
| it("should get worker info") { | ||
| val future = actor ? GetWorkerStatus() | ||
| val result = Await.result(future, ShortTimeout.duration).asInstanceOf[SparkWorkersInfo] | ||
| result.alive should equal (1) | ||
| result.dead should equal (1) | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters