[WX-1156] internal_path_prefix for TES 4.4

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

@@ -28,6 +28,15 @@
   val callInputsDockerRoot = callDockerRoot.resolve("inputs")
   val callInputsRoot = callRoot.resolve("inputs")
 
+  /*
+   * tesTaskRoot: This is the root directory that TES will use for files related to this task.
+   * We provide it to TES as a k/v pair where the key is "internal_path_prefix" (specified in TesWorkflowOptionKeys.scala)
+   * and the value is a blob path.
+   * This is not a standard TES feature, but rather related to the Azure TES implementation that Terra uses.
+   * While passing it outside of terra won't do any harm, we could consider making this optional and/or configurable.
+   */
+  val tesTaskRoot : Path = callExecutionRoot.resolve("tes_task")
+
   // Given an output path, return a path localized to the storage file system
   def storageOutput(path: String): String = {
     callExecutionRoot.resolve(path).toString

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

@@ -1,5 +1,4 @@
 package cromwell.backend.impl.tes
-
 import common.collections.EnhancedCollections._
 import common.util.StringUtil._
 import cromwell.backend.impl.tes.OutputMode.OutputMode
@@ -71,7 +70,6 @@
     path = tesPaths.callExecutionDockerRoot.resolve("script").toString,
     `type` = Option("FILE")
   )
-
   private def writeFunctionFiles: Map[FullyQualifiedName, Seq[WomFile]] =
     instantiatedCommand.createdFiles map { f => f.file.value.md5SumShort -> List(f.file) } toMap
 
@@ -231,11 +229,6 @@
       workflowExecutionIdentityOption
   )
 
-  val resources: Resources = TesTask.makeResources(
-    runtimeAttributes,
-    preferedWorkflowExecutionIdentity
-  )
-
   val executors = Seq(Executor(
     image = dockerImageUsed,
     command = Seq(jobShell, commandScript.path),
@@ -245,6 +238,12 @@
     stdin = None,
     env = None
   ))
+
+  val resources: Resources = TesTask.makeResources(
+    runtimeAttributes,
+    preferedWorkflowExecutionIdentity,
+    Option(tesPaths.tesTaskRoot.pathAsString)
+  )
 }
 
 object TesTask {
@@ -254,15 +253,21 @@
     configIdentity.map(_.value).orElse(workflowOptionsIdentity.map(_.value))
   }
   def makeResources(runtimeAttributes: TesRuntimeAttributes,
-                    workflowExecutionId: Option[String]): Resources = {
-
-    // This was added in BT-409 to let us pass information to an Azure
-    // TES server about which user identity to run tasks as.
-    // Note that we validate the type of WorkflowExecutionIdentity
-    // in TesInitializationActor.
-    val backendParameters = runtimeAttributes.backendParameters ++
+                    workflowExecutionId: Option[String], internalPathPrefix: Option[String]): Resources = {
+    /*
+     * workflowExecutionId: This was added in BT-409 to let us pass information to an Azure
+     * TES server about which user identity to run tasks as.
+     * Note that we validate the type of WorkflowExecutionIdentity in TesInitializationActor.
+     *
+     * internalPathPrefix: Added in WX-1156 to support the azure TES implementation. Specifies
+     * a working directory that the TES task can use.
+     */
+    val backendParameters : Map[String, Option[String]] = runtimeAttributes.backendParameters ++
       workflowExecutionId
         .map(TesWorkflowOptionKeys.WorkflowExecutionIdentity -> Option(_))
+        .toMap ++
+      internalPathPrefix
+        .map(TesWorkflowOptionKeys.InternalPathPrefix -> Option(_))
         .toMap
     val disk :: ram :: _ = Seq(runtimeAttributes.disk, runtimeAttributes.memory) map {
       case Some(x) =>

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

@@ -4,4 +4,5 @@ object TesWorkflowOptionKeys {
   // Communicates to the TES server which identity the tasks should execute as
   val WorkflowExecutionIdentity = "workflow_execution_identity"
   val DataAccessIdentity = "data_access_identity"
+  val InternalPathPrefix = "internal_path_prefix"
 }

supportedBackends/tes/src/test/scala/cromwell/backend/impl/tes/TesTaskSpec.scala

@@ -31,46 +31,87 @@ class TesTaskSpec
     false,
     Map.empty
   )
+  val internalPathPrefix = Option("mock/path/to/tes/task")
+  val expectedTuple = "internal_path_prefix" -> internalPathPrefix
 
   it should "create the correct resources when an identity is passed in WorkflowOptions" in {
     val wei = Option("abc123")
-    TesTask.makeResources(runtimeAttributes, wei) shouldEqual
-        Resources(None, None, None, Option(false), None, Option(Map(TesWorkflowOptionKeys.WorkflowExecutionIdentity -> Option("abc123")))
+    TesTask.makeResources(runtimeAttributes, wei, internalPathPrefix) shouldEqual
+        Resources(None, None, None, Option(false), None,
+          Option(Map(TesWorkflowOptionKeys.WorkflowExecutionIdentity -> Option("abc123"),
+            expectedTuple))
     )
   }
 
   it should "create the correct resources when an empty identity is passed in WorkflowOptions" in {
     val wei = Option("")
-    TesTask.makeResources(runtimeAttributes, wei) shouldEqual
-        Resources(None, None, None, Option(false), None, Option(Map(TesWorkflowOptionKeys.WorkflowExecutionIdentity -> Option("")))
+    TesTask.makeResources(runtimeAttributes, wei, internalPathPrefix) shouldEqual
+      Resources(None, None, None, Option(false), None,
+        Option(Map(TesWorkflowOptionKeys.WorkflowExecutionIdentity -> Option(""),
+          expectedTuple))
     )
   }
 
   it should "create the correct resources when no identity is passed in WorkflowOptions" in {
     val wei = None
-    TesTask.makeResources(runtimeAttributes, wei) shouldEqual
-        Resources(None, None, None, Option(false), None, Option(Map.empty[String, Option[String]])
-    )
+    TesTask.makeResources(runtimeAttributes, wei, internalPathPrefix) shouldEqual
+      Resources(None, None, None, Option(false), None, Option(Map(expectedTuple)))
   }
 
   it should "create the correct resources when an identity is passed in via backend config" in {
     val weic = Option(WorkflowExecutionIdentityConfig("abc123"))
     val weio = Option(WorkflowExecutionIdentityOption("def456"))
     val wei = TesTask.getPreferredWorkflowExecutionIdentity(weic, weio)
-    TesTask.makeResources(runtimeAttributes, wei) shouldEqual
-        Resources(None, None, None, Option(false), None, Option(Map(TesWorkflowOptionKeys.WorkflowExecutionIdentity -> Option("abc123")))
+    TesTask.makeResources(runtimeAttributes, wei, internalPathPrefix) shouldEqual
+        Resources(None, None, None, Option(false), None, Option(Map(TesWorkflowOptionKeys.WorkflowExecutionIdentity -> Option("abc123"),
+          expectedTuple))
     )
   }
 
   it should "create the correct resources when no identity is passed in via backend config" in {
     val weic = None
     val weio = Option(WorkflowExecutionIdentityOption("def456"))
     val wei = TesTask.getPreferredWorkflowExecutionIdentity(weic, weio)
-    TesTask.makeResources(runtimeAttributes, wei) shouldEqual
-        Resources(None, None, None, Option(false), None, Option(Map(TesWorkflowOptionKeys.WorkflowExecutionIdentity -> Option("def456")))
+    TesTask.makeResources(runtimeAttributes, wei, internalPathPrefix) shouldEqual
+        Resources(None, None, None, Option(false), None, Option(Map(TesWorkflowOptionKeys.WorkflowExecutionIdentity -> Option("def456"),
+          expectedTuple))
     )
   }
 
+  it should "correctly set the internal path prefix when provided as a backend parameter" in {
+    val wei = Option("abc123")
+    val internalPathPrefix = Option("mock/path/to/tes/task")
+    TesTask.makeResources(runtimeAttributes, wei, internalPathPrefix) shouldEqual
+      Resources(None, None, None, Option(false), None,
+        Option(Map(TesWorkflowOptionKeys.WorkflowExecutionIdentity -> Option("abc123"),
+          "internal_path_prefix" -> internalPathPrefix)
+      ))
+  }
+
+  it should "correctly resolve the path to .../tes_task and add the k/v pair to backend parameters" in {
+    val emptyWorkflowOptions = WorkflowOptions(JsObject(Map.empty[String, JsValue]))
+    val workflowDescriptor = buildWdlWorkflowDescriptor(TestWorkflows.HelloWorld,
+      labels = Labels("foo" -> "bar"))
+    val jobDescriptor = jobDescriptorFromSingleCallWorkflow(workflowDescriptor,
+      Map.empty,
+      emptyWorkflowOptions,
+      Set.empty)
+    val tesPaths = TesJobPaths(jobDescriptor.key,
+      jobDescriptor.workflowDescriptor,
+      TestConfig.emptyConfig)
+
+    val expectedKey = "internal_path_prefix"
+    val expectedValue = Option(tesPaths.tesTaskRoot.pathAsString)
+
+    //Assert path correctly ends up in the resources
+    val wei = Option("abc123")
+    TesTask.makeResources(runtimeAttributes, wei, expectedValue) shouldEqual
+      Resources(None, None, None, Option(false), None,
+        Option(Map(TesWorkflowOptionKeys.WorkflowExecutionIdentity -> Option("abc123"),
+          expectedKey -> expectedValue))
+      )
+  }
+
   it should "copy labels to tags" in {
     val jobLogger = mock[JobLogger]
     val emptyWorkflowOptions = WorkflowOptions(JsObject(Map.empty[String, JsValue]))