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 condor_status, 

52 htc_escape, 

53) 

54 

55_LOG = logging.getLogger(__name__) 

56 

57DEFAULT_HTC_EXEC_PATT = ".*worker.*" 

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

59""" 

60 

61 

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

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

64 

65 Parameters 

66 ---------- 

67 subdir_template : `str` 

68 Template for making subdirs. 

69 cached_values : `dict` 

70 Site and label specific values. 

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

72 Generic workflow that is being converted. 

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

74 The generic job to convert to a HTCondor job. 

75 out_prefix : `str` 

76 Directory prefix for HTCondor files. 

77 

78 Returns 

79 ------- 

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

81 The HTCondor job equivalent to the given generic job. 

82 """ 

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

84 

85 curvals = defaultdict(str) 

86 curvals["label"] = gwjob.label 

87 if gwjob.tags: 

88 curvals.update(gwjob.tags) 

89 

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

91 htc_job.subdir = subdir 

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

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

94 

95 htc_job_cmds = { 

96 "universe": "vanilla", 

97 "should_transfer_files": "YES", 

98 "when_to_transfer_output": "ON_EXIT_OR_EVICT", 

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

100 "transfer_executable": "False", 

101 "getenv": "True", 

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

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

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

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

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

107 "on_exit_hold_reason": "ExitBySignal == true ? " 

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

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

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

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

112 "on_exit_hold_subcode": "34", 

113 } 

114 

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

116 

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

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

119 if cached_values["overwriteJobFiles"]: 

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

121 else: 

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

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

124 

125 key = "log" 

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

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

128 

129 htc_job_cmds.update( 

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

131 ) 

132 

133 htc_job_cmds.update( 

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

135 ) 

136 

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

138 htc_job.add_job_cmds(htc_job_cmds) 

139 

140 htc_job.add_dag_cmds(_translate_dag_cmds(gwjob)) 

141 

142 # Add job attributes to job. 

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

144 htc_job.add_job_attrs(gwjob.attrs) 

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

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

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

148 

149 return htc_job 

150 

151 

152def _translate_job_cmds(cached_vals, generic_workflow, gwjob): 

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

154 

155 Parameters 

156 ---------- 

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

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

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

160 Generic workflow that contains job to being converted. 

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

162 Generic workflow job to be converted. 

163 

164 Returns 

165 ------- 

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

167 Contains commands which can appear in the HTCondor submit description 

168 file. 

169 """ 

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

171 job_translation = { 

172 "mail_to": "notify_user", 

173 "when_to_mail": "notification", 

174 "request_cpus": "request_cpus", 

175 "priority": "priority", 

176 "category": "category", 

177 "accounting_group": "accounting_group", 

178 "accounting_user": "accounting_group_user", 

179 } 

180 

181 jobcmds = {} 

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

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

184 

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

186 if not gwjob.accounting_group: 

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

188 if not gwjob.accounting_user: 

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

190 

191 # job commands that need modification 

192 if gwjob.retry_unless_exit: 

193 if isinstance(gwjob.retry_unless_exit, int): 

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

195 elif isinstance(gwjob.retry_unless_exit, list): 

196 jobcmds["retry_until"] = ( 

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

198 ) 

199 else: 

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

201 

202 if gwjob.request_disk: 

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

204 

205 if gwjob.request_memory: 

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

207 

208 memory_max = 0 

209 if gwjob.memory_multiplier: 

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

211 # string if it does not contain the key. 

212 memory_limit = cached_vals["memoryLimit"] 

213 if not memory_limit: 

214 raise RuntimeError( 

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

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

217 "search pattern 'executeMachinesPattern' might help." 

218 ) 

219 

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

221 # 

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

223 # realistically reflects actual physical limitations of a given compute 

224 # resource. 

225 memory_max = memory_limit 

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

227 memory_max = gwjob.request_memory_max 

228 

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

230 # memory requirements. 

231 jobcmds["request_memory"] = _create_request_memory_expr( 

232 gwjob.request_memory, gwjob.memory_multiplier, memory_max 

233 ) 

234 

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

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

237 jobcmds["max_retries"] = gwjob.number_of_retries 

238 

239 # No point in adding periodic_release if 0 retries 

240 if gwjob.number_of_retries > 0: 

241 periodic_release = _create_periodic_release_expr( 

242 gwjob.request_memory, gwjob.memory_multiplier, memory_max, user_release_expr 

243 ) 

244 if periodic_release: 

245 jobcmds["periodic_release"] = periodic_release 

246 

247 jobcmds["periodic_remove"] = _create_periodic_remove_expr( 

248 gwjob.request_memory, gwjob.memory_multiplier, memory_max 

249 ) 

250 

251 # Assume concurrency_limit implemented using HTCondor concurrency limits. 

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

253 # other mechanisms. 

254 if gwjob.concurrency_limit: 

255 jobcmds["concurrency_limit"] = gwjob.concurrency_limit 

256 

257 # Handle command line 

258 if gwjob.executable.transfer_executable: 

259 jobcmds["transfer_executable"] = "True" 

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

261 else: 

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

263 

264 if gwjob.arguments: 

265 arguments = gwjob.arguments 

266 arguments = _replace_cmd_vars(arguments, gwjob) 

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

268 arguments = _fix_env_var_syntax(arguments) 

269 jobcmds["arguments"] = arguments 

270 

271 if gwjob.environment: 

272 env_str = "" 

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

274 if isinstance(value, str): 

275 value2 = _replace_cmd_vars(value, gwjob) 

276 value2 = _fix_env_var_syntax(value2) 

277 value2 = htc_escape(value2) 

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

279 else: 

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

281 

282 # Process above added one trailing space 

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

284 

285 # Add extra "pass-thru" job commands 

286 if gwjob.profile: 

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

288 jobcmds[key] = htc_escape(val) 

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

290 jobcmds[key] = htc_escape(val) 

291 

292 return jobcmds 

293 

294 

295def _translate_dag_cmds(gwjob): 

296 """Translate job values into DAGMan commands. 

297 

298 Parameters 

299 ---------- 

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

301 Job containing values to be translated. 

302 

303 Returns 

304 ------- 

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

306 DAGMan commands for the job. 

307 """ 

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

309 dag_translation = {"abort_on_value": "abort_dag_on", "abort_return_value": "abort_exit"} 

310 

311 dagcmds = {} 

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

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

314 

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

316 return dagcmds 

317 

318 

319def _fix_env_var_syntax(oldstr): 

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

321 

322 Parameters 

323 ---------- 

324 oldstr : `str` 

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

326 

327 Returns 

328 ------- 

329 newstr : `str` 

330 Given string with environment variable syntax fixed. 

331 """ 

332 newstr = oldstr 

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

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

335 return newstr 

336 

337 

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

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

340 physical file names. 

341 

342 Parameters 

343 ---------- 

344 use_shared : `bool` 

345 Whether HTCondor can assume shared filesystem. 

346 arguments : `str` 

347 Arguments string in which to replace file placeholders. 

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

349 Generic workflow that contains file information. 

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

351 The job corresponding to the arguments. 

352 

353 Returns 

354 ------- 

355 arguments : `str` 

356 Given arguments string with file placeholders replaced. 

357 """ 

358 # Replace input file placeholders with paths. 

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

360 if not gwfile.wms_transfer: 

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

362 # responsible for transferring file. 

363 uri = gwfile.src_uri 

364 elif use_shared: 

365 if gwfile.job_shared: 

366 # Have shared filesystems and jobs can share file. 

367 uri = gwfile.src_uri 

368 else: 

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

370 else: # Using push transfer 

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

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

373 

374 # Replace output file placeholders with paths. 

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

376 if not gwfile.wms_transfer: 

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

378 # responsible for transferring file. 

379 uri = gwfile.src_uri 

380 elif use_shared: 

381 if gwfile.job_shared: 

382 # Have shared filesystems and jobs can share file. 

383 uri = gwfile.src_uri 

384 else: 

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

386 else: # Using push transfer 

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

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

389 return arguments 

390 

391 

392def _replace_cmd_vars(arguments, gwjob): 

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

394 

395 Parameters 

396 ---------- 

397 arguments : `str` 

398 Arguments string in which to replace placeholders. 

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

400 Job containing values to be used to replace placeholders 

401 (in particular gwjob.cmdvals). 

402 

403 Returns 

404 ------- 

405 arguments : `str` 

406 Given arguments string with placeholders replaced. 

407 """ 

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

409 try: 

410 arguments = arguments.format(**replacements) 

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

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

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

414 raise 

415 return arguments 

416 

417 

418def _handle_job_inputs( 

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

420) -> dict[str, str]: 

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

422 

423 Parameters 

424 ---------- 

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

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

427 job_name : `str` 

428 Unique name for the job. 

429 use_shared : `bool` 

430 Whether job has access to files via shared filesystem. 

431 out_prefix : `str` 

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

433 

434 Returns 

435 ------- 

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

437 HTCondor commands for the job submission script. 

438 """ 

439 inputs = [] 

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

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

442 

443 uri = Path(gwf_file.src_uri) 

444 

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

446 

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

448 inputs.append(str(uri)) 

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

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

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

452 if uri.is_dir(): 

453 raise RuntimeError( 

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

455 ) 

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

457 

458 htc_commands = {} 

459 if inputs: 

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

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

462 return htc_commands 

463 

464 

465def _handle_job_outputs( 

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

467) -> dict[str, str]: 

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

469 

470 Parameters 

471 ---------- 

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

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

474 job_name : `str` 

475 Unique name for the job. 

476 use_shared : `bool` 

477 Whether job has access to files via shared filesystem. 

478 out_prefix : `str` 

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

480 

481 Returns 

482 ------- 

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

484 HTCondor commands for the job submission script. 

485 """ 

486 outputs = [] 

487 output_remaps = [] 

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

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

490 

491 uri = Path(gwf_file.src_uri) 

492 if not use_shared: 

493 outputs.append(uri.name) 

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

495 

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

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

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

499 # by the job. 

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

501 if outputs: 

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

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

504 

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

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

507 return htc_commands 

508 

509 

510def _create_periodic_release_expr( 

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

512) -> str: 

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

514 

515 Parameters 

516 ---------- 

517 memory : `int` 

518 Requested memory in MB. 

519 multiplier : `float` or None 

520 Memory growth rate between retries. 

521 limit : `int` 

522 Memory limit. 

523 additional_expr : `str`, optional 

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

525 

526 Returns 

527 ------- 

528 expr : `str` 

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

530 """ 

531 _LOG.debug( 

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

533 memory, 

534 multiplier, 

535 limit, 

536 additional_expr, 

537 ) 

538 

539 # ctrl_bps sets multiplier to None in the GenericWorkflow if 

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

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

542 return "" 

543 

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

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

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

547 # evaluate to FALSE in this case. 

548 # 

549 # Note: 

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

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

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

553 # but better safe than sorry. 

554 is_held = "JobStatus == 5" 

555 is_retry_allowed = "NumJobStarts <= JobMaxRetries" 

556 

557 mem_expr = "" 

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

559 was_mem_exceeded = ( 

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

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

562 ) 

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

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

565 

566 user_expr = "" 

567 if additional_expr: 

568 # Never auto release a job held by user. 

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

570 

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

572 if user_expr and mem_expr: 

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

574 elif user_expr: 

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

576 elif mem_expr: 

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

578 

579 return expr 

580 

581 

582def _create_periodic_remove_expr(memory, multiplier, limit): 

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

584 

585 Parameters 

586 ---------- 

587 memory : `int` 

588 Requested memory in MB. 

589 multiplier : `float` 

590 Memory growth rate between retries. 

591 limit : `int` 

592 Memory limit. 

593 

594 Returns 

595 ------- 

596 expr : `str` 

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

598 """ 

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

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

601 # The special comparison operators ensure that all comparisons below 

602 # will evaluate to FALSE in this case. 

603 # 

604 # Note: 

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

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

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

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

609 is_held = "JobStatus == 5" 

610 is_retry_disallowed = "NumJobStarts > JobMaxRetries" 

611 

612 mem_expr = "" 

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

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

615 

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

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

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

619 ) 

620 

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

622 return expr 

623 

624 

625def _create_request_memory_expr(memory, multiplier, limit): 

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

627 

628 Parameters 

629 ---------- 

630 memory : `int` 

631 Requested memory in MB. 

632 multiplier : `float` 

633 Memory growth rate between retries. 

634 limit : `int` 

635 Memory limit. 

636 

637 Returns 

638 ------- 

639 expr : `str` 

640 A string representing an HTCondor ClassAd expression enabling safe 

641 memory scaling between job retries. 

642 """ 

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

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

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

646 # the ones describing job's current state. 

647 # 

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

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

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

651 was_mem_exceeded = ( 

652 "LastJobStatus =?= 5 " 

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

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

655 ) 

656 

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

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

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

660 # whichever is greater. 

661 expr = ( 

662 f"({was_mem_exceeded}) " 

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

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

665 ) 

666 return expr 

667 

668 

669def _gather_site_values(config, compute_site): 

670 """Gather values specific to given site. 

671 

672 Parameters 

673 ---------- 

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

675 BPS configuration that includes necessary submit/runtime 

676 information. 

677 compute_site : `str` 

678 Compute site name. 

679 

680 Returns 

681 ------- 

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

683 Values specific to the given site. 

684 """ 

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

686 search_opts = {} 

687 if compute_site: 

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

689 

690 # Determine the hard limit for the memory requirement. 

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

692 if not found: 

693 search_opts["default"] = DEFAULT_HTC_EXEC_PATT 

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

695 del search_opts["default"] 

696 

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

698 # by definition, they cannot have more memory than 

699 # the partitionable slot they are the part of. 

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

701 pool_info = condor_status(constraint=constraint) 

702 try: 

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

704 except ValueError: 

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

706 if limit: 

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

708 

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

710 site_values["memoryLimit"] = limit 

711 

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

713 if found: 

714 site_values["accountingGroup"] = value 

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

716 if found: 

717 site_values["accountingUser"] = value 

718 

719 key = f".site.{compute_site}.profile.condor" 

720 if key in config: 

721 for subkey, val in config[key].items(): 

722 if subkey.startswith("+"): 

723 site_values["attrs"][subkey[1:]] = val 

724 else: 

725 site_values["profile"][subkey] = val 

726 

727 return site_values 

728 

729 

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

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

732 

733 Parameters 

734 ---------- 

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

736 BPS configuration that includes necessary submit/runtime 

737 information. 

738 label : `str` 

739 GenericWorkflowJob label. 

740 

741 Returns 

742 ------- 

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

744 Values specific to the given job label. 

745 """ 

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

747 

748 search_opts = {} 

749 profile_key = "" 

750 if label == "finalJob": 

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

752 profile_key = ".finalJob.profile.condor" 

753 elif label in config["cluster"]: 

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

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

756 elif label in config["pipetask"]: 

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

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

759 

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

761 if found: 

762 values["releaseExpr"] = value 

763 

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

765 if found: 

766 values["overwriteJobFiles"] = value 

767 else: 

768 values["overwriteJobFiles"] = True 

769 

770 if profile_key and profile_key in config: 

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

772 if subkey.startswith("+"): 

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

774 else: 

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

776 

777 return values 

778 

779 

780def _group_to_subdag( 

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

782) -> HTCJob: 

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

784 

785 Parameters 

786 ---------- 

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

788 Workflow configuration. 

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

790 The generic workflow group to convert. 

791 out_prefix : `str` 

792 Location prefix to be used when creating jobs. 

793 

794 Returns 

795 ------- 

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

797 Job for running the HTCondor dag. 

798 """ 

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

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

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

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

803 if not generic_workflow_group.blocking: 

804 htc_job.dagcmds["post"] = { 

805 "defer": "", 

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

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

808 } 

809 return htc_job 

810 

811 

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

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

814 

815 Parameters 

816 ---------- 

817 group_job_name : `str` 

818 Name of the group job. 

819 job_label : `str` 

820 Label to use for the check status job. 

821 

822 Returns 

823 ------- 

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

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

826 """ 

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

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

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

830 

831 return htc_job 

832 

833 

834def _generic_workflow_to_htcondor_dag( 

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

836) -> HTCDag: 

837 """Convert a GenericWorkflow to a HTCDag. 

838 

839 Parameters 

840 ---------- 

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

842 Workflow configuration. 

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

844 The GenericWorkflow to convert. 

845 out_prefix : `str` 

846 Location prefix where the HTCondor files will be written. 

847 

848 Returns 

849 ------- 

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

851 The HTCDag representation of the given GenericWorkflow. 

852 """ 

853 dag = HTCDag(name=generic_workflow.name) 

854 

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

856 dag.add_attribs(generic_workflow.run_attrs) 

857 dag.add_attribs( 

858 { 

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

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

861 } 

862 ) 

863 

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

865 if isinstance(tmp_template, str): 

866 subdir_template = defaultdict(lambda: tmp_template) 

867 else: 

868 subdir_template = tmp_template 

869 

870 # Create all DAG jobs 

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

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

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

874 for job_name in generic_workflow: 

875 gwjob = generic_workflow.get_job(job_name) 

876 if gwjob.node_type == GenericWorkflowNodeType.PAYLOAD: 

877 gwjob = cast(GenericWorkflowJob, gwjob) 

878 if gwjob.compute_site not in site_values: 

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

880 if gwjob.label not in cached_values: 

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

882 cached_values[gwjob.label].update(_gather_label_values(config, gwjob.label)) 

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

884 htc_job = _create_job( 

885 subdir_template[gwjob.label], 

886 cached_values[gwjob.label], 

887 generic_workflow, 

888 gwjob, 

889 out_prefix, 

890 ) 

891 elif gwjob.node_type == GenericWorkflowNodeType.NOOP: 

892 gwjob = cast(GenericWorkflowNoopJob, gwjob) 

893 htc_job = HTCJob(f"wms_{gwjob.name}", label=gwjob.label) 

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

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

896 htc_job.add_dag_cmds({"noop": True}) 

897 elif gwjob.node_type == GenericWorkflowNodeType.GROUP: 

898 gwjob = cast(GenericWorkflowGroup, gwjob) 

899 htc_job = _group_to_subdag(config, gwjob, out_prefix) 

900 else: 

901 raise RuntimeError(f"Unsupported generic workflow node type {gwjob.node_type} ({gwjob.name})") 

902 _LOG.debug("Calling adding job %s %s", htc_job.name, htc_job.label) 

903 dag.add_job(htc_job) 

904 

905 # Add job dependencies to the DAG (be careful with wms_ jobs) 

906 for job_name in generic_workflow: 

907 gwjob = generic_workflow.get_job(job_name) 

908 parent_name = ( 

909 gwjob.name if gwjob.node_type == GenericWorkflowNodeType.PAYLOAD else f"wms_{gwjob.name}"