Skip to content

Commit 98ccc37

Browse files
AcquaDiGiorgioAcquaDiGiorgio
committed
feat: Migrate BookkeepingReport command to cwl-dirac
1 parent 91cef73 commit 98ccc37

4 files changed

Lines changed: 728 additions & 2 deletions

File tree

‎src/dirac_cwl/commands/__init__.py‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
"""Command classes for workflow pre/post-processing operations."""
22

3+
from .bookkeeping_report import BookeepingReport
34
from .core import PostProcessCommand, PreProcessCommand
45
from .upload_log_file import UploadLogFile
56

6-
__all__ = ["PreProcessCommand", "PostProcessCommand", "UploadLogFile"]
7+
__all__ = ["PreProcessCommand", "PostProcessCommand", "UploadLogFile", "BookeepingReport"]

‎src/dirac_cwl/commands/bookkeeping_report.py‎

Lines changed: 159 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,159 @@
1+
"""LHCb command for bookkeeping report file generation based on the XMLSummary and the XML catalog."""
2+
3+
import os
4+
5+
from DIRAC.Workflow.Utilities.Utils import getStepCPUTimes
6+
from LHCbDIRAC.BookkeepingSystem.Client.BookkeepingClient import BookkeepingClient
7+
from LHCbDIRAC.Core.Utilities.ProductionData import constructProductionLFNs
8+
from LHCbDIRAC.Core.Utilities.XMLSummaries import XMLSummary
9+
from LHCbDIRAC.Workflow.Modules.BookkeepingReport import (
10+
_generate_xml_object,
11+
_generateInputFiles,
12+
_generateOutputFiles,
13+
_prepare_job_info,
14+
_process_time,
15+
)
16+
from LHCbDIRAC.Workflow.Modules.ModulesUtilities import getNumberOfProcessorsToUse
17+
18+
from dirac_cwl.core.exceptions import WorkflowProcessingException
19+
20+
from .core import PostProcessCommand
21+
from .utils import prepare_lhcb_workflow_commons
22+
23+
24+
class BookeepingReport(PostProcessCommand):
25+
"""Generates a bookkeeping report file based on the XMLSummary and the pool XML catalog."""
26+
27+
def execute(self, job_path, **kwargs):
28+
"""Execute the command.
29+
30+
:param job_path: Path to the job working directory.
31+
:param kwargs: Additional keyword arguments.
32+
"""
33+
# Obtain Workflow Commons
34+
workflow_commons_path = kwargs.get("workflow-commons-path", os.path.join(job_path, "workflow_commons.json"))
35+
36+
workflow_commons = prepare_lhcb_workflow_commons(
37+
workflow_commons_path,
38+
extra_mandatory_values=[
39+
"bk_step_id",
40+
],
41+
extra_default_values={
42+
"bookkeeping_LFNs": [],
43+
"size": {},
44+
"md5": {},
45+
"guid": {},
46+
"sim_description": "NoSimConditions",
47+
},
48+
)
49+
50+
if not workflow_commons["step_status"]["OK"]:
51+
return
52+
53+
# Setup variables
54+
start_time = workflow_commons.get("start_time", None)
55+
56+
cpu_times = {}
57+
if start_time:
58+
cpu_times["StartTime"] = start_time
59+
if "start_stats" in workflow_commons:
60+
cpu_times["StartStats"] = workflow_commons["start_stats"]
61+
62+
exectime, cputime = getStepCPUTimes(cpu_times)
63+
64+
number_of_processors = getNumberOfProcessorsToUse(
65+
workflow_commons["job_id"], workflow_commons["max_number_of_processors"]
66+
)
67+
68+
bk_client = BookkeepingClient()
69+
70+
parameters = {
71+
"PRODUCTION_ID": workflow_commons["production_id"],
72+
"JOB_ID": workflow_commons["prod_job_id"],
73+
"configVersion": workflow_commons["config_version"],
74+
"outputList": workflow_commons["outputs"],
75+
"configName": workflow_commons["config_name"],
76+
"outputDataFileMask": workflow_commons["output_data_file_mask"],
77+
}
78+
79+
if "bookkeeping_LFNs" in workflow_commons and "production_output_data" in workflow_commons:
80+
bk_lfns = workflow_commons["bookkeeping_LFNs"]
81+
82+
if not isinstance(bk_lfns, list):
83+
bk_lfns = [i.strip() for i in bk_lfns.split(";")]
84+
85+
else:
86+
result = constructProductionLFNs(parameters, bk_client)
87+
if not result["OK"]:
88+
raise WorkflowProcessingException("Could not create production LFNs")
89+
90+
bk_lfns = result["Value"]["BookkeepingLFNs"]
91+
92+
ldate, ltime, ldatestart, ltimestart = _process_time(start_time)
93+
94+
# Obtain XMLSummary
95+
if "xml_summary_path" in workflow_commons:
96+
xf_o = XMLSummary(workflow_commons["xml_summary_path"])
97+
else:
98+
xf_o = _generate_xml_object(
99+
workflow_commons["cleaned_application_name"],
100+
workflow_commons["production_id"],
101+
workflow_commons["prod_job_id"],
102+
workflow_commons["command_number"],
103+
workflow_commons["command_id"],
104+
)
105+
106+
info_dict = {
107+
"exectime": exectime,
108+
"cputime": cputime,
109+
"numberOfProcessors": number_of_processors,
110+
"production_id": workflow_commons["production_id"],
111+
"jobID": workflow_commons["job_id"],
112+
"siteName": workflow_commons["site_name"],
113+
"jobType": workflow_commons["job_type"],
114+
"applicationName": workflow_commons["application_name"],
115+
"applicationVersion": workflow_commons["application_version"],
116+
"numberOfEvents": workflow_commons["number_of_events"],
117+
}
118+
119+
# Generate job_info object
120+
job_info = _prepare_job_info(
121+
info_dict,
122+
ldatestart,
123+
ltimestart,
124+
ldate,
125+
ltime,
126+
xf_o,
127+
workflow_commons["inputs"],
128+
workflow_commons["command_id"],
129+
workflow_commons["bk_step_id"],
130+
bk_client,
131+
workflow_commons["config_name"],
132+
workflow_commons["config_version"],
133+
)
134+
135+
# Add input files to job_info
136+
_generateInputFiles(job_info, bk_lfns, workflow_commons["inputs"])
137+
138+
# Add output files to job_info
139+
_generateOutputFiles(
140+
job_info,
141+
bk_lfns,
142+
workflow_commons["event_type"],
143+
workflow_commons["application_name"],
144+
xf_o,
145+
workflow_commons["outputs"],
146+
workflow_commons["inputs"],
147+
)
148+
149+
# Generate SimulationConditions
150+
if workflow_commons["application_name"] == "Gauss":
151+
job_info.simulation_condition = workflow_commons["sim_description"]
152+
153+
# Convert job_info object to XML
154+
doc = job_info.to_xml()
155+
156+
# Write to file
157+
bfilename = f"bookkeeping_{workflow_commons['command_id']}.xml"
158+
with open(bfilename, "wb") as bfile:
159+
bfile.write(doc)

‎src/dirac_cwl/commands/utils.py‎

Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,92 @@
1+
"""."""
2+
3+
import json
4+
import os
5+
6+
from DIRAC import siteName
7+
from DIRAC.Core.Utilities.ReturnValues import S_OK
8+
9+
from dirac_cwl.core.exceptions import WorkflowProcessingException
10+
11+
12+
def prepare_lhcb_workflow_commons(workflow_commons_path, extra_mandatory_values=[], extra_default_values={}):
13+
"""Return a dictionary containing the values of a workflow_commons.json file.
14+
15+
Also performs a series of checks to ensure everything is in order.
16+
"""
17+
if not os.path.exists(workflow_commons_path):
18+
raise WorkflowProcessingException(f"{workflow_commons_path} file not found")
19+
20+
with open(workflow_commons_path, "r", encoding="utf-8") as f:
21+
workflow_commons = json.load(f)
22+
23+
if not workflow_commons:
24+
raise WorkflowProcessingException(f"{workflow_commons_path} cannot be empty")
25+
26+
mandatory_values = [
27+
"job_id",
28+
"job_type",
29+
"production_id",
30+
"prod_job_id",
31+
"number_of_events",
32+
"application_name",
33+
"application_version",
34+
"inputs",
35+
"outputs", # outputList
36+
"executable",
37+
"command_id", # StepID
38+
"command_number",
39+
]
40+
41+
mandatory_values.extend(extra_mandatory_values)
42+
missing_values = []
43+
44+
for value in mandatory_values:
45+
if value not in workflow_commons:
46+
missing_values.append(value)
47+
48+
if missing_values:
49+
raise WorkflowProcessingException(
50+
f"The following values are missing in workflow_commons.json: {missing_values}"
51+
)
52+
53+
commons_defaults = {
54+
"output_data_file_mask": "",
55+
"run_metadata": {},
56+
"log_target_path": "",
57+
"output_mode": "",
58+
"production_output_data": [],
59+
"CPUe": 0,
60+
"max_number_of_events": "0",
61+
"output_SEs": {},
62+
"output_data_type": None,
63+
"application_log": "",
64+
"application_type": None,
65+
"options_file": None,
66+
"options_line": None,
67+
"extra_packages": "",
68+
"multi_core": False,
69+
"max_number_of_processors": None,
70+
"system_config": None,
71+
"mcTCK": None,
72+
"condDB_tag": None,
73+
"DQ_tag": None,
74+
"step_status": S_OK(),
75+
"config_name": None,
76+
"config_version": None,
77+
}
78+
79+
for k, v in extra_default_values.items():
80+
if k not in commons_defaults:
81+
commons_defaults[k] = v
82+
83+
for k, v in commons_defaults.items():
84+
if k not in workflow_commons:
85+
workflow_commons[k] = v
86+
87+
cleaned_application_name = workflow_commons["application_name"].replace("/", "")
88+
workflow_commons["cleaned_application_name"] = cleaned_application_name
89+
90+
workflow_commons["site_name"] = siteName()
91+
92+
return workflow_commons

0 commit comments

Comments
 (0)