fix(job-server) store error class and stack trace in database (spark-jobserver#923)

* Implement storing and fetching error class and stack trace in DAOs
* Create migrations adding columns in SQL databases
* Modify tests to take care of new columns

BREAKING CHANGES :
* Schema change: new columns in SQL databases and Cassandra
* File storage format incompatible
* Stack trace returned as string instead of list of strings

Author: krcz

README.md

@@ -353,27 +353,19 @@ It is much more type safe, separates context configuration, job ID, named object
 Let's try running our sample job with an invalid configuration:
 
     curl -i -d "bad.input=abc" "localhost:8090/jobs?appName=test&classPath=spark.jobserver.WordCountExample"
-
     HTTP/1.1 400 Bad Request
-    Server: spray-can/1.2.0
-    Date: Tue, 10 Jun 2014 22:07:18 GMT
+    Server: spray-can/1.3.4
+    Date: Thu, 14 Sep 2017 12:01:37 GMT
+    Access-Control-Allow-Origin: *
     Content-Type: application/json; charset=UTF-8
-    Content-Length: 929
+    Content-Length: 738
 
     {
       "status": "VALIDATION FAILED",
       "result": {
-        "message": "No input.string config param",
+        "message": "One(SparkJobInvalid(No input.string config param))",
         "errorClass": "java.lang.Throwable",
-        "stack": ["spark.jobserver.JobManagerActor$$anonfun$spark$jobserver$JobManagerActor$$getJobFuture$4.apply(JobManagerActor.scala:212)",
-        "scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1$1(Future.scala:24)",
-        "scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:24)",
-        "akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:42)",
-        "akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:386)",
-        "scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)",
-        "scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)",
-        "scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)",
-        "scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)"]
+        "stack": "java.lang.Throwable: One(SparkJobInvalid(No input.string config param))\n\tat spark.jobserver.JobManagerActor$$anonfun$getJobFuture$4.apply(JobManagerActor.scala:327)\n\tat scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1$1(Future.scala:24)\n\tat scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:24)\n\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)\n\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)\n\tat java.lang.Thread.run(Thread.java:748)\n"
       }
     }
 

job-server/src/main/resources/db/h2/migration/V0_7_4/V0_7_4__create_error_class_and_stacktrace.sql

@@ -0,0 +1,2 @@
+ALTER TABLE JOBS ADD COLUMN ERROR_CLASS VARCHAR(255);
+ALTER TABLE JOBS ADD COLUMN ERROR_STACK_TRACE CLOB;

job-server/src/main/resources/db/mysql/migration/V0_7_4/V0_7_4__create_error_class_and_stacktrace.sql

@@ -0,0 +1,2 @@
+ALTER TABLE JOBS ADD COLUMN ERROR_CLASS VARCHAR(255);
+ALTER TABLE JOBS ADD COLUMN ERROR_STACK_TRACE TEXT;

job-server/src/main/resources/db/postgresql/migration/V0_7_4/V0_7_4__create_error_class_and_stacktrace.sql

@@ -0,0 +1,2 @@
+ALTER TABLE JOBS ADD COLUMN ERROR_CLASS VARCHAR(255);
+ALTER TABLE JOBS ADD COLUMN ERROR_STACK_TRACE TEXT;

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

@@ -2,14 +2,13 @@ package spark.jobserver
 
 import scala.collection.mutable
 import scala.util.Try
-
 import akka.actor.{ActorRef, Props}
 import com.yammer.metrics.core.Meter
 import org.joda.time.DateTime
 import spark.jobserver.JobManagerActor.JobKilledException
 import spark.jobserver.common.akka.InstrumentedActor
 import spark.jobserver.common.akka.metrics.YammerMetrics
-import spark.jobserver.io.{JobDAOActor, JobInfo}
+import spark.jobserver.io.{ErrorData, JobDAOActor, JobInfo}
 
 object JobStatusActor {
   case class JobInit(jobInfo: JobInfo)
@@ -40,8 +39,9 @@ class JobStatusActor(jobDao: ActorRef) extends InstrumentedActor with YammerMetr
   override def postStop(): Unit = {
     val stopTime = DateTime.now()
     val stoppedInfos = infos.values.map { info =>
-      info.copy(endTime = Some(stopTime),
-                error = Some(new Exception(s"Context (${info.contextName}) for this job was terminated"))) }
+      val errorData = ErrorData(s"Context (${info.contextName}) for this job was terminated", "", "")
+      info.copy(endTime = Some(stopTime), error = Some(errorData))
+    }
     stoppedInfos.foreach({info => jobDao ! JobDAOActor.SaveJobInfo(info)})
   }
 
@@ -90,19 +90,20 @@ class JobStatusActor(jobDao: ActorRef) extends InstrumentedActor with YammerMetr
     case msg: JobValidationFailed =>
       processStatus(msg, "validation failed", remove = true) {
         case (info, msg: JobValidationFailed) =>
-          info.copy(endTime = Some(msg.endTime), error = Some(msg.err))
+          info.copy(endTime = Some(msg.endTime), error = Some(ErrorData(msg.err)))
       }
 
     case msg: JobErroredOut =>
       processStatus(msg, "finished with an error", remove = true) {
         case (info, msg: JobErroredOut) =>
-          info.copy(endTime = Some(msg.endTime), error = Some(msg.err))
+          info.copy(endTime = Some(msg.endTime), error = Some(ErrorData(msg.err)))
       }
 
     case msg: JobKilled =>
       processStatus(msg, "killed", remove = true) {
         case (info, msg: JobKilled) =>
-          info.copy(endTime = Some(msg.endTime), error = Some(JobKilledException(info.jobId)))
+          info.copy(endTime = Some(msg.endTime),
+            error = Some(ErrorData(JobKilledException(info.jobId))))
       }
   }
 

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

@@ -16,7 +16,7 @@ import org.joda.time.DateTime
 import org.slf4j.LoggerFactory
 import spark.jobserver.JobManagerActor.JobKilledException
 import spark.jobserver.auth._
-import spark.jobserver.io.{BinaryType, JobInfo, JobStatus}
+import spark.jobserver.io.{BinaryType, ErrorData, JobInfo, JobStatus}
 import spark.jobserver.routes.DataRoutes
 import spark.jobserver.util.{SSLContextFactory, SparkJobUtils}
 import spray.http.HttpHeaders.`Content-Type`
@@ -26,6 +26,7 @@ import spray.io.ServerSSLEngineProvider
 import spray.json.DefaultJsonProtocol._
 import spray.routing.directives.AuthMagnet
 import spray.routing.{HttpService, RequestContext, Route}
+
 import scala.concurrent.{Await, ExecutionContext, Future}
 import scala.util.Try
 
@@ -74,27 +75,26 @@ object WebApi {
   }
 
   def formatException(t: Throwable): Any =
-    if (t.getCause != null) {
-      Map("message" -> t.getMessage,
-        "errorClass" -> t.getClass.getName,
-        "cause" -> t.getCause.getMessage,
-        "causingClass" -> t.getCause.getClass.getName,
-        "stack" -> t.getCause.getStackTrace.map(_.toString).toSeq)
-    } else {
-      Map("message" -> t.getMessage,
-        "errorClass" -> t.getClass.getName,
-        "stack" -> t.getStackTrace.map(_.toString).toSeq)
-    }
+    Map("message" -> t.getMessage,
+      "errorClass" -> t.getClass.getName,
+      "stack" -> ErrorData.getStackTrace(t))
+
+  def formatException(t: ErrorData): Any = {
+    Map("message" -> t.message,
+      "errorClass" -> t.errorClass,
+      "stack" -> t.stackTrace
+    )
+  }
 
   def getJobReport(jobInfo: JobInfo, jobStarted: Boolean = false): Map[String, Any] = {
 
     val statusMap = if (jobStarted) Map(StatusKey -> JobStatus.Started) else jobInfo match {
       case JobInfo(_, _, _, _, _, None, _) => Map(StatusKey -> JobStatus.Running)
-      case JobInfo(_, _, _, _, _, _, Some(ex)) =>
-        ex match {
-          case e: JobKilledException => Map(StatusKey -> JobStatus.Killed, ResultKey -> formatException(ex))
-          case _ => Map(StatusKey -> JobStatus.Error, ResultKey -> formatException(ex))
-        }
+      case JobInfo(_, _, _, _, _, _, Some(err))
+        if err.errorClass == classOf[JobKilledException].getName =>
+        Map(StatusKey -> JobStatus.Killed, ResultKey -> formatException(err))
+      case JobInfo(_, _, _, _, _, _, Some(err)) =>
+        Map(StatusKey -> JobStatus.Error, ResultKey -> formatException(err))
       case JobInfo(_, _, _, _, _, Some(e), None) => Map(StatusKey -> "FINISHED")
     }
     Map("jobId" -> jobInfo.jobId,

job-server/src/main/scala/spark/jobserver/io/JobCassandraDAO.scala

@@ -44,6 +44,8 @@ object Metadata {
   val StartDate = "start_date"
   val EndTime = "end_time"
   val Error = "error"
+  val ErrorClass = "error_class"
+  val ErrorStackTrace = "error_stack_trace"
 
 }
 
@@ -160,7 +162,8 @@ class JobCassandraDAO(config: Config) extends JobDAO with FileCacher {
   override def getJobInfos(limit: Int, status: Option[String] = None): Future[Seq[JobInfo]] = {
     import Metadata._
     val query = QB.select(
-      JobId, ContextName, AppName, BType, UploadTime, Classpath, StartTime, EndTime, Error
+      JobId, ContextName, AppName, BType, UploadTime, Classpath, StartTime, EndTime,
+      Error, ErrorClass, ErrorStackTrace
     ).from(JobsChronologicalTable).where(QB.eq(StartDate, today())).limit(limit)
 
     session.executeAsync(query).map { rs =>
@@ -180,7 +183,8 @@ class JobCassandraDAO(config: Config) extends JobDAO with FileCacher {
   override def getRunningJobInfosForContextName(contextName: String): Future[Seq[JobInfo]] = {
     import Metadata._
     val query = QB.select(
-      JobId, ContextName, AppName, BType, UploadTime, Classpath, StartTime, EndTime, Error
+      JobId, ContextName, AppName, BType, UploadTime, Classpath, StartTime, EndTime,
+      Error, ErrorClass, ErrorStackTrace
     ).from(RunningJobsTable).where(QB.eq(ContextName, contextName))
 
     session.executeAsync(query).map { rs =>
@@ -222,6 +226,11 @@ class JobCassandraDAO(config: Config) extends JobDAO with FileCacher {
   }
 
   private def rowToJobInfo(row: Row): JobInfo = {
+    val errorData = Option(row.getString(Error)).map { error =>
+      val errorClass = if (row.isNull(ErrorClass)) "" else row.getString(ErrorClass)
+      val stackTrace = if (row.isNull(ErrorStackTrace)) "" else row.getString(ErrorStackTrace)
+      ErrorData(error, errorClass, stackTrace)
+    }
     JobInfo(
       row.getUUID(Metadata.JobId).toString,
       row.getString(ContextName),
@@ -233,13 +242,14 @@ class JobCassandraDAO(config: Config) extends JobDAO with FileCacher {
       row.getString(Classpath),
       new DateTime(row.getTimestamp(StartTime)),
       Option(row.getTimestamp(EndTime)).map(new DateTime(_)),
-      Option(row.getString(Error)).map(new Throwable(_))
+      errorData
     )
   }
 
   override def getJobInfo(jobId: String): Future[Option[JobInfo]] = {
     val query = QB.select(
-      JobId, ContextName, AppName, BType, UploadTime, Classpath, StartTime, EndTime, Error
+      JobId, ContextName, AppName, BType, UploadTime, Classpath, StartTime, EndTime,
+      Error, ErrorClass, ErrorStackTrace
     ).from(JobsTable).
       where(QB.eq(JobId, UUID.fromString(jobId))).
       limit(1)
@@ -252,9 +262,6 @@ class JobCassandraDAO(config: Config) extends JobDAO with FileCacher {
 
   override def saveJobInfo(jobInfo: JobInfo): Unit = {
     val JobInfo(jobId, contextName, binaryInfo, classPath, startTime, endTime, error) = jobInfo
-    val (_, endOpt, errOpt) = (startTime,
-      endTime.map(e => e),
-      error.map(_.getMessage))
 
     val localDate: LocalDate = LocalDate.fromMillisSinceEpoch(jobInfo.startTime.getMillis)
 
@@ -268,8 +275,13 @@ class JobCassandraDAO(config: Config) extends JobDAO with FileCacher {
         value(Classpath, classPath).
         value(StartTime, startTime.getMillis).
         value(StartDate, localDate)
-      endOpt.foreach{e => insert.value(EndTime, e.getMillis)}
-      errOpt.foreach(insert.value(Error, _))
+
+      endTime.foreach{e => insert.value(EndTime, e.getMillis)}
+      error.foreach { err =>
+        insert.value(Error, err.message)
+        insert.value(ErrorClass, err.errorClass)
+        insert.value(ErrorStackTrace, err.stackTrace)
+      }
       insert
     }
 
@@ -365,7 +377,9 @@ class JobCassandraDAO(config: Config) extends JobDAO with FileCacher {
       addColumn(StartTime, DataType.timestamp).
       addColumn(StartDate, DataType.date).
       addColumn(EndTime, DataType.timestamp).
-      addColumn(Error, DataType.text)
+      addColumn(Error, DataType.text).
+      addColumn(ErrorClass, DataType.text).
+      addColumn(ErrorStackTrace, DataType.text)
 
     session.execute(jobsTableStatement)
 
@@ -381,6 +395,8 @@ class JobCassandraDAO(config: Config) extends JobDAO with FileCacher {
       addColumn(Classpath, DataType.text).
       addColumn(EndTime, DataType.timestamp).
       addColumn(Error, DataType.text).
+      addColumn(ErrorClass, DataType.text).
+      addColumn(ErrorStackTrace, DataType.text).
       withOptions().clusteringOrder(StartTime, Direction.DESC)
 
     session.execute(jobsChronologicalView)
@@ -396,7 +412,9 @@ class JobCassandraDAO(config: Config) extends JobDAO with FileCacher {
       addColumn(StartTime, DataType.timestamp).
       addColumn(StartDate, DataType.date).
       addColumn(EndTime, DataType.timestamp).
-      addColumn(Error, DataType.text)
+      addColumn(Error, DataType.text).
+      addColumn(ErrorClass, DataType.text).
+      addColumn(ErrorStackTrace, DataType.text)
 
     session.execute(runningJobsView)
   }

job-server/src/main/scala/spark/jobserver/io/JobDAO.scala

@@ -1,5 +1,7 @@
 package spark.jobserver.io
 
+import java.io.{PrintWriter, StringWriter}
+
 import com.typesafe.config._
 import org.joda.time.{DateTime, Duration}
 import org.slf4j.LoggerFactory
@@ -46,13 +48,26 @@ object BinaryType {
 // Uniquely identifies the binary used to run a job
 case class BinaryInfo(appName: String, binaryType: BinaryType, uploadTime: DateTime)
 
+case class ErrorData(message: String, errorClass: String, stackTrace: String)
+
+object ErrorData {
+  def apply(ex: Throwable): ErrorData = {
+    ErrorData(ex.getMessage, ex.getClass.getName, getStackTrace(ex))
+  }
+
+  def getStackTrace(ex: Throwable): String = {
+    val stackTrace = new StringWriter()
+    ex.printStackTrace(new PrintWriter(stackTrace))
+    stackTrace.toString
+  }
+}
 
 // Both a response and used to track job progress
 // NOTE: if endTime is not None, then the job has finished.
 case class JobInfo(jobId: String, contextName: String,
                    binaryInfo: BinaryInfo, classPath: String,
                    startTime: DateTime, endTime: Option[DateTime],
-                   error: Option[Throwable]) {
+                   error: Option[ErrorData]) {
   def jobLengthMillis: Option[Long] = endTime.map { end => new Duration(startTime, end).getMillis }
 
   def isRunning: Boolean = endTime.isEmpty
@@ -143,7 +158,7 @@ trait JobDAO {
       for (info <- infos) {
         val updatedInfo = info.copy(
           endTime = Some(endTime),
-          error = Some(JobKilledException(info.jobId)))
+          error = Some(ErrorData(JobKilledException(info.jobId))))
         saveJobInfo(jobInfo = updatedInfo)
       }
     }

job-server/src/main/scala/spark/jobserver/io/JobFileDAO.scala

@@ -171,13 +171,16 @@ class JobFileDAO(config: Config) extends JobDAO {
     out.writeLong(jobInfo.startTime.getMillis)
     val time = if (jobInfo.endTime.isEmpty) jobInfo.startTime.getMillis else jobInfo.endTime.get.getMillis
     out.writeLong(time)
-    val errorStr = if (jobInfo.error.isEmpty) "" else jobInfo.error.get.toString
-    out.writeUTF(errorStr)
+    out.writeUTF(jobInfo.error.map(_.message).getOrElse(""))
+    out.writeUTF(jobInfo.error.map(_.errorClass).getOrElse(""))
+    out.writeUTF(jobInfo.error.map(_.stackTrace).getOrElse(""))
   }
 
   private def readError(in: DataInputStream) = {
-    val error = in.readUTF()
-    if (error == "") None else Some(new Throwable(error))
+    val error = Some(in.readUTF()).filter(_.isEmpty)
+    val errorClass = in.readUTF()
+    val errorStackTrace = in.readUTF()
+    error.map(ErrorData(_, errorClass, errorStackTrace))
   }
 
   private def readJobInfo(in: DataInputStream) = JobInfo(