feat: Add step information for executions of multiple steps at a time

Author: AcquaDiGiorgio

src/dirac_cwl/commands/analyze_xml_summary.py

@@ -4,42 +4,37 @@
 import os
 
 from LHCbDIRAC.Workflow.Modules.AnalyseXMLSummary import _areInputsOK, _isXMLSummaryOK
-from LHCbDIRAC.Workflow.Modules.BookkeepingReport import _generate_xml_object
 
 from dirac_cwl.core.exceptions import WorkflowProcessingException
 
 from .core import PostProcessCommand
-from .workflow_commons import StepStatus, WorkflowCommons
+from .workflow_commons import Step, StepStatus, WorkflowCommons
 
 logger = logging.getLogger(__name__)
 
 
 class AnalyseXmlSummary(PostProcessCommand):
     """Performs a series of checks on the XMLSummary output to make sure the execution was done correctly."""
 
-    def _execute(self, job_path: os.PathLike, workflow_commons: WorkflowCommons, **kwargs) -> None:
+    def _execute(self, job_path: os.PathLike, workflow_commons: WorkflowCommons, **kwargs):
         """Execute the command.
 
         :param job_path: Path to the job working directory.
-        :param workflow_commons: WorkflowCommons object.
+        :param workflow_commons: WorkflowCommons object
         :param kwargs: Additional keyword arguments.
         """
-        if not workflow_commons.xf_o:
-            workflow_commons.xf_o = _generate_xml_object(
-                workflow_commons.cleaned_application_name,
-                workflow_commons.production_id,
-                workflow_commons.prod_job_id,
-                workflow_commons.step_number,
-                workflow_commons.step_id,
-            )
+        for step in workflow_commons.steps:
+            self._execute_for_step(job_path, workflow_commons, step, **kwargs)
 
-        jobOk = _isXMLSummaryOK(workflow_commons.xf_o)
+    def _execute_for_step(self, job_path: os.PathLike, workflow_commons: WorkflowCommons, step_commons: Step, **kwargs):
+        """Execute the command for a specific step."""
+        jobOk = _isXMLSummaryOK(step_commons.xf_o)
 
         if jobOk:
             jobOk = _areInputsOK(
-                workflow_commons.xf_o,
-                workflow_commons.inputs,
-                workflow_commons.number_of_events,
+                step_commons.xf_o,
+                step_commons.inputs,
+                step_commons.number_of_events,
                 workflow_commons.production_id,
                 workflow_commons.file_report,
             )
@@ -51,4 +46,4 @@ def _execute(self, job_path: os.PathLike, workflow_commons: WorkflowCommons, **k
             logger.info("Workflow already failed")
             return
 
-        workflow_commons.job_report.setApplicationStatus(f"{workflow_commons.application_name} Step OK")
+        workflow_commons.job_report.setApplicationStatus(f"{step_commons.application_name} Step OK")

src/dirac_cwl/commands/bookkeeping_report.py

@@ -1,5 +1,6 @@
 """LHCb command for bookkeeping report file generation based on the XMLSummary and the XML catalog."""
 
+import copy
 import logging
 import os
 from typing import Any, Dict
@@ -19,7 +20,7 @@
 from dirac_cwl.core.exceptions import WorkflowProcessingException
 
 from .core import PostProcessCommand
-from .workflow_commons import StepStatus, WorkflowCommons
+from .workflow_commons import Step, StepStatus, WorkflowCommons
 
 logger = logging.getLogger(__name__)
 
@@ -34,29 +35,40 @@ def _execute(self, job_path: os.PathLike, workflow_commons: WorkflowCommons, **k
         :param workflow_commons: WorkflowCommons object
         :param kwargs: Additional keyword arguments.
         """
-        # Obtain Workflow Commons
-        if workflow_commons.step_status == StepStatus.Failed:
-            return
+        for step in workflow_commons.steps:
+            if workflow_commons.step_status == StepStatus.Failed:
+                return
 
+            self._execute_for_step(job_path, workflow_commons, step, **kwargs)
+
+    def _execute_for_step(self, job_path: os.PathLike, workflow_commons: WorkflowCommons, step_commons: Step, **kwargs):
         # Setup variables
         cpu_times: Dict[str, Any] = {}
-        if workflow_commons.start_time:
-            cpu_times["StartTime"] = workflow_commons.start_time
-        if workflow_commons.start_stats:
-            cpu_times["StartStats"] = workflow_commons.start_stats
+        if step_commons.start_time:
+            cpu_times["StartTime"] = step_commons.start_time
+        if step_commons.start_stats:
+            cpu_times["StartStats"] = step_commons.start_stats
 
         exectime, cputime = getStepCPUTimes(cpu_times)
 
-        number_of_processors = getNumberOfProcessorsToUse(
-            workflow_commons.job_id,
-            workflow_commons.max_number_of_processors,
-        )
+        number_of_processors = workflow_commons.number_of_processors
+
+        if (step_commons.multicore and workflow_commons.multicore) or (
+            workflow_commons.job_type.lower() == "user" and workflow_commons.max_number_of_processors
+        ):
+            number_of_processors = getNumberOfProcessorsToUse(
+                workflow_commons.job_id,
+                workflow_commons.max_number_of_processors,
+            )
+
+        all_outputs = copy.deepcopy(step_commons.outputs)
+        all_outputs.extend(step_commons.outputs)
 
         parameters = {
             "PRODUCTION_ID": workflow_commons.production_id,
             "JOB_ID": workflow_commons.prod_job_id,
             "configVersion": workflow_commons.config_version,
-            "outputList": workflow_commons.outputs,
+            "outputList": all_outputs,
             "configName": workflow_commons.config_name,
             "outputDataFileMask": workflow_commons.output_data_file_mask,
         }
@@ -79,16 +91,16 @@ def _execute(self, job_path: os.PathLike, workflow_commons: WorkflowCommons, **k
 
             bk_lfns = production_lfns_dict["BookkeepingLFNs"]
 
-        ldate, ltime, ldatestart, ltimestart = _process_time(workflow_commons.start_time)
+        ldate, ltime, ldatestart, ltimestart = _process_time(step_commons.start_time)
 
         # Obtain XMLSummary
-        if not workflow_commons.xf_o:
-            workflow_commons.xf_o = _generate_xml_object(
-                workflow_commons.cleaned_application_name,
+        if not step_commons.xf_o:
+            step_commons.xf_o = _generate_xml_object(
+                step_commons.cleaned_application_name,
                 workflow_commons.production_id,
                 workflow_commons.prod_job_id,
-                workflow_commons.step_number,
-                workflow_commons.step_id,
+                step_commons.number,
+                step_commons.id,
             )
 
         info_dict = {
@@ -99,9 +111,9 @@ def _execute(self, job_path: os.PathLike, workflow_commons: WorkflowCommons, **k
             "jobID": workflow_commons.job_id,
             "siteName": workflow_commons.site_name,
             "jobType": workflow_commons.job_type,
-            "applicationName": workflow_commons.application_name,
-            "applicationVersion": workflow_commons.application_version,
-            "numberOfEvents": workflow_commons.number_of_events,
+            "applicationName": step_commons.application_name,
+            "applicationVersion": step_commons.application_version,
+            "numberOfEvents": step_commons.number_of_events,
         }
 
         # Generate job_info object
@@ -111,37 +123,40 @@ def _execute(self, job_path: os.PathLike, workflow_commons: WorkflowCommons, **k
             ltimestart,
             ldate,
             ltime,
-            workflow_commons.xf_o,
-            workflow_commons.inputs,
-            workflow_commons.step_id,
-            workflow_commons.bk_step_id,
+            step_commons.xf_o,
+            step_commons.inputs,
+            step_commons.id,
+            step_commons.bk_id,
             workflow_commons.bk_client,
             workflow_commons.config_name,
             workflow_commons.config_version,
         )
 
         # Add input files to job_info
-        _generateInputFiles(job_info, bk_lfns, workflow_commons.inputs)
+        _generateInputFiles(job_info, bk_lfns, step_commons.inputs)
 
         # Add output files to job_info
         _generateOutputFiles(
             job_info,
             bk_lfns,
-            workflow_commons.event_type,
-            workflow_commons.application_name,
-            workflow_commons.xf_o,
-            workflow_commons.outputs,
-            workflow_commons.inputs,
+            step_commons.event_type,
+            step_commons.application_name,
+            step_commons.xf_o,
+            step_commons.outputs,
+            step_commons.inputs,
+            step_commons.size,
+            step_commons.md5,
+            step_commons.guid,
         )
 
         # Generate SimulationConditions
-        if workflow_commons.application_name == "Gauss":
+        if step_commons.application_name == "Gauss":
             job_info.simulation_condition = workflow_commons.sim_description
 
         # Convert job_info object to XML
         doc = job_info.to_xml()
 
         # Write to file
-        bfilename = f"bookkeeping_{workflow_commons.step_id}.xml"
+        bfilename = f"bookkeeping_{step_commons.id}.xml"
         with open(bfilename, "wb") as bfile:
             bfile.write(doc)

src/dirac_cwl/commands/core.py

@@ -29,7 +29,7 @@ def execute(self, job_path: os.PathLike, **kwargs) -> None:
         workflow_commons = None
         try:
             workflow_commons = WorkflowCommons.load(job_path)
-
+            logger.info("WorkflowCommons:\n%s", workflow_commons)
             self._execute(job_path, workflow_commons, **kwargs)
 
         except WorkflowProcessingException:

src/dirac_cwl/commands/download_config.py

@@ -4,11 +4,13 @@
 
 from dirac_cwl.commands import PreProcessCommand
 
+from .workflow_commons import WorkflowCommons
+
 
 class DownloadConfig(PreProcessCommand):
     """Example command that creates a file with named 'content.cfg'."""
 
-    def execute(self, job_path, **kwargs):
+    def _execute(self, job_path: os.PathLike, workflow_commons: WorkflowCommons, **kwargs):
         """Execute the configuration download.
 
         :param job_path: Path to the job working directory.

src/dirac_cwl/commands/group_outputs.py

@@ -5,11 +5,13 @@
 
 from dirac_cwl.commands import PostProcessCommand
 
+from .workflow_commons import WorkflowCommons
+
 
 class GroupOutputs(PostProcessCommand):
     """Example command that merges all of the outputs in a singular file."""
 
-    def execute(self, job_path, **kwargs):
+    def _execute(self, job_path: os.PathLike, workflow_commons: WorkflowCommons, **kwargs):
         """Execute the output file grouping.
 
         :param job_path: Path to the job working directory.

src/dirac_cwl/commands/workflow_accounting.py

@@ -16,27 +16,34 @@
 from dirac_cwl.core.exceptions import WorkflowProcessingException
 
 from .core import PostProcessCommand
-from .workflow_commons import WorkflowCommons
+from .workflow_commons import Step, WorkflowCommons
 
 logger = logging.getLogger(__name__)
 
 
 class WorkflowAccounting(PostProcessCommand):
     """Prepares and sends accounting information to the DIRAC Accounting system."""
 
-    def _execute(self, job_path: os.PathLike, workflow_commons: WorkflowCommons, **kwargs) -> None:
+    def _execute(self, job_path: os.PathLike, workflow_commons: WorkflowCommons, **kwargs):
         """Execute the command.
 
         :param job_path: Path to the job working directory.
-        :param workflow_commons: WorkflowCommons object.
+        :param workflow_commons: WorkflowCommons object
         :param kwargs: Additional keyword arguments.
         """
-        # Obtain Workflow Commons
+        for step in workflow_commons.steps:
+            self._execute_for_step(job_path, workflow_commons, step, **kwargs)
+
+    def _execute_for_step(self, job_path: os.PathLike, workflow_commons: WorkflowCommons, step_commons: Step, **kwargs):
         cpu_times: Dict[str, Any] = {}
-        if workflow_commons.start_time:
-            cpu_times["StartTime"] = workflow_commons.start_time
-        if workflow_commons.start_stats:
-            cpu_times["StartStats"] = workflow_commons.start_stats
+        if step_commons.start_time:
+            cpu_times["StartTime"] = step_commons.start_time
+        if step_commons.start_stats:
+            cpu_times["StartStats"] = step_commons.start_stats
+
+        if not step_commons.application_name:
+            logger.info("Not an application step: it will not be accounted")
+            return
 
         exec_time, cpu_time = getStepCPUTimes(cpu_times)
 
@@ -45,7 +52,9 @@ def _execute(self, job_path: os.PathLike, workflow_commons: WorkflowCommons, **k
 
         job_step = JobStep()
 
-        if not workflow_commons.xf_o:
+        xf_o = step_commons.xf_o
+
+        if not xf_o:
             logger.error("XML Summary object could not be found (not produced?), skipping the report")
             return
 
@@ -56,18 +65,18 @@ def _execute(self, job_path: os.PathLike, workflow_commons: WorkflowCommons, **k
         dataDict = {
             "JobGroup": str(workflow_commons.production_id),
             "RunNumber": workflow_commons.run_number,
-            "EventType": workflow_commons.event_type,
-            "ProcessingType": workflow_commons.step_proc_pass,  # this is the processing pass of the step
-            "ProcessingStep": workflow_commons.bk_step_id,  # the step ID
+            "EventType": step_commons.event_type,
+            "ProcessingType": step_commons.proc_pass,  # this is the processing pass of the step
+            "ProcessingStep": step_commons.bk_id,  # the step ID
             "Site": workflow_commons.site_name,
             "FinalStepState": workflow_commons.step_status,
             "CPUTime": cpu_time,
             "NormCPUTime": normCPU,
             "ExecTime": exec_time * workflow_commons.number_of_processors,
-            "InputData": sum(workflow_commons.xf_o.inputFileStats.values()),
-            "OutputData": sum(workflow_commons.xf_o.outputFileStats.values()),
-            "InputEvents": workflow_commons.xf_o.inputEventsTotal,
-            "OutputEvents": workflow_commons.xf_o.outputEventsTotal,
+            "InputData": sum(xf_o.inputFileStats.values()),
+            "OutputData": sum(xf_o.outputFileStats.values()),
+            "InputEvents": xf_o.inputEventsTotal,
+            "OutputEvents": xf_o.outputEventsTotal,
         }
 
         job_step.setValuesFromDict(dataDict)