Coverage for python/lsst/ctrl/bps/panda/panda_service.py: 9%
230 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-10-09 03:49 -0700
1# This file is part of ctrl_bps_panda.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (https://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# This software is dual licensed under the GNU General Public License and also
10# under a 3-clause BSD license. Recipients may choose which of these licenses
11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt,
12# respectively. If you choose the GPL option then the following text applies
13# (but note that there is still no warranty even if you opt for BSD instead):
14#
15# This program is free software: you can redistribute it and/or modify
16# it under the terms of the GNU General Public License as published by
17# the Free Software Foundation, either version 3 of the License, or
18# (at your option) any later version.
19#
20# This program is distributed in the hope that it will be useful,
21# but WITHOUT ANY WARRANTY; without even the implied warranty of
22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
23# GNU General Public License for more details.
24#
25# You should have received a copy of the GNU General Public License
26# along with this program. If not, see <https://www.gnu.org/licenses/>.
27"""Interface between generic workflow to PanDA/iDDS workflow system."""
29__all__ = ["PanDAService", "PandaBpsWmsWorkflow"]
32import json
33import logging
34import os
35import pickle
36import re
38from idds.workflowv2.workflow import Workflow as IDDS_client_workflow
40from lsst.ctrl.bps import (
41 DEFAULT_MEM_FMT,
42 DEFAULT_MEM_UNIT,
43 BaseWmsService,
44 BaseWmsWorkflow,
45 WmsRunReport,
46 WmsStates,
47)
48from lsst.ctrl.bps.panda.constants import PANDA_DEFAULT_MAX_COPY_WORKERS
49from lsst.ctrl.bps.panda.utils import (
50 add_final_idds_work,
51 add_idds_work,
52 copy_files_for_distribution,
53 create_idds_build_workflow,
54 get_idds_client,
55 get_idds_result,
56)
57from lsst.resources import ResourcePath
58from lsst.utils.timer import time_this
60_LOG = logging.getLogger(__name__)
63class PanDAService(BaseWmsService):
64 """PanDA version of WMS service."""
66 def prepare(self, config, generic_workflow, out_prefix=None):
67 # Docstring inherited from BaseWmsService.prepare.
68 _LOG.debug("out_prefix = '%s'", out_prefix)
70 _LOG.info("Starting PanDA prepare stage (creating specific implementation of workflow)")
72 with time_this(
73 log=_LOG,
74 level=logging.INFO,
75 prefix=None,
76 msg="PanDA prepare stage completed",
77 mem_usage=True,
78 mem_unit=DEFAULT_MEM_UNIT,
79 mem_fmt=DEFAULT_MEM_FMT,
80 ):
81 workflow = PandaBpsWmsWorkflow.from_generic_workflow(
82 config, generic_workflow, out_prefix, f"{self.__class__.__module__}.{self.__class__.__name__}"
83 )
84 workflow.write(out_prefix)
85 return workflow
87 def submit(self, workflow, **kwargs):
88 config = kwargs["config"] if "config" in kwargs else None
89 remote_build = kwargs["remote_build"] if "remote_build" in kwargs else None
91 if config and remote_build:
92 _LOG.info("remote build")
94 idds_build_workflow = create_idds_build_workflow(**kwargs)
95 idds_client = get_idds_client(self.config)
96 ret = idds_client.submit_build(idds_build_workflow, username=None, use_dataset_name=False)
97 _LOG.debug("iDDS client manager submit returned = %s", ret)
99 # Check submission success
100 status, result, error = get_idds_result(ret)
101 if status:
102 request_id = int(result)
103 else:
104 raise RuntimeError(f"Error submitting to PanDA service: {error}")
106 _LOG.info("Submitted into iDDs with request id=%s", request_id)
107 idds_build_workflow.run_id = request_id
108 return idds_build_workflow
110 else:
111 _, max_copy_workers = self.config.search(
112 "maxCopyWorkers", opt={"default": PANDA_DEFAULT_MAX_COPY_WORKERS}
113 )
114 file_distribution_uri = self.config["fileDistributionEndPoint"]
115 lsst_temp = "LSST_RUN_TEMP_SPACE"
116 if lsst_temp in file_distribution_uri and lsst_temp not in os.environ:
117 file_distribution_uri = self.config["fileDistributionEndPointDefault"]
118 protocol_pattern = re.compile(r"^[a-zA-Z][a-zA-Z\d+\-.]*://")
119 if not protocol_pattern.match(file_distribution_uri):
120 file_distribution_uri = "file://" + file_distribution_uri
122 submit_cmd = workflow.run_attrs.get("bps_iscustom", False)
123 if not submit_cmd:
124 copy_files_for_distribution(
125 workflow.files_to_pre_stage,
126 ResourcePath(file_distribution_uri, forceDirectory=True),
127 max_copy_workers,
128 )
130 idds_client = get_idds_client(self.config)
131 ret = idds_client.submit(workflow.idds_client_workflow, username=None, use_dataset_name=False)
132 _LOG.debug("iDDS client manager submit returned = %s", ret)
134 # Check submission success
135 status, result, error = get_idds_result(ret)
136 if status:
137 request_id = int(result)
138 else:
139 raise RuntimeError(f"Error submitting to PanDA service: {error}")
141 _LOG.info("Submitted into iDDs with request id=%s", request_id)
142 workflow.run_id = request_id
144 def restart(self, wms_workflow_id):
145 # Docstring inherited from BaseWmsService.restart.
146 idds_client = get_idds_client(self.config)
147 ret = idds_client.retry(request_id=wms_workflow_id)
148 _LOG.debug("Restart PanDA workflow returned = %s", ret)
150 status, result, error = get_idds_result(ret)
151 if status:
152 _LOG.info("Restarting PanDA workflow %s", result)
153 return wms_workflow_id, None, json.dumps(result)
155 return None, None, f"Error retry PanDA workflow: {error}"
157 def report(
158 self,
159 wms_workflow_id=None,
160 user=None,
161 hist=0,
162 pass_thru=None,
163 is_global=False,
164 return_exit_codes=False,
165 ):
166 # Docstring inherited from BaseWmsService.report.
167 message = ""
168 run_reports = []
170 if not wms_workflow_id:
171 message = "Run summary not implemented yet, use 'bps report --id <workflow_id>' instead"
172 return run_reports, message
174 idds_client = get_idds_client(self.config)
175 ret = idds_client.get_requests(request_id=wms_workflow_id, with_detail=True)
176 _LOG.debug("PanDA get workflow status returned = %s", str(ret))
178 request_status = ret[0]
179 if request_status != 0:
180 raise RuntimeError(f"Error to get workflow status: {ret} for id: {wms_workflow_id}")
182 tasks = ret[1][1]
183 if not tasks:
184 message = f"No records found for workflow id '{wms_workflow_id}'. Hint: double check the id"
185 else:
186 head = tasks[0]
187 wms_report = WmsRunReport(
188 wms_id=str(head["request_id"]),
189 operator=head["username"],
190 project="",
191 campaign="",
192 payload="",
193 run=head["name"],
194 state=WmsStates.UNKNOWN,
195 total_number_jobs=0,
196 job_state_counts=dict.fromkeys(WmsStates, 0),
197 job_summary={},
198 run_summary="",
199 exit_code_summary=[],
200 )
202 # The status of a task is taken from the first item of state_map.
203 # The workflow is in status WmsStates.FAILED when:
204 # All tasks have failed.
205 # SubFinished tasks has jobs in
206 # output_processed_files: Finished
207 # output_failed_files: Failed
208 # output_missing_files: Missing
209 state_map = {
210 "Finished": [WmsStates.SUCCEEDED],
211 "SubFinished": [
212 WmsStates.SUCCEEDED,
213 WmsStates.FAILED,
214 WmsStates.PRUNED,
215 ],
216 "Transforming": [
217 WmsStates.RUNNING,
218 WmsStates.SUCCEEDED,
219 WmsStates.FAILED,
220 WmsStates.UNREADY,
221 WmsStates.PRUNED,
222 ],
223 "Failed": [WmsStates.FAILED, WmsStates.PRUNED],
224 }
226 file_map = {
227 WmsStates.SUCCEEDED: "output_processed_files",
228 WmsStates.RUNNING: "output_processing_files",
229 WmsStates.FAILED: "output_failed_files",
230 WmsStates.UNREADY: "input_new_files",
231 WmsStates.PRUNED: "output_missing_files",
232 }
234 workflow_status = head["status"]["attributes"]["_name_"]
235 if workflow_status in ["Finished", "SubFinished"]:
236 wms_report.state = WmsStates.SUCCEEDED
237 elif workflow_status in ["Failed", "Expired"]:
238 wms_report.state = WmsStates.FAILED
239 elif workflow_status in ["Cancelled"]:
240 wms_report.state = WmsStates.DELETED
241 elif workflow_status in ["Suspended"]:
242 wms_report.state = WmsStates.HELD
243 else:
244 wms_report.state = WmsStates.RUNNING
246 try:
247 tasks.sort(key=lambda x: x["transform_workload_id"])
248 except Exception:
249 tasks.sort(key=lambda x: x["transform_id"])
251 exit_codes_all = {}
252 # Loop over all tasks data returned by idds_client
253 for task in tasks:
254 if task["transform_id"] is None:
255 # Not created task (It happens because of an outer join
256 # between requests table and transforms table).
257 continue
259 exit_codes = []
260 totaljobs = task["output_total_files"]
261 wms_report.total_number_jobs += totaljobs
262 tasklabel = task["transform_name"]
263 tasklabel = re.sub(wms_report.run + "_", "", tasklabel)
264 status = task["transform_status"]["attributes"]["_name_"]
265 taskstatus = {}
266 # if the state is failed, gather exit code information
267 if status in ["SubFinished", "Failed"]:
268 transform_workload_id = task["transform_workload_id"]
269 if not (task["transform_name"] and task["transform_name"].startswith("build_task")):
270 new_ret = idds_client.get_contents_output_ext(
271 request_id=wms_workflow_id, workload_id=transform_workload_id
272 )
273 _LOG.debug(
274 "PanDA get task %s detail returned = %s", transform_workload_id, str(new_ret)
275 )
277 request_status = new_ret[0]
278 if request_status != 0:
279 raise RuntimeError(
280 f"Error to get workflow status: {new_ret} for id: {wms_workflow_id}"
281 )
282 # task_info is a dictionary of len 1 that contains
283 # a list of dicts containing panda job info
284 task_info = new_ret[1][1]
286 if len(task_info) == 1:
287 wmskey = list(task_info.keys())[0]
288 wmsjobs = task_info[wmskey]
289 else:
290 err_msg = "Unexpected job return from PanDA: "
291 err_msg += f"{task_info} for id: {transform_workload_id}"
292 raise RuntimeError(err_msg)
293 exit_codes = [
294 wmsjob["trans_exit_code"]
295 for wmsjob in wmsjobs
296 if wmsjob["trans_exit_code"] is not None and int(wmsjob["trans_exit_code"]) != 0
297 ]
298 exit_codes_all[tasklabel] = exit_codes
299 # Fill number of jobs in all WmsStates
300 for state in WmsStates:
301 njobs = 0
302 # Each WmsState have many iDDS status mapped to it.
303 if status in state_map:
304 for mappedstate in state_map[status]:
305 if state in file_map and mappedstate == state:
306 if task[file_map[mappedstate]] is not None:
307 njobs = task[file_map[mappedstate]]
308 if state == WmsStates.RUNNING:
309 njobs += task["output_new_files"] - task["input_new_files"]
310 break
311 wms_report.job_state_counts[state] += njobs
312 taskstatus[state] = njobs
313 wms_report.job_summary[tasklabel] = taskstatus
315 # To fill the EXPECTED column
316 if wms_report.run_summary:
317 wms_report.run_summary += ";"
318 wms_report.run_summary += f"{tasklabel}:{totaljobs}"
320 wms_report.exit_code_summary = exit_codes_all
321 run_reports.append(wms_report)
323 return run_reports, message
325 def list_submitted_jobs(self, wms_id=None, user=None, require_bps=True, pass_thru=None, is_global=False):
326 # Docstring inherited from BaseWmsService.list_submitted_jobs.
327 if wms_id is None and user is not None:
328 raise RuntimeError(
329 "Error to get workflow status report: wms_id is required"
330 " and filtering workflows with 'user' is not supported."
331 )
333 idds_client = get_idds_client(self.config)
334 ret = idds_client.get_requests(request_id=wms_id)
335 _LOG.debug("PanDA get workflows returned = %s", ret)
337 status, result, error = get_idds_result(ret)
338 if status:
339 req_ids = [req["request_id"] for req in result]
340 return req_ids
342 raise RuntimeError(f"Error list PanDA workflow requests: {error}")
344 def cancel(self, wms_id, pass_thru=None):
345 # Docstring inherited from BaseWmsService.cancel.
346 idds_client = get_idds_client(self.config)
347 ret = idds_client.abort(request_id=wms_id)
348 _LOG.debug("Abort PanDA workflow returned = %s", ret)
350 status, result, error = get_idds_result(ret)
351 if status:
352 _LOG.info("Aborting PanDA workflow %s", result)
353 return True, json.dumps(result)
355 return False, f"Error abort PanDA workflow: {error}"
357 def ping(self, pass_thru=None):
358 # Docstring inherited from BaseWmsService.ping.
359 idds_client = get_idds_client(self.config)
360 ret = idds_client.ping()
361 _LOG.debug("Ping PanDA service returned = %s", ret)
363 status, result, error = get_idds_result(ret)
364 if status:
365 if "Status" in result and result["Status"] == "OK":
366 return 0, None
368 return -1, f"Error ping PanDA service: {result}"
370 return -1, f"Error ping PanDA service: {error}"
372 def run_submission_checks(self):
373 # Docstring inherited from BaseWmsService.run_submission_checks.
374 for key in ["PANDA_URL"]:
375 if key not in os.environ:
376 raise OSError(f"Missing environment variable {key}")
378 status, message = self.ping()
379 if status != 0:
380 raise RuntimeError(message)
382 def get_status(
383 self,
384 wms_workflow_id=None,
385 hist=0,
386 is_global=False,
387 ):
388 # Docstring inherited from BaseWmsService.get_status.
390 idds_client = get_idds_client(self.config)
391 ret = idds_client.get_requests(request_id=wms_workflow_id, with_detail=False)
392 _LOG.debug("PanDA get workflow status returned = %s", str(ret))
394 request_status = ret[0]
395 if request_status != 0:
396 state = WmsStates.UNKNOWN
397 message = f"Error getting workflow status for id {wms_workflow_id}: ret = {ret}"
398 else:
399 tasks = ret[1][1]
400 if not tasks:
401 state = WmsStates.UNKNOWN
402 message = f"No records found for workflow id '{wms_workflow_id}'. Hint: double check the id"
403 elif not isinstance(tasks[0], dict):
404 state = WmsStates.UNKNOWN
405 message = f"Error getting workflow status for id {wms_workflow_id}: ret = {ret}"
406 else:
407 message = ""
408 head = tasks[0]
409 workflow_status = head["status"]["attributes"]["_name_"]
410 if workflow_status in ["Finished"]:
411 state = WmsStates.SUCCEEDED
412 elif workflow_status in ["Failed", "Expired", "SubFinished"]:
413 state = WmsStates.FAILED
414 elif workflow_status in ["Cancelled"]:
415 state = WmsStates.DELETED
416 elif workflow_status in ["Suspended"]:
417 state = WmsStates.HELD
418 else:
419 state = WmsStates.RUNNING
421 return state, message
424class PandaBpsWmsWorkflow(BaseWmsWorkflow):
425 """A single Panda based workflow.
427 Parameters
428 ----------
429 name : `str`
430 Unique name for Workflow.
431 config : `lsst.ctrl.bps.BpsConfig`
432 BPS configuration that includes necessary submit/runtime information.
433 """
435 def __init__(self, name, config=None):
436 super().__init__(name, config)
437 self.files_to_pre_stage = {} # src, dest
438 self.idds_client_workflow = IDDS_client_workflow(name=name)
439 self.run_attrs = {}
441 @classmethod
442 def from_generic_workflow(cls, config, generic_workflow, out_prefix, service_class):
443 # Docstring inherited from BaseWmsWorkflow.from_generic_workflow.
444 wms_workflow = cls(generic_workflow.name, config)
446 if generic_workflow.run_attrs:
447 wms_workflow.run_attrs.update(generic_workflow.run_attrs)
449 files, dag_sink_work, task_count = add_idds_work(
450 config, generic_workflow, wms_workflow.idds_client_workflow
451 )
452 wms_workflow.files_to_pre_stage.update(files)
454 files = add_final_idds_work(
455 config, generic_workflow, wms_workflow.idds_client_workflow, dag_sink_work, task_count + 1, 1
456 )
457 wms_workflow.files_to_pre_stage.update(files)
459 return wms_workflow
461 def write(self, out_prefix):
462 # Docstring inherited from BaseWmsWorkflow.write.
463 with open(os.path.join(out_prefix, "panda_workflow.pickle"), "wb") as fh:
464 pickle.dump(self, fh)