Merge pull request #164 from wfcommons/nextflow_log_parser_tweaks

Minor tweaks to the nextflow log parser and translator

Author: henricasanova

wfcommons/wfbench/translator/nextflow.py

@@ -299,19 +299,20 @@ def _write_readme_file(self, output_folder: pathlib.Path) -> None:
 
             out.write(f"Executions of large workflows can lead to large numbers of concurrent tasks, \n")
             out.write(f"which can hit system concurrency limits and/or lead to out-of-memory errors in the JVM that runs\n")
-            out.write(f"the Nextflow engine. The f{nextflow_config_file_name} configuration file in this directory\n")
+            out.write(f"the Nextflow engine. The {nextflow_config_file_name} configuration file in this directory\n")
             out.write(f"has settings to impose (pretty stringent) limits on concurrency and memory usage. These settings\n")
-            out.write(f"many not be appropriate for your purposes, you should INSPECT AND MODIFY settings in that file.\n\n")
+            out.write(f"many not be appropriate for your purposes, you should INSPECT AND MODIFY the settings in that file.\n\n")
 
-            out.write(f"If hitting JVM out-of-memory errors one possible solution, up to a point,\n")
-            out.write(f"is to define the NXF_OPTS environment variable, e.g.:\n")
+            out.write(f"If hitting JVM out-of-memory errors one possible solution, up to a point, it to define\n")
+            out.write(f"the NXF_OPTS environment variable, e.g.:\n\n")
 
             out.write(f"\texport NXF_OPTS='-Xms2g -Xmx24g'\n\n")
 
 
-            out.write(f"Note that the workflow has been split into {len(self.subworkflows)} module file(s), ")
-            out.write(f"each containing a maximum of {self.max_tasks_per_subworkflow} tasks.\n")
-            out.write(f"\nModule files are located in the 'modules/' directory.\n")
+            out.write(f"Note that the Nextflow workflow has been split into {len(self.subworkflows)} module file(s), ")
+            out.write(f"each containing a maximum of {self.max_tasks_per_subworkflow} tasks. \n")
+            out.write(f"This is to avoid the 'code too long' limit of Groovy. ")
+            out.write(f"Module files are located in the 'modules/' directory.\n")
 
     def _write_nf_config_file(self, output_folder: pathlib.Path) -> None:
         """

wfcommons/wfinstances/logs/nextflow.py

@@ -25,8 +25,9 @@
 
 class NextflowLogsParser(LogsParser):
     """"
-    Parse Nextflow execution directory to generate a workflow trace. The workflow
-    must have been executed with two features enabled: 1) the nf-prov plugin, and
+    Parse Nextflow execution directory to generate a workflow trace. Note that the workflow
+    reconstruction is not perfect, and will likely only capture file-based data dependencies.
+    The workflow must have been executed with two features enabled: 1) the nf-prov plugin, and
     2) execution tracing. This can be achieved by invoking Nextflow with a config
     file, e.g.::
 
@@ -60,6 +61,8 @@ class NextflowLogsParser(LogsParser):
     :type execution_dir: pathlib.Path
     :param nextflow_version: The Nextflow version used to execute the workflow
     :type nextflow_version: str
+    :param trace_file_name_pattern: The trace file name pattern to find the trace file (default: "*trace*.txt")
+    :type trace_file_name_pattern: str
     :param description: Workflow instance description.
     :type description: Optional[str]
     :param logger: The logger where to log information/warning or errors (optional).
@@ -69,6 +72,7 @@ class NextflowLogsParser(LogsParser):
     def __init__(self,
                  execution_dir: pathlib.Path,
                  nextflow_version: str,
+                 trace_file_name_pattern: Optional[str] = "*trace*.txt",
                  description: Optional[str] = None,
                  logger: Optional[Logger] = None) -> None:
         """Create an object of the nextflow log parser."""
@@ -78,8 +82,8 @@ def __init__(self,
         self.execution_dir = execution_dir
 
         # Load the Nextflow execution trace, and create a task runtime dictionary
-        nextflow_execution_trace_files = list(self.execution_dir.rglob("execution_trace_*.txt"))
-        if len(nextflow_execution_trace_files) == 0:
+        self.nextflow_execution_trace_files = list(self.execution_dir.rglob(trace_file_name_pattern))
+        if len(self.nextflow_execution_trace_files) == 0:
             raise FileNotFoundError("No execution_trace_*.txt file found in Nextflow execution directory.")
 
 
@@ -94,11 +98,8 @@ def build_workflow(self, workflow_name: Optional[str] = None) -> Workflow:
         :rtype: Workflow
         """
 
-        # Parse the Nextflow execution trace to create a dict of task runtimes
-        nextflow_execution_trace_files = list(self.execution_dir.rglob("execution_trace_*.txt"))
-        if len(nextflow_execution_trace_files) == 0:
-            raise FileNotFoundError("No execution_trace_*.txt file found in Nextflow execution directory.")
-        nextflow_execution_trace_file: pathlib.Path = max(nextflow_execution_trace_files,
+        # Parse the Nextflow (most recent) execution trace file to create a dict of task runtimes
+        nextflow_execution_trace_file: pathlib.Path = max(self.nextflow_execution_trace_files,
                                                           key=lambda p: p.stat().st_mtime)
         nextflow_task_runtimes = self._load_nextflow_trace(nextflow_execution_trace_file)