AN-786: Make propagateTags a configurable AWS batch runtime attribute

CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## 92 Release Notes
 
+### AWS Batch
+* Exposed the option to propagate job tags to the underlying ECS task 
+
 ### Progress toward WDL 1.1 Support
 * WDL 1.1 support is in progress. Users that would like to try out the current partial support can do so by using WDL version `development-1.1`. In Cromwell 92, `development-1.1` has been enhanced to include:
     * Support for passthrough syntax for call inputs, e.g. `{ input: foo }` rather than `{ input: foo = foo }`.

centaur/src/main/resources/standardTestCases/aws_label_propagation.test

@@ -0,0 +1,20 @@
+name: awsbatch_jes_labels
+testFormat: workflowsuccess
+tags: [ labels ]
+backends: [AWSBATCH]
+
+files {
+  workflow: labels/jes_labels.wdl
+  labels: labels/valid.labels
+}
+
+metadata {
+  workflowName: I_hope_nobody_names_workflows_like_this_as_it_seems_very_unnecessary_
+  status: Succeeded
+  "calls.I_hope_nobody_names_workflows_like_this_as_it_seems_very_unnecessary_.my_task_aliased.backendLabels.cromwell-workflow-id": "cromwell-<<UUID>>"
+  "calls.I_hope_nobody_names_workflows_like_this_as_it_seems_very_unnecessary_.my_task_aliased.backendLabels.wdl-task-name": "my_task"
+  "calls.I_hope_nobody_names_workflows_like_this_as_it_seems_very_unnecessary_.my_task_aliased.backendLabels.wdl-call-alias": "my_task_aliased"
+  "calls.I_hope_nobody_names_workflows_like_this_as_it_seems_very_unnecessary_.my_task_aliased.labels.wdl-task-name": "MY_TASK"
+  "calls.I_hope_nobody_names_workflows_like_this_as_it_seems_very_unnecessary_.my_task_aliased.labels.wdl-call-alias": "my_task_aliased"
+  "labels.cromwell-workflow-id": "cromwell-<<UUID>>"
+}

centaur/src/main/resources/standardTestCases/labels/jes_labels.wdl

@@ -10,5 +10,7 @@ task MY_TASK {
   }
   runtime {
     docker: "ubuntu:latest"
+    tagResources: true
+    propagateTags: true
   }
 }

supportedBackends/aws/src/main/scala/cromwell/backend/impl/aws/AwsBatchAsyncBackendJobExecutionActor.scala

@@ -263,7 +263,8 @@ class AwsBatchAsyncBackendJobExecutionActor(
       Option(runtimeAttributes.tagResources),
       runtimeAttributes.logGroupName,
       runtimeAttributes.additionalTags,
-      scriptBucketPrefix
+      scriptBucketPrefix,
+      Option(runtimeAttributes.propagateTags)
     )
 
   // setup batch client to query job container info

supportedBackends/aws/src/main/scala/cromwell/backend/impl/aws/AwsBatchJob.scala

@@ -95,7 +95,8 @@ final case class AwsBatchJob(
   tagResources: Option[Boolean],
   logGroupName: String,
   additionalTags: Map[String, String],
-  scriptBucketPrefix: Option[String]
+  scriptBucketPrefix: Option[String],
+  propagateTags: Option[Boolean]
 ) {
 
   val Log: Logger = LoggerFactory.getLogger(AwsBatchJob.getClass)
@@ -683,11 +684,14 @@ final case class AwsBatchJob(
         .tags(runtimeAttributes.additionalTags.asJava)
         .jobQueue(runtimeAttributes.queueArn)
         .jobDefinition(definitionArn)
-      // tagging activated : add to request
+      // tagging activated: add metadata (custom labels) and engine tags to request
       if (tagResources.getOrElse(false)) {
         // replace invalid characters in the tags
         val invalidCharsPattern = "[^a-zA-Z0-9_.:/=+-@]+".r
-        val tags: Map[String, String] = Map(
+        val customLabels: Map[String, String] = jobDescriptor.workflowDescriptor.customLabels.asMap.map { case (k, v) =>
+          invalidCharsPattern.replaceAllIn(k, "_") -> invalidCharsPattern.replaceAllIn(v, "_")
+        }
+        val Tags: Map[String, String] = Map(
           "cromwell-workflow-name" -> invalidCharsPattern.replaceAllIn(workflowName, "_"),
           "cromwell-workflow-id" -> invalidCharsPattern.replaceAllIn(workflowId, "_"),
           "cromwell-task-id" -> invalidCharsPattern.replaceAllIn(taskId, "_"),
@@ -698,7 +702,12 @@ final case class AwsBatchJob(
             "_"
           )
         )
-        submitJobRequest = submitJobRequest.tags(tags.asJava).propagateTags(true)
+
+        // Combine both maps - Tags will override customLabels if there are duplicate keys
+        val allTags: Map[String, String] = customLabels ++ Tags
+
+        val doPropagation = propagateTags.getOrElse(false)
+        submitJobRequest = submitJobRequest.tags(allTags.asJava).propagateTags(doPropagation)
       }
       // JobTimeout provided (positive value) : add to request
       if (runtimeAttributes.jobTimeout > 0) {

supportedBackends/aws/src/main/scala/cromwell/backend/impl/aws/AwsBatchRuntimeAttributes.scala

@@ -77,6 +77,7 @@ import scala.jdk.CollectionConverters._
  * @param tagResources should we tag resources
  * @param logGroupName the CloudWatch log group name to write logs to
  * @param additionalTags a map of tags to add to the AWS Batch job submission
+  * @param propagateTags should we propagate tags to ECS tasks underlying batch jobs
  */
 case class AwsBatchRuntimeAttributes(cpu: Int Refined Positive,
                                      gpuCount: Int,
@@ -100,7 +101,8 @@ case class AwsBatchRuntimeAttributes(cpu: Int Refined Positive,
                                      additionalTags: Map[String, String],
                                      fuseMount: Boolean,
                                      fileSystem: String = "s3",
-                                     tagResources: Boolean = false
+                                     tagResources: Boolean = false,
+                                     propagateTags: Boolean = false
 )
 
 object AwsBatchRuntimeAttributes {
@@ -121,6 +123,7 @@ object AwsBatchRuntimeAttributes {
   val awsBatchefsDelocalizeKey = "efsDelocalize"
   val awsBatchefsMakeMD5Key = "efsMakeMD5"
   val tagResourcesKey = "tagResources"
+  val propagateResourcesKey = "propagateTags"
   val ZonesKey = "zones"
   private val ZonesDefaultValue = WomString("us-east-1a")
 
@@ -257,6 +260,13 @@ object AwsBatchRuntimeAttributes {
         .getOrElse(WomBoolean(false))
     )
 
+  private def awsBatchPropagateTagsValidation(runtimeConfig: Option[Config]): RuntimeAttributesValidation[Boolean] =
+    AwsBatchtagResourcesValidation(AwsBatchRuntimeAttributes.propagateResourcesKey).withDefault(
+      AwsBatchtagResourcesValidation(AwsBatchRuntimeAttributes.propagateResourcesKey)
+        .configDefaultWomValue(runtimeConfig)
+        .getOrElse(WomBoolean(false))
+    )
+
   private def ulimitsValidation(
     runtimeConfig: Option[Config]
   ): RuntimeAttributesValidation[Vector[Map[String, String]]] =
@@ -313,6 +323,7 @@ object AwsBatchRuntimeAttributes {
         awsBatchefsDelocalizeValidation(runtimeConfig),
         awsBatchefsMakeMD5Validation(runtimeConfig),
         awsBatchtagResourcesValidation(runtimeConfig),
+        awsBatchPropagateTagsValidation(runtimeConfig),
         sharedMemorySizeValidation(runtimeConfig),
         fuseMountValidation(runtimeConfig),
         jobTimeoutValidation(runtimeConfig)
@@ -336,6 +347,7 @@ object AwsBatchRuntimeAttributes {
         awsBatchefsDelocalizeValidation(runtimeConfig),
         awsBatchefsMakeMD5Validation(runtimeConfig),
         awsBatchtagResourcesValidation(runtimeConfig),
+        awsBatchPropagateTagsValidation(runtimeConfig),
         sharedMemorySizeValidation(runtimeConfig),
         fuseMountValidation(runtimeConfig),
         jobTimeoutValidation(runtimeConfig)
@@ -413,6 +425,10 @@ object AwsBatchRuntimeAttributes {
     val tagResources: Boolean = RuntimeAttributesValidation.extract(awsBatchtagResourcesValidation(runtimeAttrsConfig),
                                                                     validatedRuntimeAttributes
     )
+    val propagateTags: Boolean = RuntimeAttributesValidation.extract(
+      awsBatchPropagateTagsValidation(runtimeAttrsConfig),
+      validatedRuntimeAttributes
+    )
     val sharedMemorySize: MemorySize =
       RuntimeAttributesValidation.extract(sharedMemorySizeValidation(runtimeAttrsConfig), validatedRuntimeAttributes)
     val jobTimeout: Int =
@@ -443,7 +459,8 @@ object AwsBatchRuntimeAttributes {
       additionalTags,
       fuseMount,
       fileSystem,
-      tagResources
+      tagResources,
+      propagateTags
     )
   }
 }

supportedBackends/aws/src/main/scala/cromwell/backend/impl/aws/README.md

@@ -670,7 +670,7 @@ backend {
 
 ```
 
-Additional, custom tags can be added to jobs, using the "additionalTags" paramter in the "default-runtime-attributes" section of the job definition:
+Additional, custom tags can be added to jobs, using the "additionalTags" parameter in the "default-runtime-attributes" section of the job definition:
 
 ```
 backend {
@@ -691,7 +691,25 @@ backend {
 
 The _logGroupName_ enables you to send the logs to a custom log group name and tag the jobs that Cromwell submits.  The _additionalTags_ allows you to specify tags to be added to the jobs as <key> : <value> pairs. 
 
+Tags can be propagated to the underlying AWS ECS tasks by adding the "propagateRags = true" to the default-runtime-attributes section of your configuration:
 
+```
+backend {
+    providers {
+        AWSBatch {
+            config{
+
+                default-runtime-attributes {
+                    // enable detailed tagging
+                    tagResources = true
+                    propagateTags = true
+                }
+            }
+        }
+    }
+}
+
+```
 
 AWS Batch
 ---------

supportedBackends/aws/src/test/scala/cromwell/backend/impl/aws/AwsBatchJobSpec.scala

@@ -188,6 +188,7 @@ class AwsBatchJobSpec extends TestKitSuite with AnyFlatSpecLike with Matchers wi
       None,
       "",
       Map.empty,
+      None,
       None
     )
     job
@@ -216,6 +217,7 @@ class AwsBatchJobSpec extends TestKitSuite with AnyFlatSpecLike with Matchers wi
       None,
       "",
       Map.empty,
+      None,
       None
     )
     job
@@ -244,6 +246,7 @@ class AwsBatchJobSpec extends TestKitSuite with AnyFlatSpecLike with Matchers wi
       None,
       "",
       Map.empty,
+      None,
       None
     )
     job
@@ -652,7 +655,8 @@ class AwsBatchJobSpec extends TestKitSuite with AnyFlatSpecLike with Matchers wi
       None,
       "",
       Map.empty,
-      Some("my-project/workflow-123")
+      Some("my-project/workflow-123"),
+      None
     )
 
     // Verify the trailing slash is added to ensure proper S3 key formation
@@ -683,7 +687,8 @@ class AwsBatchJobSpec extends TestKitSuite with AnyFlatSpecLike with Matchers wi
       None,
       "",
       Map.empty,
-      Some("")
+      Some(""),
+      None
     )
 
     job.scriptKeyPrefix should be("scripts/")
@@ -716,7 +721,8 @@ class AwsBatchJobSpec extends TestKitSuite with AnyFlatSpecLike with Matchers wi
       None,
       "",
       Map.empty,
-      Some("my-project/scripts/")
+      Some("my-project/scripts/"),
+      None
     )
 
     // Verify that the existing trailing slash is preserved (not doubled)