Refactor lineage annotations as list (#6022)

Signed-off-by: jorgee <[email protected]>
Signed-off-by: Jorge Ejarque <[email protected]>
Signed-off-by: Ben Sherman <[email protected]>
Co-authored-by: Ben Sherman <[email protected]>

Author: jorgee

docs/workflow.md

@@ -627,9 +627,6 @@ See [Output directives](#output-directives) for the list of available index dire
 
 The following directives are available for each output in the output block:
 
-`annotations`
-: Specify annotations to be be applied to every published file. Can be a map or a closure that returns a map.
-
 `index`
 : Create an index file which will contain a record of each published value.
 
@@ -638,12 +635,18 @@ The following directives are available for each output in the output block:
   `header`
   : When `true`, the keys of the first record are used as the column names (default: `false`). Can also be a list of column names. Only used for CSV files.
 
+  `labels`
+  : Specify labels to be applied to the index file as a list of strings.
+
   `path`
   : The name of the index file relative to the base output directory (required). Can be a CSV, JSON, or YAML file.
 
   `sep`
   : The character used to separate values (default: `','`). Only used for CSV files.
 
+`labels`
+: Specify labels to be applied to every published file. Can be a list of strings or a closure that returns a list.
+
 `path`
 : Specify the publish path relative to the output directory (default: `'.'`). Can be a path, a closure that defines a custom directory for each published value, or a closure that publishes individual files using the `>>` operator.
 

modules/nextflow/src/main/groovy/nextflow/extension/PublishOp.groovy

@@ -23,7 +23,6 @@ import groovy.util.logging.Slf4j
 import groovyx.gpars.dataflow.DataflowReadChannel
 import nextflow.Session
 import nextflow.exception.ScriptRuntimeException
-import nextflow.file.FileHelper
 import nextflow.processor.PublishDir
 import nextflow.trace.event.FilePublishEvent
 import nextflow.trace.event.WorkflowOutputEvent
@@ -100,7 +99,7 @@ class PublishOp {
             overrides.saveAs = targetResolver
         else
             overrides.path = targetResolver
-        overrides.annotations = getAnnotations(publishOpts.annotations, value)
+        overrides.labels = getLabels(publishOpts.labels, value)
 
         final publisher = PublishDir.create(publishOpts + overrides)
 
@@ -190,22 +189,27 @@ class PublishOp {
     }
 
     /**
-     * Get or resolve the annotations of a workflow output.
+     * Get or resolve the labels of a workflow output.
      *
-     * @param annotations Map | Closure<Map>
+     * @param labels List | Closure<List>
      * @param value
      */
-    protected static Map getAnnotations(annotations, value) {
-        if( annotations == null )
-            return [:]
-        if( annotations instanceof Map )
-            return annotations
-        if( annotations instanceof Closure ) {
-            final result = annotations.call(value)
-            if( result instanceof Map )
-                return result
+    protected static List<String> getLabels(labels, value) {
+        if( labels == null )
+            return []
+        if( labels instanceof List<String> )
+            return labels
+        if( labels instanceof Closure ) {
+            try {
+                final result = labels.call(value)
+                if( result instanceof List<String> )
+                    return result
+            } catch (Throwable e) {
+                log.warn("Exception while evaluating dynamic `labels` directive for value '$value' -- ${e.getMessage()}")
+                return []
+            }
         }
-        throw new ScriptRuntimeException("Invalid output `annotations` directive -- it should be either a Map or a closure that returns a Map")
+        throw new ScriptRuntimeException("Invalid output `labels` directive -- it should be either a List or a closure that returns a List")
     }
 
     /**
@@ -238,7 +242,7 @@ class PublishOp {
             else {
                 log.warn "Invalid extension '${ext}' for index file '${indexPath}' -- should be CSV, JSON, or YAML"
             }
-            session.notifyFilePublish(new FilePublishEvent(null, indexPath))
+            session.notifyFilePublish(new FilePublishEvent(null, indexPath, indexOpts.labels))
         }
 
         log.trace "Publish operator complete"
@@ -365,13 +369,16 @@ class PublishOp {
     static class IndexOpts {
         Path path
         def /* boolean | List<String> */ header = false
+        List<String> labels
         String sep = ','
 
         IndexOpts(Path targetDir, Map opts) {
             this.path = targetDir.resolve(opts.path as String)
 
             if( opts.header != null )
                 this.header = opts.header
+            if( opts.labels )
+                this.labels = opts.labels as List<String>
             if( opts.sep )
                 this.sep = opts.sep as String
         }

modules/nextflow/src/main/groovy/nextflow/processor/PublishDir.groovy

@@ -111,9 +111,9 @@ class PublishDir {
     private def tags
 
     /**
-     * Annotations to be associated to the target file
+     * Labels to be associated to the target file
      */
-    private Map annotations
+    private List<String> labels
 
     /**
      * The content type of the file. Currently only supported by AWS S3.
@@ -217,8 +217,8 @@ class PublishDir {
         if( params.tags != null )
             result.tags = params.tags
 
-        if( params.annotations != null )
-            result.annotations = params.annotations as Map
+        if( params.labels != null )
+            result.labels = params.labels as List<String>
 
         if( params.contentType instanceof Boolean )
             result.contentType = params.contentType
@@ -590,7 +590,7 @@ class PublishDir {
     }
 
     protected void notifyFilePublish(Path destination, Path source=null) {
-        session.notifyFilePublish(new FilePublishEvent(source, destination, annotations))
+        session.notifyFilePublish(new FilePublishEvent(source, destination, labels))
     }
 
 }

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

@@ -108,14 +108,6 @@ class OutputDsl {
 
         private Map opts = [:]
 
-        void annotations(Map value) {
-            setOption('annotations', value)
-        }
-
-        void annotations(Closure value) {
-            setOption('annotations', value)
-        }
-
         void contentType(String value) {
             setOption('contentType', value)
         }
@@ -141,6 +133,14 @@ class OutputDsl {
             setOption('index', dsl.getOptions())
         }
 
+        void labels(List<String> value) {
+            setOption('labels', value)
+        }
+
+        void labels(Closure value) {
+            setOption('labels', value)
+        }
+
         void mode(String value) {
             setOption('mode', value)
         }
@@ -194,6 +194,10 @@ class OutputDsl {
             setOption('header', value)
         }
 
+        void labels(List<String> value) {
+            setOption('labels', value)
+        }
+
         void path(String value) {
             setOption('path', value)
         }

modules/nextflow/src/main/groovy/nextflow/trace/event/FilePublishEvent.groovy

@@ -38,7 +38,7 @@ class FilePublishEvent {
      */
     Path target
     /**
-     * Annotations associated with the published file.
+     * Labels associated with the published file.
      */
-    Map<String,String> annotations
+    List<String> labels
 }

modules/nextflow/src/test/groovy/nextflow/script/OutputDslTest.groovy

@@ -84,8 +84,8 @@ class OutputDslTest extends Specification {
             "${outputDir}/barbar/file2.txt"
             """.stripIndent()
         and:
-        1 * session.notifyFilePublish(new FilePublishEvent(file1, outputDir.resolve('foo/file1.txt'), [:]))
-        1 * session.notifyFilePublish(new FilePublishEvent(file2, outputDir.resolve('barbar/file2.txt'), [:]))
+        1 * session.notifyFilePublish(new FilePublishEvent(file1, outputDir.resolve('foo/file1.txt'), []))
+        1 * session.notifyFilePublish(new FilePublishEvent(file2, outputDir.resolve('barbar/file2.txt'), []))
         1 * session.notifyWorkflowOutput(new WorkflowOutputEvent('foo', [outputDir.resolve('foo/file1.txt')], null))
         1 * session.notifyWorkflowOutput(new WorkflowOutputEvent('bar', [outputDir.resolve('barbar/file2.txt')], outputDir.resolve('index.csv')))
         1 * session.notifyFilePublish(new FilePublishEvent(null, outputDir.resolve('index.csv'), null))
@@ -125,7 +125,7 @@ class OutputDslTest extends Specification {
         then:
         outputDir.resolve('file1.txt').text == 'Hello'
         and:
-        1 * session.notifyFilePublish(new FilePublishEvent(file1, outputDir.resolve('file1.txt'), [:]))
+        1 * session.notifyFilePublish(new FilePublishEvent(file1, outputDir.resolve('file1.txt'), []))
         1 * session.notifyWorkflowOutput(new WorkflowOutputEvent('foo', [outputDir.resolve('file1.txt')], null))
 
         cleanup:
@@ -149,6 +149,7 @@ class OutputDslTest extends Specification {
         dsl2.overwrite(true)
         dsl2.storageClass('someClass')
         dsl2.tags([foo:'1',bar:'2'])
+        dsl2.labels(['label'])
         then:
         dsl2.getOptions() == [
             contentType:'simple/text',
@@ -157,7 +158,8 @@ class OutputDslTest extends Specification {
             mode: 'someMode',
             overwrite: true,
             storageClass: 'someClass',
-            tags: [foo:'1',bar:'2']
+            tags: [foo:'1',bar:'2'],
+            labels: ['label']
         ]
     }
 
@@ -173,11 +175,13 @@ class OutputDslTest extends Specification {
         dsl2.header(true)
         dsl2.path('path')
         dsl2.sep(',')
+        dsl2.labels(['label'])
         then:
         dsl2.getOptions() == [
             header: true,
             path: 'path',
-            sep: ','
+            sep: ',',
+            labels: ['label']
         ]
     }
 

modules/nf-lang/src/main/java/nextflow/script/dsl/OutputDsl.java

@@ -39,12 +39,6 @@ public interface OutputDsl extends DslScope {
     """)
     Map<String,Object> getParams();
 
-    @Description("""
-        Specify annotations to be be applied to every published file. Can be a map or a closure that returns a map.
-    """)
-    /* Map | Closure */
-    void annotations(Object value);
-
     @Description("""
         *Currently only supported for S3.*
 
@@ -68,6 +62,12 @@ Enable or disable publishing (default: `true`).
     """)
     void index(Closure closure);
 
+    @Description("""
+        Specify labels to be be applied to every published file.
+    """)
+    /* List | Closure */
+    void labels(Object value);
+
     @Description("""
         The file publishing method (default: `'symlink'`).
     """)
@@ -106,6 +106,11 @@ interface IndexDsl extends DslScope {
         /* List<String> | Boolean */
         void header(Object value);
 
+        @Description("""
+             Specify labels to be applied to the index file.
+        """)
+        void labels(List<String> value);
+
         @Description("""
             Closure which defines how to transform each published value into a CSV record. The closure should return a list or map. By default, no transformation is applied.
         """)

modules/nf-lineage/src/main/nextflow/lineage/DefaultLinStore.groovy

@@ -101,7 +101,7 @@ class DefaultLinStore implements LinStore {
         return searchAllFiles(params)
     }
 
-    private Map<String, LinSerializable> searchAllFiles(Map<String,String> params) {
+    private Map<String, LinSerializable> searchAllFiles(Map<String,List<String>> params) {
         final results = new HashMap<String, LinSerializable>()
 
         Files.walkFileTree(location, new FileVisitor<Path>() {

modules/nf-lineage/src/main/nextflow/lineage/LinObserver.groovy

@@ -26,7 +26,6 @@ import java.time.OffsetDateTime
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
 import nextflow.Session
-import nextflow.lineage.model.Annotation
 import nextflow.lineage.model.Checksum
 import nextflow.lineage.model.FileOutput
 import nextflow.lineage.model.DataPath
@@ -37,7 +36,6 @@ import nextflow.lineage.model.WorkflowOutput
 import nextflow.lineage.model.WorkflowRun
 import nextflow.file.FileHelper
 import nextflow.file.FileHolder
-import nextflow.processor.TaskHandler
 import nextflow.processor.TaskRun
 import nextflow.script.ScriptMeta
 import nextflow.script.params.BaseParam
@@ -55,7 +53,6 @@ import nextflow.script.params.StdOutParam
 import nextflow.script.params.ValueInParam
 import nextflow.script.params.ValueOutParam
 import nextflow.trace.TraceObserverV2
-import nextflow.trace.TraceRecord
 import nextflow.trace.event.FilePublishEvent
 import nextflow.trace.event.TaskEvent
 import nextflow.trace.event.WorkflowOutputEvent
@@ -362,21 +359,13 @@ class LinObserver implements TraceObserverV2 {
                 attrs.size(),
                 LinUtils.toDate(attrs?.creationTime()),
                 LinUtils.toDate(attrs?.lastModifiedTime()),
-                convertAnnotations(event.annotations))
+                event.labels)
             store.save(key, value)
         } catch (Throwable e) {
             log.warn("Unexpected error storing published file '${event.target.toUriString()}' for workflow '${executionHash}'", e)
         }
     }
 
-    private static List<Annotation> convertAnnotations(Map annotations){
-        if( !annotations )
-            return null
-        final converted = new LinkedList<Annotation>()
-        annotations.forEach { Object key, Object value -> converted.add(new Annotation(key.toString(), value)) }
-        return converted
-    }
-
     String getSourceReference(Path source){
         final hash = FileHelper.getTaskHashFromPath(source, session.workDir)
         if (hash) {

modules/nf-lineage/src/main/nextflow/lineage/LinPropertyValidator.groovy

@@ -17,9 +17,9 @@
 package nextflow.lineage
 
 import groovy.transform.CompileStatic
-import nextflow.lineage.model.Annotation
 import nextflow.lineage.model.Checksum
 import nextflow.lineage.model.DataPath
+import nextflow.lineage.model.FileOutput
 import nextflow.lineage.model.Parameter
 import nextflow.lineage.model.TaskOutput
 import nextflow.lineage.model.TaskRun
@@ -36,10 +36,9 @@ import nextflow.lineage.model.WorkflowRun
 class LinPropertyValidator {
 
     private static final List<Class> LIN_MODEL_CLASSES = [
-        Annotation,
         Checksum,
-        DataOutput,
         DataPath,
+        FileOutput,
         Parameter,
         TaskOutput,
         TaskRun,
@@ -71,8 +70,8 @@ class LinPropertyValidator {
         }
     }
 
-    void validateQueryParams(Map<String, String> params) {
-        for( String key : params.keySet() ) {
+    void validateQueryParams(Collection<String> keys) {
+        for( final key : keys ) {
            validate(key.tokenize('.'))
         }
     }