Extend Task traces with hash codes of retry attempts

New feature

Currently it is impossible to track resubmitted Tasks because there is no way to identify subsequent attempts (group them to gether). One option to track Task attempts would be by storing the first attempt's hash or list of all the subsequent attempts' hashes to Task trace.

Usage scenario

Nextflow is able to resubmit Tasks in case of failure. However, Nextflow always assigns a new random hash code to each resubmitted Task and connection between the attempts is lost. I some cases (e.g. debugging) it would be invaluable to see all the Task's retry attempts to be grouped together.

Suggest implementation

The simple diff below demonstrates one possible implementation to extend current Task trace with hash codes of retried Tasks. By using the list retry_hashes one can identify all the consecutive retries from the very first to the final retry. This implementation has been tested in been in active use for a few months now and seems to work as intended.

diff --git a/modules/nextflow/src/main/groovy/nextflow/processor/TaskHandler.groovy b/modules/nextflow/src/main/groovy/nextflow/processor/TaskHandler.groovy
index fb5082b..6cd3fe9 100644
--- a/modules/nextflow/src/main/groovy/nextflow/processor/TaskHandler.groovy
+++ b/modules/nextflow/src/main/groovy/nextflow/processor/TaskHandler.groovy
@@ -170,6 +170,7 @@ abstract class TaskHandler {
         record.module = task.config.module
         record.container = task.getContainer()
         record.attempt = task.config.attempt
+	 record.retry_hashes = task.retry_hashes

         record.script = task.getScript()
         record.scratch = task.getScratch()
diff --git a/modules/nextflow/src/main/groovy/nextflow/processor/TaskProcessor.groovy b/modules/nextflow/src/main/groovy/nextflow/processor/TaskProcessor.groovy
index 80c8fbc..b1813f1 100644
--- a/modules/nextflow/src/main/groovy/nextflow/processor/TaskProcessor.groovy
+++ b/modules/nextflow/src/main/groovy/nextflow/processor/TaskProcessor.groovy
@@ -1998,6 +1998,7 @@ class TaskProcessor {
     final protected void makeTaskContextStage3( TaskRun task, HashCode hash, Path folder ) {

         // set hash-code & working directory
+        task.retry_hashes.add(hash.toString())
         task.hash = hash
         task.workDir = folder
         task.config.workDir = folder
diff --git a/modules/nextflow/src/main/groovy/nextflow/processor/TaskRun.groovy b/modules/nextflow/src/main/groovy/nextflow/processor/TaskRun.groovy
index 754f94e..005c09a 100644
--- a/modules/nextflow/src/main/groovy/nextflow/processor/TaskRun.groovy
+++ b/modules/nextflow/src/main/groovy/nextflow/processor/TaskRun.groovy
@@ -75,6 +75,11 @@ class TaskRun implements Cloneable {
      */
     HashCode hash

+    /**
+     * The list of previously submitted but failed and subsequently resubmitted task hash codes
+     */
+    ArrayList<String> retry_hashes = []
+
     /*
      * The processor that creates this 'task'
      */
diff --git a/modules/nextflow/src/main/groovy/nextflow/trace/TraceRecord.groovy b/modules/nextflow/src/main/groovy/nextflow/trace/TraceRecord.groovy
index 325bedc..c1984e7 100644
--- a/modules/nextflow/src/main/groovy/nextflow/trace/TraceRecord.groovy
+++ b/modules/nextflow/src/main/groovy/nextflow/trace/TraceRecord.groovy
@@ -101,7 +101,8 @@ class TraceRecord implements Serializable {
             error_action:'str',
             vol_ctxt: 'num',
             inv_ctxt: 'num',
-            hostname: 'str'
+            hostname: 'str',
+            retry_hashes: 'str',
     ]

     static public Map<String,Closure<String>> FORMATTER = [