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

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 

36 

37from lsst.ctrl.bps import ClusteredQuantumGraph 

38from lsst.pipe.base import QuantumGraph 

39from lsst.utils.logging import VERBOSE 

40from lsst.utils.timer import timeMethod 

41 

42from . import ( 

43 DEFAULT_MEM_RETRIES, 

44 BpsConfig, 

45 GenericWorkflow, 

46 GenericWorkflowExec, 

47 GenericWorkflowFile, 

48 GenericWorkflowJob, 

49) 

50from .bps_utils import WhenToSaveQuantumGraphs, create_job_quantum_graph_filename, save_qg_subgraph 

51 

52# All available job attributes. 

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

54 

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

56_ATTRS_MAX = frozenset( 

57 { 

58 "memory_multiplier", 

59 "number_of_retries", 

60 "request_cpus", 

61 "request_memory", 

62 "request_memory_max", 

63 } 

64) 

65 

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

67_ATTRS_SUM = frozenset( 

68 { 

69 "request_disk", 

70 "request_walltime", 

71 } 

72) 

73 

74# Job attributes do not fall into a specific category 

75_ATTRS_MISC = frozenset( 

76 { 

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

78 "cmdvals", 

79 "profile", 

80 "attrs", 

81 } 

82) 

83 

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

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

86 

87_LOG = logging.getLogger(__name__) 

88 

89 

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

91def transform( 

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

93) -> tuple[GenericWorkflow, BpsConfig]: 

94 """Transform a ClusteredQuantumGraph to a GenericWorkflow. 

95 

96 Parameters 

97 ---------- 

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

99 BPS configuration. 

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

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

102 prefix : `str` 

103 Root path for any output files. 

104 

105 Returns 

106 ------- 

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

108 The generic workflow transformed from the clustered quantum graph. 

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

110 Configuration to accompany GenericWorkflow. 

111 """ 

112 if cqgraph.name is not None: 

113 name = cqgraph.name 

114 else: 

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

116 

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

118 generic_workflow_config = create_generic_workflow_config(config, prefix) 

119 

120 return generic_workflow, generic_workflow_config 

121 

122 

123def add_workflow_init_nodes(config, qgraph, generic_workflow): 

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

125 

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

127 of the current workflow. 

128 

129 Parameters 

130 ---------- 

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

132 BPS configuration. 

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

134 The quantum graph the generic workflow represents. 

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

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

137 """ 

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

139 # initializing the pipeline execution 

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

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

142 generic_workflow.add_workflow_source(init_workflow) 

143 

144 

145def create_init_workflow( 

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

147) -> GenericWorkflow: 

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

149 

150 Parameters 

151 ---------- 

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

153 BPS configuration. 

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

155 The quantum graph the generic workflow represents. 

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

157 File object for the full run QuantumGraph file. 

158 

159 Returns 

160 ------- 

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

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

163 """ 

164 _LOG.debug("creating init subgraph") 

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

166 search_opt = { 

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

168 "replaceVars": False, 

169 "expandEnvVars": False, 

170 "replaceEnvVars": True, 

171 "required": False, 

172 } 

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

174 if found: 

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

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

177 if found: 

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

179 

180 init_workflow = GenericWorkflow("init") 

181 init_workflow.add_file(qgraph_gwfile) 

182 

183 # create job for executing --init-only 

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

185 

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

187 job_values["name"] = "pipetaskInit" 

188 job_values["label"] = "pipetaskInit" 

189 

190 # Adjust job attributes values if necessary. 

191 _handle_job_values(job_values, gwjob) 

192 

193 init_workflow.add_job(gwjob) 

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

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

196 

197 return init_workflow 

198 

199 

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

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

202 and gather command line values. 

203 

204 Parameters 

205 ---------- 

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

207 BPS configuration. 

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

209 Generic workflow that contains the job. 

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

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

212 and values should be saved. 

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

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

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

216 """ 

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

218 

219 curvals = { 

220 "curr_pipetask": gwjob.label, 

221 "curr_cluster": gwjob.label, 

222 "jobName": gwjob.name, 

223 "jobLabel": gwjob.label, 

224 } 

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

226 curvals[key] = value 

227 

228 search_opt = { 

229 "curvals": curvals, 

230 "replaceVars": False, 

231 "expandEnvVars": False, 

232 "replaceEnvVars": True, 

233 "required": False, 

234 } 

235 

236 if gwjob.label not in cached_job_values: 

237 cached_job_values[gwjob.label] = {} 

238 # Allowing whenSaveJobQgraph and useLazyCommands per pipetask label. 

239 key = "whenSaveJobQgraph" 

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

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

242 

243 key = "useLazyCommands" 

244 search_opt["default"] = True 

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

246 del search_opt["default"] 

247 

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

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

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

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

252 elif gwjob.name == "pipetaskInit": 

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

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

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

256 

257 # Replace files with special placeholders 

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

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

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

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

262 

263 # Replace wms variables with wms placeholders. 

264 gwjob.arguments = re.sub( 

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

266 ) 

267 

268 # To make yaml-specified paths easier to read, remove empty subdirs (//) 

269 gwjob.arguments = re.sub(r"//+", "/", gwjob.arguments) 

270 

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

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

273 # be different in compute job.) 

274 search_opt["replaceVars"] = True 

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

276 if key in gwjob.cmdvals: 

277 continue 

278 elif key in cached_job_values[gwjob.label]: 

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

280 else: 

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

282 

283 # backwards compatibility 

284 if not cached_job_values[gwjob.label]["useLazyCommands"]: 

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

286 key = "bpsUseShared" 

287 search_opt["default"] = True 

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

289 del search_opt["default"] 

290 

291 gwjob.arguments = _fill_arguments( 

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

293 ) 

294 

295 

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

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

298 

299 Parameters 

300 ---------- 

301 use_shared : `bool` 

302 Whether using shared filesystem. 

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

304 Generic workflow containing the job. 

305 arguments : `str` 

306 String containing placeholders. 

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

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

309 

310 Returns 

311 ------- 

312 arguments : `str` 

313 Command line with FILE and ENV placeholders replaced. 

314 """ 

315 # Replace file placeholders 

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

317 gwfile = generic_workflow.get_file(file_key) 

318 if not gwfile.wms_transfer: 

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

320 # responsible for transferring file. 

321 uri = gwfile.src_uri 

322 elif use_shared: 

323 if gwfile.job_shared: 

324 # Have shared filesystems and jobs can share file. 

325 uri = gwfile.src_uri 

326 else: 

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

328 else: # Using push transfer 

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

330 

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

332 

333 # Replace env placeholder with submit-side values 

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

335 arguments = os.path.expandvars(arguments) 

336 

337 # Replace remaining vars 

338 arguments = arguments.format(**cmdvals) 

339 

340 return arguments 

341 

342 

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

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

345 

346 Parameters 

347 ---------- 

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

349 Bps configuration. 

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

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

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

353 Job for which determining QuantumGraph file. 

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

355 File representation of the full run QuantumGraph. 

356 prefix : `str` 

357 Path prefix for any files written. 

358 

359 Returns 

360 ------- 

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

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

363 """ 

364 qgraph_gwfile = None 

365 if save_qgraph_per_job != WhenToSaveQuantumGraphs.NEVER: 

366 qgraph_gwfile = GenericWorkflowFile( 

367 f"qgraphFile_{gwjob.name}", 

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

369 wms_transfer=True, 

370 job_access_remote=True, 

371 job_shared=True, 

372 ) 

373 else: 

374 qgraph_gwfile = run_qgraph_file 

375 

376 return qgraph_gwfile 

377 

378 

379def _get_job_values(config, search_opt, cmd_line_key): 

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

381 

382 Parameters 

383 ---------- 

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

385 Bps configuration. 

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

387 Search options to be used when searching config. 

388 cmd_line_key : `str` or None 

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

390 

391 Returns 

392 ------- 

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

394 A mapping between job attributes and their values. 

395 """ 

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

397 

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

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

400 

401 job_values = {} 

402 for attr in _ATTRS_ALL: 

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

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

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

406 if found: 

407 job_values[attr] = value 

408 else: 

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

410 

411 # Need to replace all config variables in environment values 

412 # While replacing variables, convert to plain dict 

413 if job_values["environment"]: 

414 old_searchobj = search_opt.get("searchobj", None) 

415 old_replace_vars = search_opt.get("replaceVars", None) 

416 job_env = job_values["environment"] 

417 search_opt["searchobj"] = job_env 

418 search_opt["replaceVars"] = True 

419 job_values["environment"] = {} 

420 for name in job_env: 

421 job_values["environment"][name] = str(config.search(name, search_opt)[1]) 

422 if old_searchobj is None: 

423 del search_opt["searchobj"] 

424 else: 

425 search_opt["searchobj"] = old_searchobj 

426 if old_replace_vars is None: 

427 del search_opt["replaceVars"] 

428 else: 

429 search_opt["replaceVars"] = old_replace_vars 

430 

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

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

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

434 # automatic memory scaling. 

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

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

437 if job_values["number_of_retries"] is None: 

438 job_values["number_of_retries"] = DEFAULT_MEM_RETRIES 

439 else: 

440 job_values["memory_multiplier"] = None 

441 

442 if cmd_line_key: 

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

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

445 # default value in search_opt. 

446 if found and cmdline: 

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

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

449 if args: 

450 job_values["arguments"] = args 

451 

452 return job_values 

453 

454 

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

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

457 

458 Parameters 

459 ---------- 

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

461 Job values for running single Quantum. 

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

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

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

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

466 The default value is _ATTRS_ALL. 

467 """ 

468 _LOG.debug("Call to _handle_job_values") 

469 _handle_job_values_universal(quantum_job_values, gwjob, attributes) 

470 _handle_job_values_max(quantum_job_values, gwjob, attributes) 

471 _handle_job_values_sum(quantum_job_values, gwjob, attributes) 

472 

473 

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

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

476 in the cluster. 

477 

478 Parameters 

479 ---------- 

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

481 Job values for running single Quantum. 

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

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

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

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

486 The default value is _ATTRS_UNIVERSAL. 

487 """ 

488 for attr in _ATTRS_UNIVERSAL & set(attributes): 

489 _LOG.debug( 

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

491 attr, 

492 getattr(gwjob, attr), 

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

494 ) 

495 current_value = getattr(gwjob, attr) 

496 try: 

497 quantum_value = quantum_job_values[attr] 

498 except KeyError: 

499 continue 

500 else: 

501 if not current_value: 

502 setattr(gwjob, attr, quantum_value) 

503 elif current_value != quantum_value: 

504 _LOG.error( 

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

506 "Current cluster value: %s\n" 

507 "Quantum value: %s", 

508 attr, 

509 gwjob.name, 

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

511 current_value, 

512 quantum_value, 

513 ) 

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

515 

516 

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

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

519 the in cluster. 

520 

521 Parameters 

522 ---------- 

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

524 Job values for running single Quantum. 

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

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

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

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

529 The default value is _ATTR_MAX. 

530 """ 

531 for attr in _ATTRS_MAX & set(attributes): 

532 current_value = getattr(gwjob, attr) 

533 try: 

534 quantum_value = quantum_job_values[attr] 

535 except KeyError: 

536 continue 

537 else: 

538 needs_update = False 

539 if current_value is None: 

540 if quantum_value is not None: 

541 needs_update = True 

542 else: 

543 if quantum_value is not None and current_value < quantum_value: 

544 needs_update = True 

545 if needs_update: 

546 setattr(gwjob, attr, quantum_value) 

547 

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

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

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

551 # quantum. 

552 # 

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

554 # requirements will determine whether the memory autoscaling 

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

556 # memory multiplier. 

557 if attr == "request_memory": 

558 gwjob.memory_multiplier = quantum_job_values["memory_multiplier"] 

559 if gwjob.memory_multiplier is not None: 

560 gwjob.number_of_retries = quantum_job_values["number_of_retries"] 

561 

562 

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

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

565 

566 Parameters 

567 ---------- 

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

569 Job values for running single Quantum. 

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

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

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

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

574 The default value is _ATTRS_SUM. 

575 """ 

576 for attr in _ATTRS_SUM & set(attributes): 

577 current_value = getattr(gwjob, attr) 

578 if not current_value: 

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

580 else: 

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

582 

583 

584def create_generic_workflow( 

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

586) -> GenericWorkflow: 

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

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

589 

590 Parameters 

591 ---------- 

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

593 BPS configuration. 

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

595 ClusteredQuantumGraph for running a specific pipeline on a specific 

596 payload. 

597 name : `str` 

598 Name for the workflow (typically unique). 

599 prefix : `str` 

600 Root path for any output files. 

601 

602 Returns 

603 ------- 

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

605 Generic workflow for the given ClusteredQuantumGraph + config. 

606 """ 

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

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

609 save_qgraph_per_job = WhenToSaveQuantumGraphs[when_save.upper()] 

610 

611 search_opt = {"replaceVars": False, "expandEnvVars": False, "replaceEnvVars": True, "required": False} 

612 

613 generic_workflow = GenericWorkflow(name) 

614 

615 # Save full run QuantumGraph for use by jobs 

616 generic_workflow.add_file( 

617 GenericWorkflowFile( 

618 "runQgraphFile", 

619 src_uri=config["runQgraphFile"], 

620 wms_transfer=True, 

621 job_access_remote=True, 

622 job_shared=True, 

623 ) 

624 ) 

625 

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

627 # on config searches. 

628 cached_job_values = {} 

629 cached_pipetask_values = {} 

630 

631 for cluster in cqgraph.clusters(): 

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

633 _LOG.debug( 

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

635 cluster.name, 

636 len(cluster.qgraph_node_ids), 

637 cluster.label, 

638 cluster.qgraph_node_ids, 

639 ) 

640 

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

642 

643 # First get job values from cluster or cluster config 

644 search_opt["curvals"] = {"curr_cluster": cluster.label} 

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

646 if found: 

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

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

649 if found: 

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

651 

652 # If some config values are set for this cluster 

653 if cluster.label not in cached_job_values: 

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

655 cached_job_values[cluster.label] = {} 

656 

657 # Allowing whenSaveJobQgraph and useLazyCommands per cluster label. 

658 key = "whenSaveJobQgraph" 

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

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

661 

662 key = "useLazyCommands" 

663 search_opt["default"] = True 

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

665 del search_opt["default"] 

666 

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

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

669 # cluster section. 

670 cached_job_values[cluster.label].update( 

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

672 ) 

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

674 

675 cluster_job_values["name"] = cluster.name 

676 cluster_job_values["label"] = cluster.label 

677 cluster_job_values["quanta_counts"] = cluster.quanta_counts 

678 cluster_job_values["tags"] = cluster.tags 

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

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

681 

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

683 # the value evaluates to False. 

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

685 

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

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

688 

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

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

691 for node_id in iter(cluster.qgraph_node_ids): 

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

693 quantum_info = cqgraph.get_quantum_info(node_id) 

694 

695 task_label = quantum_info["task_label"] 

696 if task_label not in cached_pipetask_values: 

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

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

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

700 

701 # Update job with workflow attribute and profile values. 

702 qgraph_gwfile = _get_qgraph_gwfile( 

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

704 ) 

705 

706 generic_workflow.add_job(gwjob) 

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

708 

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

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

711 ) 

712 _enhance_command(config, generic_workflow, gwjob, cached_job_values) 

713 

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

715 # write it now while in memory. 

716 if save_qgraph_per_job == WhenToSaveQuantumGraphs.TRANSFORM: 

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

718 

719 # Create job dependencies. 

720 for parent in cqgraph.clusters(): 

721 for child in cqgraph.successors(parent): 

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

723 

724 # Add initial workflow. 

725 if config.get("runInit", "{default: False}"): 

726 add_workflow_init_nodes(config, cqgraph.qgraph, generic_workflow) 

727 

728 generic_workflow.run_attrs.update( 

729 { 

730 "bps_isjob": "True", 

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

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

733 "bps_run": generic_workflow.name, 

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

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

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

737 } 

738 ) 

739 

740 # Add final job 

741 add_final_job(config, generic_workflow, prefix) 

742 

743 if "ordering" in config: 

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

745 

746 return generic_workflow 

747 

748 

749def create_generic_workflow_config(config, prefix): 

750 """Create generic workflow configuration. 

751 

752 Parameters 

753 ---------- 

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

755 Bps configuration. 

756 prefix : `str` 

757 Root path for any output files. 

758 

759 Returns 

760 ------- 

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

762 Configuration accompanying the GenericWorkflow. 

763 """ 

764 generic_workflow_config = BpsConfig(config) 

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

766 generic_workflow_config["workflowPath"] = prefix 

767 return generic_workflow_config 

768 

769 

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

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

772 

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

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

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

776 with no errors. 

777 

778 Parameters 

779 ---------- 

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

781 Bps configuration. 

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

783 Generic workflow to which attributes should be added. 

784 prefix : `str` 

785 Directory in which to output final script. 

786 """ 

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

788 if when_run.upper() != "NEVER": 

789 gwjob = create_final_job(config, generic_workflow, prefix) 

790 if when_run.upper() == "ALWAYS": 

791 generic_workflow.add_final(gwjob) 

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

793 add_final_job_as_sink(generic_workflow, gwjob) 

794 else: 

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

796 

797 

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

799 """Create the final workflow job. 

800 

801 Parameters 

802 ---------- 

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

804 Bps configuration. 

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

806 Generic workflow to which attributes should be added. 

807 prefix : `str` 

808 Directory in which to output final script. 

809 

810 Returns 

811 ------- 

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

813 Final workflow job. 

814 """ 

815 job_name = "finalJob" 

816 gwjob = GenericWorkflowJob(job_name, job_name) 

817 

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

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

820 if found: 

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

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

823 if found: 

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

825 

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

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

828 # somewhat "special": 

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

830 # mechanism for setting them, 

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

832 job_values = _get_job_values(config, search_opt, None) 

833 for attr in _ATTRS_ALL - _ATTRS_MISC: 

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

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

836 

837 # Create script and add command line to job. 

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

839 

840 # Determine inputs from command line. 

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

842 gwfile = generic_workflow.get_file(file_key) 

843 generic_workflow.add_job_inputs(gwjob.name, gwfile) 

844 

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

846 return gwjob 

847 

848 

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

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

851 

852 Parameters 

853 ---------- 

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

855 Bps configuration. 

856 prefix : `str` 

857 Directory in which to output final script. 

858 

859 Returns 

860 ------- 

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

862 Executable object for the final script. 

863 arguments : `str` 

864 Command line needed to call the final script. 

865 

866 Raises 

867 ------ 

868 RuntimeError if no commands found. 

869 """ 

870 search_opt = { 

871 "replaceVars": True, 

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

873 "replaceEnvVars": False, 

874 "expandEnvVars": False, 

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

876 } 

877 

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

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

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

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

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

883 

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

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

886 

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

888 i = 1 

889 found, command = config.search(f"command{i}", opt=search_opt) 

890 while found: 

891 # The files will be args to script, so change to shell vars 

892 command = command.replace("{qgraphFile}", "${qgraphFile}") 

893 command = command.replace("{butlerConfig}", "${butlerConfig}") 

894 

895 print(command, file=fh) 

896 command_len += len(command.strip())