Minor improvements

Signed-off-by: Ben Sherman <[email protected]>

Author: bentsherman

README.md

@@ -4,13 +4,9 @@ Nextflow plugin for tracking provenance of pipeline output files.
 
 ## Getting Started
 
-First, `nf-prov` requires Nextflow version `22.11.0-edge` (or later). You can achieve this by running the following snippet before your `nextflow run` command. In Nextflow Tower, you can include this in your pre-run script under "Advanced Options" on the "Launch Pipeline" page.
+The `nf-prov` plugin requires Nextflow version `22.11.0-edge` or later.
 
-```sh
-export NXF_VER=22.11.0-edge
-```
-
-Second, to enable and configure `nf-prov`, you must include the following snippet to your Nextflow config and update as needed. Note that the use of `params.outdir` assumes that you are using an nf-core pipeline with that parameter.
+To enable and configure `nf-prov`, include the following snippet to your Nextflow config and update as needed.
 
 ```groovy
 plugins {
@@ -26,6 +22,26 @@ prov {
 
 Finally, run your Nextflow pipeline. You do not need to modify your pipeline script in order to use the `nf-prov` plugin. The plugin will automatically generate a JSON file with provenance information.
 
+## Configuration
+
+The following options are available:
+
+`prov.enabled`
+
+Create the provenance manifest (default: `true` if plugin is loaded).
+
+`prov.file`
+
+The path of the provenance manifest (default: `manifest.json`).
+
+`prov.patterns`
+
+List of file patterns to include in the provenance manifest, from the set of published files. By default, all published files are included.
+
+`prov.overwrite`
+
+Overwrite any existing provenance report with the same name (default: `false`).
+
 ## Development
 
 Run the following commands to build and test the nf-prov Nextflow plugin. Refer to the [nf-hello](https://github.com/nextflow-io/nf-hello) README for additional instructions (_e.g._ for publishing the plugin).

plugins/nf-prov/build.gradle

@@ -56,15 +56,15 @@ sourceSets {
 
 dependencies {
     // This dependency is exported to consumers, that is to say found on their compile classpath.
-    compileOnly 'io.nextflow:nextflow:22.04.0'
+    compileOnly 'io.nextflow:nextflow:23.04.0'
     compileOnly 'org.slf4j:slf4j-api:1.7.10'
     compileOnly 'org.pf4j:pf4j:3.4.1'
     // add here plugins depepencies
 
     // test configuration
     testImplementation "org.codehaus.groovy:groovy:3.0.8"
     testImplementation "org.codehaus.groovy:groovy-nio:3.0.8"
-    testImplementation 'io.nextflow:nextflow:22.04.0'
+    testImplementation 'io.nextflow:nextflow:23.04.0'
     testImplementation ("org.codehaus.groovy:groovy-test:3.0.8") { exclude group: 'org.codehaus.groovy' }
     testImplementation ("cglib:cglib-nodep:3.3.0")
     testImplementation ("org.objenesis:objenesis:3.1")

plugins/nf-prov/src/main/nextflow/prov/ProvObserver.groovy

@@ -17,7 +17,6 @@
 package nextflow.prov
 
 import java.nio.file.FileSystems
-import java.nio.file.Files
 import java.nio.file.Path
 import java.nio.file.PathMatcher
 
@@ -36,98 +35,86 @@ import nextflow.exception.AbortOperationException
  * Plugin observer of workflow events
  *
  * @author Bruno Grande <[email protected]>
+ * @author Ben Sherman <[email protected]>
  */
 @Slf4j
 @CompileStatic
 class ProvObserver implements TraceObserver {
 
     public static final String DEF_FILE_NAME = 'manifest.json'
 
-    private Session session
-
     private Map config
 
-    private boolean enabled
-
     private Path path
 
+    private Boolean overwrite
+
     private List<PathMatcher> matchers
 
-    private List<Map> published
+    private List<Map> published = []
+
+    private Map tasks = [:]
 
-    private Map tasks
+    ProvObserver(Path path, Boolean overwrite, List patterns) {
+        this.path = path
+        this.overwrite = overwrite
+        this.matchers = patterns.collect { pattern ->
+            FileSystems.getDefault().getPathMatcher("glob:**/${pattern}")
+        }
+    }
 
     @Override
     void onFlowCreate(Session session) {
-        this.session = session
         this.config = session.config
-        this.enabled = this.config.navigate('prov.enabled', true)
-        this.config.overwrite = this.config.navigate('prov.overwrite', false)
-        this.config.patterns = this.config.navigate('prov.patterns', [])
-        this.config.file = this.config.navigate('prov.file', DEF_FILE_NAME)
-        this.path = (this.config.file as Path).complete()
 
         // check file existance
-        final attrs = FileHelper.readAttributes(this.path)
-        if( this.enabled && attrs ) {
-            if( this.config.overwrite && (attrs.isDirectory() || !this.path.delete()) )
-                throw new AbortOperationException("Unable to overwrite existing file manifest: ${this.path.toUriString()}")
-            else if( !this.config.overwrite )
-                throw new AbortOperationException("File manifest already exists: ${this.path.toUriString()}")
+        final attrs = FileHelper.readAttributes(path)
+        if( attrs ) {
+            if( overwrite && (attrs.isDirectory() || !path.delete()) )
+                throw new AbortOperationException("Unable to overwrite existing provenance manifest: ${path.toUriString()}")
+            else if( !overwrite )
+                throw new AbortOperationException("Provenance manifest already exists: ${path.toUriString()}")
         }
-
-        this.matchers = this.config.patterns.collect { pattern ->
-            FileSystems.getDefault().getPathMatcher("glob:**/${pattern}")
-        }
-
-        this.published = []
-        this.tasks = [:]
     }
 
     static def jsonify(root) {
-        if ( root instanceof LinkedHashMap ) {
-            root.eachWithIndex { key, value, index ->
-                root[key] = jsonify(value)
-            }
-        } else if ( root instanceof Collection ) {
-            root = new ArrayList(root);
-            root.eachWithIndex { item, index ->
-                root[index] = jsonify(item)
-            }
-        } else if ( root instanceof FileHolder ) {
-            root = root.getStorePath()
-            root = jsonify(root)
-        } else if ( root instanceof Path ) {
-            root = root.toUriString()
-        } else if ( root instanceof Boolean ||
-                    root instanceof Number ) {
-            return root
-        } else {
-            return root as String
-        }
-        return root
+        if ( root instanceof Map )
+            root.collectEntries( (k, v) -> [k, jsonify(v)] )
+
+        else if ( root instanceof Collection )
+            root.collect( v -> jsonify(v) )
+
+        else if ( root instanceof FileHolder )
+            jsonify(root.storePath)
+
+        else if ( root instanceof Path )
+            root.toUriString()
+
+        else if ( root instanceof Boolean || root instanceof Number )
+            root
+
+        else
+            root.toString()
     }
 
-    void trackProcess(TaskHandler handler, TraceRecord trace){
-        def taskRun = handler.getTask()
-        def taskConfig = taskRun.config
-        def taskId = taskRun.id as String
+    void trackProcess(TaskHandler handler, TraceRecord trace) {
+        def task = handler.task
 
         // TODO: Figure out what the '$' input/output means
         //       Omitting them from manifest for now
         def taskMap = [
-            'id': taskId,
-            'name': taskRun.getName(),
-            'cached': taskRun.cached,
+            'id': task.id as String,
+            'name': task.name,
+            'cached': task.cached,
             'process': trace.getProcessName(),
-            'inputs': taskRun.inputs.findResults { inParam, object -> 
+            'inputs': task.inputs.findResults { inParam, object -> 
                 def inputMap = [ 
                     'name': inParam.getName(),
                     'value': jsonify(object) 
                 ] 
                 inputMap['name'] == '$' ? null : inputMap
             },
-            'outputs': taskRun.outputs.findResults { outParam, object -> 
+            'outputs': task.outputs.findResults { outParam, object -> 
                 def outputMap = [
                     'name': outParam.getName(),
                     'emit': outParam.getChannelEmitName(),
@@ -137,75 +124,64 @@ class ProvObserver implements TraceObserver {
             }
         ]
 
-        this.tasks.put(taskId, taskMap)
+        tasks.put(task.id, taskMap)
     }
 
     @Override
-    void onProcessComplete(TaskHandler handler, TraceRecord trace){
+    void onProcessComplete(TaskHandler handler, TraceRecord trace) {
         trackProcess(handler, trace)
     }
 
     @Override
-    void onProcessCached(TaskHandler handler, TraceRecord trace){
+    void onProcessCached(TaskHandler handler, TraceRecord trace) {
         trackProcess(handler, trace)
     }
 
     @Override
     void onFilePublish(Path destination, Path source) {
-        boolean match = this.matchers.isEmpty() || this.matchers.any { matcher ->
+        boolean match = matchers.isEmpty() || matchers.any { matcher ->
             matcher.matches(destination)
         }
 
-        def pathMap = [
+        if( !match )
+            return
+
+        published.add( [
             'source': source.toUriString(),
             'target': destination.toUriString()
-        ]
-
-        if ( match ) {
-            this.published.add(pathMap)
-        }
+        ] )
     }
 
     @Override
     void onFlowComplete() {
-        // make sure there are files to publish
-        if ( !this.enabled ) {
-            return
-        }
-
         // generate temporary output-task map
-        def outputTaskMap = [:]
-        this.tasks.each { taskId, task ->
+        def taskLookup = tasks.inject([:]) { accum, hash, task ->
             task['outputs'].each { output ->
                 // Make sure to handle tuples of outputs
                 def values = output['value']
-                if ( values instanceof Collection ) {
-                    values.each { outputTaskMap.put(it, task['id']) }
-                } else {
-                    outputTaskMap.put(values, task['id'])
-                }
+                if ( values instanceof Collection )
+                    values.each { accum.put(it, task['id']) }
+                else
+                    accum.put(values, task['id'])
             }
+            accum
         }
 
         // add task information to published files
-        this.published.each { path ->
-            path['publishingTaskId'] = outputTaskMap[path.source]
+        published.each { path ->
+            path['publishingTaskId'] = taskLookup[path.source]
         }
 
-        // generate manifest map
+        // save manifest to JSON file
         def manifest = [
-            'pipeline': this.config.manifest,
-            'published': this.published,
-            'tasks': this.tasks
+            'pipeline': config.manifest,
+            'published': published,
+            'tasks': tasks
         ]
+        def manifestJson = JsonOutput.toJson(manifest)
+        def manifestJsonPretty = JsonOutput.prettyPrint(manifestJson)
 
-        // output manifest map as JSON
-        def manifest_json = JsonOutput.toJson(manifest)
-        def manifest_json_pretty = JsonOutput.prettyPrint(manifest_json)
-
-        // create JSON file manifest
-        Path manifestFile = Files.createFile(this.path)
-        manifestFile << "${manifest_json_pretty}\n"
+        path.text = manifestJsonPretty
 
     }
 

plugins/nf-prov/src/main/nextflow/prov/ProvObserverFactory.groovy

@@ -16,6 +16,8 @@
 
 package nextflow.prov
 
+import java.nio.file.Path
+
 import groovy.transform.CompileStatic
 import nextflow.Session
 import nextflow.trace.TraceObserver
@@ -31,8 +33,18 @@ class ProvObserverFactory implements TraceObserverFactory {
 
     @Override
     Collection<TraceObserver> create(Session session) {
-        final result = new ArrayList()
-        result.add( new ProvObserver() )
-        return result
+        [ createProvObserver(session.config) ]
+    }
+
+    protected TraceObserver createProvObserver(Map config) {
+        final enabled = config.navigate('prov.enabled', true) as Boolean
+        if( !enabled )
+            return
+
+        final file = config.navigate('prov.file', ProvObserver.DEF_FILE_NAME)
+        final path = (file as Path).complete()
+        final overwrite = config.navigate('prov.overwrite') as Boolean
+        final patterns = config.navigate('prov.patterns', []) as List
+        new ProvObserver(path, format, overwrite, patterns)
     }
 }