Merge pull request #547 from Hydrospheredata/chore/handle_errors

Error handing fixes

Author: dos65

.gitignore

@@ -25,3 +25,6 @@ env-*
 dist/
 build/
 __version__.py
+.bloop
+.metals
+.vscode

mist-lib/src/main/python/setup.py

@@ -36,6 +36,6 @@
 
     packages=find_packages(exclude=['tests']),
     setup_requires=['pytest-runner'], 
-    tests_require=['pytest'],
+    tests_require=['pytest==4.6.4'],
     test_suite='tests'
 )

mist/master/src/main/scala/io/hydrosphere/mist/master/JobDetailsRequest.scala

@@ -1,20 +1,25 @@
 package io.hydrosphere.mist.master
 
+import cats.data.NonEmptyList
+import java.io.File
+
 sealed trait FilterClause
 
 object FilterClause {
   case class ByFunctionId(id: String) extends FilterClause
   case class ByWorkerId(id: String) extends FilterClause
-  case class ByStatuses(statuses: Seq[JobDetails.Status]) extends FilterClause
+  case class ByStatuses(statuses: NonEmptyList[JobDetails.Status]) extends FilterClause
 }
 
 case class JobDetailsRequest(
   limit: Int,
   offset: Int,
   filters: Seq[FilterClause]
-) {
+) { self =>
 
   def withFilter(f: FilterClause): JobDetailsRequest = copy(filters = f +: filters)
+  def withOptFilter(optF: Option[FilterClause]): JobDetailsRequest = 
+    optF.fold(self)(f => self.withFilter(f))
 }
 
 object JobDetailsRequest {

mist/master/src/main/scala/io/hydrosphere/mist/master/execution/JobActor.scala

@@ -95,6 +95,8 @@ class JobActor(
     case JobStarted(id, time) =>
       report.reportPlain(StartedEvent(id, time))
       context become completion(callback, connection)
+
+    case JobFailure(_, err) => completeFailure(err, Some(connection))
   }
 
   private def completion(callback: ActorRef, connection: PerJobConnection): Receive = {

mist/master/src/main/scala/io/hydrosphere/mist/master/interfaces/http/HttpV2Routes.scala

@@ -1,7 +1,7 @@
 package io.hydrosphere.mist.master.interfaces.http
 
 import akka.http.scaladsl.Http
-import java.io.File
+import java.io.{PrintWriter, StringWriter, File}
 
 import akka.http.scaladsl.marshalling.{ToEntityMarshaller, ToResponseMarshallable}
 import akka.http.scaladsl.model._
@@ -30,6 +30,7 @@ import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.Future
 import scala.concurrent.duration.Duration
 import scala.util._
+import cats.data.NonEmptyList
 
 case class JobRunQueryParams(
   force: Boolean,
@@ -111,11 +112,11 @@ object HttpV2Base {
     }
   }
 
-  val statusesQuery: Directive1[Seq[JobDetails.Status]] = {
+  val statusesQuery: Directive1[Option[NonEmptyList[JobDetails.Status]]] = {
     parameter('status *).flatMap(raw => {
       toStatuses(raw) match {
         case Left(errors) => reject(ValidationRejection(s"Unknown statuses: ${errors.mkString(",")}"))
-        case Right(validated) => provide(validated)
+        case Right(validated) => provide(NonEmptyList.fromList(validated))
       }
     })
   }
@@ -170,9 +171,9 @@ object HttpV2Routes extends Logger {
       }
     } ~
     path( root / "workers"/ Segment / "jobs") { workerId =>
-      get { (paginationQuery & statusesQuery) { (pagination, statuses) =>
+      get { (paginationQuery & statusesQuery) { (pagination, maybeStatuses) =>
         val req = JobDetailsRequest(pagination.limit, pagination.offset)
-          .withFilter(FilterClause.ByStatuses(statuses))
+          .withOptFilter(maybeStatuses.map(FilterClause.ByStatuses))
           .withFilter(FilterClause.ByWorkerId(workerId))
 
         onSuccess(jobService.getHistory(req))(rsp => {
@@ -231,9 +232,9 @@ object HttpV2Routes extends Logger {
 
   def functionsJobs(main: MainService): Route = {
     path( root / "functions" / Segment / "jobs" ) { functionId =>
-      get { (paginationQuery & statusesQuery) { (pagination, statuses) =>
+      get { (paginationQuery & statusesQuery) { (pagination, maybeStatuses) =>
         val req = JobDetailsRequest(pagination.limit, pagination.offset)
-          .withFilter(FilterClause.ByStatuses(statuses))
+          .withOptFilter(maybeStatuses.map(FilterClause.ByStatuses))
           .withFilter(FilterClause.ByFunctionId(functionId))
 
         onSuccess(main.execution.getHistory(req))(rsp => {
@@ -343,9 +344,9 @@ object HttpV2Routes extends Logger {
   def jobsRoutes(master: MainService): Route = {
     pathPrefix( root / "jobs") {
       pathEnd {
-        get { (paginationQuery & statusesQuery) { (pagination, statuses) =>
+        get { (paginationQuery & statusesQuery) { (pagination, maybeStatuses) =>
           val req = JobDetailsRequest(pagination.limit, pagination.offset)
-            .withFilter(FilterClause.ByStatuses(statuses))
+            .withOptFilter(maybeStatuses.map(FilterClause.ByStatuses))
 
           onSuccess(master.execution.getHistory(req))(rsp => {
             if (pagination.paginate)
@@ -432,13 +433,22 @@ object HttpV2Routes extends Logger {
   }
 
   def apiRoutes(main: MainService, artifacts: ArtifactRepository, mistHome: String): Route = {
-    val exceptionHandler =
-      ExceptionHandler {
+    val exceptionHandler = ExceptionHandler({ case e => {
+      val (status, msg) = e match {
         case ex @ (_: IllegalArgumentException  | _: IllegalStateException) =>
-          complete((StatusCodes.BadRequest, s"Bad request: ${ex.getMessage}"))
+          (StatusCodes.BadRequest, "Bad request")
         case ex =>
-          complete(HttpResponse(StatusCodes.InternalServerError, entity = s"Server error: ${ex.getMessage}"))
+          (StatusCodes.InternalServerError, "Server error")
       }
+      val errorMsg = {
+        val writer = new StringWriter()
+        e.printStackTrace(new PrintWriter(writer))
+        writer.toString
+      }
+      val fullMsg = msg + ":\n" + errorMsg
+      complete(HttpResponse(status, entity = fullMsg))
+    }})
+
     handleExceptions(exceptionHandler) {
       functionAllRoutes(main) ~
       jobsRoutes(main) ~

mist/master/src/main/scala/io/hydrosphere/mist/master/store/JobRequestSql.scala

@@ -93,25 +93,21 @@ trait JobRequestSql {
   def clear = sql"truncate table job_details"
 
   def filteredByStatuses(statuses: Seq[JobDetails.Status]): Fragment = {
-    if (statuses.nonEmpty) {
-      beginSql ++ sql"where " ++ Fragments.in(fr"status", jobDetailStatusToList(statuses))
-    } else {
-      beginSql
-    }
+    beginSql ++ statusFr(fr"where ", statuses)
   }
 
   def getAll(limit: Int, offset: Int, statuses: Seq[JobDetails.Status]): Fragment = {
     val endOfSQL = sql" limit $limit offset $offset"
 
-    if (statuses.nonEmpty) {
-      beginSql ++ sql"where " ++ Fragments.in(fr"status", jobDetailStatusToList(statuses)) ++ endOfSQL
-    } else {
-      beginSql ++ endOfSQL
-    }
+    beginSql ++ statusFr(fr"where ", statuses) ++ endOfSQL
   }
 
-  def jobDetailStatusToList(seq: Seq[JobDetails.Status]): NonEmptyList[String] =
-    NonEmptyList.fromList(seq.map(_.toString).toList).get
+  def statusFr(prefix: Fragment, statuses: Seq[JobDetails.Status]): Fragment = {
+    NonEmptyList.fromList(statuses.toList) match {
+      case None => Fragment.empty
+      case Some(inSt) => prefix ++ Fragments.in(fr"status", inSt.map(_.toString))
+    }
+  }  
 
   /**
     * Generates Doobie sql fragment with JobDetailsRequest
@@ -121,7 +117,7 @@ trait JobRequestSql {
       if (req.filters.isEmpty) sql"" else {
         fr"where" ++ req.filters.map {
           case ByFunctionId(id) => fr"function = $id"
-          case ByStatuses(statuses) => Fragments.in(fr"status", jobDetailStatusToList(statuses))
+          case ByStatuses(statuses) => Fragments.in(fr"status", statuses.map(_.toString))
           case ByWorkerId(id) => fr"worker_id = $id"
         }.reduce((a, b) => a ++ fr"and" ++ b)
       }

mist/master/src/test/scala/io/hydrosphere/mist/master/interfaces/http/HttpApiV2Spec.scala

@@ -536,7 +536,6 @@ class HttpApiV2Spec extends FunSpec
 
       val route = HttpV2Routes.apiRoutes(master, mock[ArtifactRepository], "")
       Post(s"/v2/api/functions/x/jobs", JsMap("1" -> "Hello".js)) ~> route ~> check {
-        responseAs[String] shouldBe "Bad request: argument missing"
         status shouldBe StatusCodes.BadRequest
       }
     }

mist/master/src/test/scala/io/hydrosphere/mist/master/store/JobRequestsSqlSpec.scala

@@ -8,6 +8,7 @@ import doobie.implicits._
 import io.hydrosphere.mist.core.CommonData.{Action, JobParams}
 import io.hydrosphere.mist.master.JobDetails.Source
 import mist.api.data.JsMap
+import cats.data.NonEmptyList
 
 class JobRequestsSqlSpec extends FunSpec with Matchers{
 
@@ -36,13 +37,13 @@ class JobRequestsSqlSpec extends FunSpec with Matchers{
     }
 
     it("sql filter by status") {
-      val sql = jobRequestSql.generateSqlByJobDetailsRequest(JobDetailsRequest(0, 0).withFilter(ByStatuses(Seq(Initialized, Queued))))
+      val sql = jobRequestSql.generateSqlByJobDetailsRequest(JobDetailsRequest(0, 0).withFilter(ByStatuses(NonEmptyList.of(Initialized, Queued))))
       sql.toString() shouldEqual sql"select * from job_details where status IN (?, ?) order by create_time desc limit ? offset ? ".toString()
     }
 
     it("sql complex filter") {
       val sql = jobRequestSql.generateSqlByJobDetailsRequest(JobDetailsRequest(0, 0).
-        withFilter(ByStatuses(Seq(Initialized, Queued))).
+        withFilter(ByStatuses(NonEmptyList.of(Initialized, Queued))).
         withFilter(ByWorkerId("ID")).
         withFilter(ByFunctionId("ID")))
       sql.toString() shouldEqual sql"select * from job_details where function = ? and worker_id = ? and status IN (?, ?) order by create_time desc limit ? offset ? ".toString()

mist/worker/src/main/scala/io/hydrosphere/mist/worker/WorkerActor.scala

@@ -8,6 +8,9 @@ import io.hydrosphere.mist.core.CommonData._
 import io.hydrosphere.mist.worker.runners._
 import mist.api.data.JsData
 import org.apache.spark.streaming.StreamingContext
+
+import scala.concurrent.Future
+
 import RequestSetup._
 
 class WorkerActor(

project/Ui.scala

@@ -15,7 +15,7 @@ object Ui {
 
   lazy val settings = Seq(
     uiUrl := { (s: String) => s"https://github.com/Hydrospheredata/mist-ui/releases/download/v$s/mist-ui-$s.tar.gz" },
-    uiVersion := "2.2.0",
+    uiVersion := "2.2.1",
     uiCheckoutDir := "ui_local",
     ui := {
       val local = baseDirectory.value / uiCheckoutDir.value