TES Backend Prototype. Closes #1294

backend/src/main/scala/cromwell/backend/wdl/WdlFileMapper.scala

@@ -18,6 +18,7 @@ object WdlFileMapper {
         val mappedMap = map.value map {
           case (key, value) => mapWdlFiles(mapper)(key) -> mapWdlFiles(mapper)(value)
         }
+
         TryUtil.sequenceKeyValues(mappedMap) map {
           WdlMap(map.wdlType, _)
         }

build.sbt

@@ -44,6 +44,12 @@ lazy val sfsBackend = (project in backendRoot / "sfs")
   .dependsOn(gcsFileSystem)
   .dependsOn(backend % "test->test")
 
+lazy val tesBackend = (project in backendRoot / "tes")
+  .settings(tesBackendSettings:_*)
+  .withTestSettings
+  .dependsOn(sfsBackend)
+  .dependsOn(backend % "test->test")
+
 lazy val htCondorBackend = (project in backendRoot / "htcondor")
   .settings(htCondorBackendSettings:_*)
   .withTestSettings
@@ -93,10 +99,12 @@ lazy val root = (project in file("."))
   .aggregate(htCondorBackend)
   .aggregate(sparkBackend)
   .aggregate(jesBackend)
+  .aggregate(tesBackend)
   .aggregate(engine)
   // Next level of projects to include in the fat jar (their dependsOn will be transitively included)
   .dependsOn(engine)
   .dependsOn(jesBackend)
+  .dependsOn(tesBackend)
   .dependsOn(htCondorBackend)
   .dependsOn(sparkBackend)
   // Dependencies for tests

core/src/main/resources/reference.conf

@@ -102,7 +102,7 @@ workflow-options {
   #workflow-failure-mode: "ContinueWhilePossible"
 }
 
-// Optional call-caching configuration.
+# Optional call-caching configuration.
 call-caching {
   # Allows re-use of existing results for jobs you've already run
   # (default: false)
@@ -175,7 +175,7 @@ backend {
         # Root directory where Cromwell writes job results.  This directory must be
         # visible and writeable by the Cromwell process as well as the jobs that Cromwell
         # launches.
-        root: "cromwell-executions"
+        root = "cromwell-executions"
 
         filesystems {
           local {
@@ -204,6 +204,15 @@ backend {
       }
     }
 
+    #TES {
+    #  actor-factory = "cromwell.backend.impl.tes.TesBackendLifecycleActorFactory"
+    #  config {
+    #    root = "cromwell-executions"
+    #    endpoint = "http://127.0.0.1:8000/v1/jobs"
+    #    poll-interval = 10
+    #  }
+    #}
+
     #SGE {
     #  actor-factory = "cromwell.backend.impl.sfs.config.ConfigBackendLifecycleActorFactory"
     #  config {

project/Dependencies.scala

@@ -132,6 +132,10 @@ object Dependencies {
     "org.mongodb" %% "casbah" % "3.0.0"
   )
 
+  val tesBackendDependencies = List(
+    "io.spray" %% "spray-client" % sprayV
+  ) ++ sprayServerDependencies
+
   val sparkBackendDependencies = List(
     "io.spray" %% "spray-client" % sprayV
   ) ++ sprayServerDependencies

project/Settings.scala

@@ -142,6 +142,11 @@ object Settings {
     name := "cromwell-sfs-backend"
   ) ++ commonSettings
 
+  val tesBackendSettings = List(
+    name := "cromwell-tes-backend",
+    libraryDependencies ++= tesBackendDependencies
+  ) ++ commonSettings
+
   val htCondorBackendSettings = List(
     name := "cromwell-htcondor-backend",
     libraryDependencies ++= htCondorBackendDependencies

supportedBackends/jes/src/test/scala/cromwell/backend/impl/jes/JesRuntimeAttributesSpec.scala

@@ -13,7 +13,7 @@ import wdl4s.parser.MemoryUnit
 import wdl4s.types.{WdlArrayType, WdlIntegerType, WdlStringType}
 import wdl4s.values.{WdlArray, WdlBoolean, WdlInteger, WdlString, WdlValue}
 
-class JesRuntimeAttributesSpec extends WordSpecLike with Matchers with Mockito {
+class  JesRuntimeAttributesSpec extends WordSpecLike with Matchers with Mockito {
 
   def workflowOptionsWithDefaultRA(defaults: Map[String, JsValue]): WorkflowOptions = {
     WorkflowOptions(JsObject(Map(

supportedBackends/tes/src/main/scala/cromwell/backend/impl/tes/TesAsyncBackendJobExecutionActor.scala

@@ -0,0 +1,194 @@
+package cromwell.backend.impl.tes
+
+import java.nio.file.FileAlreadyExistsException
+
+import cromwell.backend.BackendJobLifecycleActor
+import cromwell.backend.async.{ExecutionHandle, FailedNonRetryableExecutionHandle, PendingExecutionHandle}
+import cromwell.backend.impl.tes.TesResponseJsonFormatter._
+import cromwell.backend.standard.{StandardAsyncExecutionActor, StandardAsyncExecutionActorParams, StandardAsyncJob}
+import cromwell.core.path.Obsolete._
+import cromwell.core.path.Path
+import cromwell.core.retry.SimpleExponentialBackoff
+import spray.client.pipelining._
+import spray.http.HttpRequest
+import spray.httpx.SprayJsonSupport._
+import spray.httpx.unmarshalling._
+import wdl4s.values.WdlFile
+
+import scala.concurrent.duration._
+import scala.concurrent.{ExecutionContext, Future}
+import scala.language.postfixOps
+import scala.util.{Failure, Success}
+
+sealed trait TesRunStatus {
+  def isTerminal: Boolean
+}
+
+case object Running extends TesRunStatus {
+  def isTerminal = false
+}
+
+case object Complete extends TesRunStatus {
+  def isTerminal = true
+}
+
+case object FailedOrError extends TesRunStatus {
+  def isTerminal = true
+}
+
+object TesAsyncBackendJobExecutionActor {
+  val JobIdKey = "tes_job_id"
+}
+
+class TesAsyncBackendJobExecutionActor(override val standardParams: StandardAsyncExecutionActorParams)
+  extends BackendJobLifecycleActor with StandardAsyncExecutionActor with TesJobCachingActorHelper {
+
+  override type StandardAsyncRunInfo = Any
+
+  override type StandardAsyncRunStatus = TesRunStatus
+
+  override lazy val pollBackOff = SimpleExponentialBackoff(
+    initialInterval = 1 seconds,
+    maxInterval = 5 minutes,
+    multiplier = 1.1
+  )
+
+  override lazy val executeOrRecoverBackOff = SimpleExponentialBackoff(
+    initialInterval = 3 seconds,
+    maxInterval = 30 seconds,
+    multiplier = 1.1
+  )
+
+  override lazy val retryable: Boolean = false
+
+  private val tesEndpoint = tesConfiguration.endpointURL
+
+  override lazy val jobTag = jobDescriptor.key.tag
+
+  private def pipeline[T: FromResponseUnmarshaller]: HttpRequest => Future[T] = sendReceive ~> unmarshal[T]
+
+  // Utility for converting a WdlValue so that the path is localized to the
+  // container's filesystem.
+  override def mapCommandLineWdlFile(wdlFile: WdlFile): WdlFile = {
+    val localPath = Paths.get(wdlFile.valueString).toAbsolutePath
+    // Workaround since each input seemed to hit this function twice?
+    if (localPath.startsWith(tesJobPaths.callInputsDockerRoot)) {
+      WdlFile(localPath.pathAsString)
+    } else {
+      val containerPath = tesJobPaths.containerInput(localPath.pathAsString)
+      WdlFile(containerPath)
+    }
+  }
+
+  override lazy val commandDirectory: Path = tesJobPaths.callExecutionDockerRoot
+
+  def createTaskMessage(): TesTaskMessage = {
+    tesJobPaths.script.write(commandScriptContents)
+
+    val task = TesTask(jobDescriptor, configurationDescriptor, jobLogger, tesJobPaths, runtimeAttributes)
+
+    TesTaskMessage(
+      Option(task.name),
+      Option(task.description),
+      Option(task.project),
+      Option(task.inputs),
+      Option(task.outputs),
+      task.resources,
+      task.dockerExecutor
+    )
+  }
+
+  override def executeAsync()(implicit ec: ExecutionContext): Future[ExecutionHandle] = {
+    // create call exec dir
+    File(tesJobPaths.callExecutionRoot).createPermissionedDirectories()
+    val taskMessage = createTaskMessage()
+
+    val submitTask = pipeline[TesPostResponse]
+      .apply(Post(tesEndpoint, taskMessage))
+
+    submitTask.map {
+      response =>
+        val jobID = response.value
+        PendingExecutionHandle(jobDescriptor, StandardAsyncJob(jobID), None, previousStatus = None)
+    }
+  }
+
+  override def tryAbort(job: StandardAsyncJob): Unit = {
+
+    val returnCodeTmp = jobPaths.returnCode.plusExt("kill")
+    returnCodeTmp.write(s"$SIGTERM\n")
+    try {
+      returnCodeTmp.moveTo(jobPaths.returnCode)
+    } catch {
+      case _: FileAlreadyExistsException =>
+        // If the process has already completed, there will be an existing rc file.
+        returnCodeTmp.delete(true)
+    }
+
+    val abortRequest = pipeline[TesGetResponse]
+      .apply(Delete(s"$tesEndpoint/${job.jobId}"))
+    abortRequest onComplete {
+      case Success(_) => jobLogger.info("{} Aborted {}", tag: Any, job.jobId)
+      case Failure(ex) => jobLogger.warn("{} Failed to abort {}: {}", tag, job.jobId, ex.getMessage)
+    }
+    ()
+  }
+
+  override def requestsAbortAndDiesImmediately: Boolean = true
+
+  override def pollStatusAsync(handle: StandardAsyncPendingExecutionHandle)(implicit ec: ExecutionContext): Future[TesRunStatus] = {
+    val pollTask = pipeline[TesGetResponse].apply(Get(s"$tesEndpoint/${handle.pendingJob.jobId}"))
+
+    pollTask.map {
+      response =>
+        val state = response.state
+        state match {
+          case s if s.contains("Complete") => {
+            jobLogger.info(s"Job ${handle.pendingJob.jobId} is complete")
+            Complete
+          }
+          case s if s.contains("Cancel") => {
+            jobLogger.info(s"Job ${handle.pendingJob.jobId} was canceled")
+            FailedOrError
+          }
+          case s if s.contains("Error") => {
+            jobLogger.info(s"TES reported an error for Job ${handle.pendingJob.jobId}")
+            FailedOrError
+          }
+          case _ => Running
+        }
+    }
+  }
+
+  override def customPollStatusFailure: PartialFunction[(ExecutionHandle, Exception), ExecutionHandle] = {
+    case (oldHandle: StandardAsyncPendingExecutionHandle@unchecked, e: Exception) => {
+      jobLogger.error(s"$tag TES Job ${oldHandle.pendingJob.jobId} has not been found, failing call")
+      FailedNonRetryableExecutionHandle(e)
+    }
+  }
+
+  override def isTerminal(runStatus: TesRunStatus): Boolean = {
+    runStatus.isTerminal
+  }
+
+  override def isSuccess(runStatus: TesRunStatus): Boolean = {
+    runStatus match {
+      case Complete => true
+      case _ => false
+    }
+  }
+
+  // Everything below was 'borrowed' from the SFS backend
+  private def hostAbsoluteFilePath(wdlFile: WdlFile): File = {
+    tesJobPaths.callExecutionRoot.resolve(wdlFile.value).toAbsolutePath
+  }
+
+  override def mapOutputWdlFile(wdlFile: WdlFile): WdlFile = {
+    if (!hostAbsoluteFilePath(wdlFile).exists) {
+      throw new RuntimeException("Could not process output, file not found: " +
+        s"${hostAbsoluteFilePath(wdlFile).pathAsString}")
+    } else {
+      WdlFile(hostAbsoluteFilePath(wdlFile).pathAsString)
+    }
+  }
+}

supportedBackends/tes/src/main/scala/cromwell/backend/impl/tes/TesBackendInitializationData.scala

@@ -0,0 +1,10 @@
+package cromwell.backend.impl.tes
+
+import cromwell.backend.standard.{StandardExpressionFunctions, StandardInitializationData, StandardValidatedRuntimeAttributesBuilder}
+
+case class TesBackendInitializationData
+(
+  override val workflowPaths: TesWorkflowPaths,
+  override val runtimeAttributesBuilder: StandardValidatedRuntimeAttributesBuilder,
+  tesConfiguration: TesConfiguration
+) extends StandardInitializationData(workflowPaths, runtimeAttributesBuilder, classOf[StandardExpressionFunctions])

supportedBackends/tes/src/main/scala/cromwell/backend/impl/tes/TesBackendLifecycleActorFactory.scala

@@ -0,0 +1,31 @@
+package cromwell.backend.impl.tes
+
+import akka.actor.ActorRef
+import cromwell.backend._
+import cromwell.backend.standard._
+import cromwell.core.JobExecutionToken.JobExecutionTokenType
+import net.ceedubs.ficus.Ficus._
+import wdl4s.TaskCall
+
+case class TesBackendLifecycleActorFactory(name: String, configurationDescriptor: BackendConfigurationDescriptor)
+  extends StandardLifecycleActorFactory {
+
+  override lazy val initializationActorClass: Class[_ <: StandardInitializationActor] = classOf[TesInitializationActor]
+
+  override lazy val  asyncExecutionActorClass: Class[_ <: StandardAsyncExecutionActor] =
+    classOf[TesAsyncBackendJobExecutionActor]
+
+  override def jobIdKey: String = TesAsyncBackendJobExecutionActor.JobIdKey
+
+  val tesConfiguration = new TesConfiguration(configurationDescriptor)
+
+  override val jobExecutionTokenType: JobExecutionTokenType = {
+    val concurrentJobLimit = configurationDescriptor.backendConfig.as[Option[Int]]("concurrent-job-limit")
+    JobExecutionTokenType(name, concurrentJobLimit)
+  }
+
+  override def workflowInitializationActorParams(workflowDescriptor: BackendWorkflowDescriptor, calls: Set[TaskCall],
+                                                 serviceRegistryActor: ActorRef): StandardInitializationActorParams = {
+    TesInitializationActorParams(workflowDescriptor, calls, tesConfiguration, serviceRegistryActor)
+  }
+}

supportedBackends/tes/src/main/scala/cromwell/backend/impl/tes/TesConfiguration.scala

@@ -0,0 +1,9 @@
+package cromwell.backend.impl.tes
+
+import cromwell.backend.BackendConfigurationDescriptor
+
+
+class TesConfiguration(val configurationDescriptor: BackendConfigurationDescriptor) {
+  val root = configurationDescriptor.backendConfig.getString("root")
+  val endpointURL = configurationDescriptor.backendConfig.getString("endpoint")
+}

supportedBackends/tes/src/main/scala/cromwell/backend/impl/tes/TesInitializationActor.scala

@@ -0,0 +1,48 @@
+package cromwell.backend.impl.tes
+
+import akka.actor.ActorRef
+import cromwell.backend.standard._
+import cromwell.backend.{BackendConfigurationDescriptor, BackendInitializationData, BackendWorkflowDescriptor}
+import cromwell.core.path.Obsolete._
+import cromwell.core.path.{DefaultPathBuilderFactory, PathBuilder, PathBuilderFactory}
+import wdl4s.TaskCall
+
+import scala.concurrent.Future
+import scala.util.Try
+
+case class TesInitializationActorParams
+(
+  workflowDescriptor: BackendWorkflowDescriptor,
+  calls: Set[TaskCall],
+  tesConfiguration: TesConfiguration,
+  serviceRegistryActor: ActorRef
+) extends StandardInitializationActorParams {
+  override val configurationDescriptor: BackendConfigurationDescriptor = tesConfiguration.configurationDescriptor
+}
+
+class TesInitializationActor(params: TesInitializationActorParams)
+  extends StandardInitializationActor(params) {
+
+  private val tesConfiguration = params.tesConfiguration
+
+  lazy val pathBuilderFactories: List[PathBuilderFactory] = List(Option(DefaultPathBuilderFactory)).flatten
+
+  override lazy val pathBuilders: List[PathBuilder] =
+    pathBuilderFactories map {
+      _.withOptions(workflowDescriptor.workflowOptions)(context.system)
+    }
+
+  override lazy val workflowPaths: TesWorkflowPaths =
+    new TesWorkflowPaths(workflowDescriptor, tesConfiguration.configurationDescriptor.backendConfig, pathBuilders)
+
+  override lazy val runtimeAttributesBuilder: StandardValidatedRuntimeAttributesBuilder =
+    TesRuntimeAttributes.runtimeAttributesBuilder
+
+  override def beforeAll(): Future[Option[BackendInitializationData]] = {
+    Future.fromTry(Try {
+      publishWorkflowRoot(workflowPaths.workflowRoot.toString)
+      File(workflowPaths.workflowRoot).createPermissionedDirectories()
+      Option(TesBackendInitializationData(workflowPaths, runtimeAttributesBuilder, tesConfiguration))
+    })
+  }
+}