Coverage for python/lsst/ctrl/bps/transform.py: 85%

1# This file is part of ctrl_bps. 

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"""Driver for the transformation of a QuantumGraph into a generic workflow.""" 

29 

30import copy 

31import dataclasses 

32import logging 

33import math 

34import os 

35import re 

36from typing import Any 

37 

38from lsst.ctrl.bps import ClusteredQuantumGraph 

39from lsst.daf.butler import Config 

40from lsst.pipe.base import QuantumGraph 

41from lsst.utils.logging import VERBOSE 

42from lsst.utils.timer import timeMethod 

43 

44from . import ( 

45 BPS_NONE, 

46 BPS_SEARCH_ORDER, 

47 DEFAULT_MEM_RETRIES, 

48 BpsConfig, 

49 GenericWorkflow, 

50 GenericWorkflowExec, 

51 GenericWorkflowFile, 

52 GenericWorkflowJob, 

53) 

54from .bps_utils import ( 

55 WhenToSaveQuantumGraphs, 

56 create_job_quantum_graph_filename, 

57 save_qg_subgraph, 

58) 

59 

60# All available job attributes. 

61_ATTRS_ALL = frozenset([field.name for field in dataclasses.fields(GenericWorkflowJob)]) 

62 

63# Job attributes that need to be set to their maximal value in the cluster. 

64_ATTRS_MAX = frozenset( 

65 { 

66 "memory_multiplier", 

67 "number_of_retries", 

68 "request_cpus", 

69 "request_memory", 

70 "request_memory_max", 

71 } 

72) 

73 

74# Job attributes that need to be set to sum of their values in the cluster. 

75_ATTRS_SUM = frozenset( 

76 { 

77 "request_disk", 

78 "request_walltime", 

79 } 

80) 

81 

82# Job attributes do not fall into a specific category 

83_ATTRS_MISC = frozenset( 

84 { 

85 "label", # taskDef labels aren't same in job and may not match job label 

86 "cmdvals", 

87 "profile", 

88 "attrs", 

89 } 

90) 

91 

92# Attributes that need to be the same for each quanta in the cluster. 

93_ATTRS_UNIVERSAL = frozenset(_ATTRS_ALL - (_ATTRS_MAX | _ATTRS_MISC | _ATTRS_SUM)) 

94 

95_LOG = logging.getLogger(__name__) 

96 

97 

98@timeMethod(logger=_LOG, logLevel=VERBOSE) 

99def transform( 

100 config: BpsConfig, cqgraph: ClusteredQuantumGraph, prefix: str 

101) -> tuple[GenericWorkflow, BpsConfig]: 

102 """Transform a ClusteredQuantumGraph to a GenericWorkflow. 

103 

104 Parameters 

105 ---------- 

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

107 BPS configuration. 

108 cqgraph : `lsst.ctrl.bps.ClusteredQuantumGraph` 

109 A clustered quantum graph to transform into a generic workflow. 

110 prefix : `str` 

111 Root path for any output files. 

112 

113 Returns 

114 ------- 

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

116 The generic workflow transformed from the clustered quantum graph. 

117 generic_workflow_config : `lsst.ctrl.bps.BpsConfig` 

118 Configuration to accompany GenericWorkflow. 

119 """ 

120 if cqgraph.name is not None: 

121 name = cqgraph.name 

122 else: 

123 _, name = config.search("uniqProcName", opt={"required": True}) 

124 

125 generic_workflow = create_generic_workflow(config, cqgraph, name, prefix) 

126 generic_workflow_config = create_generic_workflow_config(config, prefix) 

127 

128 return generic_workflow, generic_workflow_config 

129 

130 

131def add_workflow_init_nodes(config, qgraph, generic_workflow): 

132 """Add nodes to workflow graph that perform initialization steps. 

133 

134 Assumes that all of the initialization should be executed prior to any 

135 of the current workflow. 

136 

137 Parameters 

138 ---------- 

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

140 BPS configuration. 

141 qgraph : `lsst.pipe.base.graph.QuantumGraph` 

142 The quantum graph the generic workflow represents. 

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

144 Generic workflow to which the initialization steps should be added. 

145 """ 

146 # Create a workflow graph that will have task and file nodes necessary for 

147 # initializing the pipeline execution 

148 init_workflow = create_init_workflow(config, qgraph, generic_workflow.get_file("runQgraphFile")) 

149 _LOG.debug("init_workflow nodes = %s", init_workflow.nodes()) 

150 generic_workflow.add_workflow_source(init_workflow) 

151 

152 

153def create_init_workflow( 

154 config: BpsConfig, qgraph: QuantumGraph, qgraph_gwfile: GenericWorkflowFile 

155) -> GenericWorkflow: 

156 """Create workflow for running initialization job(s). 

157 

158 Parameters 

159 ---------- 

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

161 BPS configuration. 

162 qgraph : `lsst.pipe.base.graph.QuantumGraph` 

163 The quantum graph the generic workflow represents. 

164 qgraph_gwfile : `lsst.ctrl.bps.GenericWorkflowFile` 

165 File object for the full run QuantumGraph file. 

166 

167 Returns 

168 ------- 

169 init_workflow : `lsst.ctrl.bps.GenericWorkflow` 

170 GenericWorkflow consisting of job(s) to initialize workflow. 

171 """ 

172 _LOG.debug("creating init subgraph") 

173 _LOG.debug("creating init task input(s)") 

174 search_opt = { 

175 "curvals": {"curr_pipetask": "pipetaskInit"}, 

176 "replaceVars": False, 

177 "expandEnvVars": False, 

178 "replaceEnvVars": True, 

179 "required": False, 

180 } 

181 found, value = config.search("computeSite", opt=search_opt) 

182 if found: 182 ↛ 184line 182 didn't jump to line 184 because the condition on line 182 was always true

183 search_opt["curvals"]["curr_site"] = value 

184 found, value = config.search("computeCloud", opt=search_opt) 

185 if found: 185 ↛ 186line 185 didn't jump to line 186 because the condition on line 185 was never true

186 search_opt["curvals"]["curr_cloud"] = value 

187 

188 init_workflow = GenericWorkflow("init") 

189 init_workflow.add_file(qgraph_gwfile) 

190 

191 # create job for executing --init-only 

192 gwjob = GenericWorkflowJob("pipetaskInit", "pipetaskInit") 

193 

194 job_values = _get_job_values(config, search_opt, "runQuantumCommand") 

195 job_values["name"] = "pipetaskInit" 

196 job_values["label"] = "pipetaskInit" 

197 

198 # Adjust job attributes values if necessary. 

199 _handle_job_values(job_values, gwjob) 

200 

201 init_workflow.add_job(gwjob) 

202 init_workflow.add_job_inputs(gwjob.name, [qgraph_gwfile]) 

203 _enhance_command(config, init_workflow, gwjob, {}) 

204 

205 return init_workflow 

206 

207 

208def _enhance_command(config, generic_workflow, gwjob, cached_job_values): 

209 """Enhance command line with env and file placeholders 

210 and gather command line values. 

211 

212 Parameters 

213 ---------- 

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

215 BPS configuration. 

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

217 Generic workflow that contains the job. 

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

219 Generic workflow job to which the updated executable, arguments, 

220 and values should be saved. 

221 cached_job_values : `dict` [`str`, dict[`str`, `~typing.Any`]] 

222 Cached values common across jobs with same label. Updated if values 

223 aren't already saved for given gwjob's label. 

224 """ 

225 _LOG.debug("gwjob given to _enhance_command: %s", gwjob) 

226 

227 search_opt = config.get_search_opts(gwjob.label) 

228 

229 search_opt["curvals"]["jobName"] = gwjob.name 

230 search_opt["curvals"]["jobLabel"] = gwjob.label 

231 for key, value in gwjob.tags.items(): 

232 search_opt["curvals"][key] = value 

233 

234 search_opt.update( 

235 { 

236 "replaceVars": False, 

237 "expandEnvVars": False, 

238 "replaceEnvVars": True, 

239 "required": False, 

240 } 

241 ) 

242 

243 if gwjob.label not in cached_job_values: 

244 cached_job_values[gwjob.label] = {} 

245 # Allowing whenSaveJobQgraph and useLazyCommands per pipetask label. 

246 key = "whenSaveJobQgraph" 

247 _, when_save = config.search(key, opt=search_opt) 

248 cached_job_values[gwjob.label][key] = WhenToSaveQuantumGraphs[when_save.upper()] 

249 

250 key = "useLazyCommands" 

251 search_opt["default"] = True 

252 _, cached_job_values[gwjob.label][key] = config.search(key, opt=search_opt) 

253 del search_opt["default"] 

254 

255 # Change qgraph variable to match whether using run or per-job qgraph 

256 # Note: these are lookup keys, not actual physical filenames. 

257 if cached_job_values[gwjob.label]["whenSaveJobQgraph"] == WhenToSaveQuantumGraphs.NEVER: 

258 gwjob.arguments = gwjob.arguments.replace("{qgraphFile}", "{runQgraphFile}") 

259 elif gwjob.name == "pipetaskInit": 259 ↛ 260line 259 didn't jump to line 260 because the condition on line 259 was never true

260 gwjob.arguments = gwjob.arguments.replace("{qgraphFile}", "{runQgraphFile}") 

261 else: # Needed unique file keys for per-job QuantumGraphs 

262 gwjob.arguments = gwjob.arguments.replace("{qgraphFile}", f"{{qgraphFile_{gwjob.name}}}") 

263 

264 # Replace files with special placeholders 

265 for gwfile in generic_workflow.get_job_inputs(gwjob.name): 

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

267 for gwfile in generic_workflow.get_job_outputs(gwjob.name): 

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

269 

270 # Replace wms variables with wms placeholders. 

271 gwjob.arguments = re.sub( 

272 r"{wms([^}]+)}", lambda x: f"<WMS:{x[1][0].lower() + x[1][1:]}>", gwjob.arguments 

273 ) 

274 

275 # Save dict of other values needed to complete command line. 

276 # (Be careful to not replace env variables as they may 

277 # be different in compute job.) 

278 search_opt["replaceVars"] = True 

279 _LOG.debug("before cmdvals = %s (search_opt = %s)", gwjob.cmdvals, search_opt) 

280 for key in re.findall(r"{([^}]+)}", gwjob.arguments): 

281 _LOG.debug("looking for %s in cmdvals", key) 

282 if key in gwjob.cmdvals: 

283 continue 

284 elif key in cached_job_values[gwjob.label]: 

285 gwjob.cmdvals[key] = cached_job_values[gwjob.label][key] 

286 else: 

287 _, gwjob.cmdvals[key] = config.search(key, opt=search_opt) 

288 _LOG.debug("after cmdvals = %s", gwjob.cmdvals) 

289 

290 # backwards compatibility 

291 if not cached_job_values[gwjob.label]["useLazyCommands"]: 291 ↛ 292line 291 didn't jump to line 292 because the condition on line 291 was never true

292 if "bpsUseShared" not in cached_job_values[gwjob.label]: 

293 key = "bpsUseShared" 

294 search_opt["default"] = True 

295 _, cached_job_values[gwjob.label][key] = config.search(key, opt=search_opt) 

296 del search_opt["default"] 

297 

298 gwjob.arguments = _fill_arguments( 

299 cached_job_values[gwjob.label]["bpsUseShared"], generic_workflow, gwjob.arguments, gwjob.cmdvals 

300 ) 

301 

302 

303def _fill_arguments(use_shared, generic_workflow, arguments, cmdvals): 

304 """Replace placeholders in command line string in job. 

305 

306 Parameters 

307 ---------- 

308 use_shared : `bool` 

309 Whether using shared filesystem. 

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

311 Generic workflow containing the job. 

312 arguments : `str` 

313 String containing placeholders. 

314 cmdvals : `dict` [`str`, `~typing.Any`] 

315 Any command line values that can be used to replace placeholders. 

316 

317 Returns 

318 ------- 

319 arguments : `str` 

320 Command line with FILE and ENV placeholders replaced. 

321 """ 

322 # Replace file placeholders 

323 for file_key in re.findall(r"<FILE:([^>]+)>", arguments): 

324 gwfile = generic_workflow.get_file(file_key) 

325 if not gwfile.wms_transfer: 

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

327 # responsible for transferring file. 

328 uri = gwfile.src_uri 

329 elif use_shared: 

330 if gwfile.job_shared: 

331 # Have shared filesystems and jobs can share file. 

332 uri = gwfile.src_uri 

333 else: 

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

335 else: # Using push transfer 

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

337 

338 arguments = arguments.replace(f"<FILE:{file_key}>", uri) 

339 

340 # Replace env placeholder with submit-side values 

341 arguments = re.sub(r"<ENV:([^>]+)>", r"$\1", arguments) 

342 arguments = os.path.expandvars(arguments) 

343 

344 # Replace remaining vars 

345 arguments = arguments.format(**cmdvals) 

346 

347 return arguments 

348 

349 

350def _get_qgraph_gwfile(config, save_qgraph_per_job, gwjob, run_qgraph_file, prefix): 

351 """Get qgraph location to be used by job. 

352 

353 Parameters 

354 ---------- 

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

356 Bps configuration. 

357 save_qgraph_per_job : `lsst.ctrl.bps.bps_utils.WhenToSaveQuantumGraphs` 

358 What submission stage to save per-job qgraph files (or NEVER) 

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

360 Job for which determining QuantumGraph file. 

361 run_qgraph_file : `lsst.ctrl.bps.GenericWorkflowFile` 

362 File representation of the full run QuantumGraph. 

363 prefix : `str` 

364 Path prefix for any files written. 

365 

366 Returns 

367 ------- 

368 gwfile : `lsst.ctrl.bps.GenericWorkflowFile` 

369 Representation of butler location (may not include filename). 

370 """ 

371 qgraph_gwfile = None 

372 if save_qgraph_per_job != WhenToSaveQuantumGraphs.NEVER: 372 ↛ 373line 372 didn't jump to line 373 because the condition on line 372 was never true

373 qgraph_gwfile = GenericWorkflowFile( 

374 f"qgraphFile_{gwjob.name}", 

375 src_uri=create_job_quantum_graph_filename(config, gwjob, prefix), 

376 wms_transfer=True, 

377 job_access_remote=True, 

378 job_shared=True, 

379 ) 

380 else: 

381 qgraph_gwfile = run_qgraph_file 

382 

383 return qgraph_gwfile 

384 

385 

386def _get_job_values(config, search_opt, cmd_line_key): 

387 """Gather generic workflow job values from the bps config. 

388 

389 Parameters 

390 ---------- 

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

392 Bps configuration. 

393 search_opt : `dict` [`str`, `~typing.Any`] 

394 Search options to be used when searching config. 

395 cmd_line_key : `str` or None 

396 Which command line key to search for (e.g., "runQuantumCommand"). 

397 

398 Returns 

399 ------- 

400 job_values : `dict` [ `str`, `~typing.Any` ]` 

401 A mapping between job attributes and their values. 

402 """ 

403 _LOG.debug("cmd_line_key=%s, search_opt=%s", cmd_line_key, search_opt) 

404 

405 # Create a dummy job to easily access the default values. 

406 default_gwjob = GenericWorkflowJob("default_job", "default_label") 

407 

408 job_values = {} 

409 for attr in _ATTRS_ALL: 

410 # Variable names in yaml are camel case instead of snake case. 

411 yaml_name = re.sub(r"_(\S)", lambda match: match.group(1).upper(), attr) 

412 found, value = config.search(yaml_name, opt=search_opt) 

413 if found: 

414 job_values[attr] = value 

415 else: 

416 job_values[attr] = getattr(default_gwjob, attr) 

417 

418 job_values["environment"] = gather_job_environment(config, search_opt) 

419 

420 # If the automatic memory scaling is enabled (i.e. the memory multiplier 

421 # is set and it is a positive number greater than 1.0), adjust number 

422 # of retries when necessary. If the memory multiplier is invalid, disable 

423 # automatic memory scaling. 

424 if job_values["memory_multiplier"] is not None: 

425 if math.ceil(float(job_values["memory_multiplier"])) > 1: 

426 if job_values["number_of_retries"] is None: 

427 job_values["number_of_retries"] = DEFAULT_MEM_RETRIES 

428 else: 

429 job_values["memory_multiplier"] = None 

430 

431 if cmd_line_key: 

432 found, cmdline = config.search(cmd_line_key, opt=search_opt) 

433 # Make sure cmdline isn't None as that could be sent in as a 

434 # default value in search_opt. 

435 if found and cmdline: 

436 cmd, args = cmdline.split(" ", 1) 

437 job_values["executable"] = GenericWorkflowExec(os.path.basename(cmd), cmd, False) 

438 if args: 438 ↛ 441line 438 didn't jump to line 441 because the condition on line 438 was always true

439 job_values["arguments"] = args 

440 

441 return job_values 

442 

443 

444def _handle_job_values(quantum_job_values, gwjob, attributes=_ATTRS_ALL): 

445 """Set the job attributes in the cluster to their correct values. 

446 

447 Parameters 

448 ---------- 

449 quantum_job_values : `dict` [`str`, Any] 

450 Job values for running single Quantum. 

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

452 Generic workflow job in which to store the universal values. 

453 attributes : `~collections.abc.Iterable` [`str`], optional 

454 Job attributes to be set in the job following different rules. 

455 The default value is _ATTRS_ALL. 

456 """ 

457 _LOG.debug("Call to _handle_job_values") 

458 _handle_job_values_universal(quantum_job_values, gwjob, attributes) 

459 _handle_job_values_max(quantum_job_values, gwjob, attributes) 

460 _handle_job_values_sum(quantum_job_values, gwjob, attributes) 

461 

462 

463def _handle_job_values_universal(quantum_job_values, gwjob, attributes=_ATTRS_UNIVERSAL): 

464 """Handle job attributes that must have the same value for every quantum 

465 in the cluster. 

466 

467 Parameters 

468 ---------- 

469 quantum_job_values : `dict` [`str`, Any] 

470 Job values for running single Quantum. 

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

472 Generic workflow job in which to store the universal values. 

473 attributes : `~collections.abc.Iterable` [`str`], optional 

474 Job attributes to be set in the job following different rules. 

475 The default value is _ATTRS_UNIVERSAL. 

476 """ 

477 for attr in _ATTRS_UNIVERSAL & set(attributes): 

478 _LOG.debug( 

479 "Handling job %s (job=%s, quantum=%s)", 

480 attr, 

481 getattr(gwjob, attr), 

482 quantum_job_values.get(attr, "MISSING"), 

483 ) 

484 current_value = getattr(gwjob, attr) 

485 try: 

486 quantum_value = quantum_job_values[attr] 

487 except KeyError: 

488 continue 

489 else: 

490 if not current_value: 

491 setattr(gwjob, attr, quantum_value) 

492 elif current_value != quantum_value: 492 ↛ 493line 492 didn't jump to line 493 because the condition on line 492 was never true

493 _LOG.error( 

494 "Inconsistent value for %s in Cluster %s Quantum Number %s\n" 

495 "Current cluster value: %s\n" 

496 "Quantum value: %s", 

497 attr, 

498 gwjob.name, 

499 quantum_job_values.get("qgraphNodeId", "MISSING"), 

500 current_value, 

501 quantum_value, 

502 ) 

503 raise RuntimeError(f"Inconsistent value for {attr} in cluster {gwjob.name}.") 

504 

505 

506def _handle_job_values_max(quantum_job_values, gwjob, attributes=_ATTRS_MAX): 

507 """Handle job attributes that should be set to their maximum value in 

508 the in cluster. 

509 

510 Parameters 

511 ---------- 

512 quantum_job_values : `dict` [`str`, `~typing.Any`] 

513 Job values for running single Quantum. 

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

515 Generic workflow job in which to store the aggregate values. 

516 attributes : `~collections.abc.Iterable` [`str`], optional 

517 Job attributes to be set in the job following different rules. 

518 The default value is _ATTR_MAX. 

519 """ 

520 for attr in _ATTRS_MAX & set(attributes): 

521 current_value = getattr(gwjob, attr) 

522 try: 

523 quantum_value = quantum_job_values[attr] 

524 except KeyError: 

525 continue 

526 else: 

527 needs_update = False 

528 if current_value is None: 

529 if quantum_value is not None: 

530 needs_update = True 

531 else: 

532 if quantum_value is not None and current_value < quantum_value: 

533 needs_update = True 

534 if needs_update: 

535 setattr(gwjob, attr, quantum_value) 

536 

537 # When updating memory requirements for a job, check if memory 

538 # autoscaling is enabled. If it is, always use the memory 

539 # multiplier and the number of retries which comes with the 

540 # quantum. 

541 # 

542 # Note that as a result, the quantum with the biggest memory 

543 # requirements will determine whether the memory autoscaling 

544 # will be enabled (or disabled) depending on the value of its 

545 # memory multiplier. 

546 if attr == "request_memory": 

547 gwjob.memory_multiplier = quantum_job_values["memory_multiplier"] 

548 if gwjob.memory_multiplier is not None: 

549 gwjob.number_of_retries = quantum_job_values["number_of_retries"] 

550 

551 

552def _handle_job_values_sum(quantum_job_values, gwjob, attributes=_ATTRS_SUM): 

553 """Handle job attributes that are the sum of their values in the cluster. 

554 

555 Parameters 

556 ---------- 

557 quantum_job_values : `dict` [`str`, `~typing.Any`] 

558 Job values for running single Quantum. 

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

560 Generic workflow job in which to store the aggregate values. 

561 attributes : `~collections.abc.Iterable` [`str`], optional 

562 Job attributes to be set in the job following different rules. 

563 The default value is _ATTRS_SUM. 

564 """ 

565 for attr in _ATTRS_SUM & set(attributes): 

566 current_value = getattr(gwjob, attr) 

567 if not current_value: 567 ↛ 570line 567 didn't jump to line 570 because the condition on line 567 was always true

568 setattr(gwjob, attr, quantum_job_values[attr]) 

569 else: 

570 setattr(gwjob, attr, current_value + quantum_job_values[attr]) 

571 

572 

573def create_generic_workflow( 

574 config: BpsConfig, cqgraph: ClusteredQuantumGraph, name: str, prefix: str 

575) -> GenericWorkflow: 

576 """Create a generic workflow from a ClusteredQuantumGraph such that it 

577 has information needed for WMS (e.g., command lines). 

578 

579 Parameters 

580 ---------- 

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

582 BPS configuration. 

583 cqgraph : `lsst.ctrl.bps.ClusteredQuantumGraph` 

584 ClusteredQuantumGraph for running a specific pipeline on a specific 

585 payload. 

586 name : `str` 

587 Name for the workflow (typically unique). 

588 prefix : `str` 

589 Root path for any output files. 

590 

591 Returns 

592 ------- 

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

594 Generic workflow for the given ClusteredQuantumGraph + config. 

595 """ 

596 # Determine whether saving per-job QuantumGraph files in the loop. 

597 _, when_save = config.search("whenSaveJobQgraph", {"default": WhenToSaveQuantumGraphs.TRANSFORM.name}) 

598 save_qgraph_per_job = WhenToSaveQuantumGraphs[when_save.upper()] 

599 

600 generic_workflow = GenericWorkflow(name) 

601 

602 # Save full run QuantumGraph for use by jobs 

603 generic_workflow.add_file( 

604 GenericWorkflowFile( 

605 "runQgraphFile", 

606 src_uri=config["runQgraphFile"], 

607 wms_transfer=True, 

608 job_access_remote=True, 

609 job_shared=True, 

610 ) 

611 ) 

612 

613 # Cache pipetask specific or more generic job values to minimize number 

614 # on config searches. 

615 cached_job_values = {} 

616 cached_pipetask_values = {} 

617 

618 for cluster in cqgraph.clusters(): 

619 _LOG.debug("Loop over clusters: %s, %s", cluster, type(cluster)) 

620 _LOG.debug( 

621 "cqgraph: name=%s, len=%s, label=%s, ids=%s", 

622 cluster.name, 

623 len(cluster.qgraph_node_ids), 

624 cluster.label, 

625 cluster.qgraph_node_ids, 

626 ) 

627 

628 gwjob = GenericWorkflowJob(cluster.name, cluster.label) 

629 

630 # First get job values from cluster or cluster config 

631 search_opt = config.get_search_opts(cluster.label) 

632 search_opt.update( 

633 { 

634 "replaceVars": False, 

635 "expandEnvVars": False, 

636 "replaceEnvVars": True, 

637 "required": False, 

638 } 

639 ) 

640 

641 # If some config values are set for this cluster 

642 if cluster.label not in cached_job_values: 

643 _LOG.debug("config['cluster'][%s] = %s", cluster.label, config["cluster"][cluster.label]) 

644 cached_job_values[cluster.label] = {} 

645 

646 # Allowing whenSaveJobQgraph and useLazyCommands per cluster label. 

647 key = "whenSaveJobQgraph" 

648 _, when_save = config.search(key, opt=search_opt) 

649 cached_job_values[cluster.label][key] = WhenToSaveQuantumGraphs[when_save.upper()] 

650 

651 key = "useLazyCommands" 

652 search_opt["default"] = True 

653 _, cached_job_values[cluster.label][key] = config.search(key, opt=search_opt) 

654 del search_opt["default"] 

655 

656 if cluster.label in config["cluster"]: 

657 # Don't want to get global defaults here so only look in 

658 # cluster section. 

659 cached_job_values[cluster.label].update( 

660 _get_job_values(config["cluster"][cluster.label], search_opt, "runQuantumCommand") 

661 ) 

662 cluster_job_values = copy.copy(cached_job_values[cluster.label]) 

663 # Environment is special because of the way it is merged. 

664 # It needs to be set in the cached_pipetask_values, so 

665 # don't include it here. 

666 cluster_job_values.pop("environment", None) 

667 

668 cluster_job_values["name"] = cluster.name 

669 cluster_job_values["label"] = cluster.label 

670 cluster_job_values["quanta_counts"] = cluster.quanta_counts 

671 cluster_job_values["tags"] = cluster.tags 

672 _LOG.debug("cluster_job_values = %s", cluster_job_values) 

673 _handle_job_values(cluster_job_values, gwjob, cluster_job_values.keys()) 

674 _LOG.debug( 

675 "After _handle_job_values cluster %s: gwjob.environment = %s", cluster.label, gwjob.environment 

676 ) 

677 _LOG.debug( 

678 "After _handle_job_values cluster %s: gwjob.arguments = %s", cluster.label, gwjob.arguments 

679 ) 

680 

681 # For purposes of whether to continue searching for a value is whether 

682 # the value evaluates to False. 

683 unset_attributes = {attr for attr in _ATTRS_ALL if not getattr(gwjob, attr)} 

684 

685 _LOG.debug("unset_attributes=%s", unset_attributes) 

686 _LOG.debug("set=%s", _ATTRS_ALL - unset_attributes) 

687 

688 # For job info not defined at cluster level, attempt to get job info 

689 # either common or aggregate for all Quanta in cluster. 

690 for node_id in iter(cluster.qgraph_node_ids): 

691 _LOG.debug("node_id=%s", node_id) 

692 quantum_info = cqgraph.get_quantum_info(node_id) 

693 

694 task_label = quantum_info["task_label"] 

695 if task_label not in cached_pipetask_values: 

696 search_opt["curvals"]["curr_pipetask"] = task_label 

697 cached_pipetask_values[task_label] = _get_job_values(config, search_opt, "runQuantumCommand") 

698 _LOG.debug( 

699 "cached_pipetask_values[%s]['environment'] = %s", 

700 task_label, 

701 cached_pipetask_values[task_label].get("environment", None), 

702 ) 

703 _LOG.debug( 

704 "cached_pipetask_values[%s]['arguments'] = %s", 

705 task_label, 

706 cached_pipetask_values[task_label].get("arguments", None), 

707 ) 

708 _handle_job_values(cached_pipetask_values[task_label], gwjob, unset_attributes) 

709 _LOG.debug("After _handle_job_values pipetask: gwjob.environment = %s", gwjob.environment) 

710 _LOG.debug("After _handle_job_values pipetask: gwjob.arguments = %s", gwjob.arguments) 

711 

712 # Update job with workflow attribute and profile values. 

713 qgraph_gwfile = _get_qgraph_gwfile( 

714 config, save_qgraph_per_job, gwjob, generic_workflow.get_file("runQgraphFile"), prefix 

715 ) 

716 

717 generic_workflow.add_job(gwjob) 

718 generic_workflow.add_job_inputs(gwjob.name, [qgraph_gwfile]) 

719 

720 gwjob.cmdvals["qgraphNodeId"] = ",".join( 

721 sorted([f"{node_id}" for node_id in cluster.qgraph_node_ids]) 

722 ) 

723 _LOG.debug("Before _enhance_command: gwjob.arguments = %s", gwjob.arguments) 

724 _LOG.debug("Before _enhance_command: gwjob.cmdvals = %s", gwjob.cmdvals) 

725 _enhance_command(config, generic_workflow, gwjob, cached_job_values) 

726 _LOG.debug("After _enhance_command: gwjob.environment = %s", gwjob.environment) 

727 

728 # If writing per-job QuantumGraph files during TRANSFORM stage, 

729 # write it now while in memory. 

730 if save_qgraph_per_job == WhenToSaveQuantumGraphs.TRANSFORM: 730 ↛ 731line 730 didn't jump to line 731 because the condition on line 730 was never true

731 save_qg_subgraph(cqgraph.qgraph, qgraph_gwfile.src_uri, cluster.qgraph_node_ids) 

732 

733 # Create job dependencies. 

734 for parent in cqgraph.clusters(): 

735 for child in cqgraph.successors(parent): 

736 generic_workflow.add_job_relationships(parent.name, child.name) 

737 

738 # Add initial workflow. 

739 if config.get("runInit", "{default: False}"): 739 ↛ 742line 739 didn't jump to line 742 because the condition on line 739 was always true

740 add_workflow_init_nodes(config, cqgraph.qgraph, generic_workflow) 

741 

742 generic_workflow.run_attrs.update( 

743 { 

744 "bps_isjob": "True", 

745 "bps_project": config["project"], 

746 "bps_campaign": config["campaign"], 

747 "bps_run": generic_workflow.name, 

748 "bps_operator": config["operator"], 

749 "bps_payload": config["payloadName"], 

750 "bps_runsite": config["computeSite"], 

751 } 

752 ) 

753 

754 # Add final job 

755 add_final_job(config, generic_workflow, prefix) 

756 

757 if "ordering" in config: 757 ↛ 758line 757 didn't jump to line 758 because the condition on line 757 was never true

758 generic_workflow.add_special_job_ordering(config["ordering"]) 

759 

760 return generic_workflow 

761 

762 

763def create_generic_workflow_config(config, prefix): 

764 """Create generic workflow configuration. 

765 

766 Parameters 

767 ---------- 

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

769 Bps configuration. 

770 prefix : `str` 

771 Root path for any output files. 

772 

773 Returns 

774 ------- 

775 generic_workflow_config : `lsst.ctrl.bps.BpsConfig` 

776 Configuration accompanying the GenericWorkflow. 

777 """ 

778 generic_workflow_config = BpsConfig(config) 

779 generic_workflow_config["workflowName"] = config["uniqProcName"] 

780 generic_workflow_config["workflowPath"] = prefix 

781 return generic_workflow_config 

782 

783 

784def add_final_job(config: BpsConfig, generic_workflow: GenericWorkflow, prefix: str) -> None: 

785 """Add final workflow job depending upon configuration. 

786 

787 Depending on configuration, the final job will be added as a special job 

788 which will always run regardless of the exit status of the workflow or 

789 a regular sink node which will only run if the workflow execution finished 

790 with no errors. 

791 

792 Parameters 

793 ---------- 

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

795 Bps configuration. 

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

797 Generic workflow to which attributes should be added. 

798 prefix : `str` 

799 Directory in which to output final script. 

800 """ 

801 _, when_run = config.search(".finalJob.whenRun") 

802 if when_run.upper() != "NEVER": 802 ↛ exitline 802 didn't return from function 'add_final_job' because the condition on line 802 was always true

803 gwjob = create_final_job(config, generic_workflow, prefix) 

804 if when_run.upper() == "ALWAYS": 804 ↛ 806line 804 didn't jump to line 806 because the condition on line 804 was always true

805 generic_workflow.add_final(gwjob) 

806 elif when_run.upper() == "SUCCESS": 

807 add_final_job_as_sink(generic_workflow, gwjob) 

808 else: 

809 raise ValueError(f"Invalid value for finalJob.whenRun: {when_run}") 

810 

811 

812def create_final_job(config: BpsConfig, generic_workflow: GenericWorkflow, prefix: str) -> GenericWorkflowJob: 

813 """Create the final workflow job. 

814 

815 Parameters 

816 ---------- 

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

818 Bps configuration. 

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

820 Generic workflow to which attributes should be added. 

821 prefix : `str` 

822 Directory in which to output final script. 

823 

824 Returns 

825 ------- 

826 final_job : `lsst.ctrl.bps.GenericWorkflowJob` 

827 Final workflow job. 

828 """ 

829 job_name = "finalJob" 

830 gwjob = GenericWorkflowJob(job_name, job_name) 

831 

832 search_opt = {"searchobj": config[job_name], "curvals": {}, "default": None} 

833 found, value = config.search("computeSite", opt=search_opt) 

834 if found: 834 ↛ 836line 834 didn't jump to line 836 because the condition on line 834 was always true

835 search_opt["curvals"]["curr_site"] = value 

836 found, value = config.search("computeCloud", opt=search_opt) 

837 if found: 837 ↛ 846line 837 didn't jump to line 846 because the condition on line 837 was always true

838 search_opt["curvals"]["curr_cloud"] = value 

839 

840 # Set job attributes based on the values find in the config excluding 

841 # the ones in the _ATTRS_MISC group. The attributes in this group are 

842 # somewhat "special": 

843 # * HTCondor plugin, which uses 'attrs' and 'profile', has its own 

844 # mechanism for setting them, 

845 # * 'cmdvals' is being set internally, not via config. 

846 job_values = _get_job_values(config, search_opt, None) 

847 for attr in _ATTRS_ALL - _ATTRS_MISC: 

848 if not getattr(gwjob, attr) and job_values.get(attr, None): 

849 setattr(gwjob, attr, job_values[attr]) 

850 

851 # Create script and add command line to job. 

852 gwjob.executable, gwjob.arguments = create_final_command(config, prefix) 

853 

854 # Determine inputs from command line. 

855 for file_key in re.findall(r"<FILE:([^>]+)>", gwjob.arguments): 

856 gwfile = generic_workflow.get_file(file_key) 

857 generic_workflow.add_job_inputs(gwjob.name, gwfile) 

858 

859 _enhance_command(config, generic_workflow, gwjob, {}) 

860 return gwjob 

861 

862 

863def create_final_command(config: BpsConfig, prefix: str) -> tuple[GenericWorkflowExec, str]: 

864 """Create the command and shell script for the final job. 

865 

866 Parameters 

867 ---------- 

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

869 Bps configuration. 

870 prefix : `str` 

871 Directory in which to output final script. 

872 

873 Returns 

874 ------- 

875 executable : `lsst.ctrl.bps.GenericWorkflowExec` 

876 Executable object for the final script. 

877 arguments : `str` 

878 Command line needed to call the final script. 

879 

880 Raises 

881 ------ 

882 RuntimeError if no commands found. 

883 """ 

884 search_opt = { 

885 "replaceVars": True, 

886 "skipNames": ["butlerConfig", "qgraphFile"], 

887 "replaceEnvVars": False, 

888 "expandEnvVars": False, 

889 "searchobj": config["finalJob"], 

890 } 

891 

892 script_file = os.path.join(prefix, "final_job.bash") 

893 with open(script_file, "w", encoding="utf8") as fh: 

894 print("#!/bin/bash\n", file=fh) 

895 print("set -e", file=fh) 

896 print("set -x", file=fh) 

897 

898 print("qgraphFile=$1", file=fh) 

899 print("butlerConfig=$2", file=fh) 

900 

901 command_len = 0 # Make sure at least write one actual command