Coverage for python / lsst / ctrl / bps / panda / utils.py: 5%

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 

28"""Utilities for bps PanDA plugin.""" 

29 

30__all__ = [ 

31 "add_decoder_prefix", 

32 "aggregate_by_basename", 

33 "convert_exec_string_to_hex", 

34 "copy_files_for_distribution", 

35 "extract_taskname", 

36 "get_idds_client", 

37 "get_idds_result", 

38 "idds_call_with_check", 

39] 

40 

41import binascii 

42import concurrent.futures 

43import json 

44import logging 

45import os 

46import random 

47import re 

48import tarfile 

49import time 

50import uuid 

51 

52import idds.common.utils as idds_utils 

53import pandaclient.idds_api 

54from idds.doma.workflowv2.domapandawork import DomaPanDAWork 

55from idds.workflowv2.workflow import AndCondition 

56from idds.workflowv2.workflow import Workflow as IDDS_client_workflow 

57 

58from lsst.ctrl.bps import BpsConfig, GenericWorkflow, GenericWorkflowJob, WmsStates 

59from lsst.ctrl.bps.panda.cmd_line_embedder import CommandLineEmbedder 

60from lsst.ctrl.bps.panda.constants import ( 

61 PANDA_DEFAULT_CLOUD, 

62 PANDA_DEFAULT_CORE_COUNT, 

63 PANDA_DEFAULT_MAX_ATTEMPTS, 

64 PANDA_DEFAULT_MAX_JOBS_PER_TASK, 

65 PANDA_DEFAULT_MAX_PAYLOADS_PER_PANDA_JOB, 

66 PANDA_DEFAULT_MAX_WALLTIME, 

67 PANDA_DEFAULT_NAME_LENGTH, 

68 PANDA_DEFAULT_ORDER_ID_MAP_FILE, 

69 PANDA_DEFAULT_PRIORITY, 

70 PANDA_DEFAULT_PROCESSING_TYPE, 

71 PANDA_DEFAULT_PROD_SOURCE_LABEL, 

72 PANDA_DEFAULT_RSS, 

73 PANDA_DEFAULT_RSS_MAX, 

74 PANDA_DEFAULT_TASK_TYPE, 

75 PANDA_DEFAULT_VO, 

76) 

77from lsst.resources import ResourcePath 

78 

79_LOG = logging.getLogger(__name__) 

80 

81 

82def extract_taskname(s: str) -> str: 

83 """Extract the task name from a string that follows a pattern 

84 CampaignName_timestamp_TaskNumber_TaskLabel_ChunkNumber. 

85 

86 Parameters 

87 ---------- 

88 s : `str` 

89 The input string from which to extract the task name. 

90 

91 Returns 

92 ------- 

93 taskname : `str` 

94 The extracted task name as per the rules above. 

95 """ 

96 # remove surrounding quotes/spaces if present 

97 s = s.strip().strip("'\"") 

98 

99 # find all occurrences of underscore + digits + underscore, 

100 # take the last one 

101 matches = re.findall(r"_(\d+)_", s) 

102 if matches: 

103 last_number = matches[-1] 

104 last_pos = s.rfind(f"_{last_number}_") + len(f"_{last_number}_") 

105 taskname = s[last_pos:] 

106 return taskname 

107 

108 # fallback: if no such pattern, return everything 

109 taskname = s 

110 return taskname 

111 

112 

113def aggregate_by_basename(job_summary, exit_code_summary, run_summary): 

114 """Aggregate job exit code and run summaries by 

115 their base label (basename). 

116 

117 Parameters 

118 ---------- 

119 job_summary : `dict` [`str`, `dict` [`str`, `int`]] 

120 A mapping of job labels to state-count mappings. 

121 exit_code_summary : `dict` [`str`, `list` [`int`]] 

122 A mapping of job labels to lists of exit codes. 

123 run_summary : `str` 

124 A semicolon-separated string of job summaries 

125 where each entry has the format "<label>:<count>". 

126 

127 Returns 

128 ------- 

129 aggregated_jobs : `dict` [`str`, `dict` [`str`, `int`]] 

130 A dictionary mapping each base label to the summed job state counts 

131 across all matching labels. 

132 aggregated_exits : `dict` [`str`, `list` [`int`]] 

133 A dictionary mapping each base label to a combined list of exit codes 

134 from all matching labels. 

135 aggregated_run : `str` 

136 A semicolon-separated string with aggregated job counts by base label. 

137 """ 

138 

139 def base_label(label): 

140 return re.sub(r"_\d+$", "", label) 

141 

142 aggregated_jobs = {} 

143 aggregated_exits = {} 

144 

145 for label, states in job_summary.items(): 

146 base = base_label(label) 

147 if base not in aggregated_jobs: 

148 aggregated_jobs[base] = dict.fromkeys(WmsStates, 0) 

149 for state, count in states.items(): 

150 aggregated_jobs[base][state] += count 

151 

152 for label, codes in exit_code_summary.items(): 

153 base = base_label(label) 

154 aggregated_exits.setdefault(base, []).extend(codes) 

155 

156 aggregated = {} 

157 for entry in run_summary.split(";"): 

158 entry = entry.strip() 

159 if not entry: 

160 continue 

161 try: 

162 label, num = entry.split(":") 

163 num = int(num) 

164 except ValueError: 

165 continue 

166 

167 base = base_label(label) 

168 aggregated[base] = aggregated.get(base, 0) + num 

169 

170 aggregated_run = ";".join(f"{base}:{count}" for base, count in aggregated.items()) 

171 return aggregated_jobs, aggregated_exits, aggregated_run 

172 

173 

174def copy_files_for_distribution(files_to_stage, file_distribution_uri, max_copy_workers): 

175 """Brings locally generated files into Cloud for further 

176 utilization them on the edge nodes. 

177 

178 Parameters 

179 ---------- 

180 files_to_stage : `dict` [`str`, `str`] 

181 Files which need to be copied to a workflow staging area. 

182 file_distribution_uri : `ResourcePath` 

183 Path on the edge node accessed storage, 

184 including access protocol, bucket name to place files. 

185 max_copy_workers : `int` 

186 Maximum number of workers for copying files. 

187 

188 Raises 

189 ------ 

190 RuntimeError 

191 Raised when error copying files to the distribution point. 

192 """ 

193 files_to_copy = {} 

194 

195 # In case there are folders we iterate over its content 

196 for local_pfn in files_to_stage.values(): 

197 folder_name = os.path.basename(os.path.normpath(local_pfn)) 

198 if os.path.isdir(local_pfn): 

199 folder_uri = file_distribution_uri.join(folder_name, forceDirectory=True) 

200 files_in_folder = ResourcePath.findFileResources([local_pfn]) 

201 for file in files_in_folder: 

202 file_name = file.basename() 

203 files_to_copy[file] = folder_uri.join(file_name, forceDirectory=False) 

204 else: 

205 folder_uri = file_distribution_uri.join(folder_name, forceDirectory=False) 

206 files_to_copy[ResourcePath(local_pfn, forceDirectory=False)] = folder_uri 

207 

208 copy_executor = concurrent.futures.ThreadPoolExecutor(max_workers=max_copy_workers) 

209 future_file_copy = [] 

210 for src, trgt in files_to_copy.items(): 

211 _LOG.debug("Staging %s to %s", src, trgt) 

212 # S3 clients explicitly instantiate here to overpass this 

213 # https://stackoverflow.com/questions/52820971/is-boto3-client-thread-safe 

214 trgt.exists() 

215 future_file_copy.append(copy_executor.submit(trgt.transfer_from, src, transfer="copy")) 

216 

217 for future in concurrent.futures.as_completed(future_file_copy): 

218 if future.result() is not None: 

219 raise RuntimeError("Error of placing files to the distribution point") 

220 

221 

222def get_idds_client(config): 

223 """Get the idds client. 

224 

225 Parameters 

226 ---------- 

227 config : `lsst.ctrl.bps.BpsConfig` 

228 BPS configuration. 

229 

230 Returns 

231 ------- 

232 idds_client: `idds.client.clientmanager.ClientManager` 

233 The iDDS ClientManager object. 

234 """ 

235 idds_server = None 

236 if isinstance(config, BpsConfig): 

237 _, idds_server = config.search("iddsServer", opt={"default": None}) 

238 elif isinstance(config, dict) and "iddsServer" in config: 

239 idds_server = config["iddsServer"] 

240 # if idds_server is None, a default value on the panda relay service 

241 # will be used 

242 idds_client = pandaclient.idds_api.get_api( 

243 idds_utils.json_dumps, idds_host=idds_server, compress=True, manager=True 

244 ) 

245 return idds_client 

246 

247 

248def get_idds_result(ret): 

249 """Parse the results returned from iDDS. 

250 

251 Parameters 

252 ---------- 

253 ret : `tuple` [`int`, `tuple` [`bool`, payload ]] 

254 The first part ``ret[0]`` is the status of PanDA relay service. 

255 The part of ``ret[1][0]`` is the status of iDDS service. 

256 The part of ``ret[1][1]`` is the returned payload. 

257 If ``ret[1][0]`` is `False`, ``ret[1][1]`` can be error messages. 

258 

259 Returns 

260 ------- 

261 status: `bool` 

262 The status of iDDS calls. 

263 result: `int` or `list` or `dict` or `None` 

264 The result returned from iDDS. `None` if error state. 

265 error: `str` or `None` 

266 Error messages. `None` if no error state. 

267 """ 

268 # https://panda-wms.readthedocs.io/en/latest/client/rest_idds.html 

269 if not isinstance(ret, list | tuple) or ret[0] != 0: 

270 # Something wrong with the PanDA relay service. 

271 # The call may not be delivered to iDDS. 

272 status = False 

273 result = None 

274 error = f"PanDA relay service returns errors: {ret}" 

275 else: 

276 if ret[1][0]: 

277 status = True 

278 result = ret[1][1] 

279 error = None 

280 if isinstance(result, str) and "Authentication no permission" in result: 

281 status = False 

282 result = None 

283 error = result 

284 else: 

285 # iDDS returns errors 

286 status = False 

287 result = None 

288 error = f"iDDS returns errors: {ret[1][1]}" 

289 return status, result, error 

290 

291 

292def idds_call_with_check(func, *, func_name: str, request_id: int, **kwargs): 

293 """Call an iDDS client function, log, and check the return code. 

294 

295 Parameters 

296 ---------- 

297 func : callable 

298 The iDDS client function to call. 

299 func_name : `str` 

300 Name used for logging. 

301 request_id : `int` 

302 The request or workflow ID. 

303 **kwargs 

304 Additional keyword arguments passed to the function. 

305 

306 Returns 

307 ------- 

308 ret : `Any` 

309 The return value from the iDDS client function. 

310 """ 

311 call_kwargs = dict(kwargs) 

312 if request_id is not None: 

313 call_kwargs["request_id"] = request_id 

314 

315 ret = func(**call_kwargs) 

316 

317 _LOG.debug("PanDA %s returned = %s", func_name, str(ret)) 

318 

319 request_status = ret[0] 

320 if request_status != 0: 

321 raise RuntimeError(f"Error calling {func_name}: {ret} for id: {request_id}") 

322 

323 return ret 

324 

325 

326def _make_pseudo_filename(config, gwjob): 

327 """Make the job pseudo filename. 

328 

329 Parameters 

330 ---------- 

331 config : `lsst.ctrl.bps.BpsConfig` 

332 BPS configuration. 

333 gwjob : `lsst.ctrl.bps.GenericWorkflowJob` 

334 Job for which to create the pseudo filename. 

335 

336 Returns 

337 ------- 

338 pseudo_filename : `str` 

339 The pseudo filename for the given job. 

340 """ 

341 cmd_line_embedder = CommandLineEmbedder(config) 

342 _, pseudo_filename = cmd_line_embedder.substitute_command_line( 

343 gwjob.executable.src_uri + " " + gwjob.arguments, gwjob.cmdvals, gwjob.name, [] 

344 ) 

345 return pseudo_filename 

346 

347 

348def _make_doma_work( 

349 config, 

350 generic_workflow, 

351 gwjob, 

352 task_count, 

353 task_chunk, 

354 enable_event_service=False, 

355 enable_job_name_map=False, 

356 order_id_map_files=None, 

357 es_label=None, 

358 max_payloads_per_panda_job=PANDA_DEFAULT_MAX_PAYLOADS_PER_PANDA_JOB, 

359 max_wms_job_wall_time=None, 

360 remote_filename=None, 

361 qnode_map_filename=None, 

362): 

363 """Make the DOMA Work object for a PanDA task. 

364 

365 Parameters 

366 ---------- 

367 config : `lsst.ctrl.bps.BpsConfig` 

368 BPS configuration. 

369 gwjob : `lsst.ctrl.bps.GenericWorkflowJob` 

370 Job representing the jobs for the PanDA task. 

371 task_count : `int` 

372 Count of PanDA tasks used when making unique names. 

373 task_chunk : `int` 

374 Count of chunk of a PanDA tasks used when making unique names. 

375 

376 Returns 

377 ------- 

378 work : `idds.doma.workflowv2.domapandawork.DomaPanDAWork` 

379 The client representation of a PanDA task. 

380 local_pfns : `dict` [`str`, `str`] 

381 Files which need to be copied to a workflow staging area. 

382 """ 

383 if order_id_map_files is None: 

384 order_id_map_files = {} 

385 _LOG.debug("Using gwjob %s to create new PanDA task (gwjob=%s)", gwjob.name, gwjob) 

386 cvals = {"curr_cluster": gwjob.label} 

387 _, site = config.search("computeSite", opt={"curvals": cvals, "required": True}) 

388 cvals["curr_site"] = site 

389 cvals["curr_pipetask"] = gwjob.label 

390 _, processing_type = config.search( 

391 "processingType", opt={"curvals": cvals, "default": PANDA_DEFAULT_PROCESSING_TYPE} 

392 ) 

393 if gwjob.label in ["finalJob", "customJob"]: 

394 _, nonpipetask = config.search(gwjob.label) 

395 default_type = "Rubin_Merge" 

396 if gwjob.label == "customJob": 

397 default_type = PANDA_DEFAULT_PROCESSING_TYPE 

398 processing_type = nonpipetask["processingType"] if nonpipetask["processingType"] else default_type 

399 _, task_type = config.search("taskType", opt={"curvals": cvals, "default": PANDA_DEFAULT_TASK_TYPE}) 

400 _, prod_source_label = config.search( 

401 "prodSourceLabel", opt={"curvals": cvals, "default": PANDA_DEFAULT_PROD_SOURCE_LABEL} 

402 ) 

403 _, vo = config.search("vo", opt={"curvals": cvals, "default": PANDA_DEFAULT_VO}) 

404 

405 _, file_distribution_end_point = config.search( 

406 "fileDistributionEndPoint", opt={"curvals": cvals, "default": None} 

407 ) 

408 

409 _, file_distribution_end_point_default = config.search( 

410 "fileDistributionEndPointDefault", opt={"curvals": cvals, "default": None} 

411 ) 

412 

413 task_rss = gwjob.request_memory if gwjob.request_memory else PANDA_DEFAULT_RSS 

414 task_rss_retry_step = task_rss * gwjob.memory_multiplier if gwjob.memory_multiplier else 0 

415 task_rss_retry_offset = 0 if task_rss_retry_step else task_rss 

416 

417 # Assume input files are same across task 

418 local_pfns = {} 

419 direct_io_files = set() 

420 

421 if gwjob.executable.transfer_executable: 

422 local_pfns["job_executable"] = gwjob.executable.src_uri 

423 job_executable = f"./{os.path.basename(gwjob.executable.src_uri)}" 

424 else: 

425 job_executable = gwjob.executable.src_uri 

426 cmd_line_embedder = CommandLineEmbedder(config) 

427 _LOG.debug( 

428 "job %s inputs = %s, outputs = %s", 

429 gwjob.name, 

430 generic_workflow.get_job_inputs(gwjob.name), 

431 generic_workflow.get_job_outputs(gwjob.name), 

432 ) 

433 

434 job_env = "" 

435 if gwjob.environment: 

436 for key, value in gwjob.environment.items(): 

437 try: 

438 sub_value = value.format_map(gwjob.cmdvals) 

439 except (KeyError, TypeError) as exc: 

440 _LOG.error("Could not replace command variables: replacement for %s not provided", str(exc)) 

441 raise 

442 job_env += f"export {key}={sub_value}; " 

443 

444 cmd_line, _ = cmd_line_embedder.substitute_command_line( 

445 job_env + job_executable + " " + gwjob.arguments, 

446 gwjob.cmdvals, 

447 gwjob.name, 

448 generic_workflow.get_job_inputs(gwjob.name) + generic_workflow.get_job_outputs(gwjob.name), 

449 ) 

450 

451 my_log = f"enable_event_service {enable_event_service} for {gwjob.label}" 

452 _LOG.info(my_log) 

453 if enable_event_service: 

454 if gwjob.request_walltime and max_wms_job_wall_time: 

455 my_log = ( 

456 f"requestWalltime({gwjob.request_walltime}) " 

457 f"and maxWmsJobWalltime({max_wms_job_wall_time}) are set, " 

458 "max_payloads_per_panda_job is int(max_wms_job_wall_time / gwjob.request_walltime), " 

459 "ignore maxPayloadsPerPandaJob." 

460 ) 

461 _LOG.info(my_log) 

462 max_payloads_per_panda_job = int(max_wms_job_wall_time / gwjob.request_walltime) 

463 if max_payloads_per_panda_job < 2: 

464 my_log = ( 

465 f"max_payloads_per_panda_job ({max_payloads_per_panda_job}) is too small, " 

466 "disable EventService" 

467 ) 

468 _LOG.info(my_log) 

469 enable_event_service = False 

470 

471 maxwalltime = gwjob.request_walltime if gwjob.request_walltime else PANDA_DEFAULT_MAX_WALLTIME 

472 if enable_event_service: 

473 if gwjob.request_walltime and max_payloads_per_panda_job: 

474 maxwalltime = gwjob.request_walltime * max_payloads_per_panda_job 

475 elif max_wms_job_wall_time: 

476 maxwalltime = max_wms_job_wall_time 

477 

478 if enable_event_service or enable_job_name_map: 

479 for es_name in order_id_map_files: 

480 local_pfns[es_name] = order_id_map_files[es_name] 

481 

482 for gwfile in generic_workflow.get_job_inputs(gwjob.name, transfer_only=True): 

483 local_pfns[gwfile.name] = gwfile.src_uri 

484 if os.path.isdir(gwfile.src_uri): 

485 # this is needed to make isdir function working 

486 # properly in ButlerURL instance on the edge node 

487 local_pfns[gwfile.name] += "/" 

488 

489 if gwfile.job_access_remote: 

490 direct_io_files.add(gwfile.name) 

491 

492 if qnode_map_filename: 

493 local_pfns.update(qnode_map_filename) 

494 

495 submit_cmd = generic_workflow.run_attrs.get("bps_iscustom", False) 

496 

497 if not direct_io_files: 

498 if submit_cmd: 

499 direct_io_files.add(remote_filename) 

500 else: 

501 direct_io_files.add("cmdlineplaceholder") 

502 

503 lsst_temp = "LSST_RUN_TEMP_SPACE" 

504 if lsst_temp in file_distribution_end_point and lsst_temp not in os.environ: 

505 file_distribution_end_point = file_distribution_end_point_default 

506 if submit_cmd and not file_distribution_end_point: 

507 file_distribution_end_point = "FileDistribution" 

508 

509 executable = add_decoder_prefix( 

510 config, cmd_line, file_distribution_end_point, (local_pfns, direct_io_files) 

511 ) 

512 work = DomaPanDAWork( 

513 executable=executable, 

514 primary_input_collection={ 

515 "scope": "pseudo_dataset", 

516 "name": f"pseudo_input_collection#{task_count}", 

517 }, 

518 output_collections=[{"scope": "pseudo_dataset", "name": f"pseudo_output_collection#{task_count}"}], 

519 log_collections=[], 

520 dependency_map=[], 

521 task_name=f"{generic_workflow.name}_{task_count:02d}_{gwjob.label}_{task_chunk:02d}", 

522 task_queue=gwjob.queue, 

523 task_log={ 

524 "destination": "local", 

525 "value": "log.tgz", 

526 "dataset": "PandaJob_#{pandaid}/", 

527 "token": "local", 

528 "param_type": "log", 

529 "type": "template", 

530 }, 

531 encode_command_line=True, 

532 task_rss=task_rss, 

533 task_rss_retry_offset=task_rss_retry_offset, 

534 task_rss_retry_step=task_rss_retry_step, 

535 task_rss_max=gwjob.request_memory_max if gwjob.request_memory_max else PANDA_DEFAULT_RSS_MAX, 

536 task_cloud=gwjob.compute_cloud if gwjob.compute_cloud else PANDA_DEFAULT_CLOUD, 

537 task_site=site, 

538 task_priority=int(gwjob.priority) if gwjob.priority else PANDA_DEFAULT_PRIORITY, 

539 core_count=gwjob.request_cpus if gwjob.request_cpus else PANDA_DEFAULT_CORE_COUNT, 

540 working_group=gwjob.accounting_group, 

541 processing_type=processing_type, 

542 task_type=task_type, 

543 prodSourceLabel=prod_source_label, 

544 vo=vo, 

545 es=enable_event_service, 

546 es_label=es_label, 

547 max_events_per_job=max_payloads_per_panda_job, 

548 maxattempt=gwjob.number_of_retries if gwjob.number_of_retries else PANDA_DEFAULT_MAX_ATTEMPTS, 

549 maxwalltime=maxwalltime, 

550 ) 

551 return work, local_pfns 

552 

553 

554def add_final_idds_work( 

555 config, generic_workflow, idds_client_workflow, dag_sink_work, task_count, task_chunk 

556): 

557 """Add the special final PanDA task to the client workflow. 

558 

559 Parameters 

560 ---------- 

561 config : `lsst.ctrl.bps.BpsConfig` 

562 BPS configuration. 

563 generic_workflow : `lsst.ctrl.bps.GenericWorkflow` 

564 Generic workflow in which to find the final job. 

565 idds_client_workflow : `idds.workflowv2.workflow.Workflow` 

566 The iDDS client representation of the workflow to which the final task 

567 is added. 

568 dag_sink_work : `list` [`idds.doma.workflowv2.domapandawork.DomaPanDAWork`] 

569 The work nodes in the client workflow which have no successors. 

570 task_count : `int` 

571 Count of PanDA tasks used when making unique names. 

572 task_chunk : `int` 

573 Count of chunk of a PanDA tasks used when making unique names. 

574 

575 Returns 

576 ------- 

577 files : `dict` [`str`, `str`] 

578 Files which need to be copied to a workflow staging area. 

579 

580 Raises 

581 ------ 

582 NotImplementedError 

583 Raised if final job in GenericWorkflow is itself a workflow. 

584 TypeError 

585 Raised if final job in GenericWorkflow is invalid type. 

586 """ 

587 files = {} 

588 

589 # If final job exists in generic workflow, create DAG final job 

590 final = generic_workflow.get_final() 

591 if final: 

592 if isinstance(final, GenericWorkflow): 

593 raise NotImplementedError("PanDA plugin does not support a workflow as the final job") 

594 

595 if not isinstance(final, GenericWorkflowJob): 

596 raise TypeError(f"Invalid type for GenericWorkflow.get_final() results ({type(final)})") 

597 

598 dag_final_work, files = _make_doma_work( 

599 config, 

600 generic_workflow, 

601 final, 

602 task_count, 

603 task_chunk, 

604 ) 

605 pseudo_filename = "pure_pseudoinput+qgraphNodeId:+qgraphId:" 

606 dag_final_work.dependency_map.append( 

607 {"name": pseudo_filename, "submitted": False, "dependencies": []} 

608 ) 

609 idds_client_workflow.add_work(dag_final_work) 

610 conditions = [] 

611 for work in dag_sink_work: 

612 conditions.append(work.is_terminated) 

613 and_cond = AndCondition(conditions=conditions, true_works=[dag_final_work]) 

614 idds_client_workflow.add_condition(and_cond) 

615 else: 

616 _LOG.debug("No final job in GenericWorkflow") 

617 return files 

618 

619 

620def convert_exec_string_to_hex(cmdline): 

621 """Convert the command line into hex representation. 

622 

623 This step is currently involved because large blocks of command lines 

624 including special symbols passed to the pilot/container. To make sure 

625 the 1 to 1 matching and pass by the special symbol stripping 

626 performed by the Pilot we applied the hexing. 

627 

628 Parameters 

629 ---------- 

630 cmdline : `str` 

631 UTF-8 command line string. 

632 

633 Returns 

634 ------- 

635 hex : `str` 

636 Hex representation of string. 

637 """ 

638 return binascii.hexlify(cmdline.encode()).decode("utf-8") 

639 

640 

641def add_decoder_prefix(config, cmd_line, distribution_path, files): 

642 """Compose the command line sent to the pilot from the functional part 

643 (the actual SW running) and the middleware part (containers invocation). 

644 

645 Parameters 

646 ---------- 

647 config : `lsst.ctrl.bps.BpsConfig` 

648 Configuration information. 

649 cmd_line : `str` 

650 UTF-8 based functional part of the command line. 

651 distribution_path : `str` 

652 URI of path where all files are located for distribution. 

653 files : `tuple` [`dict` [`str`, `str`], `list` [`str`]] 

654 File names needed for a task (copied local, direct access). 

655 

656 Returns 

657 ------- 

658 decoder_prefix : `str` 

659 Full command line to be executed on the edge node. 

660 """ 

661 # Manipulate file paths for placement on cmdline 

662 files_plc_hldr = {} 

663 for key, pfn in files[0].items(): 

664 if pfn.endswith("/"): 

665 files_plc_hldr[key] = os.path.basename(pfn[:-1]) 

666 isdir = True 

667 else: 

668 files_plc_hldr[key] = os.path.basename(pfn) 

669 _, extension = os.path.splitext(pfn) 

670 isdir = os.path.isdir(pfn) or (key == "butlerConfig" and extension != "yaml") 

671 if isdir: 

672 # this is needed to make isdir function working 

673 # properly in ButlerURL instance on the egde node 

674 files_plc_hldr[key] += "/" 

675 _LOG.debug("files_plc_hldr[%s] = %s", key, files_plc_hldr[key]) 

676 

677 cmdline_hex = convert_exec_string_to_hex(cmd_line) 

678 _, runner_command = config.search("runnerCommand", opt={"replaceEnvVars": False, "expandEnvVars": False}) 

679 order_id_map_filename = files[0].get("orderIdMapFilename", None) 

680 if order_id_map_filename: 

681 order_id_map_filename = os.path.basename(order_id_map_filename) 

682 order_id_map_filename = os.path.join(distribution_path, order_id_map_filename) 

683 runner_command = runner_command.replace("orderIdMapFilename", order_id_map_filename) 

684 runner_command = runner_command.replace("\n", " ") 

685 decoder_prefix = runner_command.replace( 

686 "_cmd_line_", 

687 str(cmdline_hex) 

688 + " ${IN/L} " 

689 + distribution_path 

690 + " " 

691 + "+".join(f"{k}:{v}" for k, v in files_plc_hldr.items()) 

692 + " " 

693 + "+".join(files[1]), 

694 ) 

695 return decoder_prefix 

696 

697 

698def add_idds_work(config, generic_workflow, idds_workflow): 

699 """Convert GenericWorkflowJobs to iDDS work and add them to the iDDS 

700 workflow. 

701 

702 Parameters 

703 ---------- 

704 config : `lsst.ctrl.bps.BpsConfig` 

705 BPS configuration. 

706 generic_workflow : `lsst.ctrl.bps.GenericWorkflow` 

707 Generic workflow containing jobs to convert. 

708 idds_workflow : `idds.workflowv2.workflow.Workflow` 

709 The iDDS workflow to which the converted jobs should be added. 

710 

711 Returns 

712 ------- 

713 files_to_pre_stage : `dict` [`str`, `str`] 

714 Files that need to be copied to the staging area before submission. 

715 dag_sink_work : `list` [`idds.doma.workflowv2.domapandawork.DomaPanDAWork`] 

716 The work nodes in the client workflow which have no successors. 

717 task_count : `int` 

718 Number of tasks in iDDS workflow used for unique task names. 

719 

720 Raises 

721 ------ 

722 RuntimeError 

723 If cannot recover from dependency issues after pass through workflow. 

724 """ 

725 # event service 

726 _, enable_event_service = config.search("enableEventService", opt={"default": None}) 

727 _, enable_qnode_map = config.search("enableQnodeMap", opt={"default": None}) 

728 _, max_payloads_per_panda_job = config.search( 

729 "maxPayloadsPerPandaJob", opt={"default": PANDA_DEFAULT_MAX_PAYLOADS_PER_PANDA_JOB} 

730 ) 

731 _, max_wms_job_wall_time = config.search("maxWmsJobWalltime", opt={"default": None}) 

732 my_log = ( 

733 f"enableEventService: {enable_event_service}, maxPayloadsPerPandaJob: {max_payloads_per_panda_job}" 

734 ) 

735 _LOG.info(my_log) 

736 

737 # job name map: Use a short job name to map the long job name 

738 _, enable_job_name_map = config.search("enableJobNameMap", opt={"default": None}) 

739 _LOG.info(f"enable_job_name_map: {enable_job_name_map}, {type(enable_job_name_map)}") 

740 if enable_event_service and not enable_job_name_map: 

741 enable_job_name_map = True 

742 my_log = "enable_event_service is set, set enable_job_name_map True." 

743 _LOG.info(my_log) 

744 

745 # Limit number of jobs in single PanDA task 

746 _, max_jobs_per_task = config.search("maxJobsPerTask", opt={"default": PANDA_DEFAULT_MAX_JOBS_PER_TASK}) 

747 

748 files_to_pre_stage = {} 

749 dag_sink_work = [] # Workflow sink nodes that need to be connected to final task 

750 job_to_task = {} 

751 job_to_pseudo_filename = {} 

752 task_count = 0 # Task number/ID in idds workflow used for unique name 

753 remote_archive_filename = None 

754 

755 submit_path = config["submitPath"] 

756 

757 submit_cmd = generic_workflow.run_attrs.get("bps_iscustom", False) 

758 if submit_cmd: 

759 files = generic_workflow.get_executables(data=False, transfer_only=True) 

760 archive_filename = f"jobO.{uuid.uuid4()}.tar.gz" 

761 archive_filename = create_archive_file(submit_path, archive_filename, files) 

762 remote_archive_filename = copy_files_to_pandacache(archive_filename) 

763 

764 order_id_map_files = {} 

765 name_works = {} 

766 order_id_map = {} 

767 job_name_to_order_id_map = {} 

768 order_id_map_file = None 

769 max_payloads_per_panda_job_by_label = {} 

770 if enable_event_service: 

771 enable_event_service = enable_event_service.split(",") 

772 enable_event_service_tmp = [] 

773 for es_def in enable_event_service: 

774 if ":" in es_def: 

775 es_label, m_payloads = es_def.split(":") 

776 else: 

777 es_label, m_payloads = es_def, max_payloads_per_panda_job 

778 es_label = es_label.strip() 

779 enable_event_service_tmp.append(es_label) 

780 max_payloads_per_panda_job_by_label[es_label] = int(m_payloads) 

781 enable_event_service = enable_event_service_tmp 

782 if enable_job_name_map: 

783 _, order_id_map_filename = config.search( 

784 "orderIdMapFilename", opt={"default": PANDA_DEFAULT_ORDER_ID_MAP_FILE} 

785 ) 

786 order_id_map_file = os.path.join(submit_path, order_id_map_filename) 

787 order_id_map_files = {"orderIdMapFilename": order_id_map_file} 

788 files_to_pre_stage.update(order_id_map_files) 

789 

790 # To avoid dying due to optimizing number of times through workflow, 

791 # catch dependency issues to loop through again later. 

792 jobs_with_dependency_issues = {} 

793 

794 # Initialize quantum node map 

795 qnode_map = {} 

796 qnode_map_filename = None 

797 if enable_qnode_map: 

798 qnode_map_file = os.path.join(submit_path, "qnode_map.json") 

799 qnode_map_filename = {"qnodemap": qnode_map_file} 

800 files_to_pre_stage.update(qnode_map_filename) 

801 

802 # Assume jobs with same label share config values 

803 for job_label in generic_workflow.labels: 

804 _LOG.debug("job_label = %s", job_label) 

805 

806 if enable_job_name_map: 

807 order_id_map[job_label] = {} 

808 job_name_to_order_id_map[job_label] = {} 

809 

810 # Add each job with a particular label to a corresponding PanDA task 

811 # A PanDA task has a limit on number of jobs, so break into multiple 

812 # PanDA tasks if needed. 

813 job_count = 0 # Number of jobs in idds task used for task chunking 

814 task_chunk = 1 # Task chunk number within job label used for unique name 

815 work = None 

816 order_id = -1 

817 

818 # Instead of changing code to make chunks up front and round-robin 

819 # assign jobs to chunks, for now keeping chunk creation in loop 

820 # but using knowledge of how many chunks there will be to set better 

821 # maximum number of jobs in a chunk for more even distribution. 

822 jobs_by_label = generic_workflow.get_jobs_by_label(job_label) 

823 num_chunks = -(-len(jobs_by_label) // max_jobs_per_task) # ceil 

824 max_jobs_per_task_this_label = -(-len(jobs_by_label) // num_chunks) 

825 _LOG.debug( 

826 "For job_label = %s, num jobs = %s, num_chunks = %s, max_jobs = %s", 

827 job_label, 

828 len(jobs_by_label), 

829 num_chunks, 

830 max_jobs_per_task_this_label, 

831 ) 

832 for gwjob in jobs_by_label: 

833 order_id += 1 

834 pseudo_filename = _make_pseudo_filename(config, gwjob) 

835 job_to_pseudo_filename[gwjob.name] = pseudo_filename 

836 if enable_job_name_map: 

837 order_id_map[job_label][str(order_id)] = pseudo_filename 

838 job_name_to_order_id_map[job_label][gwjob.name] = str(order_id) 

839 

840 job_count += 1 

841 if job_count > max_jobs_per_task_this_label: 

842 job_count = 1 

843 task_chunk += 1 

844 

845 if job_count == 1: 

846 # Create new PanDA task object 

847 task_count += 1 

848 work_enable_event_service = False 

849 if enable_event_service and job_label in enable_event_service: 

850 work_enable_event_service = True 

851 max_payloads_per_panda_job_current = max_payloads_per_panda_job_by_label.get( 

852 job_label, max_payloads_per_panda_job 

853 ) 

854 work, files = _make_doma_work( 

855 config, 

856 generic_workflow, 

857 gwjob, 

858 task_count, 

859 task_chunk, 

860 enable_event_service=work_enable_event_service, 

861 enable_job_name_map=enable_job_name_map, 

862 order_id_map_files=order_id_map_files, 

863 es_label=job_label, 

864 max_payloads_per_panda_job=max_payloads_per_panda_job_current, 

865 max_wms_job_wall_time=max_wms_job_wall_time, 

866 remote_filename=remote_archive_filename, 

867 qnode_map_filename=qnode_map_filename, 

868 ) 

869 work.dependency_tasks = [] 

870 name_works[work.task_name] = work 

871 files_to_pre_stage.update(files) 

872 idds_workflow.add_work(work) 

873 if generic_workflow.out_degree(gwjob.name) == 0: 

874 dag_sink_work.append(work) 

875 

876 if enable_qnode_map: 

877 job_name_PH = "PH:" + gwjob.name 

878 job_to_pseudo_filename[gwjob.name] = job_name_PH 

879 qnode_map[job_name_PH] = pseudo_filename 

880 

881 job_to_task[gwjob.name] = work.get_work_name() 

882 deps = [] 

883 missing_deps = False 

884 for parent_job_name in generic_workflow.predecessors(gwjob.name): 

885 if parent_job_name not in job_to_task: