Merge pull request #382 from JosuaCarl/adapt-to-nextflow-26

Chore: Adapt to Nextflow 26.04.0

Author: JosuaCarl

CHANGELOG.md

@@ -6,6 +6,7 @@
 ## Misc:
 - Improved workflow reporting form extension/CLI by deriving and injecting workflow metadata from the provided trace file
 - Added full integration test
+- Adapted to Nextflow 26
 
 ## Features:
 - Transformation of data file to provenance file with schema.org / bioschemas.org type annotation in JSON-LD data format

build.gradle

@@ -58,7 +58,7 @@ sourceSets {
 
 // Versioning of important packages
 ext {
-    nextflowVersion = '24.10.6'
+    nextflowVersion = '26.04.0'
 }
 String groovyVersion = '4.0.27'
 String pf4jVersion = '3.13.0'

src/main/nextflow/co2footprint/CO2FootprintCLI.groovy

@@ -1,7 +1,6 @@
 package nextflow.co2footprint
 
 import groovy.util.logging.Slf4j
-import nextflow.BuildInfo
 import nextflow.Session
 import nextflow.co2footprint.DataContainers.CIDataMatrix
 import nextflow.co2footprint.DataContainers.TDPDataMatrix
@@ -13,12 +12,9 @@ import nextflow.script.ScriptFile
 import nextflow.script.WorkflowMetadata
 import nextflow.trace.TraceRecord
 import nextflow.util.Duration
-import nextflow.util.VersionNumber
 
 import java.nio.file.Path
-import java.nio.file.Paths
 import java.time.Instant
-import java.time.OffsetDateTime
 import java.time.ZoneOffset
 
 /**
@@ -72,7 +68,6 @@ class CO2FootprintCLI {
         // Collect CO2Records from traces & optionally write the corresponding files
         List<CO2Record> co2Records = []
         traceRecords.each { TraceRecord traceRecord ->
-            observer.recordStarted(traceRecord)
             co2Records.add(observer.aggregateRecords(traceRecord))
         }
 

src/main/nextflow/co2footprint/CO2FootprintConfig.groovy

@@ -6,14 +6,14 @@ import nextflow.co2footprint.Config.ReportFileConfig
 import nextflow.co2footprint.Config.SummaryFileConfig
 import nextflow.co2footprint.Config.TraceFileConfig
 import nextflow.co2footprint.DataContainers.AWSRegionsDataMatrix
-import nextflow.co2footprint.DataContainers.MachineTypeDataMatrix
-import nextflow.config.spec.ScopeName
-import nextflow.config.spec.ConfigScope
-import nextflow.config.spec.ConfigOption
-import nextflow.script.dsl.Description
 import nextflow.co2footprint.DataContainers.CIDataMatrix
+import nextflow.co2footprint.DataContainers.MachineTypeDataMatrix
 import nextflow.co2footprint.DataContainers.TDPDataMatrix
 import nextflow.co2footprint.Records.CiRecord
+import nextflow.config.spec.ConfigOption
+import nextflow.config.spec.ConfigScope
+import nextflow.config.spec.ScopeName
+import nextflow.script.dsl.Description
 import nextflow.trace.TraceHelper
 
 import java.nio.file.Path

src/main/nextflow/co2footprint/CO2FootprintExtension.groovy

@@ -77,7 +77,6 @@ class CO2FootprintExtension extends PluginExtensionPoint {
         // Collect CO2Records from traces & optionally write the corresponding files
         List<CO2Record> co2Records = []
         traceRecords.each { TraceRecord traceRecord ->
-            observer.recordStarted(traceRecord)
             co2Records.add(observer.aggregateRecords(traceRecord))
         }
 

src/main/nextflow/co2footprint/CO2FootprintFactory.groovy

@@ -1,18 +1,15 @@
 package nextflow.co2footprint
 
-import nextflow.co2footprint.DataContainers.CIDataMatrix
-import nextflow.co2footprint.DataContainers.TDPDataMatrix
-import nextflow.co2footprint.Logging.LoggingAdapter
-
+import groovy.transform.CompileStatic
 import groovy.transform.PackageScope
 import groovy.transform.PackageScopeTarget
-import groovy.transform.CompileStatic
-
 import groovy.util.logging.Slf4j
-
 import nextflow.Session
-import nextflow.trace.TraceObserver
-import nextflow.trace.TraceObserverFactory
+import nextflow.co2footprint.DataContainers.CIDataMatrix
+import nextflow.co2footprint.DataContainers.TDPDataMatrix
+import nextflow.co2footprint.Logging.LoggingAdapter
+import nextflow.trace.TraceObserverFactoryV2
+import nextflow.trace.TraceObserverV2
 
 /**
  * Factory class for creating the CO2Footprint trace observer.
@@ -25,7 +22,7 @@ import nextflow.trace.TraceObserverFactory
 @Slf4j
 @CompileStatic
 @PackageScope(PackageScopeTarget.FIELDS)
-class CO2FootprintFactory implements TraceObserverFactory {
+class CO2FootprintFactory implements TraceObserverFactoryV2 {
     // Nextflow Session
     private Session session = null
 
@@ -71,10 +68,10 @@ class CO2FootprintFactory implements TraceObserverFactory {
      * Loads configuration, sets up the observer, and injects all required data.
      *
      * @param session The Nextflow session
-     * @return Collection of TraceObserver (with one CO2FootprintObserver)
+     * @return Collection of TraceObserverV2 (with one CO2FootprintObserver)
      */
     @Override
-    Collection<TraceObserver> create(Session session) {
+    Collection<TraceObserverV2> create(Session session) {
         this.session = session
         adaptLogging()
 
@@ -85,10 +82,10 @@ class CO2FootprintFactory implements TraceObserverFactory {
         CO2FootprintCalculator co2FootprintCalculator = new CO2FootprintCalculator(TDPDataMatrix.tdpDataMatrix, config)
 
         // Define list of observers
-        TraceObserver observer = new CO2FootprintObserver(config, co2FootprintCalculator)
+        TraceObserverV2 observer = new CO2FootprintObserver(config, co2FootprintCalculator)
         CO2FootprintPlugin co2FootprintPlugin = CO2FootprintPlugin.getPlugin()
         co2FootprintPlugin?.observer = observer
-        final ArrayList<TraceObserver> result = [ observer ]
+        final ArrayList<TraceObserverV2> result = [ observer ]
         return result
     }
 }

src/main/nextflow/co2footprint/CO2FootprintObserver.groovy

@@ -10,12 +10,11 @@ import nextflow.co2footprint.FileCreation.TraceFileCreator
 import nextflow.co2footprint.Records.CO2Record
 import nextflow.co2footprint.Records.CO2RecordTree
 import nextflow.co2footprint.Records.CiRecordCollector
-import nextflow.processor.TaskHandler
 import nextflow.processor.TaskId
-import nextflow.processor.TaskProcessor
 import nextflow.script.WorkflowMetadata
-import nextflow.trace.TraceObserver
+import nextflow.trace.TraceObserverV2
 import nextflow.trace.TraceRecord
+import nextflow.trace.event.TaskEvent
 
 import java.util.concurrent.ConcurrentHashMap
 
@@ -30,7 +29,7 @@ import java.util.concurrent.ConcurrentHashMap
  *         Josua Carl <josua.carl@uni-tuebingen.de>
  */
 @Slf4j
-class CO2FootprintObserver implements TraceObserver {
+class CO2FootprintObserver implements TraceObserverV2 {
     // Holds workflow session
     Session session
 
@@ -97,20 +96,15 @@ class CO2FootprintObserver implements TraceObserver {
     boolean enableMetrics() { return true }
 
     // ------ HELPER METHODS ------
-
+    
     /**
      * Records the start of a task by storing its {@link TraceRecord}.
      *
      * @param trace the TraceRecord of the task that just started
      */
-    synchronized void recordStarted(TraceRecord traceRecord) {
+    synchronized void recordStartedTask(TraceRecord traceRecord) {
         // Keep started tasks
         runningTasks[traceRecord.taskId] = traceRecord
-
-        // Add a process node under the workflow if it doesn’t exist yet
-        if(!workflowStats.getChild(traceRecord.processName)) {
-            workflowStats.addChild(new CO2RecordTree(traceRecord.processName, [level: 'process']))
-        }
     }
 
     /**
@@ -136,8 +130,13 @@ class CO2FootprintObserver implements TraceObserver {
         // Optionally write to trace file
         this.traceFile.write(co2Record)
 
-        // Add a task node with its CO2Record to the corresponding process
+        // Add a process node under the workflow if it doesn’t exist yet
         CO2RecordTree processNode = workflowStats.getChild(traceRecord.processName)
+        if(!processNode) {
+            processNode = workflowStats.addChild(new CO2RecordTree(traceRecord.processName, [level: 'process']))
+        }
+
+        // Add a task node with its CO2Record to the corresponding process
         processNode.addChild(new CO2RecordTree(traceRecord.taskId, [level: 'task'], co2Record))
 
         return co2Record
@@ -155,15 +154,14 @@ class CO2FootprintObserver implements TraceObserver {
         co2RecordTree.collectAdditionalMetrics()
 
         // Create report and summary if any content exists to write to the file
-        if (workflowStats) {
+        if (co2RecordTree) {
             summaryFile.create()
             summaryFile.write(co2RecordTree, co2FootprintCalculator, config)
 
             reportFile.create()
             reportFile.addEntries(co2RecordTree, co2FootprintCalculator, config, timeCiRecordCollector, workflowMetadata)
             reportFile.write()
-        }
-        if (co2RecordTree) {
+
             provenanceFile.create()
             provenanceFile.write(co2RecordTree)
         }
@@ -226,77 +224,68 @@ class CO2FootprintObserver implements TraceObserver {
         )
     }
 
-
-    // ---- PROCESS LEVEL ----
+    // ---- TASK LEVEL ----
 
     /**
-     * This method is invoked when a process is created
+     * Invoked when a task is submitted by an executor to the
+     * underlying execution backend.
      *
-     * @param process The process created ({@link nextflow.processor.TaskProcessor})
+     * @param event A task event containing the ({@link nextflow.processor.TaskHandler})
+     *              and ({@link nextflow.trace.TraceRecord}) for the task
      */
     @Override
-    void onProcessCreate(TaskProcessor process) {}
+    void onTaskSubmit(TaskEvent event) {
+        log.trace("Trace report - submit process > ${event.handler}")
 
-    /**
-     * This method is invoked before a process run is going to be submitted.
-     *
-     * @param handler The task handler ({@link nextflow.processor.TaskHandler})
-     * @param trace   The trace record for the task ({@link nextflow.trace.TraceRecord})
-     */
-    @Override
-    void onProcessSubmit(TaskHandler handler, TraceRecord trace) {
-        log.trace("Trace report - submit process > ${handler}")
-
-        recordStarted(trace)
+        recordStartedTask(event.trace)
     }
 
     /**
-     * This method is invoked when a process run is going to start.
+     * Invoked when a task is running in the underlying execution backend.
      *
-     * @param handler The task handler ({@link nextflow.processor.TaskHandler})
-     * @param trace   The trace record for the task ({@link nextflow.trace.TraceRecord})
+     * @param event A task event containing the ({@link nextflow.processor.TaskHandler})
+     *              and ({@link nextflow.trace.TraceRecord}) for the task
      */
-    @Override
-    void onProcessStart(TaskHandler handler, TraceRecord trace) {
-        log.trace("Trace report - start process > ${handler}")
+    void onTaskStart(TaskEvent event) {
+        log.trace("Trace report - start process > ${event.handler}")
 
-        recordStarted(trace)
+        recordStartedTask(event.trace)
     }
 
     /**
-     * This method is invoked when a process run completes.
+     * Invoked when a task completes.
      *
-     * @param handler The task handler ({@link nextflow.processor.TaskHandler})
-     * @param trace   The trace record for the task ({@link nextflow.trace.TraceRecord})
+     * @param event A task event containing the ({@link nextflow.processor.TaskHandler})
+     *              and ({@link nextflow.trace.TraceRecord}) for the task
      */
     @Override
-    void onProcessComplete(TaskHandler handler, TraceRecord trace) {
-        log.trace("Trace report - complete process > ${handler}")
+    void onTaskComplete(TaskEvent event) {
+        log.trace("Trace report - complete process > ${event.handler}")
 
         // Ensure the presence of a Trace BaseRecord
-        if (!trace) {
-            log.warn("Unable to find TraceRecord for task with id: ${handler.task.id}")
+        if (!event.trace) {
+            log.warn("Unable to find TraceRecord for task with id: ${event.handler.task.id}")
             return
         }
 
-        aggregateRecords(trace)
+        aggregateRecords(event.trace)
     }
 
     /**
-     * This method is invoked when a process was cached.
+     * Invoked when a task execution is skipped because the result is cached (already computed)
+     * or stored (using the `storeDir` directive).
      *
-     * @param handler The task handler ({@link nextflow.processor.TaskHandler})
-     * @param trace   The trace record for the task ({@link nextflow.trace.TraceRecord})
+     * @param event A task event containing the ({@link nextflow.processor.TaskHandler})
+     *              and ({@link nextflow.trace.TraceRecord}) for the task
      */
     @Override
-    void onProcessCached(TaskHandler handler, TraceRecord trace) {
-        log.trace("Trace report - cached process > ${handler}")
+    void onTaskCached(TaskEvent event) {
+        log.trace("Trace report - cached process > ${event.handler}")
 
         // Event was triggered by a stored task, ignore it
-        if (trace == null) { return }
-        
-        recordStarted(trace) // add also cashed tasks to the runningTasks to be able to report them in the output files
-        aggregateRecords(trace)
-    }
+        if (event.trace == null) { return }
 
+        recordStartedTask(event.trace) // add also cashed tasks to the runningTasks to be able to report them in the output files
+        aggregateRecords(event.trace)
+    }
 }

src/main/nextflow/co2footprint/CO2FootprintPlugin.groovy

@@ -19,9 +19,8 @@ package nextflow.co2footprint
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
 import nextflow.cli.PluginAbstractExec
-
-import nextflow.co2footprint.Recorders.SessionTraceRecorder
 import nextflow.co2footprint.Parsers.ArgsParser
+import nextflow.co2footprint.Recorders.SessionTraceRecorder
 import nextflow.co2footprint.Records.CO2Record
 import nextflow.co2footprint.Records.CO2RecordTree
 import nextflow.plugin.BasePlugin

src/main/nextflow/co2footprint/Config/BaseFileConfig.groovy

@@ -4,6 +4,7 @@ import groovy.util.logging.Slf4j
 import nextflow.co2footprint.CO2FootprintConfig
 import nextflow.config.spec.ConfigOption
 import nextflow.script.dsl.Description
+
 import java.nio.file.Path
 
 /**

src/main/nextflow/co2footprint/FileCreation/ProvenanceFileCreator.groovy

@@ -1,10 +1,10 @@
 package nextflow.co2footprint.FileCreation
 
+import groovy.json.JsonBuilder
 import groovyx.gpars.agent.Agent
 import nextflow.co2footprint.Config.ProvenanceFileConfig
 import nextflow.co2footprint.Records.CO2RecordTree
 import nextflow.trace.TraceHelper
-import groovy.json.JsonBuilder
 
 import java.time.Duration
 import java.time.Instant