Fixed RO-Crate parser to preserve original task names (easier debugging)

Augmented TO-Crate parser to take in an optional list of step names to ignore

Author: henricasanova

tests/translators_loggers/Dockerfile.streamflow

@@ -24,6 +24,7 @@ RUN apt-get -y install vim --fix-missing
 RUN apt-get -y install gcc 
 RUN apt-get -y install gcc-multilib
 RUN apt-get -y install graphviz libgraphviz-dev
+RUN apt-get -y install zip
 
 
 

tests/translators_loggers/test_translators_loggers.py

@@ -209,6 +209,10 @@ def run_workflow_streamflow(container, num_tasks, str_dirpath):
     uuid = output.decode().splitlines()[1].strip().split(" ")[0]
     exit_code, output = container.exec_run(cmd=f"streamflow prov {uuid}",
                                            user="wfcommons", stdout=True, stderr=True)
+    exit_code, output = container.exec_run(cmd=f"mkdir RO-Crate",
+                                           user="wfcommons", stdout=True, stderr=True)
+    exit_code, output = container.exec_run(cmd=f"unzip *.zip -d ./RO-Crate",
+                                           user="wfcommons", stdout=True, stderr=True)
 
 def run_workflow_pegasus(container, num_tasks, str_dirpath):
     # Run the workflow!
@@ -264,18 +268,18 @@ class TestTranslators:
     @pytest.mark.parametrize(
         "backend",
         [
-           "swiftt",
-           "dask",
-           "parsl",
-           "nextflow",
-           "nextflow_subworkflow",
-           "airflow",
-           "bash",
-           "taskvine",
-           "makeflow",
-           "cwl",
+           # "swiftt",
+           # "dask",
+           # "parsl",
+           # "nextflow",
+           # "nextflow_subworkflow",
+           # "airflow",
+           # "bash",
+           # "taskvine",
+           # "makeflow",
+           # "cwl",
            "streamflow",
-           "pegasus",
+           # "pegasus",
         ])
     @pytest.mark.unit
     # @pytest.mark.skip(reason="tmp")
@@ -330,8 +334,8 @@ def test_translator(self, backend) -> None:
             parser = TaskVineLogsParser(dirpath / "vine-run-info/most-recent/vine-logs", filenames_to_ignore=["cpu-benchmark","stress-ng", "wfbench"])
         elif backend == "makeflow":
             parser = MakeflowLogsParser(execution_dir = dirpath, resource_monitor_logs_dir = dirpath / "monitor_data/")
-        # elif backend == "streamflow":
-        #     parser =ROCrateLogsParser(dirpath / "work/wfcommons/most-recent/wfbench")
+        elif backend == "streamflow":
+            parser = ROCrateLogsParser(dirpath / "RO-Crate", steps_to_ignore=["main.cwl#compile_output_files", "main.cwl#compile_log_files"])
 
         if parser is not None:
             sys.stderr.write(f"[{backend}] Parsing the logs...\n")
@@ -340,6 +344,18 @@ def test_translator(self, backend) -> None:
 
             original_workflow : Workflow = benchmark.workflow
 
+            # print(original_workflow.tasks)
+            # print("======")
+            # print(reconstructed_workflow.tasks)
+            for task_name in original_workflow.tasks.keys():
+                original_task = original_workflow.tasks[task_name]
+                reconstructed_task = reconstructed_workflow.tasks[task_name]
+                print("ORIGINAL:", original_task.task_id, "RECONSTRUCTED:", reconstructed_task.task_id)
+                print("   NUM_INPUT_FILES: ", len(original_task.input_files), len(reconstructed_task.input_files))
+                print("   NUM_OUTPUT_FILES: ", len(original_task.output_files), len(reconstructed_task.output_files))
+                print("   INPUT FILES: ", [f.file_id for f in original_task.input_files], [f.file_id for f in reconstructed_task.input_files])
+                print("   OUTPUT FILES: ", [f.file_id for f in original_task.output_files], [f.file_id for f in reconstructed_task.output_files])
+
             _compare_workflows(original_workflow, reconstructed_workflow)
 
         # Shutdown the container (weirdly, container is already shutdown by now... not sure how)

wfcommons/wfinstances/logs/ro_crate.py

@@ -44,13 +44,16 @@ class ROCrateLogsParser(LogsParser):
     def __init__(self,
                  crate_dir: pathlib.Path,
                  description: Optional[str] = None,
-                 logger: Optional[Logger] = None) -> None:
+                 logger: Optional[Logger] = None,
+                 steps_to_ignore: Optional[list[str]]=None) -> None:
         """Create an object of the RO crate parser."""
 
         # TODO: Decide if these should be RO crate or Streamflow or whatev
         super().__init__('Streamflow-ROCrate', 'https://w3id.org/workflowhub/workflow-ro-crate/1.0', description, logger)
 
         # Sanity check
+        if steps_to_ignore is None:
+            steps_to_ignore = []
         if not crate_dir.is_dir():
             raise OSError(f'The provided path does not exist or is not a folder: {crate_dir}')
 
@@ -63,6 +66,10 @@ def __init__(self,
 
         self.file_objects = {}
 
+        self.task_id_name_map: dict[str, str] = {}
+
+        self.steps_to_ignore = steps_to_ignore
+
 
     def build_workflow(self, workflow_name: Optional[str] = None) -> Workflow:
         """
@@ -113,6 +120,16 @@ def _create_tasks(self, create_actions, main_workflow_id):
                 self._process_main_workflow(create_action)
                 continue
 
+            create_action['name'] = create_action['name'].removeprefix("Run of workflow/")
+
+            # Below would remove the "file.cwl#" tag, which runs the risk
+            # of non-uniqueness of action names perhaps
+            # create_action['name'] = create_action['name'].split('#', 1)[-1]
+
+            # Check if we should ignore this step
+            if create_action["name"] in self.steps_to_ignore:
+                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']]
@@ -121,17 +138,19 @@ def _create_tasks(self, create_actions, main_workflow_id):
             input_files = self._filter_file_ids(input)
             output_files = self._filter_file_ids(output)
 
-            create_action['name'] = create_action['name'].removeprefix("Run of workflow/")
 
             task = Task(name=create_action['name'],
-                        task_id=create_action['@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'],
                         input_files=self._get_file_objects(input_files),
                         output_files=self._get_file_objects(output_files),
                         logger=self.logger)
             self.workflow.add_task(task)
+            self.task_id_name_map[create_action['@id']] = create_action['name']
+            # self.task_id_name_map[create_action['@id']] = create_action['name'] + "_" + create_action['@id']
 
             # For each file, track which task(s) it is in/output for
             for infile in input_files:
@@ -162,20 +181,29 @@ def _add_dependencies(self, files, instruments):
         for file in files.values():
             for parent in file.get('out', []):
                 for child in file.get('in', []):
-                    self.workflow.add_dependency(parent, child)
+                    # self.workflow.add_dependency(parent, child)
+                    self.workflow.add_dependency(self.task_id_name_map[parent], self.task_id_name_map[child])
 
         # Assumes
         parameter_connections = list(filter((lambda x: x.get('@type') == "ParameterConnection"), self.graph_data))
         for parameter_connection in parameter_connections:
-            source = parameter_connection["sourceParameter"]["@id"]
-            source = source.rsplit("#", 1)[0]   # Trim to get instrument
+            # parameter_connection["sourceParameter"] is either a single dict or a list of dicts,
+            # which is bad design but whatever
+            source_parameters = parameter_connection["sourceParameter"]
+            if not isinstance(source_parameters, list):
+                source_parameters = [source_parameters]
+
+            for item in source_parameters:
+                source = item["@id"]
+                source = source.rsplit("#", 1)[0]   # Trim to get instrument
+
+                target = parameter_connection["targetParameter"]["@id"]
+                target = target.rsplit("#", 1)[0]   # Trim to get instrument
 
-            target = parameter_connection["targetParameter"]["@id"]
-            target = target.rsplit("#", 1)[0]   # Trim to get instrument
+                for parent in instruments.get(source, []):
+                    for child in instruments.get(target, []):
+                        self.workflow.add_dependency(self.task_id_name_map[parent], self.task_id_name_map[child])
 
-            for parent in instruments.get(source, []):
-                for child in instruments.get(target, []):
-                    self.workflow.add_dependency(parent, child)
 
     def _time_diff(self, start_time, end_time):
         diff = datetime.fromisoformat(end_time) - datetime.fromisoformat(start_time)