release_candidate_0.22

.pullapprove.yml

@@ -7,7 +7,6 @@ reviewers:
     members:
         - cjllanwarne
         - Horneth
-        - scottfrazer
         - mcovarr
         - geoffjentry
         - kshakir

CHANGELOG.md

@@ -1,23 +1,68 @@
 # Cromwell Change Log
 
-## 0.20
+## 0.22
 
-* The default per-upload bytes size for GCS is now the minumum 256K
-instead of 64M. There is also an undocumented config key
-`google.upload-buffer-bytes` that allows adjusting this internal value.
+* Improved retries for Call Caching and general bug fixes.
+* Users will experience better scalability of status polling for Google JES.
+* Now there are configurable caching strategies for a SharedFileSystem backend (i.e. Local, SFS) in the backend's stanza:
+  See below for detailed descriptions of each configurable key.
 
-* Updated Docker Hub hash retriever to parse json with [custom media
-types](https://github.com/docker/distribution/blob/05b0ab0/docs/spec/manifest-v2-1.md).
+```
+backend {
+  ...
+  providers {
+    SFS_BackendName {
+      actor-factory = ...
+      config {
+        ...
+        filesystems {
+          local {
+            localization: [
+               ...
+            ]
+            caching {
+              duplication-strategy: [
+                "hard-link", "soft-link", "copy"
+              ]
+              # Possible values: file, path
+              # "file" will compute an md5 hash of the file content.
+              # "path" will compute an md5 hash of the file path. This strategy will only be effective if the duplication-strategy (above) is set to "soft-link",
+              # in order to allow for the original file path to be hashed.
+              hashing-strategy: "file"
+
+              # When true, will check if a sibling file with the same name and the .md5 extension exists, and if it does, use the content of this file as a hash.
+              # If false or the md5 does not exist, will proceed with the above-defined hashing strategy.
+              check-sibling-md5: false
+            }
+```
+* Multiple Input JSON files can now be submitted in server mode through the existing submission endpoint: /api/workflows/:version.
+    This endpoint accepts a POST request with a multipart/form-data encoded body. You can now include multiple keys for workflow inputs.
 
-* Added a `/batch` submit endpoint that accepts a single wdl with
-multiple input files.
+        Each key below can contain an optional JSON file of the workflow inputs. A skeleton file can be generated from wdltool using the "inputs" subcommand.
+        NOTE: In case of key conflicts between multiple JSON files, higher values of x in workflowInputs_x override lower values. For example, an input
+        specified in workflowInputs_3 will override an input with the same name that was given in workflowInputs or workflowInputs_2. Similarly, an input
+        specified in workflowInputs_5 will override an input with the same name in any other input file.
 
-* The `/query` endpoint now supports querying by `id`, and submitting
-parameters as a HTTP POST.
+        workflowInputs
+        workflowInputs_2
+        workflowInputs_3
+        workflowInputs_4
+        workflowInputs_5
 
-## 0.21
+* You can now limit the number of concurrent jobs for a backend by specifying the following option in the backend's config stanza:
+```
+backend {
+  ...
+  providers {
+    BackendName {
+      actor-factory = ...
+      config {
+        concurrent-job-limit = 5
+```
 
 
+## 0.21
+
 * Warning: Significant database updates when you switch from version 0.19 to 0.21 of Cromwell.
   There may be a long wait period for the migration to finish for large databases.
   Please refer to MIGRATION.md for more details.
@@ -71,7 +116,7 @@ task {
     command {
         echo "I'm private !"
     }
-    
+
     runtime {
         docker: "ubuntu:latest"
         noAddress: true
@@ -94,7 +139,7 @@ passed absolute paths for input `File`s.
 * Override the default database configuration by setting the keys
 `database.driver`, `database.db.driver`, `database.db.url`, etc.
 * Override the default database configuration by setting the keys
-`database.driver`, `database.db.driver`, `database.db.url`, etc. 
+`database.driver`, `database.db.driver`, `database.db.url`, etc.
 
 For example:
 ```
@@ -111,3 +156,18 @@ database {
 }
 ```
 
+## 0.20
+
+* The default per-upload bytes size for GCS is now the minimum 256K
+instead of 64M. There is also an undocumented config key
+`google.upload-buffer-bytes` that allows adjusting this internal value.
+
+* Updated Docker Hub hash retriever to parse json with [custom media
+types](https://github.com/docker/distribution/blob/05b0ab0/docs/spec/manifest-v2-1.md).
+
+* Added a `/batch` submit endpoint that accepts a single wdl with
+multiple input files.
+
+* The `/query` endpoint now supports querying by `id`, and submitting
+parameters as a HTTP POST.
+

README.md

@@ -101,8 +101,8 @@ There is a [Cromwell gitter channel](https://gitter.im/broadinstitute/cromwell)
 
 The following is the toolchain used for development of Cromwell.  Other versions may work, but these are recommended.
 
-* [Scala 2.11.7](http://www.scala-lang.org/news/2.11.7/)
-* [SBT 0.13.8](https://github.com/sbt/sbt/releases/tag/v0.13.8)
+* [Scala 2.11.8](http://www.scala-lang.org/news/2.11.8/)
+* [SBT 0.13.12](https://github.com/sbt/sbt/releases/tag/v0.13.12)
 * [Java 8](http://www.oracle.com/technetwork/java/javase/overview/java8-2100321.html)
 
 # Building

backend/src/main/scala/cromwell/backend/BackendJobExecutionActor.scala

@@ -24,8 +24,9 @@ object BackendJobExecutionActor {
   sealed trait BackendJobExecutionResponse extends BackendJobExecutionActorResponse { def jobKey: BackendJobDescriptorKey }
   case class SucceededResponse(jobKey: BackendJobDescriptorKey, returnCode: Option[Int], jobOutputs: JobOutputs, jobDetritusFiles: Option[Map[String, String]], executionEvents: Seq[ExecutionEvent]) extends BackendJobExecutionResponse
   case class AbortedResponse(jobKey: BackendJobDescriptorKey) extends BackendJobExecutionResponse
-  case class FailedNonRetryableResponse(jobKey: BackendJobDescriptorKey, throwable: Throwable, returnCode: Option[Int]) extends BackendJobExecutionResponse
-  case class FailedRetryableResponse(jobKey: BackendJobDescriptorKey, throwable: Throwable, returnCode: Option[Int]) extends BackendJobExecutionResponse
+  sealed trait BackendJobFailedResponse extends BackendJobExecutionResponse {  def throwable: Throwable; def returnCode: Option[Int] }
+  case class FailedNonRetryableResponse(jobKey: BackendJobDescriptorKey, throwable: Throwable, returnCode: Option[Int]) extends BackendJobFailedResponse
+  case class FailedRetryableResponse(jobKey: BackendJobDescriptorKey, throwable: Throwable, returnCode: Option[Int]) extends BackendJobFailedResponse
 }
 
 /**

backend/src/main/scala/cromwell/backend/BackendLifecycleActorFactory.scala

@@ -7,6 +7,7 @@ import com.typesafe.config.Config
 import cromwell.backend.callcaching.FileHashingActor
 import cromwell.backend.callcaching.FileHashingActor.FileHashingFunction
 import cromwell.backend.io.WorkflowPaths
+import cromwell.core.JobExecutionToken.JobExecutionTokenType
 import cromwell.core.{ExecutionStore, OutputStore}
 import wdl4s.Call
 import wdl4s.expression.WdlStandardLibraryFunctions
@@ -19,7 +20,8 @@ trait BackendLifecycleActorFactory {
 
   def jobExecutionActorProps(jobDescriptor: BackendJobDescriptor,
                              initializationData: Option[BackendInitializationData],
-                             serviceRegistryActor: ActorRef): Props
+                             serviceRegistryActor: ActorRef,
+                             backendSingletonActor: Option[ActorRef]): Props
 
   /**
     * Providing this method to generate Props for a cache hit copying actor is optional.
@@ -32,6 +34,8 @@ trait BackendLifecycleActorFactory {
     */
   def cacheHitCopyingActorProps: Option[(BackendJobDescriptor, Option[BackendInitializationData], ActorRef) => Props] = None
 
+  def backendSingletonActorProps: Option[Props] = None
+
   def workflowFinalizationActorProps(workflowDescriptor: BackendWorkflowDescriptor,
                                      calls: Seq[Call],
                                      executionStore: ExecutionStore,
@@ -52,4 +56,6 @@ trait BackendLifecycleActorFactory {
   lazy val fileHashingActorCount: Int = 50
 
   def fileHashingActorProps: Props = FileHashingActor.props(fileHashingFunction)
+
+  def jobExecutionTokenType: JobExecutionTokenType = JobExecutionTokenType("Default", None)
 }

backend/src/main/scala/cromwell/backend/BackendWorkflowInitializationActor.scala

@@ -142,7 +142,7 @@ trait BackendWorkflowInitializationActor extends BackendWorkflowLifecycleActor w
         def badRuntimeAttrsForTask(task: Task) = {
           runtimeAttributeValidators map { case (attributeName, validator) =>
             val value = task.runtimeAttributes.attrs.get(attributeName) orElse defaultRuntimeAttribute(attributeName)
-            attributeName -> (value, validator(value))
+            attributeName -> ((value, validator(value)))
           } collect {
             case (name, (value, false)) => s"Task ${task.name} has an invalid runtime attribute $name = ${value map { _.valueString} getOrElse "!! NOT FOUND !!"}"
           }

backend/src/main/scala/cromwell/backend/MemorySize.scala

@@ -1,29 +1,35 @@
 package cromwell.backend
 
-import wdl4s.parser.MemoryUnit
+
+import cats.data.Validated._
+import cats.syntax.cartesian._
+import cats.syntax.validated._
+import cromwell.core.ErrorOr._
+import mouse.string._
 
 import scala.language.postfixOps
 import scala.util.{Failure, Success, Try}
-import scalaz.Scalaz._
+import wdl4s.parser.MemoryUnit
+
 
 object MemorySize {
   val memoryPattern = """(\d+(?:\.\d+)?)\s*(\w+)""".r
 
   def parse(unparsed: String): Try[MemorySize] = {
     unparsed match {
       case memoryPattern(amountString, unitString) =>
-        val amount = amountString.parseDouble leftMap {
+        val amount: ErrorOr[Double] = amountString.parseDouble leftMap {
           _.getMessage
-        } toValidationNel
-        val unit = MemoryUnit.values find {
+        } toValidatedNel
+        val unit: ErrorOr[MemoryUnit] = MemoryUnit.values find {
           _.suffixes.contains(unitString)
         } match {
-          case Some(s) => s.successNel[String]
-          case None => s"$unitString is an invalid memory unit".failureNel
+          case Some(s) => s.validNel
+          case None => s"$unitString is an invalid memory unit".invalidNel
         }
-        (amount |@| unit) { (a, u) => new MemorySize(a, u) } match {
-          case scalaz.Success(memorySize) => Success(memorySize)
-          case scalaz.Failure(nel) => Failure(new UnsupportedOperationException(nel.list.toList.mkString("\n")))
+        (amount |@| unit) map { (a, u) => new MemorySize(a, u) } match {
+          case Valid(memorySize) => Success(memorySize)
+          case Invalid(nel) => Failure(new UnsupportedOperationException(nel.toList.mkString("\n")))
         }
       case _ => Failure(new UnsupportedOperationException(s"$unparsed should be of the form 'X Unit' where X is a number, e.g. 8 GB"))
     }

backend/src/main/scala/cromwell/backend/RuntimeAttributeDefinition.scala

@@ -39,7 +39,6 @@ object RuntimeAttributeDefinition {
   def addDefaultsToAttributes(runtimeAttributeDefinitions: Set[RuntimeAttributeDefinition], workflowOptions: WorkflowOptions)
                              (specifiedAttributes: Map[LocallyQualifiedName, WdlValue]): Map[LocallyQualifiedName, WdlValue] = {
     import WdlValueJsonFormatter._
-    import spray.json._   // IGNORE INTELLIJ - this *is* required (unless it isn't any more, who will ever know...!)
 
     def isUnspecifiedAttribute(name: String) = !specifiedAttributes.contains(name)
 

backend/src/main/scala/cromwell/backend/async/AsyncBackendJobExecutionActor.scala

@@ -70,7 +70,10 @@ trait AsyncBackendJobExecutionActor { this: Actor with ActorLogging =>
     case IssuePollRequest(handle) => robustPoll(handle)
     case PollResponseReceived(handle) if handle.isDone => self ! Finish(handle)
     case PollResponseReceived(handle) =>
+      // This should stash the Cancellable someplace so it can be cancelled once polling is complete.
+      // -Ywarn-value-discard
       context.system.scheduler.scheduleOnce(pollBackOff.backoffMillis.millis, self, IssuePollRequest(handle))
+      ()
     case Finish(SuccessfulExecutionHandle(outputs, returnCode, jobDetritusFiles, executionEvents, resultsClonedFrom)) =>
       completionPromise.success(SucceededResponse(jobDescriptor.key, Some(returnCode), outputs, Option(jobDetritusFiles), executionEvents))
       context.stop(self)

backend/src/main/scala/cromwell/backend/backend.scala

@@ -0,0 +1,53 @@
+package cromwell.backend
+
+import com.typesafe.config.Config
+import cromwell.core.WorkflowOptions.WorkflowOption
+import cromwell.core.{JobKey, WorkflowId, WorkflowOptions}
+import wdl4s.values.WdlValue
+import wdl4s.{Call, NamespaceWithWorkflow, _}
+
+import scala.util.Try
+
+/**
+  * For uniquely identifying a job which has been or will be sent to the backend.
+  */
+case class BackendJobDescriptorKey(call: Call, index: Option[Int], attempt: Int) extends JobKey {
+  def scope = call
+  private val indexString = index map { _.toString } getOrElse "NA"
+  val tag = s"${call.fullyQualifiedName}:$indexString:$attempt"
+  val isShard = index.isDefined
+  def mkTag(workflowId: WorkflowId) = s"$workflowId:$this"
+}
+
+/**
+  * For passing to a BackendWorkflowActor for job execution or recovery
+  */
+case class BackendJobDescriptor(workflowDescriptor: BackendWorkflowDescriptor,
+                                key: BackendJobDescriptorKey,
+                                runtimeAttributes: Map[LocallyQualifiedName, WdlValue],
+                                inputs: Map[LocallyQualifiedName, WdlValue]) {
+  val call = key.call
+  override val toString = s"${key.mkTag(workflowDescriptor.id)}"
+}
+
+/**
+  * For passing to a BackendActor construction time
+  */
+case class BackendWorkflowDescriptor(id: WorkflowId,
+                                     workflowNamespace: NamespaceWithWorkflow,
+                                     inputs: Map[FullyQualifiedName, WdlValue],
+                                     workflowOptions: WorkflowOptions) {
+  override def toString: String = s"[BackendWorkflowDescriptor id=${id.shortString} workflowName=${workflowNamespace.workflow.unqualifiedName}]"
+  def getWorkflowOption(key: WorkflowOption) = workflowOptions.get(key).toOption
+}
+
+/**
+  * For passing to a BackendActor construction time
+  */
+case class BackendConfigurationDescriptor(backendConfig: Config, globalConfig: Config)
+
+final case class AttemptedLookupResult(name: String, value: Try[WdlValue]) {
+  def toPair = name -> value
+}
+
+case class PreemptedException(msg: String) extends Exception(msg)

backend/src/main/scala/cromwell/backend/callcaching/CacheHitDuplicating.scala

@@ -48,7 +48,7 @@ trait CacheHitDuplicating {
 
   private def lookupSourceCallRootPath(sourceJobDetritusFiles: Map[String, String]): Path = {
     sourceJobDetritusFiles.get(JobPaths.CallRootPathKey).map(getPath).getOrElse(throw new RuntimeException(
-      s"The call detritus files for source cache hit aren't found for call ${jobDescriptor.call.fullyQualifiedName}")
+      s"${JobPaths.CallRootPathKey} wasn't found for call ${jobDescriptor.call.fullyQualifiedName}")
     )
   }
 

backend/src/main/scala/cromwell/backend/io/WorkflowPaths.scala

@@ -5,14 +5,14 @@ import java.nio.file.{FileSystem, FileSystems, Path, Paths}
 import com.typesafe.config.Config
 import cromwell.backend.{BackendJobDescriptorKey, BackendWorkflowDescriptor}
 import cromwell.core.PathFactory
-import lenthall.config.ScalaConfig._
+import net.ceedubs.ficus.Ficus._
 
 object WorkflowPaths{
   val DockerRoot = Paths.get("/root")
 }
 
 class WorkflowPaths(workflowDescriptor: BackendWorkflowDescriptor, config: Config, val fileSystems: List[FileSystem] = List(FileSystems.getDefault)) extends PathFactory {
-  val executionRoot = Paths.get(config.getStringOr("root", "cromwell-executions")).toAbsolutePath
+  val executionRoot = Paths.get(config.as[Option[String]]("root").getOrElse("cromwell-executions")).toAbsolutePath
 
   private def workflowPathBuilder(root: Path) = {
     root.resolve(workflowDescriptor.workflowNamespace.workflow.unqualifiedName)