From 567e8f89834bee506229ef7720a2348552388aa1 Mon Sep 17 00:00:00 2001 From: Henri Casanova Date: Fri, 10 Jul 2026 11:55:14 -1000 Subject: [PATCH] Small tweaks to the nextflow log parser and translator --- wfcommons/wfbench/translator/nextflow.py | 15 ++++++++------- wfcommons/wfinstances/logs/nextflow.py | 19 ++++++++++--------- 2 files changed, 18 insertions(+), 16 deletions(-) diff --git a/wfcommons/wfbench/translator/nextflow.py b/wfcommons/wfbench/translator/nextflow.py index 23438608..753bb0f5 100644 --- a/wfcommons/wfbench/translator/nextflow.py +++ b/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: """ diff --git a/wfcommons/wfinstances/logs/nextflow.py b/wfcommons/wfinstances/logs/nextflow.py index 67c9e9f3..fb5eb2f2 100644 --- a/wfcommons/wfinstances/logs/nextflow.py +++ b/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)