[batch/qob] propagates errors from failed QoB worker jobs without res…

…ult files

Currently, when a QoB worker job fails in a way that prevents it from
writing a result file with the error's stack trace in it, the Batch
worker that started the job tries to get the result file anyway, and
when it fails, raises a 404. This change retrieves the stack trace from
the QoB job and raises an error with the stack trace included.

hail/Makefile

@@ -214,7 +214,7 @@ pytest-qob: upload-qob-jar upload-qob-test-resources install-editable
             --durations=50 \
             --self-contained-html \
             --html=../build/reports/pytest.html \
-            --timeout=120 \
+            --timeout=1200 \
             $(PYTEST_TARGET) \
             $(PYTEST_ARGS)
 
@@ -316,8 +316,8 @@ upload-artifacts: $(WHEEL)
 upload-qob-test-resources: $(shell git ls-files src/test/resources)
 upload-qob-test-resources: $(shell git ls-files python/hail/docs/data)
 	! [ -z $(NAMESPACE) ]  # call this like: make upload-qob-test-resources NAMESPACE=default
-	gcloud storage cp src/test/resources/\* $(CLOUD_HAIL_TEST_RESOURCES_DIR)
-	gcloud storage cp python/hail/docs/data/\* $(CLOUD_HAIL_DOCTEST_DATA_DIR)
+	gcloud storage cp -R src/test/resources/\* $(CLOUD_HAIL_TEST_RESOURCES_DIR)
+	gcloud storage cp -R python/hail/docs/data/\* $(CLOUD_HAIL_DOCTEST_DATA_DIR)
 	# # In Azure, use the following instead of gcloud storage cp
 	# python3 -m hailtop.aiotools.copy -vvv 'null' '[\
         #      {"from":"src/test/resources","to":"$(CLOUD_HAIL_TEST_RESOURCES_DIR)"},\

hail/scripts/upload_qob_jar.sh

@@ -22,5 +22,7 @@ else
     JAR_LOCATION="${TEST_STORAGE_URI}/${NAMESPACE}/jars/${TOKEN}/${REVISION}.jar"
 fi
 
-python3 -m hailtop.aiotools.copy -vvv 'null' '[{"from":"'${SHADOW_JAR}'", "to":"'${JAR_LOCATION}'"}]'
+# FIXME revert when done testing
+gcloud storage cp ${SHADOW_JAR} ${JAR_LOCATION}
+# python3 -m hailtop.aiotools.copy -vvv 'null' '[{"from":"'${SHADOW_JAR}'", "to":"'${JAR_LOCATION}'"}]'
 echo ${JAR_LOCATION} > ${PATH_FILE}

hail/src/main/scala/is/hail/backend/service/ServiceBackend.scala

@@ -268,10 +268,6 @@ class ServiceBackend(
 
     stageCount += 1
     implicit val formats: Formats = DefaultFormats
-    val batchState = (batch \ "state").extract[String]
-    if (batchState == "failed") {
-      throw new HailBatchFailure(s"Update $updateId for batch $batchId failed")
-    }
 
     log.info(s"parallelizeAndComputeWithIndex: $token: reading results")
 
@@ -281,16 +277,29 @@ class ServiceBackend(
       collection.map { case (_, i) =>
         (
           () => {
-            val bytes = fs.readNoCompression(s"$root/result.$i")
-            if (bytes(0) != 0) {
-              bytes.slice(1, bytes.length)
-            } else {
-              val errorInformationBytes = bytes.slice(1, bytes.length)
-              val is = new DataInputStream(new ByteArrayInputStream(errorInformationBytes))
-              val shortMessage = readString(is)
-              val expandedMessage = readString(is)
-              val errorId = is.readInt()
-              throw new HailWorkerException(i, shortMessage, expandedMessage, errorId)
+            try {
+              val bytes = fs.readNoCompression(s"$root/result.$i")
+              if (bytes(0) != 0) {
+                bytes.slice(1, bytes.length)
+              } else {
+                val errorInformationBytes = bytes.slice(1, bytes.length)
+                val is = new DataInputStream(new ByteArrayInputStream(errorInformationBytes))
+                val shortMessage = readString(is)
+                val expandedMessage = readString(is)
+                val errorId = is.readInt()
+                throw new HailWorkerException(i, shortMessage, expandedMessage, errorId)
+              }
+            } catch {
+              case e: HailWorkerException => throw e
+              case e: Exception => {
+                val jobId = (jobs(i) \ "job_id").extract[Int]
+                val err = s"Update $updateId for batch $batchId failed due to error in job $jobId."
+                val (jobData, jobLog) = batchClient.getJob(batchId, jobId)
+                (jobData \ "state").extract[String].toLowerCase() match {
+                  case "failed" => throw new HailWorkerException(i, err, s"""Log for job $jobId:\n\n${(jobLog \ "main").extract[String]}\n\n""", -1)
+                  case _ => throw new HailBatchFailure(err, e)
+                }
+              }
             }
           },
           i
@@ -407,7 +416,7 @@ class ServiceBackend(
 }
 
 class EndOfInputException extends RuntimeException
-class HailBatchFailure(message: String) extends RuntimeException(message)
+class HailBatchFailure(message: String, error: Exception) extends RuntimeException(message, error)
 
 object ServiceBackendAPI {
   private[this] val log = Logger.getLogger(getClass.getName())

hail/src/main/scala/is/hail/backend/service/Worker.scala

@@ -179,6 +179,7 @@ object Worker {
     timer.end("executeFunction")
     timer.start("writeOutputs")
 
+    /* TODO uncomment after testing
     retryTransientErrors {
       write(s"$root/result.$i") { dos =>
         result match {
@@ -194,6 +195,7 @@ object Worker {
         }
       }
     }
+    */
 
     timer.end("writeOutputs")
     timer.end(s"Job $i")

hail/src/main/scala/is/hail/services/DeployConfig.scala

@@ -121,4 +121,14 @@ class DeployConfig(
   def baseUrl(service: String, baseScheme: String = "http"): String = {
     s"${ scheme(baseScheme) }://${ domain(service) }${ basePath(service) }"
   }
+
+  def externalUrl(service: String, path: String, baseScheme: String = "http"): String = {
+    val ns = getServiceNamespace(service)
+    val prefix = s"${baseScheme}s://"
+    (ns, service) match {
+      case ("default", "www") => s"$prefix$domain$path"
+      case ("default", _) => s"$prefix$service$domain$path"
+      case _ => s"${prefix}internal.$domain/$ns/$service$path"
+    }
+  }
 }

hail/src/main/scala/is/hail/services/batch_client/BatchClient.scala

@@ -195,6 +195,13 @@ class BatchClient(
     throw new AssertionError("unreachable")
   }
 
+  def getJob(batchId: Long, jobId: Int): (JValue, JValue) = {
+    (
+      get(s"/api/v1alpha/batches/$batchId/jobs/$jobId"),
+      get(s"/api/v1alpha/batches/$batchId/jobs/$jobId/log")
+    )
+  }
+
   private def createBunches(jobs: IndexedSeq[JObject]): BoxedArrayBuilder[Array[Array[Byte]]] = {
     val bunches = new BoxedArrayBuilder[Array[Array[Byte]]]()
     val bunchb = new BoxedArrayBuilder[Array[Byte]]()