Translator / Logger updates (RO-Crate, Nextflow)

Author: henricasanova

tests/translators_loggers/test_translators_loggers.py

@@ -46,6 +46,7 @@
 from wfcommons.wfinstances.logs import MakeflowLogsParser
 from wfcommons.wfinstances.logs import ROCrateLogsParser
 from wfcommons.wfinstances.logs import SnakemakeLogsParser
+from wfcommons.wfinstances.logs import NextflowLogsParser
 
 
 def _create_workflow_benchmark() -> (WorkflowBenchmark, int):
@@ -172,7 +173,14 @@ def run_workflow_parsl(container, num_tasks, str_dirpath):
 
 def run_workflow_nextflow(container, num_tasks, str_dirpath):
     # Run the workflow!
-    exit_code, output = container.exec_run(f"nextflow run ./workflow.nf --pwd .", user="wfcommons", stdout=True, stderr=True)
+    exit_code, output = container.exec_run(f"nextflow run ./workflow.nf --pwd . "
+                                           # f"-with-report execution_report.html "
+                                           # f"-with-timeline execution_timeline.html "
+                                           # f"-with-trace trace_file "
+                                           # f"-plugins nf-prov -with-prov prov.json "
+                                           # f"-with-dag dag_file.html"
+                                           f"-c plugin.config ",
+                                           user="wfcommons", stdout=True, stderr=True)
     ignored, task_exit_codes = container.exec_run("find . -name .exitcode -exec cat {} \;", user="wfcommons", stdout=True, stderr=True)
     # Check sanity
     if exit_code != 0:
@@ -323,19 +331,19 @@ class TestTranslators:
     @pytest.mark.parametrize(
         "backend",
         [
-           "swiftt",
-           "dask",
-           "parsl",
+           # "swiftt",
+           # "dask",
+           # "parsl",
            "nextflow",
-           "nextflow_subworkflow",
-           "airflow",
-           "bash",
-           "taskvine",
-           "makeflow",
-           "snakemake",
-           "cwl",
-           "streamflow",
-           "pegasus",
+           # "nextflow_subworkflow",
+           # "airflow",
+           # "bash",
+           # "taskvine",
+           # "makeflow",
+           # "snakemake",
+           # "cwl",
+           # "streamflow",
+           # "pegasus",
         ])
     @pytest.mark.unit
     # @pytest.mark.skip(reason="tmp")
@@ -394,6 +402,13 @@ def test_translator(self, backend) -> None:
                                        instruments_to_ignore=["shell.cwl"])
         elif backend == "snakemake":
             parser = SnakemakeLogsParser(dirpath, snkmt_db=dirpath / "snkmt.sqlite", rules_to_ignore=["all_wfbench_tasks"])
+        elif backend == "nextflow":
+            # parser = NextflowLogsParser(execution_dir = dirpath)
+            parser = ROCrateLogsParser(dirpath,
+                                       steps_to_ignore=["main.cwl#compile_output_files", "main.cwl#compile_log_files"],
+                                       file_extensions_to_ignore=[".out", ".err"],
+                                       instruments_to_ignore=["shell.cwl"])
+
 
         if parser is not None:
             sys.stderr.write(f"[{backend}] Parsing the logs...\n")

wfcommons/wfbench/translator/nextflow.py

@@ -92,6 +92,9 @@ def translate(self, output_folder: pathlib.Path) -> None:
         else:
             self._translate_single_file(output_folder, sorted_tasks)
 
+        # Create the config file for the nf-prov plugin
+        self._write_nf_prov_plugin_config_file(output_folder)
+
         # Create the README file
         self._write_readme_file(output_folder, self.use_subworkflows)
 
@@ -258,6 +261,24 @@ def _generate_task_function(self, task: Task) -> str:
 
         return code
 
+    def _write_nf_prov_plugin_config_file(selfself, output_folder: pathlib.Path):
+        nf_prov_plugin_config_file = output_folder.joinpath("plugin.config")
+        with open(nf_prov_plugin_config_file, "w") as out:
+            out.write("""plugins {
+    id 'nf-prov'
+}
+
+prov {
+    enabled = true
+    formats {
+        wrroc {
+            file = 'ro-crate-metadata.json'
+            overwrite = true
+        }
+    }
+}
+""")
+
     def _write_readme_file(self, output_folder: pathlib.Path, use_subworkflows: bool) -> None:
         """
         Write the README file.

wfcommons/wfinstances/logs/nextflow.py

@@ -45,6 +45,7 @@ def __init__(self,
 
         self.execution_dir = execution_dir
         self.files_map = {}
+        self.tasknames_to_taskids = {}
         self.text_files = None
         self.line_count = None
 
@@ -74,7 +75,7 @@ def build_workflow(self, workflow_name: Optional[str] = None) -> Workflow:
 
     def _parse_execution_report_file(self) -> None:
         """Parse the Nextflow execution report file and gather the tasks information."""
-        trace_data = self._read_data('execution_report_*.html')
+        trace_data = self._read_data('execution_report*.html')
 
         for t in trace_data['trace']:
             task_id = "ID{:06d}".format(int(t['task_id']))
@@ -96,11 +97,12 @@ def _parse_execution_report_file(self) -> None:
                             (int(_parse_number(t['wchar'])) + int(_parse_number(t['write_bytes']))) / 1024),
                         memory=round(int(_parse_number(t['rss'])) / 1024),
                         logger=self.logger)
-            self.workflow.add_node(task_name, task=task)
+            self.workflow.add_task(task)
+            self.tasknames_to_taskids[task_name] = task_id
 
     def _parse_execution_timeline_file(self) -> None:
         """Parse the Nextflow execution timeline file and build the workflow structure."""
-        timeline_data = self._read_data('execution_timeline_*.html')
+        timeline_data = self._read_data('execution_timeline*.html')
         tasks_map = {}
         max_index = 0
 
@@ -116,7 +118,7 @@ def _parse_execution_timeline_file(self) -> None:
             if index > 0:
                 for c in tasks_map[index]:
                     for p in tasks_map[index - 1]:
-                        self.workflow.add_edge(p, c)
+                        self.workflow.add_edge(self.tasknames_to_taskids[p], self.tasknames_to_taskids[c])
 
         self.workflow.makespan = float(
             (int(timeline_data['endingMillis']) - int(timeline_data['beginningMillis'])) / 1000)
@@ -133,7 +135,7 @@ def _read_data(self, file_format: str) -> Dict:
         """
         files = glob.glob(f'{self.execution_dir}/{file_format}')
         if len(files) == 0:
-            raise OSError(f'Unable to find {self.execution_dir} file in: {file_format}')
+            raise OSError(f'Unable to find {file_format} in {self.execution_dir}')
 
         data = None
 

wfcommons/wfinstances/logs/ro_crate.py

@@ -123,9 +123,10 @@ def _construct_data_file_id_name_map(self):
             if item["@type"] != "File":
                 continue
             id = item["@id"]
-            if "alternateName" not in item:
-                continue
-            alternate_name = item["alternateName"]
+            #if "alternateName" not in item:
+            #    continue
+            #alternate_name = item["alternateName"]
+            alternate_name = item.get("alternateName", id)
             self.data_file_id_name_map[id] = alternate_name
 
 
@@ -137,7 +138,8 @@ def _create_tasks(self, create_actions, main_workflow_id):
 
         for create_action in create_actions:
             # Handle overall workflow create_action then skip
-            if create_action["name"] == f"Run of workflow/{main_workflow_id}":
+            if ("Run of workflow" in create_action["name"] or
+                    "workflow run" in create_action["name"]):
                 self._process_main_workflow(create_action)
                 continue
 
@@ -154,8 +156,8 @@ def _create_tasks(self, create_actions, main_workflow_id):
                 continue
 
             # Get all input & output for the create_action
-            input = [obj['@id'] for obj in create_action['object']]
-            output = [obj['@id'] for obj in create_action['result']]
+            input = [obj['@id'] if isinstance(obj, dict) else obj for obj in create_action['object']]
+            output = [obj['@id'] if isinstance(obj, dict) else obj for obj in create_action['result']]
 
             # Filter for actual files
             input_files = self._filter_file_ids(input)
@@ -166,8 +168,8 @@ def _create_tasks(self, create_actions, main_workflow_id):
                         # task_id=create_action['name'],
                         task_id=create_action['name'] + "_" + create_action['@id'],
                         task_type=TaskType.COMPUTE,
-                        runtime=self._time_diff(create_action['startTime'], create_action['endTime']),
-                        executed_at=create_action['startTime'],
+                        runtime=self._time_diff(create_action.get('startTime'), create_action.get('endTime')),
+                        executed_at=create_action.get('startTime',''),
                         input_files=self._get_file_objects(input_files),
                         output_files=self._get_file_objects(output_files),
                         logger=self.logger)
@@ -193,10 +195,11 @@ def _create_tasks(self, create_actions, main_workflow_id):
                 files[outfile]['out'].append(create_action['@id'])
 
             # For each task, track which 'instrument' it uses
-            instrument = create_action['instrument']['@id']
-            if instrument not in instruments:
-                instruments[instrument] = []
-            instruments[instrument].append(create_action['@id'])
+            if create_action.get('instrument'):
+                instrument = create_action['instrument']['@id']
+                if instrument not in instruments:
+                    instruments[instrument] = []
+                instruments[instrument].append(create_action['@id'])
 
         self._add_dependencies(files, instruments)
 
@@ -231,6 +234,8 @@ def _add_dependencies(self, files, instruments):
                         self.workflow.add_dependency(self.task_id_name_map[parent], self.task_id_name_map[child])
 
     def _time_diff(self, start_time, end_time):
+        if not start_time or not end_time:
+            return 0.0
         diff = datetime.fromisoformat(end_time) - datetime.fromisoformat(start_time)
         return diff.total_seconds()
 
@@ -239,19 +244,38 @@ def _get_file_objects(self, files):
         output = []
         for file in files:
             if file not in self.file_objects:
-                self.file_objects[file] = File(file_id=self.data_file_id_name_map[file],
-                                               size=os.path.getsize(f"{self.crate_dir}/{file}"),
-                                               logger=self.logger)
+                #self.file_objects[file] = File(file_id=self.data_file_id_name_map[file],
+                #                               size=os.path.getsize(f"{self.crate_dir}/{file}"),
+                #                               logger=self.logger)
+                if file not in self.data_file_id_name_map:
+                    # File is referenced but not in the map — use its @id as the name
+                    self.logger.warning(f"File not in data_file_id_name_map, using @id as name: {file}") if self.logger else None
+                    file_name = file
+                else:
+                    file_name = self.data_file_id_name_map[file]
+                try:
+                    size = os.path.getsize(f"{self.crate_dir}/{file}")
+                except (OSError, ValueError):
+                    size = 0  # file:// absolute paths won't resolve relative to crate_dir
+                self.file_objects[file] = File(file_id=file_name,
+                                            size=size,
+                                            logger=self.logger)
             output.append(self.file_objects[file])
         return output
 
     def _filter_file_ids(self, ids):
 
-        file_ids = list(filter(lambda x: self.lookup.get(x)['@type'] == 'File', ids))
-        property_value_ids = list(filter(lambda x: self.lookup.get(x)['@type'] == 'PropertyValue', ids))
+        file_ids = list(filter(lambda x: (self.lookup.get(x) or {}).get('@type') == 'File', ids))
+        # Ignore the files that start with http:// or https://
+        file_ids = [x for x in file_ids if not x.startswith("http://")]
+        file_ids = [x for x in file_ids if not x.startswith("https://")]
+        property_value_ids = list(filter(lambda x: (self.lookup.get(x) or {}).get('@type') == 'PropertyValue', ids))
         for property_value_id in property_value_ids:
             property_values = self.lookup.get(property_value_id)['value']
-            if isinstance(property_values, dict):
+            # If the lookup fails, ignore
+            if not property_values:
+                    continue
+            if not isinstance(property_values, list):
                 property_values = [property_values]
 
             # Filter out values without "@id"s (i.e. int values, etc.)
@@ -279,4 +303,4 @@ def _filter_file_ids(self, ids):
 
     def _process_main_workflow(self, main_workflow):
         self.workflow.makespan = self._time_diff(main_workflow['startTime'], main_workflow['endTime'])
-        self.workflow.executed_at = main_workflow['startTime']
+        self.workflow.executed_at = main_workflow['startTime']