Add support for tower_workflow_id

Author: pditommaso

modules/nextflow/src/main/groovy/nextflow/Session.groovy

@@ -298,7 +298,7 @@ class Session implements ISession {
             uniqueId = UUID.fromString(config.resume as String)
         }
         else {
-           uniqueId = UUID.randomUUID()
+           uniqueId = systemEnv.get('NXF_UUID') ? UUID.fromString(systemEnv.get('NXF_UUID')) : UUID.randomUUID()
         }
         log.debug "Session uuid: $uniqueId"
 

modules/nextflow/src/main/groovy/nextflow/script/ScriptRunner.groovy

@@ -236,8 +236,9 @@ class ScriptRunner {
     void verifyAndTrackHistory(String cli, String name) {
         assert cli, 'Missing launch command line'
 
+        final ignore = System.getenv('NXF_IGNORE_RESUME_HISTORY') as Boolean
         // -- when resume, make sure the session id exists in the executions history
-        if( session.resumeMode && !HistoryFile.DEFAULT.checkExistsById(session.uniqueId.toString())) {
+        if( session.resumeMode && !ignore && !HistoryFile.DEFAULT.checkExistsById(session.uniqueId.toString()) ) {
             throw new AbortOperationException("Can't find a run with the specified id: ${session.uniqueId} -- Execution can't be resumed")
         }
 

modules/nextflow/src/main/groovy/nextflow/script/WorkflowMetadata.groovy

@@ -202,7 +202,7 @@ class WorkflowMetadata {
      */
     Manifest manifest
 
-    final private Session session
+    private Session session
 
     final private List<Closure> onCompleteActions = []
 
@@ -462,4 +462,4 @@ class WorkflowMetadata {
         session.statsObserver.getStats()
     }
 
-}
+}

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

@@ -97,7 +97,7 @@ class TowerClient implements TraceObserver {
 
     private ResourcesAggregator aggregator
 
-    private Map<String,String> env = System.getenv()
+    protected Map<String,String> env = System.getenv()
 
     private LinkedBlockingQueue<ProcessEvent> events = new LinkedBlockingQueue()
 
@@ -119,6 +119,8 @@ class TowerClient implements TraceObserver {
 
     private int backOffBase
 
+    private boolean towerLaunch
+
     /**
      * Constructor that consumes a URL and creates
      * a basic HTTP client.
@@ -137,16 +139,14 @@ class TowerClient implements TraceObserver {
         this.generator = TowerJsonGenerator.create(Collections.EMPTY_MAP)
     }
 
-    protected String getLaunchId() {
-        env.get('TOWER_LAUNCH_ID')
-    }
-
     boolean enableMetrics() { true }
 
     String getEndpoint() { endpoint }
 
     String getWorkflowId() { workflowId }
 
+    boolean getTowerLaunch() { towerLaunch }
+
     void setAliveInterval(Duration d) {
         this.aliveInterval = d
     }
@@ -245,7 +245,8 @@ class TowerClient implements TraceObserver {
         result.runName = session.runName
         result.projectName = session.workflowMetadata.projectName
         result.repository = session.workflowMetadata.repository
-        result.launchId = launchId
+        result.workflowId = env.get('TOWER_WORKFLOW_ID')
+        this.towerLaunch = result.workflowId != null
         return result
     }
 
@@ -412,7 +413,7 @@ class TowerClient implements TraceObserver {
         def result = new LinkedHashMap(5)
         result.workflow = workflow
         result.processNames = new ArrayList(processNames)
-        result.launchId = launchId
+        result.towerLaunch = towerLaunch
         return result
     }
 

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

@@ -11,6 +11,8 @@
 
 package io.seqera.tower.plugin
 
+import spock.lang.Specification
+
 import java.nio.file.Files
 import java.time.Instant
 import java.time.OffsetDateTime
@@ -22,14 +24,11 @@ import nextflow.cloud.types.PriceModel
 import nextflow.container.ContainerConfig
 import nextflow.exception.AbortOperationException
 import nextflow.script.ScriptBinding
-import nextflow.script.ScriptFile
 import nextflow.script.WorkflowMetadata
 import nextflow.trace.TraceRecord
 import nextflow.trace.WorkflowStats
 import nextflow.trace.WorkflowStatsObserver
 import nextflow.util.SimpleHttpClient
-import org.junit.Ignore
-import spock.lang.Specification
 /**
  *
  * @author Paolo Di Tommaso <[email protected]>
@@ -162,14 +161,6 @@ class TowerClientTest extends Specification {
 
     }
 
-    def 'should get launchId from env' () {
-        when:
-        def observer = new TowerClient(env:[TOWER_LAUNCH_ID:'foo-123'])
-
-        then:
-        observer.launchId == 'foo-123'
-    }
-
     def 'should post task records' () {
         given:
         def URL = 'http://foo.com'
@@ -240,35 +231,56 @@ class TowerClientTest extends Specification {
     }
 
 
-    @Ignore
     def 'should create workflow json' () {
 
         given:
+        def sessionId = UUID.randomUUID()
         def dir = Files.createTempDirectory('test')
-        def client = Mock(SimpleHttpClient)
-        TowerClient tower = Spy(TowerClient, constructorArgs: [ [httpClient: client] ])
+        def http = Mock(SimpleHttpClient)
+        Map args = [httpClient: http, env: ENV]
+        TowerClient client = Spy(TowerClient, constructorArgs: [ args ])
         and:
         def session = Mock(Session)
+        session.getUniqueId() >> sessionId
+        session.getRunName() >> 'foo'
         session.config >> [:]
         session.containerConfig >> new ContainerConfig()
         session.getParams() >> new ScriptBinding.ParamsMap([foo:'Hello', bar:'World'])
 
-        def meta = new WorkflowMetadata(session, Mock(ScriptFile))
+        def meta = new WorkflowMetadata(
+                session: session,
+                projectName: 'the-project-name',
+                repository: 'git://repo.com/foo' )
         session.getWorkflowMetadata() >> meta
         session.getStatsObserver() >> Mock(WorkflowStatsObserver) { getStats() >> new WorkflowStats() }
 
         when:
-        def req = tower.makeBeginReq(session)
+        def req1 = client.makeCreateReq(session)
         then:
-        tower.getWorkflowId() >> '12345'
-        tower.getLaunchId() >> 'x123'
+        req1.sessionId == sessionId.toString()
+        req1.runName == 'foo'
+        req1.projectName == 'the-project-name'
+        req1.repository == 'git://repo.com/foo'
+        req1.workflowId == WORKFLOW_ID
+
+        when:
+        def req = client.makeBeginReq(session)
+        then:
+        client.getWorkflowId() >> '12345'
         and:
         req.workflow.id == '12345'
         req.workflow.params == [foo:'Hello', bar:'World']
-        req.launchId == 'x123'
+        and:
+        req.towerLaunch == TOWER_LAUNCH
+
         cleanup:
         dir?.deleteDir()
 
+        where:
+        ENV                         | WORKFLOW_ID   | TOWER_LAUNCH
+        [:]                         | null          | false
+        [TOWER_WORKFLOW_ID: '1234'] | '1234'        | true
+
     }
 
     def 'should convert map' () {
@@ -298,7 +310,7 @@ class TowerClientTest extends Specification {
     def 'should create init request' () {
         given:
         def uuid = UUID.randomUUID()
-        def tower = new TowerClient(env: [TOWER_LAUNCH_ID: 'x123'])
+        def client = new TowerClient(env: [TOWER_WORKFLOW_ID: 'x123'])
         def meta = Mock(WorkflowMetadata) {
             getProjectName() >> 'the-project-name'
             getRepository() >> 'git://repo.com/foo'
@@ -310,13 +322,15 @@ class TowerClientTest extends Specification {
         }
 
         when:
-        def req = tower.makeCreateReq(session)
+        def req = client.makeCreateReq(session)
         then:
         req.sessionId == uuid.toString()
         req.runName == 'foo_bar'
         req.projectName == 'the-project-name'
         req.repository == 'git://repo.com/foo'
-        req.launchId == 'x123'
+        req.workflowId == 'x123'
+        and:
+        client.towerLaunch
     }
 
     def 'should post create request' () {
@@ -332,19 +346,20 @@ class TowerClientTest extends Specification {
             getWorkflowMetadata() >> meta
         }
 
-        TowerClient tower = Spy(TowerClient, constructorArgs: ['https://tower.nf'])
+        TowerClient client = Spy(TowerClient, constructorArgs: ['https://tower.nf'])
 
         when:
-        tower.onFlowCreate(session)
+        client.onFlowCreate(session)
         then:
-        1 * tower.getAccessToken() >> 'secret'
-        1 * tower.makeCreateReq(session) >> [runName: 'foo']
-        1 * tower.sendHttpMessage('https://tower.nf/trace/create', [runName: 'foo'], 'POST') >> new TowerClient.Response(200, '{"workflowId":"xyz123"}')
+        1 * client.getAccessToken() >> 'secret'
+        1 * client.makeCreateReq(session) >> [runName: 'foo']
+        1 * client.sendHttpMessage('https://tower.nf/trace/create', [runName: 'foo'], 'POST') >> new TowerClient.Response(200, '{"workflowId":"xyz123"}')
         and:
-        tower.runName == 'foo_bar'
-        tower.runId == uuid.toString()
+        client.runName == 'foo_bar'
+        client.runId == uuid.toString()
         and:
-        tower.workflowId == 'xyz123'
+        client.workflowId == 'xyz123'
+        !client.towerLaunch
 
     }