Add nextflow tags to AWS Batch job def

Signed-off-by: Paolo Di Tommaso <paolo.ditommaso@gmail.com>

plugins/nf-amazon/src/main/nextflow/cloud/aws/batch/AwsBatchTaskHandler.groovy

@@ -21,6 +21,7 @@ import static nextflow.cloud.aws.batch.AwsContainerOptionsMapper.*
 
 import java.nio.file.Path
 import java.nio.file.Paths
+import java.time.Instant
 
 import com.amazonaws.services.batch.AWSBatch
 import com.amazonaws.services.batch.model.AWSBatchException
@@ -53,13 +54,14 @@ import com.amazonaws.services.batch.model.Volume
 import groovy.transform.CompileStatic
 import groovy.transform.Memoized
 import groovy.util.logging.Slf4j
+import nextflow.Const
 import nextflow.cloud.types.CloudMachineInfo
 import nextflow.container.ContainerNameValidator
 import nextflow.exception.ProcessSubmitException
 import nextflow.exception.ProcessUnrecoverableException
 import nextflow.executor.BashWrapperBuilder
-import nextflow.fusion.FusionAwareTask
 import nextflow.executor.res.AcceleratorResource
+import nextflow.fusion.FusionAwareTask
 import nextflow.processor.BatchContext
 import nextflow.processor.BatchHandler
 import nextflow.processor.TaskHandler
@@ -593,6 +595,10 @@ class AwsBatchTaskHandler extends TaskHandler implements BatchHandler<String,Job
      * @return The fully qualified Batch job definition name eg {@code my-job-definition:3}
      */
     protected String createJobDef(RegisterJobDefinitionRequest req) {
+        // add nextflow tags
+        req.addTagsEntry('nextflow.io/createdAt', Instant.now().toString())
+        req.addTagsEntry('nextflow.io/version', Const.APP_VER)
+        // create the job def
         final res = createJobDef0(bypassProxy(client), req) // bypass the client proxy! see #1024
         return "${res.jobDefinitionName}:$res.revision"
     }

plugins/nf-amazon/src/test/nextflow/cloud/aws/batch/AwsBatchTaskHandlerTest.groovy

@@ -18,6 +18,7 @@
 package nextflow.cloud.aws.batch
 
 import java.nio.file.Paths
+import java.time.Instant
 
 import com.amazonaws.services.batch.AWSBatch
 import com.amazonaws.services.batch.model.ContainerProperties
@@ -35,6 +36,7 @@ import com.amazonaws.services.batch.model.RetryStrategy
 import com.amazonaws.services.batch.model.SubmitJobRequest
 import com.amazonaws.services.batch.model.SubmitJobResult
 import com.amazonaws.services.batch.model.TerminateJobRequest
+import nextflow.Const
 import nextflow.cloud.aws.util.S3PathFactory
 import nextflow.cloud.types.CloudMachineInfo
 import nextflow.cloud.types.PriceModel
@@ -484,7 +486,7 @@ class AwsBatchTaskHandlerTest extends Specification {
         def handler = Spy(AwsBatchTaskHandler)
         handler.@client = client
 
-        def req = Mock(RegisterJobDefinitionRequest)
+        def req = new RegisterJobDefinitionRequest()
         def res = Mock(RegisterJobDefinitionResult)
 
         when:
@@ -493,7 +495,11 @@ class AwsBatchTaskHandlerTest extends Specification {
         1 * client.registerJobDefinition(req) >> res
         1 * res.getJobDefinitionName() >> JOB_NAME
         1 * res.getRevision() >> 10
+        and:
         result == "$JOB_NAME:10"
+        and:
+        req.getTags().get('nextflow.io/version') == Const.APP_VER
+        Instant.parse(req.getTags().get('nextflow.io/createdAt'))
 
     }