Add workflow outputs to WRROC report (#47)

Author: bentsherman

nf-prov-test/main.nf

@@ -1,9 +1,10 @@
 
+nextflow.preview.output = true
+
 params.constant = "foo"
 
 process ECHO_SCRIPT {
     tag "${prefix}"
-    publishDir params.outdir, mode: 'copy'
 
     input:
     tuple val(prefix), val(constant)
@@ -20,7 +21,6 @@ process ECHO_SCRIPT {
 
 process ECHO_EXEC {
     tag "${prefix}"
-    publishDir params.outdir, mode: 'copy'
 
     input:
     tuple val(prefix), val(constant)
@@ -45,6 +45,7 @@ process WC_SAMPLE {
 }
 
 workflow {
+    main:
     prefixes_ch = channel.of('r1', 'r2', 'r3')
     constant_ch = channel.value(params.constant)
     inputs_ch   = prefixes_ch.combine(constant_ch)
@@ -53,4 +54,24 @@ workflow {
 
     samples_ch = channel.fromPath(params.input).splitCsv(header: true)
     WC_SAMPLE(samples_ch)
+
+    publish:
+    script = ECHO_SCRIPT.out
+    exec = ECHO_EXEC.out
+}
+
+output {
+    script {
+        path 'script'
+        index {
+            path 'script.json'
+        }
+    }
+
+    exec {
+        path 'exec'
+        index {
+            path 'exec.json'
+        }
+    }
 }

nf-prov-test/nextflow.config

@@ -7,6 +7,8 @@ params {
     outdir = 'results'
 }
 
+outputDir = params.outdir
+
 prov {
     formats {
         bco {
@@ -43,7 +45,7 @@ manifest {
     homePage = "https://github.com/nextflow-io/nf-prov"
     description = "Test pipeline for nf-prov"
     mainScript = "main.nf"
-    nextflowVersion = "!>=24.10.0"
-    version = "0.3.0"
+    nextflowVersion = "!>=25.04.0"
+    version = "0.4.0"
     license = "https://spdx.org/licenses/Apache-2.0"
 }

src/main/groovy/nextflow/prov/ProvConfig.groovy

@@ -28,27 +28,52 @@ import nextflow.script.dsl.Description
 class ProvConfig implements ConfigScope {
 
     @ConfigOption
-    @Description('Create the provenance report (default: `true` if plugin is loaded).')
-    boolean enabled
+    @Description('''
+        Create the provenance report (default: `true` if plugin is loaded).
+    ''')
+    final boolean enabled
+
+    @Description('''
+        Configuration scope for the desired output formats.
+    ''')
+    final ProvFormatsConfig formats
 
-    @Description('Configuration scope for the desired output formats.')
-    ProvFormatsConfig formats
+    @Description('''
+        List of file patterns to include in the provenance report, from the set of published files. By default, all published files are included.
+    ''')
+    final List<String> patterns
+
+    /* required by extension point -- do not remove */
+    ProvConfig() {}
+
+    ProvConfig(Map opts) {
+        enabled = opts.enabled != null ? opts.enabled as boolean : true
+        formats = opts.formats ? new ProvFormatsConfig(opts.formats as Map) : null
+        patterns = opts.patterns as List<String> ?: []
+    }
 }
 
 
 class ProvFormatsConfig implements ConfigScope {
 
     @Description('Configuration scope for the BCO output format.')
-    ProvBcoConfig bco
+    final ProvBcoConfig bco
 
     @Description('Configuration scope for the DAG output format.')
-    ProvDagConfig dag
+    final ProvDagConfig dag
 
     @Description('Configuration scope for the legacy output format.')
-    ProvLegacyConfig legacy
+    final ProvLegacyConfig legacy
 
     @Description('Configuration scope for the WRROC output format.')
-    ProvWrrocConfig wrroc
+    final ProvWrrocConfig wrroc
+
+    ProvFormatsConfig(Map opts) {
+        bco = opts.bco ? new ProvBcoConfig(opts.bco as Map) : null
+        dag = opts.dag ? new ProvDagConfig(opts.dag as Map) : null
+        legacy = opts.legacy ? new ProvLegacyConfig(opts.legacy as Map) : null
+        wrroc = opts.wrroc ? new ProvWrrocConfig(opts.wrroc as Map) : null
+    }
 }
 
 
@@ -57,14 +82,19 @@ class ProvBcoConfig implements ConfigScope {
     @ConfigOption
     @Description('''
         The file name of the BCO manifest.
-        ''')
-    String file
+    ''')
+    final String file
 
     @ConfigOption
     @Description('''
         When `true` overwrites any existing BCO manifest with the same name.
-        ''')
-    boolean overwrite
+    ''')
+    final boolean overwrite
+
+    ProvBcoConfig(Map opts) {
+        file = opts.file
+        overwrite = opts.overwrite as boolean
+    }
 }
 
 
@@ -73,14 +103,19 @@ class ProvDagConfig implements ConfigScope {
     @ConfigOption
     @Description('''
         The file name of the DAG diagram.
-        ''')
+    ''')
     String file
 
     @ConfigOption
     @Description('''
         When `true` overwrites any existing DAG diagram with the same name.
-        ''')
+    ''')
     boolean overwrite
+
+    ProvDagConfig(Map opts) {
+        file = opts.file
+        overwrite = opts.overwrite as boolean
+    }
 }
 
 
@@ -89,14 +124,19 @@ class ProvLegacyConfig implements ConfigScope {
     @ConfigOption
     @Description('''
         The file name of the legacy manifest.
-        ''')
-    String file
+    ''')
+    final String file
 
     @ConfigOption
     @Description('''
         When `true` overwrites any existing legacy manifest with the same name.
-        ''')
-    boolean overwrite
+    ''')
+    final boolean overwrite
+
+    ProvLegacyConfig(Map opts) {
+        file = opts.file
+        overwrite = opts.overwrite as boolean
+    }
 }
 
 
@@ -105,18 +145,24 @@ class ProvWrrocConfig implements ConfigScope {
     @ConfigOption
     @Description('''
         The file name of the Workflow Run RO-Crate.
-        ''')
-    String file
+    ''')
+    final String file
 
     @ConfigOption
     @Description('''
         When `true` overwrites any existing Workflow Run RO-Crate with the same name.
-        ''')
-    boolean overwrite
+    ''')
+    final boolean overwrite
 
     @ConfigOption
     @Description('''
         The license for the Workflow Run RO-Crate.
-        ''')
-    String license
+    ''')
+    final String license
+
+    ProvWrrocConfig(Map opts) {
+        file = opts.file
+        overwrite = opts.overwrite as boolean
+        license = opts.license
+    }
 }

src/main/groovy/nextflow/prov/ProvObserver.groovy

@@ -25,14 +25,15 @@ import java.util.concurrent.locks.ReentrantLock
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
 import nextflow.Session
-import nextflow.processor.TaskHandler
 import nextflow.processor.TaskRun
 import nextflow.prov.renderers.BcoRenderer
 import nextflow.prov.renderers.DagRenderer
 import nextflow.prov.renderers.LegacyRenderer
 import nextflow.prov.renderers.WrrocRenderer
-import nextflow.trace.TraceObserver
-import nextflow.trace.TraceRecord
+import nextflow.trace.TraceObserverV2
+import nextflow.trace.event.FilePublishEvent
+import nextflow.trace.event.TaskEvent
+import nextflow.trace.event.WorkflowOutputEvent
 
 /**
  * Plugin observer of workflow events
@@ -42,7 +43,7 @@ import nextflow.trace.TraceRecord
  */
 @Slf4j
 @CompileStatic
-class ProvObserver implements TraceObserver {
+class ProvObserver implements TraceObserverV2 {
 
     public static final List<String> VALID_FORMATS = ['bco', 'dag', 'legacy', 'wrroc']
 
@@ -54,33 +55,37 @@ class ProvObserver implements TraceObserver {
 
     private Set<TaskRun> tasks = []
 
-    private Map<Path,Path> workflowOutputs = [:]
+    private Map<String,Path> workflowOutputs
+
+    private Map<Path,Path> publishedFiles = [:]
 
     private Lock lock = new ReentrantLock()
 
-    ProvObserver(Map<String,Map> formats, List<String> patterns) {
-        this.renderers = formats.collect( (name, config) -> createRenderer(name, config) )
+    ProvObserver(ProvFormatsConfig formats, List<String> patterns) {
+        this.renderers = createRenderers(formats)
         this.matchers = patterns.collect( pattern ->
             FileSystems.getDefault().getPathMatcher("glob:**/${pattern}")
         )
     }
 
-    private Renderer createRenderer(String name, Map opts) {
-        if( name == 'bco' )
-            return new BcoRenderer(opts)
+    private List<Renderer> createRenderers(ProvFormatsConfig config) {
+        final List<Renderer> result = []
+
+        if( config.bco )
+            result.add(new BcoRenderer(config.bco))
 
-        if( name == 'dag' )
-            return new DagRenderer(opts)
+        if( config.dag )
+            result.add(new DagRenderer(config.dag))
 
-        if( name == 'legacy' ) {
+        if( config.legacy ) {
             log.warn "The legacy format is deprecated -- it will be removed in a future version"
-            return new LegacyRenderer(opts)
+            result.add(new LegacyRenderer(config.legacy))
         }
 
-        if( name == 'wrroc' )
-            return new WrrocRenderer(opts)
+        if( config.wrroc )
+            result.add(new WrrocRenderer(config.wrroc))
 
-        throw new IllegalArgumentException("Invalid provenance format -- valid formats are ${VALID_FORMATS.join(', ')}")
+        return result
     }
 
     @Override
@@ -89,9 +94,9 @@ class ProvObserver implements TraceObserver {
     }
 
     @Override
-    void onProcessComplete(TaskHandler handler, TraceRecord trace) {
+    void onTaskComplete(TaskEvent event) {
         // skip failed tasks
-        final task = handler.task
+        final task = event.handler.task
         if( !task.isSuccess() )
             return
 
@@ -101,20 +106,35 @@ class ProvObserver implements TraceObserver {
     }
 
     @Override
-    void onProcessCached(TaskHandler handler, TraceRecord trace) {
+    void onTaskCached(TaskEvent event) {
         lock.withLock {
-            tasks << handler.task
+            tasks << event.handler.task
         }
     }
 
     @Override
-    void onFilePublish(Path destination, Path source) {
-        final match = matchers.isEmpty() || matchers.any { matcher -> matcher.matches(destination) }
+    void onWorkflowOutput(WorkflowOutputEvent event) {
+        if( workflowOutputs == null )
+            workflowOutputs = [:]
+
+        final value = event.value instanceof Path
+            ? event.value as Path
+            : event.index
+
+        if( !value )
+            log.warn "Workflow output `${event.name}` should either be a single path or declare an index file in order to be included in provenance reports"
+
+        workflowOutputs[event.name] = value
+    }
+
+    @Override
+    void onFilePublish(FilePublishEvent event) {
+        final match = matchers.isEmpty() || matchers.any { matcher -> matcher.matches(event.target) }
         if( !match )
             return
 
         lock.withLock {
-            workflowOutputs[source] = destination
+            publishedFiles[event.source] = event.target
         }
     }
 
@@ -123,8 +143,25 @@ class ProvObserver implements TraceObserver {
         if( !session.isSuccess() )
             return
 
-        for( final renderer : renderers )
-            renderer.render(session, tasks, workflowOutputs)
+        for( final renderer : renderers ) {
+            try {
+                renderer.render(session, tasks, workflowOutputs, publishedFiles)
+            }
+            catch( Exception e ) {
+                log.warn "Error occurred while rendering ${rendererName(renderer)} provenance report -- see Nextflow log for details"
+                log.debug "${e}"
+            }
+        }
+    }
+
+    private static String rendererName(Renderer renderer) {
+        return switch( renderer ) {
+            case BcoRenderer -> 'BCO';
+            case DagRenderer -> 'DAG';
+            case LegacyRenderer -> 'legacy';
+            case WrrocRenderer -> 'WRROC';
+            default -> null;
+        }
     }
 
 }