Merge pull request #153 from broadinstitute/kc_jes_fixes

added a default local SSD disk to be used for localization, also need…
broadinstitute · Aug 24, 2015 · ed38985 · ed38985
2 parents 0ec6b84 + 2611191
commit ed38985
Show file tree

Hide file tree

Showing 4 changed files with 17 additions and 6 deletions.
diff --git a/src/main/scala/cromwell/binding/RuntimeAttributes.scala b/src/main/scala/cromwell/binding/RuntimeAttributes.scala
@@ -22,7 +22,7 @@ object RuntimeAttributes {
   /** Fallback values if values for these keys are not specified in a task "runtime" stanza. */
   object Defaults {
     val Cpu = 1
-    val Disk = Seq(LocalDisk("local-disk", 100L, "LOCAL_SSD").toDisk)
+    val LocalizationDisk = LocalDisk("local-disk", 100L, "LOCAL_SSD").toDisk
     val FailOnStderr = false
     val MemoryInBytes = MemorySize.GB.toBytes(2)
     val Preemptible = false
@@ -109,7 +109,8 @@ case class RuntimeAttributes private(attributes: Map[String, String], warnings:
         Array(name, sizeGb, diskType) = diskString.split("\\s+")
       } yield LocalDisk(name, sizeGb.toLong, diskType).toDisk
 
-    if (taskSpecifiedAttributes.isEmpty) Defaults.Disk else taskSpecifiedAttributes
+    // additional disks will be added to the default localization disk
+    taskSpecifiedAttributes :+ Defaults.LocalizationDisk
   }
 
   val memoryGB: Double = {

diff --git a/src/main/scala/cromwell/engine/backend/jes/JesBackend.scala b/src/main/scala/cromwell/engine/backend/jes/JesBackend.scala
@@ -1,5 +1,6 @@
 package cromwell.engine.backend.jes
 
+import java.io.File
 import java.math.BigInteger
 import java.net.URL
 import java.nio.file.{Path, Paths}
@@ -43,6 +44,8 @@ object JesBackend {
   val LocalStdoutValue = "job.stdout.txt"
   val LocalStderrValue = "job.stderr.txt"
 
+  val JesCromwellRoot = "/cromwell_root"
+
   private def callGcsPath(workflowId: String, workflowName: String, callName: String): String =
     s"$CromwellExecutionBucket/$workflowName/$workflowId/call-$callName"
 
@@ -53,11 +56,13 @@ object JesBackend {
 
   def stderrJesOutput(callGcsPath: String): JesOutput = JesOutput(LocalStderrParamName, s"$callGcsPath/$LocalStderrValue", Paths.get(LocalStderrValue))
   def stdoutJesOutput(callGcsPath: String): JesOutput = JesOutput(LocalStdoutParamName, s"$callGcsPath/$LocalStdoutValue", Paths.get(LocalStdoutValue))
+  def localizationDiskInput(): JesInput = JesInput("working_disk", "disk://local-disk", new File(JesCromwellRoot).toPath)
 
   // For now we want to always redirect stdout and stderr. This could be problematic if that's what the WDL calls stuff, but oh well
   def standardParameters(callGcsPath: String): Seq[JesParameter] = Seq(
     stdoutJesOutput(callGcsPath),
-    stderrJesOutput(callGcsPath)
+    stderrJesOutput(callGcsPath),
+    localizationDiskInput
   )
 
   sealed trait JesParameter {
@@ -82,7 +87,7 @@ class JesBackend extends Backend with LazyLogging {
    * @return A path which is unique per input path
    */
   def localFilePathFromCloudStoragePath(gcsPath: GoogleCloudStoragePath): Path = {
-    Paths.get("/cromwell_root/" + gcsPath.bucket + "/" + gcsPath.objectName)
+    Paths.get(JesCromwellRoot + "/" + gcsPath.bucket + "/" + gcsPath.objectName)
   }
 
   /**

diff --git a/src/main/scala/cromwell/engine/backend/jes/Run.scala b/src/main/scala/cromwell/engine/backend/jes/Run.scala
@@ -27,6 +27,11 @@ object Run  {
     logging.setGcsPath(pipeline.gcsPath)
     rpr.setLogging(logging)
 
+    // Currently, disk resources need to be specified both at pipeline creation and pipeline run time
+    val resources = new Resources()
+    resources.setDisks( scala.collection.JavaConversions.seqAsJavaList(pipeline.call.task.runtimeAttributes.defaultDisks))
+    rpr.setResources(resources)
+
     val id = pipeline.genomicsService.pipelines().run(rpr).execute().getName
     Log.info(s"$tag JES ID is $id")
     new Run(id, pipeline, tag)

diff --git a/src/test/scala/cromwell/binding/RuntimeAttributeSpec.scala b/src/test/scala/cromwell/binding/RuntimeAttributeSpec.scala
@@ -180,7 +180,7 @@ class RuntimeAttributeSpec extends FlatSpec with Matchers {
     attributes.cpu shouldBe 3
     val firstDisk = new Disk().setName("Disk1").setSizeGb(3L).setType("SSD")
     val secondDisk = new Disk().setName("Disk2").setSizeGb(500L).setType("OldSpinnyKind")
-    attributes.defaultDisks shouldEqual Seq(firstDisk, secondDisk)
+    attributes.defaultDisks shouldEqual Seq(firstDisk, secondDisk, RuntimeAttributes.Defaults.LocalizationDisk)
     attributes.defaultZones shouldEqual Seq("US_Metro", "US_Backwater")
     attributes.memoryGB shouldBe 4
   }
@@ -194,7 +194,7 @@ class RuntimeAttributeSpec extends FlatSpec with Matchers {
     val googlyCall = calls(callIndex)
     val attributes = googlyCall.task.runtimeAttributes
     attributes.cpu shouldBe RuntimeAttributes.Defaults.Cpu
-    attributes.defaultDisks shouldBe RuntimeAttributes.Defaults.Disk
+    attributes.defaultDisks shouldEqual Seq(RuntimeAttributes.Defaults.LocalizationDisk)
     attributes.defaultZones shouldBe RuntimeAttributes.Defaults.Zones
     attributes.memoryGB shouldBe MemorySize.GB.fromBytes(RuntimeAttributes.Defaults.MemoryInBytes)
   }