Fix container meta deduplication

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

Author: pditommaso

plugins/nf-tower/src/main/io/seqera/tower/plugin/TowerClient.groovy

@@ -141,7 +141,7 @@ class TowerClient implements TraceObserver {
 
     private TowerReports reports
 
-    private Map<String,Map<String,Object>> allContainers = new ConcurrentHashMap<>()
+    private Map<String,Boolean> allContainers = new ConcurrentHashMap<>()
 
     /**
      * Constructor that consumes a URL and creates
@@ -720,8 +720,10 @@ class TowerClient implements TraceObserver {
         final result = new ArrayList<ContainerMeta>()
         for( TraceRecord it : tasks ) {
             final meta = it.getContainerMeta()
-            if( meta && !allContainers.containsKey(meta.targetImage) )
+            if( meta && !allContainers.get(meta.targetImage) ) {
+                allContainers.put(meta.targetImage, Boolean.TRUE)
                 result.add(meta)
+            }
         }
         return result
     }

plugins/nf-tower/src/test/io/seqera/tower/plugin/TowerClientTest.groovy

@@ -479,4 +479,49 @@ class TowerClientTest extends Specification {
         "local-platform::${ProcessHelper.selfPid()}"    | null          | null        | [TOWER_ALLOW_NEXTFLOW_LOGS:'true']
         'aws-batch::1234z'                              | 'xyz.out'     | 'hola.log'  | [TOWER_ALLOW_NEXTFLOW_LOGS:'true', AWS_BATCH_JOB_ID: '1234z', NXF_OUT_FILE: 'xyz.out', NXF_LOG_FILE: 'hola.log']
     }
+
+    def 'should deduplicate containers' () {
+        given:
+        def client = Spy(new TowerClient())
+        and:
+        def c1 = new ContainerMeta(requestId: '12345', sourceImage: 'ubuntu:latest', targetImage: 'wave.io/12345/ubuntu:latest')
+        def c2 = new ContainerMeta(requestId: '54321', sourceImage: 'ubuntu:latest', targetImage: 'wave.io/54321/ubuntu:latest')
+        and:
+        def trace1 = new TraceRecord(
+                taskId: 1,
+                process: 'foo',
+                workdir: "/work/dir",
+                cpus: 1,
+                submit: System.currentTimeMillis(),
+                start: System.currentTimeMillis(),
+                complete: System.currentTimeMillis())
+        trace1.containerMeta = c1
+        and:
+        def trace2 = new TraceRecord(
+            taskId: 2,
+            process: 'foo',
+            workdir: "/work/dir",
+            cpus: 1,
+            submit: System.currentTimeMillis(),
+            start: System.currentTimeMillis(),
+            complete: System.currentTimeMillis())
+        trace2.containerMeta = c2
+        and:
+        def trace3 = new TraceRecord(
+            taskId: 3,
+            process: 'foo',
+            workdir: "/work/dir",
+            cpus: 1,
+            submit: System.currentTimeMillis(),
+            start: System.currentTimeMillis(),
+            complete: System.currentTimeMillis())
+        trace3.containerMeta = c2
+
+        expect:
+        client.getNewContainers([trace1]) == [c1]
+        and:
+        client.getNewContainers([trace1]) == []
+        and:
+        client.getNewContainers([trace1, trace2, trace3]) == [c2]
+    }
 }