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 = { 

310 "abort_on_value": "abort_dag_on", 

311 "abort_return_value": "abort_exit", 

312 "priority": "priority", 

313 } 

314 

315 dagcmds = {} 

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

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

318 

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

320 return dagcmds 

321 

322 

323def _fix_env_var_syntax(oldstr): 

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

325 

326 Parameters 

327 ---------- 

328 oldstr : `str` 

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

330 

331 Returns 

332 ------- 

333 newstr : `str` 

334 Given string with environment variable syntax fixed. 

335 """ 

336 newstr = oldstr 

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

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

339 return newstr 

340 

341 

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

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

344 physical file names. 

345 

346 Parameters 

347 ---------- 

348 use_shared : `bool` 

349 Whether HTCondor can assume shared filesystem. 

350 arguments : `str` 

351 Arguments string in which to replace file placeholders. 

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

353 Generic workflow that contains file information. 

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

355 The job corresponding to the arguments. 

356 

357 Returns 

358 ------- 

359 arguments : `str` 

360 Given arguments string with file placeholders replaced. 

361 """ 

362 # Replace input file placeholders with paths. 

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

364 if not gwfile.wms_transfer: 

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

366 # responsible for transferring file. 

367 uri = gwfile.src_uri 

368 elif use_shared: 

369 if gwfile.job_shared: 

370 # Have shared filesystems and jobs can share file. 

371 uri = gwfile.src_uri 

372 else: 

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

374 else: # Using push transfer 

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

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

377 

378 # Replace output file placeholders with paths. 

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

380 if not gwfile.wms_transfer: 

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

382 # responsible for transferring file. 

383 uri = gwfile.src_uri 

384 elif use_shared: 

385 if gwfile.job_shared: 

386 # Have shared filesystems and jobs can share file. 

387 uri = gwfile.src_uri 

388 else: 

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

390 else: # Using push transfer 

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

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

393 return arguments 

394 

395 

396def _replace_cmd_vars(arguments, gwjob): 

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

398 

399 Parameters 

400 ---------- 

401 arguments : `str` 

402 Arguments string in which to replace placeholders. 

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

404 Job containing values to be used to replace placeholders 

405 (in particular gwjob.cmdvals). 

406 

407 Returns 

408 ------- 

409 arguments : `str` 

410 Given arguments string with placeholders replaced. 

411 """ 

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

413 try: 

414 arguments = arguments.format(**replacements) 

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

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

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

418 raise 

419 return arguments 

420 

421 

422def _handle_job_inputs( 

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

424) -> dict[str, str]: 

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

426 

427 Parameters 

428 ---------- 

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

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

431 job_name : `str` 

432 Unique name for the job. 

433 use_shared : `bool` 

434 Whether job has access to files via shared filesystem. 

435 out_prefix : `str` 

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

437 

438 Returns 

439 ------- 

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

441 HTCondor commands for the job submission script. 

442 """ 

443 inputs = [] 

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

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

446 

447 uri = Path(gwf_file.src_uri) 

448 

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

450 

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

452 inputs.append(str(uri)) 

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

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

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

456 if uri.is_dir(): 

457 raise RuntimeError( 

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

459 ) 

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

461 

462 htc_commands = {} 

463 if inputs: 

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

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

466 return htc_commands 

467 

468 

469def _handle_job_outputs( 

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

471) -> dict[str, str]: 

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

473 

474 Parameters 

475 ---------- 

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

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

478 job_name : `str` 

479 Unique name for the job. 

480 use_shared : `bool` 

481 Whether job has access to files via shared filesystem. 

482 out_prefix : `str` 

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

484 

485 Returns 

486 ------- 

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

488 HTCondor commands for the job submission script. 

489 """ 

490 outputs = [] 

491 output_remaps = [] 

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

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

494 

495 uri = Path(gwf_file.src_uri) 

496 if not use_shared: 

497 outputs.append(uri.name) 

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

499 

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

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

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

503 # by the job. 

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

505 if outputs: 

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

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

508 

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

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

511 return htc_commands 

512 

513 

514def _create_periodic_release_expr( 

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

516) -> str: 

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

518 

519 Parameters 

520 ---------- 

521 memory : `int` 

522 Requested memory in MB. 

523 multiplier : `float` or None 

524 Memory growth rate between retries. 

525 limit : `int` 

526 Memory limit. 

527 additional_expr : `str`, optional 

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

529 

530 Returns 

531 ------- 

532 expr : `str` 

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

534 """ 

535 _LOG.debug( 

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

537 memory, 

538 multiplier, 

539 limit, 

540 additional_expr, 

541 ) 

542 

543 # ctrl_bps sets multiplier to None in the GenericWorkflow if 

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

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

546 return "" 

547 

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

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

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

551 # evaluate to FALSE in this case. 

552 # 

553 # Note: 

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

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

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

557 # but better safe than sorry. 

558 is_held = "JobStatus == 5" 

559 is_retry_allowed = "NumJobStarts <= JobMaxRetries" 

560 

561 mem_expr = "" 

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

563 was_mem_exceeded = ( 

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

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

566 ) 

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

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

569 

570 user_expr = "" 

571 if additional_expr: 

572 # Never auto release a job held by user. 

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

574 

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

576 if user_expr and mem_expr: 

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

578 elif user_expr: 

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

580 elif mem_expr: 

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

582 

583 return expr 

584 

585 

586def _create_periodic_remove_expr(memory, multiplier, limit): 

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

588 

589 Parameters 

590 ---------- 

591 memory : `int` 

592 Requested memory in MB. 

593 multiplier : `float` 

594 Memory growth rate between retries. 

595 limit : `int` 

596 Memory limit. 

597 

598 Returns 

599 ------- 

600 expr : `str` 

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

602 """ 

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

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

605 # The special comparison operators ensure that all comparisons below 

606 # will evaluate to FALSE in this case. 

607 # 

608 # Note: 

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

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

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

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

613 is_held = "JobStatus == 5" 

614 is_retry_disallowed = "NumJobStarts > JobMaxRetries" 

615 

616 mem_expr = "" 

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

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

619 

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

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

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

623 ) 

624 

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

626 return expr 

627 

628 

629def _create_request_memory_expr(memory, multiplier, limit): 

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

631 

632 Parameters 

633 ---------- 

634 memory : `int` 

635 Requested memory in MB. 

636 multiplier : `float` 

637 Memory growth rate between retries. 

638 limit : `int` 

639 Memory limit. 

640 

641 Returns 

642 ------- 

643 expr : `str` 

644 A string representing an HTCondor ClassAd expression enabling safe 

645 memory scaling between job retries. 

646 """ 

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

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

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

650 # the ones describing job's current state. 

651 # 

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

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

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

655 was_mem_exceeded = ( 

656 "LastJobStatus =?= 5 " 

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

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

659 ) 

660 

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

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

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

664 # whichever is greater. 

665 expr = ( 

666 f"({was_mem_exceeded}) " 

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

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

669 ) 

670 return expr 

671 

672 

673def _gather_site_values(config, compute_site): 

674 """Gather values specific to given site. 

675 

676 Parameters 

677 ---------- 

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

679 BPS configuration that includes necessary submit/runtime 

680 information. 

681 compute_site : `str` 

682 Compute site name. 

683 

684 Returns 

685 ------- 

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

687 Values specific to the given site. 

688 """ 

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

690 search_opts = {} 

691 if compute_site: 

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

693 

694 # Determine the hard limit for the memory requirement. 

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

696 if not found: 

697 search_opts["default"] = DEFAULT_HTC_EXEC_PATT 

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

699 del search_opts["default"] 

700 

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

702 # by definition, they cannot have more memory than 

703 # the partitionable slot they are the part of. 

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

705 pool_info = condor_status(constraint=constraint) 

706 try: 

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

708 except ValueError: 

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

710 if limit: 

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

712 

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

714 site_values["memoryLimit"] = limit 

715 

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

717 if found: 

718 site_values["accountingGroup"] = value 

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

720 if found: 

721 site_values["accountingUser"] = value 

722 

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

724 if key in config: 

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

726 if subkey.startswith("+"): 

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

728 else: 

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

730 

731 return site_values 

732 

733 

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

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

736 

737 Parameters 

738 ---------- 

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

740 BPS configuration that includes necessary submit/runtime 

741 information. 

742 label : `str` 

743 GenericWorkflowJob label. 

744 

745 Returns 

746 ------- 

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

748 Values specific to the given job label. 

749 """ 

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

751 

752 search_opts = {} 

753 profile_key = "" 

754 if label == "finalJob": 

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

756 profile_key = ".finalJob.profile.condor" 

757 elif label in config["cluster"]: 

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

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

760 elif label in config["pipetask"]: 

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

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

763 

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

765 if found: 

766 values["releaseExpr"] = value 

767 

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

769 if found: 

770 values["overwriteJobFiles"] = value 

771 else: 

772 values["overwriteJobFiles"] = True 

773 

774 if profile_key and profile_key in config: 

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

776 if subkey.startswith("+"): 

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

778 else: 

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

780 

781 return values 

782 

783 

784def _group_to_subdag( 

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

786) -> HTCJob: 

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

788 

789 Parameters 

790 ---------- 

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

792 Workflow configuration. 

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

794 The generic workflow group to convert. 

795 out_prefix : `str` 

796 Location prefix to be used when creating jobs. 

797 

798 Returns 

799 ------- 

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

801 Job for running the HTCondor dag. 

802 """ 

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

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

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

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

807 if not generic_workflow_group.blocking: 

808 htc_job.dagcmds["post"] = { 

809 "defer": "", 

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

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

812 } 

813 return htc_job 

814 

815 

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

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

818 

819 Parameters 

820 ---------- 

821 group_job_name : `str` 

822 Name of the group job. 

823 job_label : `str` 

824 Label to use for the check status job. 

825 

826 Returns 

827 ------- 

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

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

830 """ 

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

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

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

834 

835 return htc_job 

836 

837 

838def _generic_workflow_to_htcondor_dag( 

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

840) -> HTCDag: 

841 """Convert a GenericWorkflow to a HTCDag. 

842 

843 Parameters 

844 ---------- 

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

846 Workflow configuration. 

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

848 The GenericWorkflow to convert. 

849 out_prefix : `str` 

850 Location prefix where the HTCondor files will be written. 

851 

852 Returns 

853 ------- 

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

855 The HTCDag representation of the given GenericWorkflow. 

856 """ 

857 dag = HTCDag(name=generic_workflow.name) 

858 

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

860 dag.add_attribs(generic_workflow.run_attrs) 

861 dag.add_attribs( 

862 { 

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

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

865 } 

866 ) 

867 

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

869 if isinstance(tmp_template, str): 

870 subdir_template = defaultdict(lambda: tmp_template) 

871 else: 

872 subdir_template = tmp_template 

873 

874 # Create all DAG jobs 

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

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

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

878 for job_name in generic_workflow: 

879 gwjob = generic_workflow.get_job(job_name) 

880 if gwjob.node_type == GenericWorkflowNodeType.PAYLOAD: 

881 gwjob = cast(GenericWorkflowJob, gwjob) 

882 if gwjob.compute_site not in site_values: 

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

884 if gwjob.label not in cached_values: 

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

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

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

888 htc_job = _create_job( 

889 subdir_template[gwjob.label], 

890 cached_values[gwjob.label], 

891 generic_workflow, 

892 gwjob, 

893 out_prefix, 

894 ) 

895 elif gwjob.node_type == GenericWorkflowNodeType.NOOP: 

896 gwjob = cast(GenericWorkflowNoopJob, gwjob) 

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

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

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

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

901 elif gwjob.node_type == GenericWorkflowNodeType.GROUP: 

902 gwjob = cast(GenericWorkflowGroup, gwjob) 

903 htc_job = _group_to_subdag(config, gwjob, out_prefix) 

904 else: 

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

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

907 dag.add_job(htc_job) 

908 

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

910 for job_name in generic_workflow: 

911 gwjob = generic_workflow.get_job(job_name)