Coverage for python / lsst / ctrl / bps / htcondor / prepare_utils.py: 5%

1# This file is part of ctrl_bps_htcondor. 

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"""Utility functions for preparing the HTCondor workflow.""" 

29 

30import logging 

31import os 

32import re 

33from collections import defaultdict 

34from copy import deepcopy 

35from pathlib import Path 

36from typing import Any, cast 

37 

38from lsst.ctrl.bps import ( 

39 BpsConfig, 

40 GenericWorkflow, 

41 GenericWorkflowGroup, 

42 GenericWorkflowJob, 

43 GenericWorkflowNodeType, 

44 GenericWorkflowNoopJob, 

45) 

46from lsst.ctrl.bps.bps_utils import create_count_summary 

47 

48from .lssthtc import ( 

49 HTCDag, 

50 HTCJob, 

51 _update_dicts, 

52 condor_status, 

53 htc_escape, 

54) 

55 

56_LOG = logging.getLogger(__name__) 

57 

58DEFAULT_HTC_EXEC_PATT = ".*worker.*" 

59"""Default pattern for searching execute machines in an HTCondor pool. 

60""" 

61 

62 

63def _create_job(subdir_template, cached_values, generic_workflow, gwjob, out_prefix): 

64 """Convert GenericWorkflow job nodes to DAG jobs. 

65 

66 Parameters 

67 ---------- 

68 subdir_template : `str` 

69 Template for making subdirs. 

70 cached_values : `dict` 

71 Site and label specific values. 

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

73 Generic workflow that is being converted. 

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

75 The generic job to convert to a HTCondor job. 

76 out_prefix : `str` 

77 Directory prefix for HTCondor files. 

78 

79 Returns 

80 ------- 

81 htc_job : `lsst.ctrl.bps.wms.htcondor.HTCJob` 

82 The HTCondor job equivalent to the given generic job. 

83 """ 

84 htc_job = HTCJob(gwjob.name, label=gwjob.label) 

85 

86 curvals = defaultdict(str) 

87 curvals["label"] = gwjob.label 

88 if gwjob.tags: 

89 curvals.update(gwjob.tags) 

90 

91 subdir = Path("jobs") / subdir_template.format_map(curvals) 

92 htc_job.subdir = subdir 

93 htc_job.subfile = f"{gwjob.name}.sub" 

94 htc_job.add_dag_cmds({"dir": subdir}) 

95 

96 htc_job_cmds = { 

97 "universe": "vanilla", 

98 "should_transfer_files": "YES", 

99 "when_to_transfer_output": "ON_EXIT_OR_EVICT", 

100 "transfer_output_files": '""', # Set to empty string to disable 

101 "transfer_executable": "False", 

102 "getenv": "True", 

103 # Exceeding memory sometimes triggers SIGBUS or SIGSEGV error. Tell 

104 # htcondor to put on hold any jobs which exited by a signal. If 

105 # executed in a bash script, like finalJob, the signals will become 

106 # exit codes above 128 (exit code = 128 + signal number). 

107 "on_exit_hold": "ExitBySignal == true || ExitCode > 128", 

108 "on_exit_hold_reason": "ExitBySignal == true ? " 

109 'strcat("Job raised a signal ", string(ExitSignal), ' 

110 '". Handling job as if it has gone over memory limit.") : ' 

111 'strcat("Job exit code (", string(ExitCode), ") > 128. ' 

112 'Handling job as if it has gone over memory limit.")', 

113 "on_exit_hold_subcode": "34", 

114 } 

115 

116 htc_job_cmds.update(_translate_job_cmds(cached_values, generic_workflow, gwjob)) 

117 

118 # Combine stdout and stderr to reduce the number of files. 

119 for key in ("output", "error"): 

120 if cached_values["overwriteJobFiles"]: 

121 htc_job_cmds[key] = f"{gwjob.name}.$(Cluster).out" 

122 else: 

123 htc_job_cmds[key] = f"{gwjob.name}.$(Cluster).$$([NumJobStarts ?: 0]).out" 

124 _LOG.debug("HTCondor %s = %s", key, htc_job_cmds[key]) 

125 

126 key = "log" 

127 htc_job_cmds[key] = f"{gwjob.name}.$(Cluster).{key}" 

128 _LOG.debug("HTCondor %s = %s", key, htc_job_cmds[key]) 

129 

130 htc_job_cmds.update( 

131 _handle_job_inputs(generic_workflow, gwjob.name, cached_values["bpsUseShared"], out_prefix) 

132 ) 

133 

134 htc_job_cmds.update( 

135 _handle_job_outputs(generic_workflow, gwjob.name, cached_values["bpsUseShared"], out_prefix) 

136 ) 

137 

138 # If specified, add nodeset to the job 

139 if "nodeset" in cached_values: 

140 htc_job.add_job_attrs({"JobNodeset": cached_values["nodeset"]}) 

141 clause = f'( Target.Nodeset == "{cached_values["nodeset"]}" )' 

142 if "requirements" in htc_job_cmds: 

143 htc_job_cmds["requirements"] = f"({htc_job_cmds['requirements']}) && {clause}" 

144 else: 

145 htc_job_cmds["requirements"] = clause 

146 

147 # Add the job cmds dict to the job object. 

148 htc_job.add_job_cmds(htc_job_cmds) 

149 

150 # Add job-related cmds to the DAG (e.g., VARS) 

151 htc_job.add_dag_cmds(_translate_dag_cmds(gwjob)) 

152 

153 # Add job attributes to job. 

154 _LOG.debug("gwjob.attrs = %s", gwjob.attrs) 

155 htc_job.add_job_attrs(gwjob.attrs) 

156 htc_job.add_job_attrs(cached_values["attrs"]) 

157 htc_job.add_job_attrs({"bps_job_quanta": create_count_summary(gwjob.quanta_counts)}) 

158 htc_job.add_job_attrs({"bps_job_name": gwjob.name, "bps_job_label": gwjob.label}) 

159 

160 return htc_job 

161 

162 

163def _translate_job_cmds(cached_vals, generic_workflow, gwjob): 

164 """Translate the job data that are one to one mapping 

165 

166 Parameters 

167 ---------- 

168 cached_vals : `dict` [`str`, `~typing.Any`] 

169 Config values common to jobs with same site or label. 

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

171 Generic workflow that contains job to being converted. 

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

173 Generic workflow job to be converted. 

174 

175 Returns 

176 ------- 

177 htc_job_commands : `dict` [`str`, `~typing.Any`] 

178 Contains commands which can appear in the HTCondor submit description 

179 file. 

180 """ 

181 # Values in the job script that just are name mappings. 

182 job_translation = { 

183 "mail_to": "notify_user", 

184 "when_to_mail": "notification", 

185 "request_cpus": "request_cpus", 

186 "priority": "priority", 

187 "category": "category", 

188 "accounting_group": "accounting_group", 

189 "accounting_user": "accounting_group_user", 

190 } 

191 

192 jobcmds = {} 

193 for gwkey, htckey in job_translation.items(): 

194 jobcmds[htckey] = getattr(gwjob, gwkey, None) 

195 

196 # If accounting info was not set explicitly, use site settings if any. 

197 if not gwjob.accounting_group: 

198 jobcmds["accounting_group"] = cached_vals.get("accountingGroup") 

199 if not gwjob.accounting_user: 

200 jobcmds["accounting_group_user"] = cached_vals.get("accountingUser") 

201 

202 # job commands that need modification 

203 if gwjob.retry_unless_exit: 

204 if isinstance(gwjob.retry_unless_exit, int): 

205 jobcmds["retry_until"] = f"{gwjob.retry_unless_exit}" 

206 elif isinstance(gwjob.retry_unless_exit, list): 

207 jobcmds["retry_until"] = ( 

208 f"member(ExitCode, {{{','.join([str(x) for x in gwjob.retry_unless_exit])}}})" 

209 ) 

210 else: 

211 raise ValueError("retryUnlessExit must be an integer or a list of integers.") 

212 

213 if gwjob.request_disk: 

214 jobcmds["request_disk"] = f"{gwjob.request_disk}MB" 

215 

216 if gwjob.request_memory: 

217 jobcmds["request_memory"] = f"{gwjob.request_memory}" 

218 

219 memory_max = 0 

220 if gwjob.memory_multiplier: 

221 # Do not use try-except! At the moment, BpsConfig returns an empty 

222 # string if it does not contain the key. 

223 memory_limit = cached_vals["memoryLimit"] 

224 if not memory_limit: 

225 raise RuntimeError( 

226 "Memory autoscaling enabled, but automatic detection of the memory limit " 

227 "failed; setting it explicitly with 'memoryLimit' or changing worker node " 

228 "search pattern 'executeMachinesPattern' might help." 

229 ) 

230 

231 # Set maximal amount of memory job can ask for. 

232 # 

233 # The check below assumes that 'memory_limit' was set to a value which 

234 # realistically reflects actual physical limitations of a given compute 

235 # resource. 

236 memory_max = memory_limit 

237 if gwjob.request_memory_max and gwjob.request_memory_max < memory_limit: 

238 memory_max = gwjob.request_memory_max 

239 

240 # Make job ask for more memory each time it failed due to insufficient 

241 # memory requirements. 

242 jobcmds["request_memory"] = _create_request_memory_expr( 

243 gwjob.request_memory, gwjob.memory_multiplier, memory_max 

244 ) 

245 

246 user_release_expr = cached_vals.get("releaseExpr", "") 

247 if gwjob.number_of_retries is not None and gwjob.number_of_retries >= 0: 

248 jobcmds["max_retries"] = gwjob.number_of_retries 

249 

250 # No point in adding periodic_release if 0 retries 

251 if gwjob.number_of_retries > 0: 

252 periodic_release = _create_periodic_release_expr( 

253 gwjob.request_memory, gwjob.memory_multiplier, memory_max, user_release_expr 

254 ) 

255 if periodic_release: 

256 jobcmds["periodic_release"] = periodic_release 

257 

258 jobcmds["periodic_remove"] = _create_periodic_remove_expr( 

259 gwjob.request_memory, gwjob.memory_multiplier, memory_max 

260 ) 

261 

262 # Assume concurrency_limit implemented using HTCondor concurrency limits. 

263 # May need to move to special site-specific implementation if sites use 

264 # other mechanisms. 

265 if gwjob.concurrency_limit: 

266 jobcmds["concurrency_limit"] = gwjob.concurrency_limit 

267 

268 # Handle command line 

269 if gwjob.executable.transfer_executable: 

270 jobcmds["transfer_executable"] = "True" 

271 jobcmds["executable"] = gwjob.executable.src_uri 

272 else: 

273 jobcmds["executable"] = _fix_env_var_syntax(gwjob.executable.src_uri) 

274 

275 if gwjob.arguments: 

276 arguments = gwjob.arguments 

277 arguments = _replace_cmd_vars(arguments, gwjob) 

278 arguments = _replace_file_vars(cached_vals["bpsUseShared"], arguments, generic_workflow, gwjob) 

279 arguments = _fix_env_var_syntax(arguments) 

280 jobcmds["arguments"] = arguments 

281 

282 if gwjob.environment: 

283 env_str = "" 

284 for name, value in gwjob.environment.items(): 

285 if isinstance(value, str): 

286 value2 = _replace_cmd_vars(value, gwjob) 

287 value2 = _fix_env_var_syntax(value2) 

288 value2 = htc_escape(value2) 

289 env_str += f"{name}='{value2}' " # Add single quotes to allow internal spaces 

290 else: 

291 env_str += f"{name}={value} " 

292 

293 # Process above added one trailing space 

294 jobcmds["environment"] = env_str.rstrip() 

295 

296 # Add extra "pass-thru" job commands 

297 if gwjob.profile: 

298 for key, val in gwjob.profile.items(): 

299 jobcmds[key] = val 

300 for key, val in cached_vals["profile"].items(): 

301 jobcmds[key] = val 

302 

303 return jobcmds 

304 

305 

306def _translate_dag_cmds(gwjob): 

307 """Translate job values into DAGMan commands. 

308 

309 Parameters 

310 ---------- 

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

312 Job containing values to be translated. 

313 

314 Returns 

315 ------- 

316 dagcmds : `dict` [`str`, `~typing.Any`] 

317 DAGMan commands for the job. 

318 """ 

319 # Values in the dag script that just are name mappings. 

320 dag_translation = { 

321 "abort_on_value": "abort_dag_on", 

322 "abort_return_value": "abort_exit", 

323 "priority": "priority", 

324 } 

325 

326 dagcmds = {} 

327 for gwkey, htckey in dag_translation.items(): 

328 dagcmds[htckey] = getattr(gwjob, gwkey, None) 

329 

330 # Still to be coded: vars "pre_cmdline", "post_cmdline" 

331 return dagcmds 

332 

333 

334def _fix_env_var_syntax(oldstr): 

335 """Change ENV place holders to HTCondor Env var syntax. 

336 

337 Parameters 

338 ---------- 

339 oldstr : `str` 

340 String in which environment variable syntax is to be fixed. 

341 

342 Returns 

343 ------- 

344 newstr : `str` 

345 Given string with environment variable syntax fixed. 

346 """ 

347 newstr = oldstr 

348 for key in re.findall(r"<ENV:([^>]+)>", oldstr): 

349 newstr = newstr.replace(rf"<ENV:{key}>", f"$ENV({key})") 

350 return newstr 

351 

352 

353def _replace_file_vars(use_shared, arguments, workflow, gwjob): 

354 """Replace file placeholders in command line arguments with correct 

355 physical file names. 

356 

357 Parameters 

358 ---------- 

359 use_shared : `bool` 

360 Whether HTCondor can assume shared filesystem. 

361 arguments : `str` 

362 Arguments string in which to replace file placeholders. 

363 workflow : `lsst.ctrl.bps.GenericWorkflow` 

364 Generic workflow that contains file information. 

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

366 The job corresponding to the arguments. 

367 

368 Returns 

369 ------- 

370 arguments : `str` 

371 Given arguments string with file placeholders replaced. 

372 """ 

373 # Replace input file placeholders with paths. 

374 for gwfile in workflow.get_job_inputs(gwjob.name, data=True, transfer_only=False): 

375 if not gwfile.wms_transfer: 

376 # Must assume full URI if in command line and told WMS is not 

377 # responsible for transferring file. 

378 uri = gwfile.src_uri 

379 elif use_shared: 

380 if gwfile.job_shared: 

381 # Have shared filesystems and jobs can share file. 

382 uri = gwfile.src_uri 

383 else: 

384 uri = os.path.basename(gwfile.src_uri) 

385 else: # Using push transfer 

386 uri = os.path.basename(gwfile.src_uri) 

387 arguments = arguments.replace(f"<FILE:{gwfile.name}>", uri) 

388 

389 # Replace output file placeholders with paths. 

390 for gwfile in workflow.get_job_outputs(gwjob.name, data=True, transfer_only=False): 

391 if not gwfile.wms_transfer: 

392 # Must assume full URI if in command line and told WMS is not 

393 # responsible for transferring file. 

394 uri = gwfile.src_uri 

395 elif use_shared: 

396 if gwfile.job_shared: 

397 # Have shared filesystems and jobs can share file. 

398 uri = gwfile.src_uri 

399 else: 

400 uri = os.path.basename(gwfile.src_uri) 

401 else: # Using push transfer 

402 uri = os.path.basename(gwfile.src_uri) 

403 arguments = arguments.replace(f"<FILE:{gwfile.name}>", uri) 

404 return arguments 

405 

406 

407def _replace_cmd_vars(arguments, gwjob): 

408 """Replace format-style placeholders in arguments. 

409 

410 Parameters 

411 ---------- 

412 arguments : `str` 

413 Arguments string in which to replace placeholders. 

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

415 Job containing values to be used to replace placeholders 

416 (in particular gwjob.cmdvals). 

417 

418 Returns 

419 ------- 

420 arguments : `str` 

421 Given arguments string with placeholders replaced. 

422 """ 

423 replacements = gwjob.cmdvals if gwjob.cmdvals is not None else {} 

424 try: 

425 arguments = arguments.format(**replacements) 

426 except (KeyError, TypeError) as exc: # TypeError in case None instead of {} 

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

428 _LOG.debug("arguments: %s\ncmdvals: %s", arguments, replacements) 

429 raise 

430 return arguments 

431 

432 

433def _handle_job_inputs( 

434 generic_workflow: GenericWorkflow, job_name: str, use_shared: bool, out_prefix: str 

435) -> dict[str, str]: 

436 """Add job input files from generic workflow to job. 

437 

438 Parameters 

439 ---------- 

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

441 The generic workflow (e.g., has executable name and arguments). 

442 job_name : `str` 

443 Unique name for the job. 

444 use_shared : `bool` 

445 Whether job has access to files via shared filesystem. 

446 out_prefix : `str` 

447 The root directory into which all WMS-specific files are written. 

448 

449 Returns 

450 ------- 

451 htc_commands : `dict` [`str`, `str`] 

452 HTCondor commands for the job submission script. 

453 """ 

454 inputs = [] 

455 for gwf_file in generic_workflow.get_job_inputs(job_name, data=True, transfer_only=True): 

456 _LOG.debug("src_uri=%s", gwf_file.src_uri) 

457 

458 uri = Path(gwf_file.src_uri) 

459 

460 # Note if use_shared and job_shared, don't need to transfer file. 

461 

462 if not use_shared: # Copy file using push to job 

463 inputs.append(str(uri)) 

464 elif not gwf_file.job_shared: # Jobs require own copy 

465 # if using shared filesystem, but still need copy in job. Use 

466 # HTCondor's curl plugin for a local copy. 

467 if uri.is_dir(): 

468 raise RuntimeError( 

469 f"HTCondor plugin cannot transfer directories locally within job {gwf_file.src_uri}" 

470 ) 

471 inputs.append(f"file://{uri}") 

472 

473 htc_commands = {} 

474 if inputs: 

475 htc_commands["transfer_input_files"] = ",".join(inputs) 

476 _LOG.debug("transfer_input_files=%s", htc_commands["transfer_input_files"]) 

477 return htc_commands 

478 

479 

480def _handle_job_outputs( 

481 generic_workflow: GenericWorkflow, job_name: str, use_shared: bool, out_prefix: str 

482) -> dict[str, str]: 

483 """Add job output files from generic workflow to the job if any. 

484 

485 Parameters 

486 ---------- 

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

488 The generic workflow (e.g., has executable name and arguments). 

489 job_name : `str` 

490 Unique name for the job. 

491 use_shared : `bool` 

492 Whether job has access to files via shared filesystem. 

493 out_prefix : `str` 

494 The root directory into which all WMS-specific files are written. 

495 

496 Returns 

497 ------- 

498 htc_commands : `dict` [`str`, `str`] 

499 HTCondor commands for the job submission script. 

500 """ 

501 outputs = [] 

502 output_remaps = [] 

503 for gwf_file in generic_workflow.get_job_outputs(job_name, data=True, transfer_only=True): 

504 _LOG.debug("src_uri=%s", gwf_file.src_uri) 

505 

506 uri = Path(gwf_file.src_uri) 

507 if not use_shared: 

508 outputs.append(uri.name) 

509 output_remaps.append(f"{uri.name}={str(uri)}") 

510 

511 # Set to an empty string to disable and only update if there are output 

512 # files to transfer. Otherwise, HTCondor will transfer back all files in 

513 # the job’s temporary working directory that have been modified or created 

514 # by the job. 

515 htc_commands = {"transfer_output_files": '""'} 

516 if outputs: 

517 htc_commands["transfer_output_files"] = ",".join(outputs) 

518 _LOG.debug("transfer_output_files=%s", htc_commands["transfer_output_files"]) 

519 

520 htc_commands["transfer_output_remaps"] = f'"{";".join(output_remaps)}"' 

521 _LOG.debug("transfer_output_remaps=%s", htc_commands["transfer_output_remaps"]) 

522 return htc_commands 

523 

524 

525def _create_periodic_release_expr( 

526 memory: int, multiplier: float | None, limit: int, additional_expr: str = "" 

527) -> str: 

528 """Construct an HTCondorAd expression for releasing held jobs. 

529 

530 Parameters 

531 ---------- 

532 memory : `int` 

533 Requested memory in MB. 

534 multiplier : `float` or None 

535 Memory growth rate between retries. 

536 limit : `int` 

537 Memory limit. 

538 additional_expr : `str`, optional 

539 Expression to add to periodic_release. Defaults to empty string. 

540 

541 Returns 

542 ------- 

543 expr : `str` 

544 A string representing an HTCondor ClassAd expression for releasing job. 

545 """ 

546 _LOG.debug( 

547 "periodic_release: memory: %s, multiplier: %s, limit: %s, additional_expr: %s", 

548 memory, 

549 multiplier, 

550 limit, 

551 additional_expr, 

552 ) 

553 

554 # ctrl_bps sets multiplier to None in the GenericWorkflow if 

555 # memoryMultiplier <= 1, but checking value just in case. 

556 if (not multiplier or multiplier <= 1) and not additional_expr: 

557 return "" 

558 

559 # Job ClassAds attributes 'HoldReasonCode' and 'HoldReasonSubCode' are 

560 # UNDEFINED if job is not HELD (i.e. when 'JobStatus' is not 5). 

561 # The special comparison operators ensure that all comparisons below will 

562 # evaluate to FALSE in this case. 

563 # 

564 # Note: 

565 # May not be strictly necessary. Operators '&&' and '||' are not strict so 

566 # the entire expression should evaluate to FALSE when the job is not HELD. 

567 # According to ClassAd evaluation semantics FALSE && UNDEFINED is FALSE, 

568 # but better safe than sorry. 

569 is_held = "JobStatus == 5" 

570 is_retry_allowed = "NumJobStarts <= JobMaxRetries" 

571 

572 mem_expr = "" 

573 if memory and multiplier and multiplier > 1 and limit: 

574 was_mem_exceeded = ( 

575 "(HoldReasonCode =?= 34 && HoldReasonSubCode =?= 0 " 

576 "|| HoldReasonCode =?= 3 && HoldReasonSubCode =?= 34)" 

577 ) 

578 was_below_limit = f"min({{int({memory} * pow({multiplier}, NumJobStarts - 1)), {limit}}}) < {limit}" 

579 mem_expr = f"{was_mem_exceeded} && {was_below_limit}" 

580 

581 user_expr = "" 

582 if additional_expr: 

583 # Never auto release a job held by user. 

584 user_expr = f"HoldReasonCode =!= 1 && {additional_expr}" 

585 

586 expr = f"{is_held} && {is_retry_allowed}" 

587 if user_expr and mem_expr: 

588 expr += f" && ({mem_expr} || {user_expr})" 

589 elif user_expr: 

590 expr += f" && {user_expr}" 

591 elif mem_expr: 

592 expr += f" && {mem_expr}" 

593 

594 return expr 

595 

596 

597def _create_periodic_remove_expr(memory, multiplier, limit): 

598 """Construct an HTCondorAd expression for removing jobs from the queue. 

599 

600 Parameters 

601 ---------- 

602 memory : `int` 

603 Requested memory in MB. 

604 multiplier : `float` 

605 Memory growth rate between retries. 

606 limit : `int` 

607 Memory limit. 

608 

609 Returns 

610 ------- 

611 expr : `str` 

612 A string representing an HTCondor ClassAd expression for removing jobs. 

613 """ 

614 # Job ClassAds attributes 'HoldReasonCode' and 'HoldReasonSubCode' 

615 # are UNDEFINED if job is not HELD (i.e. when 'JobStatus' is not 5). 

616 # The special comparison operators ensure that all comparisons below 

617 # will evaluate to FALSE in this case. 

618 # 

619 # Note: 

620 # May not be strictly necessary. Operators '&&' and '||' are not 

621 # strict so the entire expression should evaluate to FALSE when the 

622 # job is not HELD. According to ClassAd evaluation semantics 

623 # FALSE && UNDEFINED is FALSE, but better safe than sorry. 

624 is_held = "JobStatus == 5" 

625 is_retry_disallowed = "NumJobStarts > JobMaxRetries" 

626 

627 mem_expr = "" 

628 if memory and multiplier and multiplier > 1 and limit: 

629 mem_limit_expr = f"min({{int({memory} * pow({multiplier}, NumJobStarts - 1)), {limit}}}) == {limit}" 

630 

631 mem_expr = ( # Add || here so only added if adding memory expr 

632 " || ((HoldReasonCode =?= 34 && HoldReasonSubCode =?= 0 " 

633 f"|| HoldReasonCode =?= 3 && HoldReasonSubCode =?= 34) && {mem_limit_expr})" 

634 ) 

635 

636 expr = f"{is_held} && ({is_retry_disallowed}{mem_expr})" 

637 return expr 

638 

639 

640def _create_request_memory_expr(memory, multiplier, limit): 

641 """Construct an HTCondor ClassAd expression for safe memory scaling. 

642 

643 Parameters 

644 ---------- 

645 memory : `int` 

646 Requested memory in MB. 

647 multiplier : `float` 

648 Memory growth rate between retries. 

649 limit : `int` 

650 Memory limit. 

651 

652 Returns 

653 ------- 

654 expr : `str` 

655 A string representing an HTCondor ClassAd expression enabling safe 

656 memory scaling between job retries. 

657 """ 

658 # The check if the job was held due to exceeding memory requirements 

659 # will be made *after* job was released back to the job queue (is in 

660 # the IDLE state), hence the need to use `Last*` job ClassAds instead of 

661 # the ones describing job's current state. 

662 # 

663 # Also, 'Last*' job ClassAds attributes are UNDEFINED when a job is 

664 # initially put in the job queue. The special comparison operators ensure 

665 # that all comparisons below will evaluate to FALSE in this case. 

666 was_mem_exceeded = ( 

667 "LastJobStatus =?= 5 " 

668 "&& (LastHoldReasonCode =?= 34 && LastHoldReasonSubCode =?= 0 " 

669 "|| LastHoldReasonCode =?= 3 && LastHoldReasonSubCode =?= 34)" 

670 ) 

671 

672 # If job runs the first time or was held for reasons other than exceeding 

673 # the memory, set the required memory to the requested value or use 

674 # the memory value measured by HTCondor (MemoryUsage) depending on 

675 # whichever is greater. 

676 expr = ( 

677 f"({was_mem_exceeded}) " 

678 f"? min({{int({memory} * pow({multiplier}, NumJobStarts)), {limit}}}) " 

679 f": max({{{memory}, MemoryUsage ?: 0}})" 

680 ) 

681 return expr 

682 

683 

684def _gather_site_values(config, compute_site): 

685 """Gather values specific to given site. 

686 

687 Parameters 

688 ---------- 

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

690 BPS configuration that includes necessary submit/runtime 

691 information. 

692 compute_site : `str` 

693 Compute site name. 

694 

695 Returns 

696 ------- 

697 site_values : `dict` [`str`, `~typing.Any`] 

698 Values specific to the given site. 

699 """ 

700 site_values = {"attrs": {}, "profile": {}} 

701 search_opts = {} 

702 if compute_site: 

703 search_opts["curvals"] = {"curr_site": compute_site} 

704 

705 # Determine the hard limit for the memory requirement. 

706 found, limit = config.search("memoryLimit", opt=search_opts) 

707 if not found: 

708 search_opts["default"] = DEFAULT_HTC_EXEC_PATT 

709 _, patt = config.search("executeMachinesPattern", opt=search_opts) 

710 del search_opts["default"] 

711 

712 # To reduce the amount of data, ignore dynamic slots (if any) as, 

713 # by definition, they cannot have more memory than 

714 # the partitionable slot they are the part of. 

715 constraint = f'SlotType != "Dynamic" && regexp("{patt}", Machine)' 

716 pool_info = condor_status(constraint=constraint) 

717 try: 

718 limit = max(int(info["TotalSlotMemory"]) for info in pool_info.values()) 

719 except ValueError: 

720 _LOG.debug("No execute machine in the pool matches %s", patt) 

721 if limit: 

722 config[".bps_defined.memory_limit"] = limit 

723 

724 _, site_values["bpsUseShared"] = config.search("bpsUseShared", opt={"default": False}) 

725 site_values["memoryLimit"] = limit 

726 

727 found, value = config.search("accountingGroup", opt=search_opts) 

728 if found: 

729 site_values["accountingGroup"] = value 

730 found, value = config.search("accountingUser", opt=search_opts) 

731 if found: 

732 site_values["accountingUser"] = value 

733 

734 found, nodeset = config.search("nodeset", opt=search_opts) 

735 if found: 

736 site_values["nodeset"] = nodeset 

737 

738 searchobj = config[f".site.{compute_site}.profile.condor"] 

739 if searchobj: 

740 search_opts["searchobj"] = searchobj 

741 search_opts["replaceVars"] = True 

742 for key in searchobj: 

743 if key.startswith("+"): 

744 _, val = config.search(key, opt=search_opts) 

745 site_values["attrs"][key[1:]] = val 

746 else: 

747 _, val = config.search(key, opt=search_opts) 

748 site_values["profile"][key] = val 

749 

750 _LOG.debug("site_values = %s", site_values) 

751 return site_values 

752 

753 

754def _gather_label_values(config: BpsConfig, label: str) -> dict[str, Any]: 

755 """Gather values specific to given job label. 

756 

757 Parameters 

758 ---------- 

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

760 BPS configuration that includes necessary submit/runtime 

761 information. 

762 label : `str` 

763 GenericWorkflowJob label. 

764 

765 Returns 

766 ------- 

767 values : `dict` [`str`, `~typing.Any`] 

768 Values specific to the given job label. 

769 """ 

770 values: dict[str, Any] = {"attrs": {}, "profile": {}} 

771 

772 search_opts = {} 

773 profile_key = "" 

774 if label == "finalJob" and "finalJob" in config: 

775 search_opts["searchobj"] = config["finalJob"] 

776 profile_key = ".finalJob.profile.condor" 

777 elif label in config["cluster"]: 

778 search_opts["curvals"] = {"curr_cluster": label} 

779 profile_key = f".cluster.{label}.profile.condor" 

780 elif label in config["pipetask"]: 

781 search_opts["curvals"] = {"curr_pipetask": label} 

782 profile_key = f".pipetask.{label}.profile.condor" 

783 

784 found, value = config.search("releaseExpr", opt=search_opts) 

785 if found: 

786 values["releaseExpr"] = value 

787 

788 found, value = config.search("overwriteJobFiles", opt=search_opts) 

789 if found: 

790 values["overwriteJobFiles"] = value 

791 else: 

792 values["overwriteJobFiles"] = True 

793 

794 if profile_key and profile_key in config: 

795 for subkey, val in config[profile_key].items(): 

796 if subkey.startswith("+"): 

797 values["attrs"][subkey[1:]] = val 

798 else: 

799 values["profile"][subkey] = val 

800 

801 return values 

802 

803 

804def _group_to_subdag( 

805 config: BpsConfig, generic_workflow_group: GenericWorkflowGroup, out_prefix: str 

806) -> HTCJob: 

807 """Convert a generic workflow group to an HTCondor dag. 

808 

809 Parameters 

810 ---------- 

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

812 Workflow configuration. 

813 generic_workflow_group : `lsst.ctrl.bps.GenericWorkflowGroup` 

814 The generic workflow group to convert. 

815 out_prefix : `str` 

816 Location prefix to be used when creating jobs. 

817 

818 Returns 

819 ------- 

820 htc_job : `lsst.ctrl.bps.htcondor.HTCJob` 

821 Job for running the HTCondor dag. 

822 """ 

823 jobname = f"wms_{generic_workflow_group.name}" 

824 htc_job = HTCJob(name=jobname, label=generic_workflow_group.label) 

825 htc_job.add_dag_cmds({"dir": f"subdags/{jobname}"}) 

826 htc_job.subdag = _generic_workflow_to_htcondor_dag(config, generic_workflow_group, out_prefix) 

827 if not generic_workflow_group.blocking: 

828 htc_job.dagcmds["post"] = { 

829 "defer": "", 

830 "executable": f"{os.path.dirname(__file__)}/subdag_post.sh", 

831 "arguments": f"{jobname} $RETURN", 

832 } 

833 return htc_job 

834 

835 

836def _create_check_job(group_job_name: str, job_label: str) -> HTCJob: 

837 """Create a job to check status of a group job. 

838 

839 Parameters 

840 ---------- 

841 group_job_name : `str` 

842 Name of the group job. 

843 job_label : `str` 

844 Label to use for the check status job. 

845 

846 Returns 

847 ------- 

848 htc_job : `lsst.ctrl.bps.htcondor.HTCJob` 

849 Job description for the job to check group job status. 

850 """ 

851 htc_job = HTCJob(name=f"wms_check_status_{group_job_name}", label=job_label) 

852 htc_job.subfile = "${CTRL_BPS_HTCONDOR_DIR}/python/lsst/ctrl/bps/htcondor/check_group_status.sub" 

853 htc_job.add_dag_cmds({"dir": f"subdags/{group_job_name}", "vars": {"group_job_name": group_job_name}}) 

854 

855 return htc_job 

856 

857 

858def _generic_workflow_to_htcondor_dag( 

859 config: BpsConfig, generic_workflow: GenericWorkflow, out_prefix: str 

860) -> HTCDag: 

861 """Convert a GenericWorkflow to a HTCDag. 

862 

863 Parameters 

864 ---------- 

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

866 Workflow configuration. 

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

868 The GenericWorkflow to convert. 

869 out_prefix : `str` 

870 Location prefix where the HTCondor files will be written. 

871 

872 Returns 

873 ------- 

874 dag : `lsst.ctrl.bps.htcondor.HTCDag` 

875 The HTCDag representation of the given GenericWorkflow. 

876 """ 

877 dag = HTCDag(name=generic_workflow.name) 

878 

879 _LOG.debug("htcondor dag attribs %s", generic_workflow.run_attrs) 

880 dag.add_attribs(generic_workflow.run_attrs) 

881 dag.add_attribs( 

882 { 

883 "bps_run_quanta": create_count_summary(generic_workflow.quanta_counts), 

884 "bps_job_summary": create_count_summary(generic_workflow.job_counts), 

885 } 

886 ) 

887 

888 _, tmp_template = config.search("subDirTemplate", opt={"replaceVars": False, "default": ""}) 

889 if isinstance(tmp_template, str): 

890 subdir_template = defaultdict(lambda: tmp_template) 

891 else: 

892 subdir_template = tmp_template 

893 

894 # Create all DAG jobs 

895 site_values = {} # Cache compute site specific values to reduce config lookups. 

896 cached_values = {} # Cache label-specific values to reduce config lookups. 

897 # Note: Can't use get_job_by_label because those only include payload jobs. 

898 for job_name in generic_workflow: 

899 gwjob = generic_workflow.get_job(job_name) 

900 if gwjob.node_type == GenericWorkflowNodeType.PAYLOAD: 

901 gwjob = cast(GenericWorkflowJob, gwjob) 

902 if gwjob.compute_site not in site_values: 

903 site_values[gwjob.compute_site] = _gather_site_values(config, gwjob.compute_site) 

904 if gwjob.label not in cached_values: 

905 cached_values[gwjob.label] = deepcopy(site_values[gwjob.compute_site]) 

906 _update_dicts(cached_values[gwjob.label], _gather_label_values(config, gwjob.label)) 

907 _LOG.debug("cached: %s= %s", gwjob.label, cached_values[gwjob.label]) 

908 htc_job = _create_job( 

909 subdir_template[gwjob.label], 

910 cached_values[gwjob.label], 

911 generic_workflow, 

912 gwjob, 

913 out_prefix, 

914 ) 

915 elif gwjob.node_type == GenericWorkflowNodeType.NOOP: