Updated the code for the Hierarchical Pegasus logger

Excluded the Hierarchical Pegasus logger from testing

Author: henricasanova

pyproject.toml

@@ -61,6 +61,7 @@ addopts="""
 --cov-report term-missing \
 --cov ./wfcommons \
 --ignore wfcommons/wfbench/translator/templates \
+--ignore wfcommons/wfinstances/logs/pegasusrec.py \
 --no-cov-on-fail \
 -ra \
 -W ignore"""

wfcommons/wfinstances/instance_analyzer.py

@@ -116,7 +116,7 @@ def build_summary(self,
                     if extension[1:].isnumeric():
                         extension = path.splitext(infile.file_id.replace(extension, ''))[1]
 
-                    # Check if the file is definetly an input 
+                    # Check if the file is definitely an input
                     assert infile.link == FileLink.INPUT, f"{infile.file_id} is not set as input"
                     _append_file_to_dict(extension, inputs_dict, infile.size)
 

wfcommons/wfinstances/logs/pegasusrec.py

@@ -30,9 +30,11 @@
 
 class HierarchicalPegasusLogsParser(LogsParser):
     """
-    Parse Pegasus submit directory to generate workflow instance. 
-    This specific parser targets Pegasus submit dir with Hierachical workflows
-    Which means that some jobs are sub-workflows and must be parse as well.
+    [OUT OF DATE AND NO LONGER FUNCTIONING]
+    [LIKELY TO BE DEPRECATED/REMOVED AT SOME POINT ANYWAY]
+    Parse Pegasus submit directory to generate workflow instance.
+    This specific parser targets Pegasus submit dir with Hierarchical workflows
+    Which means that some jobs are sub-workflows and must be parsed as well.
     This parser recursively parse each sub workflow and rebuild a coherent workflow.
 
     WARNING: one level of recursion for now (i.e., sub-workflow cannot have sub-workflow)
@@ -204,9 +206,9 @@ def _parse_braindump(self):
         # create base workflow instance object
         self.workflow = Workflow(name=self.workflow_name,
                                  description=self.description,
-                                 wms_name=self.wms_name,
-                                 wms_version=wms_version,
-                                 wms_url=self.wms_url,
+                                 runtime_system_name=self.wms_name,
+                                 runtime_system_version=wms_version,
+                                 runtime_system_url=self.wms_url,
                                  executed_at=executed_at)
 
     def _parse_workflow(self):
@@ -232,12 +234,15 @@ def _parse_workflow(self):
                     task_name = f"{j['name']}_{j['id']}"
 
                     list_files = [File(
-                        name=f['lfn'],
+                        file_id=f['lfn'],
                         size=0,
                         link=FileLink(f['type']),
                         logger=self.logger
                     ) for f in j['uses']]
 
+                    input_files = [f for f in list_files if f.link == FileLink.INPUT]
+                    output_files = [f for f in list_files if f.link == FileLink.OUTPUT]
+
                     self.workflow.add_node(
                         task_name,
                         task=Task(
@@ -248,7 +253,8 @@ def _parse_workflow(self):
                             runtime=0,
                             args=j['arguments'],
                             cores=0,
-                            files=list_files,
+                            input_files=input_files,
+                            output_files=output_files,
                             logger=self.logger
                         )
                     )
@@ -267,7 +273,8 @@ def _parse_workflow(self):
                             runtime=0,
                             args=j['arguments'],
                             cores=0,
-                            files=list_files,
+                            input_files=input_files,
+                            output_files=output_files,
                             logger=self.logger
                         )
                     )
@@ -311,12 +318,15 @@ def _parse_dax(self):
                 task_name = str(j.get('name')) + '_' + str(j.get('id'))
 
                 list_files = [File(
-                    name=f.get('name') if not f.get('name') is None else f.get('file'),
+                    file_id=f.get('name') if not f.get('name') is None else f.get('file'),
                     size=0,
                     link=FileLink(f.get('link')),
                     logger=self.logger
                 ) for f in j.findall('{http://pegasus.isi.edu/schema/DAX}uses')]
 
+                input_files = [f for f in list_files if f.link == FileLink.INPUT]
+                output_files = [f for f in list_files if f.link == FileLink.OUTPUT]
+
                 self.workflow.add_node(
                     task_name,
                     task=Task(
@@ -327,7 +337,8 @@ def _parse_dax(self):
                         runtime=0,
                         args=[],
                         cores=0,
-                        files=list_files,
+                        input_files=input_files,
+                        output_files=output_files,
                         logger=self.logger
                     )
                 )
@@ -391,7 +402,8 @@ def _parse_dag(self):
                                 runtime=0,
                                 args=[],
                                 cores=0,
-                                files=[],
+                                input_files=[],
+                                output_files=[],
                                 logger=self.logger
                             )
                         )
@@ -646,15 +658,15 @@ def _parse_job_output_legacy(self, task: Task, output_file_path: pathlib.Path) -
                             machine['memory'] = int(r.get('total'))
                         for c in u.findall('{http://pegasus.isi.edu/schema/invocation}cpu'):
                             machine['cpu'] = {
-                                'count': int(c.get('count')),
-                                'speed': int(c.get('speed')),
+                                'coreCount': int(c.get('count')),
+                                'speedInMHz': int(c.get('speed')),
                                 'vendor': c.get('vendor')
                             }
                     task.machine = Machine(
                         name=machine['nodeName'],
                         cpu={
-                            'count': machine['cpu']['count'],
-                            'speed': machine['cpu']['speed'],
+                            'coreCount': machine['cpu']['count'],
+                            'speedInMHz': machine['cpu']['speed'],
                             'vendor': machine['cpu']['vendor']
                         },
                         system=machine['system'],