Coverage for python/lsst/ctrl/bps/htcondor/prepare_utils.py: 83%
518 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-14 18:41 +0000
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/>.
28"""Utility functions for preparing the HTCondor workflow."""
30import logging
31import os
32import re
33import shutil
34from collections import defaultdict
35from pathlib import Path
36from typing import Any, cast
38from lsst.ctrl.bps import (
39 BpsConfig,
40 GenericWorkflow,
41 GenericWorkflowFile,
42 GenericWorkflowGroup,
43 GenericWorkflowJob,
44 GenericWorkflowNodeType,
45 GenericWorkflowNoopJob,
46)
47from lsst.ctrl.bps.bps_utils import create_count_summary
49from .lssthtc import (
50 HTCDag,
51 HTCJob,
52 condor_status,
53 htc_escape,
54 read_dag_info,
55 write_dag_info,
56)
58_LOG = logging.getLogger(__name__)
60DEFAULT_HTC_EXEC_PATT = ".*worker.*"
61"""Default pattern for searching execute machines in an HTCondor pool.
62"""
65def _create_job(subdir_template, cached_values, generic_workflow, gwjob, out_prefix):
66 """Convert GenericWorkflow job nodes to DAG jobs.
68 Parameters
69 ----------
70 subdir_template : `str`
71 Template for making subdirs.
72 cached_values : `dict`
73 Site and label specific values.
74 generic_workflow : `lsst.ctrl.bps.GenericWorkflow`
75 Generic workflow that is being converted.
76 gwjob : `lsst.ctrl.bps.GenericWorkflowJob`
77 The generic job to convert to a HTCondor job.
78 out_prefix : `str`
79 Directory prefix for HTCondor files.
81 Returns
82 -------
83 htc_job : `lsst.ctrl.bps.wms.htcondor.HTCJob`
84 The HTCondor job equivalent to the given generic job.
85 """
86 _LOG.debug("_create_job: cached_values = %s", cached_values)
87 htc_job = HTCJob(gwjob.name, label=gwjob.label)
89 curvals = defaultdict(str)
90 curvals["label"] = gwjob.label
91 if gwjob.tags:
92 curvals.update(gwjob.tags)
94 subdir = Path("jobs") / subdir_template.format_map(curvals)
95 htc_job.subdir = subdir
96 htc_job.subfile = f"{gwjob.name}.sub"
97 htc_job.add_dag_cmds({"dir": subdir})
99 htc_job_cmds = {
100 "universe": "vanilla",
101 "should_transfer_files": "YES",
102 "when_to_transfer_output": "ON_EXIT_OR_EVICT",
103 "transfer_output_files": '""', # Set to empty string to disable
104 "transfer_executable": "False",
105 # Exceeding memory sometimes triggers SIGBUS or SIGSEGV error. Tell
106 # htcondor to put on hold any jobs which exited by a signal. If
107 # executed in a bash script, like finalJob, the signals will become
108 # exit codes above 128 (exit code = 128 + signal number).
109 "on_exit_hold": "ExitBySignal == true || ExitCode > 128",
110 "on_exit_hold_reason": "ExitBySignal == true ? "
111 'strcat("Job raised a signal ", string(ExitSignal), '
112 '". Handling job as if it has gone over memory limit.") : '
113 'strcat("Job exit code (", string(ExitCode), ") > 128. '
114 'Handling job as if it has gone over memory limit.")',
115 "on_exit_hold_subcode": "34",
116 }
118 htc_job_cmds.update(_translate_job_cmds(cached_values, generic_workflow, gwjob))
120 # Combine stdout and stderr to reduce the number of files.
121 for key in ("output", "error"):
122 if cached_values["overwriteJobFiles"]:
123 htc_job_cmds[key] = f"{gwjob.name}.$(Cluster).out"
124 else:
125 htc_job_cmds[key] = f"{gwjob.name}.$(Cluster).$$([NumJobStarts ?: 0]).out"
126 _LOG.debug("HTCondor %s = %s", key, htc_job_cmds[key])
128 key = "log"
129 htc_job_cmds[key] = f"{gwjob.name}.$(Cluster).{key}"
130 _LOG.debug("HTCondor %s = %s", key, htc_job_cmds[key])
132 htc_job_cmds.update(
133 _handle_job_inputs(generic_workflow, gwjob.name, cached_values["bpsUseShared"], out_prefix)
134 )
136 htc_job_cmds.update(
137 _handle_job_outputs(generic_workflow, gwjob.name, cached_values["bpsUseShared"], out_prefix)
138 )
140 # If specified, add nodeset to the job
141 if "nodeset" in cached_values:
142 htc_job.add_job_attrs({"JobNodeset": cached_values["nodeset"]})
143 clause = f'( Target.Nodeset == "{cached_values["nodeset"]}" )'
144 if "requirements" in htc_job_cmds:
145 htc_job_cmds["requirements"] = f"({htc_job_cmds['requirements']}) && {clause}"
146 else:
147 htc_job_cmds["requirements"] = clause
149 # Add the job cmds dict to the job object.
150 htc_job.add_job_cmds(htc_job_cmds)
152 # Add job-related cmds to the DAG (e.g., VARS)
153 htc_job.add_dag_cmds(_translate_dag_cmds(gwjob))
155 # Add job attributes to job.
156 _LOG.debug("gwjob.attrs = %s", gwjob.attrs)
157 htc_job.add_job_attrs(gwjob.attrs)
158 htc_job.add_job_attrs(cached_values["attrs"])
159 htc_job.add_job_attrs({"bps_job_quanta": create_count_summary(gwjob.quanta_counts)})
160 htc_job.add_job_attrs({"bps_job_name": gwjob.name, "bps_job_label": gwjob.label})
162 return htc_job
165def _translate_job_cmds(cached_vals, generic_workflow, gwjob):
166 """Translate the job data that are one to one mapping
168 Parameters
169 ----------
170 cached_vals : `dict` [`str`, `~typing.Any`]
171 Config values common to jobs with same site or label.
172 generic_workflow : `lsst.ctrl.bps.GenericWorkflow`
173 Generic workflow that contains job to being converted.
174 gwjob : `lsst.ctrl.bps.GenericWorkflowJob`
175 Generic workflow job to be converted.
177 Returns
178 -------
179 htc_job_commands : `dict` [`str`, `~typing.Any`]
180 Contains commands which can appear in the HTCondor submit description
181 file.
182 """
183 # Values in the job script that just are name mappings.
184 job_translation = {
185 "mail_to": "notify_user",
186 "when_to_mail": "notification",
187 "request_cpus": "request_cpus",
188 "priority": "priority",
189 "category": "category",
190 "accounting_group": "accounting_group",
191 "accounting_user": "accounting_group_user",
192 }
194 jobcmds = {}
195 for gwkey, htckey in job_translation.items():
196 jobcmds[htckey] = getattr(gwjob, gwkey, None)
198 # If accounting info was not set explicitly, use site settings if any.
199 if not gwjob.accounting_group: 199 ↛ 201line 199 didn't jump to line 201 because the condition on line 199 was always true
200 jobcmds["accounting_group"] = cached_vals.get("accountingGroup")
201 if not gwjob.accounting_user: 201 ↛ 205line 201 didn't jump to line 205 because the condition on line 201 was always true
202 jobcmds["accounting_group_user"] = cached_vals.get("accountingUser")
204 # job commands that need modification
205 if gwjob.retry_unless_exit:
206 if isinstance(gwjob.retry_unless_exit, int):
207 jobcmds["retry_until"] = f"{gwjob.retry_unless_exit}"
208 elif isinstance(gwjob.retry_unless_exit, list):
209 jobcmds["retry_until"] = (
210 f"member(ExitCode, {{{','.join([str(x) for x in gwjob.retry_unless_exit])}}})"
211 )
212 else:
213 raise ValueError("retryUnlessExit must be an integer or a list of integers.")
215 if gwjob.request_disk: 215 ↛ 216line 215 didn't jump to line 216 because the condition on line 215 was never true
216 jobcmds["request_disk"] = f"{gwjob.request_disk}MB"
218 if gwjob.request_memory:
219 jobcmds["request_memory"] = f"{gwjob.request_memory}"
221 memory_max = 0
222 if gwjob.memory_multiplier:
223 # Do not use try-except! At the moment, BpsConfig returns an empty
224 # string if it does not contain the key.
225 memory_limit = cached_vals["memoryLimit"]
226 if not memory_limit: 226 ↛ 227line 226 didn't jump to line 227 because the condition on line 226 was never true
227 raise RuntimeError(
228 "Memory autoscaling enabled, but automatic detection of the memory limit "
229 "failed; setting it explicitly with 'memoryLimit' or changing worker node "
230 "search pattern 'executeMachinesPattern' might help."
231 )
233 # Set maximal amount of memory job can ask for.
234 #
235 # The check below assumes that 'memory_limit' was set to a value which
236 # realistically reflects actual physical limitations of a given compute
237 # resource.
238 memory_max = memory_limit
239 if gwjob.request_memory_max and gwjob.request_memory_max < memory_limit: 239 ↛ 240line 239 didn't jump to line 240 because the condition on line 239 was never true
240 memory_max = gwjob.request_memory_max
242 # Make job ask for more memory each time it failed due to insufficient
243 # memory requirements.
244 jobcmds["request_memory"] = _create_request_memory_expr(
245 gwjob.request_memory, gwjob.memory_multiplier, memory_max
246 )
248 user_release_expr = cached_vals.get("releaseExpr", "")
249 if gwjob.number_of_retries is not None and gwjob.number_of_retries >= 0:
250 jobcmds["max_retries"] = gwjob.number_of_retries
252 # No point in adding periodic_release if 0 retries
253 if gwjob.number_of_retries > 0:
254 periodic_release = _create_periodic_release_expr(
255 gwjob.request_memory,
256 gwjob.memory_multiplier,
257 memory_max,
258 user_release_expr,
259 )
260 if periodic_release: 260 ↛ 263line 260 didn't jump to line 263 because the condition on line 260 was always true
261 jobcmds["periodic_release"] = periodic_release
263 jobcmds["periodic_remove"] = _create_periodic_remove_expr(
264 gwjob.request_memory, gwjob.memory_multiplier, memory_max
265 )
267 # Assume concurrency_limit implemented using HTCondor concurrency limits.
268 # May need to move to special site-specific implementation if sites use
269 # other mechanisms.
270 if gwjob.concurrency_limit: 270 ↛ 271line 270 didn't jump to line 271 because the condition on line 270 was never true
271 jobcmds["concurrency_limit"] = gwjob.concurrency_limit
273 # Handle command line
274 new_job_cmds = _translate_command_line(cached_vals, generic_workflow, gwjob)
275 jobcmds.update(new_job_cmds)
277 # Add extra "pass-thru" job commands
278 if gwjob.profile:
279 for key, val in gwjob.profile.items():
280 jobcmds[key] = val
281 for key, val in cached_vals["profile"].items():
282 jobcmds[key] = val
284 return jobcmds
287def _resolve_executable(executable: str) -> str:
288 """Resolve an executable name to its full path via PATH.
290 HTCondor resolves a relative ``executable`` in the submit description
291 relative to the job's sandbox directory rather than searching ``PATH``.
292 So a bare or relative command name (e.g., ``pipetask``) must be turned
293 into a full path at submit time, otherwise the job is held with a
294 "No such file or directory" error on the compute node.
296 Parameters
297 ----------
298 executable : `str`
299 The executable value from the job description.
301 Returns
302 -------
303 executable : `str`
304 A full path if ``executable`` was a bare name found on ``PATH``,
305 otherwise the value unchanged (already absolute/relative paths and
306 values containing HTCondor env syntax are left as is).
307 """
308 # Leave anything that already starts with a path separator or
309 # environment-variable syntax (e.g., $ENV(...), $(...)) untouched.
310 if executable.startswith(("/", "$")):
311 _LOG.debug("Executable started with / or $. Using as-is (%s)", executable)
312 return executable
314 exec_path = shutil.which(executable)
315 if exec_path:
316 _LOG.debug("results of 'which %s' = %s", executable, exec_path)
317 # Still have to make sure it is not a relative path as
318 # shutil.which would skip checking PATH if executable
319 # contains '/' and then only check existence.
320 exec_path = os.path.abspath(exec_path)
321 return exec_path
323 _LOG.warning(
324 "Could not resolve executable '%s' on PATH; HTCondor resolves an "
325 "executable name relative to the job sandbox, so the job may be held.",
326 executable,
327 )
328 return executable
331def _translate_command_line(
332 cached_vals: dict[str, Any], generic_workflow: GenericWorkflow, gwjob: GenericWorkflowJob
333) -> dict[str, Any]:
334 """Make environment, executable and argument settings for job.
336 Parameters
337 ----------
338 cached_vals : `dict` [`str`, `~typing.Any`]
339 Config values common to jobs with same site or label.
340 generic_workflow : `lsst.ctrl.bps.GenericWorkflow`
341 Generic workflow that contains job to being converted.
342 gwjob : `lsst.ctrl.bps.GenericWorkflowJob`
343 Generic workflow job to be converted.
345 Returns
346 -------
347 jobcmds : `dict` [`str` `Any`]
348 Commands to add to HTC submit description.
349 """
350 jobcmds = {}
352 job_exports = ""
353 htc_envs = ""
354 if gwjob.environment:
355 use_htc_env = cached_vals.get("bpsUseHTCEnvironment", False)
356 _LOG.debug("_translate_command_line: use_htc_env = %s", use_htc_env)
357 if use_htc_env:
358 # Even though it seems like just using getenv environment will
359 # work, we must use HTCondor env syntax to get submit side value.
360 # An environment variable defined in the job description just
361 # overrides any value from getenv. So we can't ask the job
362 # description to prepend/append to the value from getenv.
363 _fix_env = _fix_env_var_syntax
364 else:
365 # If not using environment in the job description, setting the
366 # environment is implemented as exports in the commands run via
367 # the shell.
368 _fix_env = _fix_env_var_syntax_shell
369 for name, value in gwjob.environment.items():
370 if isinstance(value, str):
371 value = _replace_wms_vars(value)
372 value = _fix_env(value)
373 value = htc_escape(value)
374 if use_htc_env:
375 htc_envs += f"{name}='{value}' " # Add single quotes to allow internal spaces
376 else:
377 job_exports += f"export {name}='{value}';"
379 # Process above added one trailing space
380 if use_htc_env:
381 jobcmds["environment"] = htc_envs.rstrip()
382 _LOG.debug("_translate_command_line: saving htc environment = %s", jobcmds["environment"])
384 arguments = ""
385 if gwjob.arguments:
386 arguments = gwjob.arguments
387 arguments = _replace_cmd_vars(arguments, gwjob)
388 arguments = _replace_wms_vars(arguments)
389 arguments = _replace_file_vars(cached_vals["bpsUseShared"], arguments, generic_workflow, gwjob)
391 if cached_vals.get("bpsMakeCommand", True):
392 # Way to have fallback to previous behavior as well as
393 # a way forward to centralize logic in bps.
395 jobcmds["getenv"] = "True"
397 if gwjob.executable.transfer_executable:
398 jobcmds["transfer_executable"] = "True"
399 jobcmds["executable"] = gwjob.executable.src_uri
400 else:
401 jobcmds["executable"] = _resolve_executable(_fix_env_var_syntax(gwjob.executable.src_uri))
403 if arguments:
404 arguments = _fix_env_var_syntax(arguments)
405 jobcmds["arguments"] = arguments
407 else:
408 # Instead of making a bash script, run /bin/bash -c <commands>
409 # HTCondor v25 has a job command called shell that can replace the
410 # /bin/bash when we get to that version.
412 # Don't set getenv as setting up the environment is assumed to be
413 # part of the payloadCommand.
415 if arguments:
416 arguments = _fix_env_var_syntax_shell(arguments)
418 if gwjob.executable.transfer_executable:
419 # Since replacing executable need to add this executable to the
420 # file transfer list.
421 gwfile = GenericWorkflowFile(
422 name=gwjob.executable.name, src_uri=gwjob.executable.src_uri, wms_transfer=True
423 )
424 generic_workflow.add_job_inputs(gwjob.name, [gwfile])
425 exec_name = os.path.basename(gwjob.executable.src_uri)
426 # Ensure the executable copy is executable.
427 gwjob_command = f"chmod u+x {exec_name}; ./{exec_name} {arguments}"
428 else:
429 exec_name = _fix_env_var_syntax_shell(gwjob.executable.src_uri)
430 gwjob_command = f"{exec_name} {arguments}"
432 payload_command = cached_vals["payloadCommand"]
433 _LOG.debug("%s payload_command pre-format: %s", gwjob.label, payload_command)
434 payload_command = re.sub("{gwjobCommand}", gwjob_command, payload_command)
435 payload_command = re.sub("{gwjobExports}", job_exports, payload_command)
437 # Remove newlines
438 payload_command = re.sub("\n", "", payload_command)
440 _LOG.debug("%s payload_command post-format: %s", gwjob.label, payload_command)
442 jobcmds["arguments"] = f"-c '{payload_command}'"
444 jobcmds["executable"] = "/bin/bash"
445 # Don't need to transfer /bin/bash
446 jobcmds["transfer_executable"] = "False"
448 return jobcmds
451def _translate_dag_cmds(gwjob):
452 """Translate job values into DAGMan commands.
454 Parameters
455 ----------
456 gwjob : `lsst.ctrl.bps.GenericWorkflowJob`
457 Job containing values to be translated.
459 Returns
460 -------
461 dagcmds : `dict` [`str`, `~typing.Any`]
462 DAGMan commands for the job.
463 """
464 # Values in the dag script that just are name mappings.
465 dag_translation = {
466 "abort_on_value": "abort_dag_on",
467 "abort_return_value": "abort_exit",
468 "priority": "priority",
469 }
471 dagcmds = {}
472 for gwkey, htckey in dag_translation.items():
473 dagcmds[htckey] = getattr(gwjob, gwkey, None)
475 # Still to be coded: vars "pre_cmdline", "post_cmdline"
476 return dagcmds
479def _fix_env_var_syntax(oldstr):
480 """Change ENV place holders to HTCondor Env var syntax.
482 Parameters
483 ----------
484 oldstr : `str`
485 String in which environment variable syntax is to be fixed.
487 Returns
488 -------
489 newstr : `str`
490 Given string with environment variable syntax fixed.
491 """
492 newstr = oldstr
493 for key in re.findall(r"<ENV:([^>]+)>", oldstr):
494 newstr = newstr.replace(rf"<ENV:{key}>", f"$ENV({key})")
495 return newstr
498def _fix_env_var_syntax_shell(oldstr):
499 """Change ENV place holders to shell var syntax.
501 Parameters
502 ----------
503 oldstr : `str`
504 String in which environment variable syntax is to be fixed.
506 Returns
507 -------
508 newstr : `str`
509 Given string with environment variable syntax fixed.
510 """
511 newstr = oldstr
512 for key in re.findall(r"<ENV:([^>]+)>", oldstr):
513 newstr = newstr.replace(rf"<ENV:{key}>", f"${{{key}}}")
514 return newstr
517def _replace_file_vars(use_shared, arguments, workflow, gwjob):
518 """Replace file placeholders in command line arguments with correct
519 physical file names.
521 Parameters
522 ----------
523 use_shared : `bool`
524 Whether HTCondor can assume shared filesystem.
525 arguments : `str`
526 Arguments string in which to replace file placeholders.
527 workflow : `lsst.ctrl.bps.GenericWorkflow`
528 Generic workflow that contains file information.
529 gwjob : `lsst.ctrl.bps.GenericWorkflowJob`
530 The job corresponding to the arguments.
532 Returns
533 -------
534 arguments : `str`
535 Given arguments string with file placeholders replaced.
536 """
537 # Replace input file placeholders with paths.
538 for gwfile in workflow.get_job_inputs(gwjob.name, data=True, transfer_only=False):
539 if not gwfile.wms_transfer: 539 ↛ 542line 539 didn't jump to line 542 because the condition on line 539 was never true
540 # Must assume full URI if in command line and told WMS is not
541 # responsible for transferring file.
542 uri = gwfile.src_uri
543 elif use_shared: 543 ↛ 550line 543 didn't jump to line 550 because the condition on line 543 was always true
544 if gwfile.job_shared: 544 ↛ 548line 544 didn't jump to line 548 because the condition on line 544 was always true
545 # Have shared filesystems and jobs can share file.
546 uri = gwfile.src_uri
547 else:
548 uri = os.path.basename(gwfile.src_uri)
549 else: # Using push transfer
550 uri = os.path.basename(gwfile.src_uri)
551 arguments = arguments.replace(f"<FILE:{gwfile.name}>", uri)
553 # Replace output file placeholders with paths.
554 for gwfile in workflow.get_job_outputs(gwjob.name, data=True, transfer_only=False): 554 ↛ 555line 554 didn't jump to line 555 because the loop on line 554 never started
555 if not gwfile.wms_transfer:
556 # Must assume full URI if in command line and told WMS is not
557 # responsible for transferring file.
558 uri = gwfile.src_uri
559 elif use_shared:
560 if gwfile.job_shared:
561 # Have shared filesystems and jobs can share file.
562 uri = gwfile.src_uri
563 else:
564 uri = os.path.basename(gwfile.src_uri)
565 else: # Using push transfer
566 uri = os.path.basename(gwfile.src_uri)
567 arguments = arguments.replace(f"<FILE:{gwfile.name}>", uri)
568 return arguments
571def _replace_cmd_vars(arguments, gwjob):
572 """Replace format-style placeholders in arguments.
574 Parameters
575 ----------
576 arguments : `str`
577 Arguments string in which to replace placeholders.
578 gwjob : `lsst.ctrl.bps.GenericWorkflowJob`
579 Job containing values to be used to replace placeholders
580 (in particular gwjob.cmdvals).
582 Returns
583 -------
584 arguments : `str`
585 Given arguments string with placeholders replaced.
586 """
587 replacements = gwjob.cmdvals if gwjob.cmdvals is not None else {}
588 try:
589 arguments = arguments.format(**replacements)
590 except (KeyError, TypeError) as exc: # TypeError in case None instead of {}
591 _LOG.error(
592 "Could not replace command variables for job %s: replacement for %s not provided",
593 gwjob.name,
594 str(exc),
595 )
596 _LOG.debug("arguments: %s\ncmdvals: %s", arguments, replacements)
597 raise
598 return arguments
601def _replace_wms_vars(orig_string: str) -> str:
602 """Replace special wms placeholders in given string.
604 Parameters
605 ----------
606 orig_string : `str`
607 String in which to replace wms placeholders.
609 Returns
610 -------
611 updated_string : `str`
612 Given string with wms placeholders replaced.
613 """
614 values = {"attemptNum": "$$([NumJobStarts])"}
615 updated_string = orig_string
616 for key in re.findall(r"<WMS:([^>]+)>", orig_string):
617 try:
618 updated_string = updated_string.replace(rf"<WMS:{key}>", values[key])
619 except KeyError:
620 _LOG.error("Unrecognized WMS placeholder: %s in %s", key, orig_string)
621 raise
622 return updated_string
625def _handle_job_inputs(
626 generic_workflow: GenericWorkflow, job_name: str, use_shared: bool, out_prefix: str
627) -> dict[str, str]:
628 """Add job input files from generic workflow to job.
630 Parameters
631 ----------
632 generic_workflow : `lsst.ctrl.bps.GenericWorkflow`
633 The generic workflow (e.g., has executable name and arguments).
634 job_name : `str`
635 Unique name for the job.
636 use_shared : `bool`
637 Whether job has access to files via shared filesystem.
638 out_prefix : `str`
639 The root directory into which all WMS-specific files are written.
641 Returns
642 -------
643 htc_commands : `dict` [`str`, `str`]
644 HTCondor commands for the job submission script.
645 """
646 inputs = []
647 for gwf_file in generic_workflow.get_job_inputs(job_name, data=True, transfer_only=True): 647 ↛ 648line 647 didn't jump to line 648 because the loop on line 647 never started
648 _LOG.debug("src_uri=%s", gwf_file.src_uri)
650 uri = Path(gwf_file.src_uri)
652 # Note if use_shared and job_shared, don't need to transfer file.
654 if not use_shared: # Copy file using push to job
655 inputs.append(str(uri))
656 elif not gwf_file.job_shared: # Jobs require own copy
657 # if using shared filesystem, but still need copy in job. Use
658 # HTCondor's curl plugin for a local copy.
659 if uri.is_dir():
660 raise RuntimeError(
661 f"HTCondor plugin cannot transfer directories locally within job {gwf_file.src_uri}"
662 )
663 inputs.append(f"file://{uri}")
665 htc_commands = {}
666 if inputs: 666 ↛ 667line 666 didn't jump to line 667 because the condition on line 666 was never true
667 htc_commands["transfer_input_files"] = ",".join(inputs)
668 _LOG.debug("transfer_input_files=%s", htc_commands["transfer_input_files"])
669 return htc_commands
672def _handle_job_outputs(
673 generic_workflow: GenericWorkflow, job_name: str, use_shared: bool, out_prefix: str
674) -> dict[str, str]:
675 """Add job output files from generic workflow to the job if any.
677 Parameters
678 ----------
679 generic_workflow : `lsst.ctrl.bps.GenericWorkflow`
680 The generic workflow (e.g., has executable name and arguments).
681 job_name : `str`
682 Unique name for the job.
683 use_shared : `bool`
684 Whether job has access to files via shared filesystem.
685 out_prefix : `str`
686 The root directory into which all WMS-specific files are written.
688 Returns
689 -------
690 htc_commands : `dict` [`str`, `str`]
691 HTCondor commands for the job submission script.
692 """
693 outputs = []
694 output_remaps = []
695 for gwf_file in generic_workflow.get_job_outputs(job_name, data=True, transfer_only=True):
696 _LOG.debug("src_uri=%s", gwf_file.src_uri)
698 uri = Path(gwf_file.src_uri)
699 if not use_shared:
700 outputs.append(uri.name)
701 output_remaps.append(f"{uri.name}={str(uri)}")
703 # Set to an empty string to disable and only update if there are output
704 # files to transfer. Otherwise, HTCondor will transfer back all files in
705 # the job’s temporary working directory that have been modified or created
706 # by the job.
707 htc_commands = {"transfer_output_files": '""'}
708 if outputs:
709 htc_commands["transfer_output_files"] = ",".join(outputs)
710 _LOG.debug("transfer_output_files=%s", htc_commands["transfer_output_files"])
712 htc_commands["transfer_output_remaps"] = f'"{";".join(output_remaps)}"'
713 _LOG.debug("transfer_output_remaps=%s", htc_commands["transfer_output_remaps"])
714 return htc_commands
717def _create_periodic_release_expr(
718 memory: int, multiplier: float | None, limit: int, additional_expr: str = ""
719) -> str:
720 """Construct an HTCondorAd expression for releasing held jobs.
722 Parameters
723 ----------
724 memory : `int`
725 Requested memory in MB.
726 multiplier : `float` or None
727 Memory growth rate between retries.
728 limit : `int`
729 Memory limit.
730 additional_expr : `str`, optional
731 Expression to add to periodic_release. Defaults to empty string.
733 Returns
734 -------
735 expr : `str`
736 A string representing an HTCondor ClassAd expression for releasing job.
737 """
738 _LOG.debug(
739 "periodic_release: memory: %s, multiplier: %s, limit: %s, additional_expr: %s",
740 memory,
741 multiplier,
742 limit,
743 additional_expr,
744 )
746 # ctrl_bps sets multiplier to None in the GenericWorkflow if
747 # memoryMultiplier <= 1, but checking value just in case.
748 if (not multiplier or multiplier <= 1) and not additional_expr:
749 return ""
751 # Job ClassAds attributes 'HoldReasonCode' and 'HoldReasonSubCode' are
752 # UNDEFINED if job is not HELD (i.e. when 'JobStatus' is not 5).
753 # The special comparison operators ensure that all comparisons below will
754 # evaluate to FALSE in this case.
755 #
756 # Note:
757 # May not be strictly necessary. Operators '&&' and '||' are not strict so
758 # the entire expression should evaluate to FALSE when the job is not HELD.
759 # According to ClassAd evaluation semantics FALSE && UNDEFINED is FALSE,
760 # but better safe than sorry.
761 is_held = "JobStatus == 5"
762 is_retry_allowed = "NumJobStarts <= JobMaxRetries"
764 mem_expr = ""
765 if memory and multiplier and multiplier > 1 and limit:
766 was_mem_exceeded = (
767 "(HoldReasonCode =?= 34 && HoldReasonSubCode =?= 0 "
768 "|| HoldReasonCode =?= 3 && HoldReasonSubCode =?= 34)"
769 )
770 was_below_limit = f"min({{int({memory} * pow({multiplier}, NumJobStarts - 1)), {limit}}}) < {limit}"
771 mem_expr = f"{was_mem_exceeded} && {was_below_limit}"
773 user_expr = ""
774 if additional_expr:
775 # Never auto release a job held by user.
776 user_expr = f"HoldReasonCode =!= 1 && {additional_expr}"
778 # Automatically release job if held because output file not found
779 # (e.g., job failed so didn't produce output file).
780 transfer_expr = "HoldReasonCode =?= 12"
782 expr = f"{is_held} && {is_retry_allowed}"
783 if user_expr and mem_expr:
784 expr += f" && ({transfer_expr} || {mem_expr} || {user_expr})"
785 elif user_expr:
786 expr += f" && ({transfer_expr} || {user_expr})"
787 elif mem_expr: 787 ↛ 790line 787 didn't jump to line 790 because the condition on line 787 was always true
788 expr += f" && ({transfer_expr} || {mem_expr})"
790 return expr
793def _create_periodic_remove_expr(memory, multiplier, limit):
794 """Construct an HTCondorAd expression for removing jobs from the queue.
796 Parameters
797 ----------
798 memory : `int`
799 Requested memory in MB.
800 multiplier : `float`
801 Memory growth rate between retries.
802 limit : `int`
803 Memory limit.
805 Returns
806 -------
807 expr : `str`
808 A string representing an HTCondor ClassAd expression for removing jobs.
809 """
810 # Job ClassAds attributes 'HoldReasonCode' and 'HoldReasonSubCode'
811 # are UNDEFINED if job is not HELD (i.e. when 'JobStatus' is not 5).
812 # The special comparison operators ensure that all comparisons below
813 # will evaluate to FALSE in this case.
814 #
815 # Note:
816 # May not be strictly necessary. Operators '&&' and '||' are not
817 # strict so the entire expression should evaluate to FALSE when the
818 # job is not HELD. According to ClassAd evaluation semantics
819 # FALSE && UNDEFINED is FALSE, but better safe than sorry.
820 is_held = "JobStatus == 5"
821 is_retry_disallowed = "NumJobStarts > JobMaxRetries"
823 mem_expr = ""
824 if memory and multiplier and multiplier > 1 and limit:
825 mem_limit_expr = f"min({{int({memory} * pow({multiplier}, NumJobStarts - 1)), {limit}}}) == {limit}"
827 mem_expr = ( # Add || here so only added if adding memory expr
828 " || ((HoldReasonCode =?= 34 && HoldReasonSubCode =?= 0 "
829 f"|| HoldReasonCode =?= 3 && HoldReasonSubCode =?= 34) && {mem_limit_expr})"
830 )
832 expr = f"{is_held} && ({is_retry_disallowed}{mem_expr})"
833 return expr
836def _create_request_memory_expr(memory, multiplier, limit):
837 """Construct an HTCondor ClassAd expression for safe memory scaling.
839 Parameters
840 ----------
841 memory : `int`
842 Requested memory in MB.
843 multiplier : `float`
844 Memory growth rate between retries.
845 limit : `int`
846 Memory limit.
848 Returns
849 -------
850 expr : `str`
851 A string representing an HTCondor ClassAd expression enabling safe
852 memory scaling between job retries.
853 """
854 # The check if the job was held due to exceeding memory requirements
855 # will be made *after* job was released back to the job queue (is in
856 # the IDLE state), hence the need to use `Last*` job ClassAds instead of
857 # the ones describing job's current state.
858 #
859 # Also, 'Last*' job ClassAds attributes are UNDEFINED when a job is
860 # initially put in the job queue. The special comparison operators ensure
861 # that all comparisons below will evaluate to FALSE in this case.
862 was_mem_exceeded = (
863 "LastJobStatus =?= 5 "
864 "&& (LastHoldReasonCode =?= 34 && LastHoldReasonSubCode =?= 0 "
865 "|| LastHoldReasonCode =?= 3 && LastHoldReasonSubCode =?= 34)"
866 )
868 # If job runs the first time or was held for reasons other than exceeding
869 # the memory, set the required memory to the requested value or use
870 # the memory value measured by HTCondor (MemoryUsage) depending on
871 # whichever is greater.
872 expr = (
873 f"({was_mem_exceeded}) "
874 f"? min({{int({memory} * pow({multiplier}, NumJobStarts)), {limit}}}) "
875 f": min({{max({{{memory}, MemoryUsage ?: 0}}), {limit}}})"
876 )
877 return expr
880def _gather_site_values(config, compute_site):
881 """Gather values specific to given site.
883 Parameters
884 ----------
885 config : `lsst.ctrl.bps.BpsConfig`
886 BPS configuration that includes necessary submit/runtime
887 information.
888 compute_site : `str`
889 Compute site name.
891 Returns
892 -------
893 site_values : `dict` [`str`, `~typing.Any`]
894 Values specific to the given site.
895 """
896 site_values = {"attrs": {}, "profile": {}}
897 search_opts = {}
898 if compute_site: 898 ↛ 902line 898 didn't jump to line 902 because the condition on line 898 was always true
899 search_opts["curvals"] = {"curr_site": compute_site}
901 # Determine the hard limit for the memory requirement.
902 found, limit = config.search("memoryLimit", opt=search_opts)
903 if not found: 903 ↛ 904line 903 didn't jump to line 904 because the condition on line 903 was never true
904 search_opts["default"] = DEFAULT_HTC_EXEC_PATT
905 _, patt = config.search("executeMachinesPattern", opt=search_opts)
906 del search_opts["default"]
908 # To reduce the amount of data, ignore dynamic slots (if any) as,
909 # by definition, they cannot have more memory than
910 # the partitionable slot they are the part of.
911 constraint = f'SlotType != "Dynamic" && regexp("{patt}", Machine)'
912 pool_info = condor_status(constraint=constraint)
913 try:
914 limit = max(int(info["TotalSlotMemory"]) for info in pool_info.values())
915 except ValueError:
916 _LOG.debug("No execute machine in the pool matches %s", patt)
917 if limit: 917 ↛ 919line 917 didn't jump to line 919 because the condition on line 917 was always true
918 config[".bps_defined.memory_limit"] = limit
919 site_values["memoryLimit"] = limit
921 _, site_values["bpsUseShared"] = config.search("bpsUseShared", opt={"default": False})
923 searchobj = config[f".site.{compute_site}.profile.condor"]
924 if searchobj:
925 search_opts["searchobj"] = searchobj
926 search_opts["replaceVars"] = True
927 for key in searchobj:
928 if key.startswith("+"):
929 _, val = config.search(key, opt=search_opts)
930 site_values["attrs"][key[1:]] = val
931 else:
932 _, val = config.search(key, opt=search_opts)
933 site_values["profile"][key] = val
935 searchobj = config[f".site.{compute_site}"]
936 if searchobj:
937 for key, value in searchobj.items():
938 if key not in site_values and key not in ["attrs", "profile"]: 938 ↛ 939line 938 didn't jump to line 939 because the condition on line 938 was never true
939 site_values[key] = value
941 _LOG.debug("site_values = %s", site_values)
942 return site_values
945def _gather_label_values(config: BpsConfig, label: str) -> dict[str, Any]:
946 """Gather values specific to given job label.
948 Parameters
949 ----------
950 config : `lsst.ctrl.bps.BpsConfig`
951 BPS configuration that includes necessary submit/runtime
952 information.
953 label : `str`
954 GenericWorkflowJob label.
956 Returns
957 -------
958 values : `dict` [`str`, `~typing.Any`]
959 Values specific to the given job label.
960 """
961 _LOG.debug("_gather_label_values: label = %s", label)
962 values: dict[str, Any] = {"attrs": {}, "profile": {}}
964 search_opts = config.get_search_opts(label)
966 # Determine the hard limit for the memory requirement.
967 found, limit = config.search("memoryLimit", opt=search_opts)
968 if not found: 968 ↛ 969line 968 didn't jump to line 969 because the condition on line 968 was never true
969 search_opts["default"] = DEFAULT_HTC_EXEC_PATT
970 _, patt = config.search("executeMachinesPattern", opt=search_opts)
971 del search_opts["default"]
973 # To reduce the amount of data, ignore dynamic slots (if any) as,
974 # by definition, they cannot have more memory than
975 # the partitionable slot they are the part of.
976 constraint = f'SlotType != "Dynamic" && regexp("{patt}", Machine)'
977 pool_info = condor_status(constraint=constraint)
978 try:
979 limit = max(int(info["TotalSlotMemory"]) for info in pool_info.values())
980 except ValueError:
981 _LOG.debug("No execute machine in the pool matches %s", patt)
983 if limit: 983 ↛ 987line 983 didn't jump to line 987 because the condition on line 983 was always true
984 config[".bps_defined.memory_limit"] = limit
985 values["memoryLimit"] = limit
987 values["bpsUseShared"] = False
988 found, value = config.search("bpsUseShared", opt=search_opts)
989 if found:
990 values["bpsUseShared"] = value
992 found, value = config.search("releaseExpr", opt=search_opts)
993 if found:
994 values["releaseExpr"] = value
996 values["overwriteJobFiles"] = True
997 found, value = config.search("overwriteJobFiles", opt=search_opts)
998 if found:
999 values["overwriteJobFiles"] = value
1001 found, value = config.search("releaseExpr", opt=search_opts)
1002 if found:
1003 values["releaseExpr"] = value
1005 found, value = config.search("bpsMakeCommand", opt=search_opts)
1006 values["bpsMakeCommand"] = value if found else True
1007 if found and not value:
1008 search_opts["skipNames"] = {"gwjobCommand", "gwjobExports"}
1009 _LOG.debug("_gather_label_values: search_opts = %s", search_opts)
1010 found, value = config.search("payloadCommand", opt=search_opts)
1011 if found: 1011 ↛ 1015line 1011 didn't jump to line 1015 because the condition on line 1011 was always true
1012 values["payloadCommand"] = value
1013 _LOG.debug("payloadCommand = %s", value)
1015 found, value = config.search("bpsUseHTCEnvironment", opt=search_opts)
1016 values["bpsUseHTCEnvironment"] = value if found else values["bpsMakeCommand"]
1018 found, value = config.search("nodeset", opt=search_opts)
1019 if found:
1020 values["nodeset"] = value
1022 found, profile_sect = config.search("profile", opt=search_opts)
1023 if found and "condor" in profile_sect:
1024 for subkey, val in profile_sect["condor"].items():
1025 if subkey.startswith("+"):
1026 values["attrs"][subkey[1:]] = val
1027 else:
1028 values["profile"][subkey] = val
1030 # Copy all of the site values.
1031 if "curr_site" in search_opts["curvals"]:
1032 site_obj = config[f".site.{search_opts['curvals']['curr_site']}"]
1033 if site_obj: 1033 ↛ 1038line 1033 didn't jump to line 1038 because the condition on line 1033 was always true
1034 for key, value in site_obj.items():
1035 if key not in values and key not in ["attrs", "profile"]:
1036 values[key] = value
1038 _LOG.debug("_gather_label_values: label = %s, values = %s", label, values)
1040 return values
1043def _group_to_subdag(
1044 config: BpsConfig, generic_workflow_group: GenericWorkflowGroup, out_prefix: str
1045) -> HTCJob:
1046 """Convert a generic workflow group to an HTCondor dag.
1048 Parameters
1049 ----------
1050 config : `lsst.ctrl.bps.BpsConfig`
1051 Workflow configuration.
1052 generic_workflow_group : `lsst.ctrl.bps.GenericWorkflowGroup`
1053 The generic workflow group to convert.
1054 out_prefix : `str`
1055 Location prefix to be used when creating jobs.
1057 Returns
1058 -------
1059 htc_job : `lsst.ctrl.bps.htcondor.HTCJob`
1060 Job for running the HTCondor dag.
1061 """
1062 jobname = f"wms_{generic_workflow_group.name}"
1063 htc_job = HTCJob(name=jobname, label=generic_workflow_group.label)
1064 htc_job.add_dag_cmds({"dir": f"subdags/{jobname}"})
1065 htc_job.subdag = _generic_workflow_to_htcondor_dag(config, generic_workflow_group, out_prefix)
1066 if not generic_workflow_group.blocking: 1066 ↛ 1072line 1066 didn't jump to line 1072 because the condition on line 1066 was always true
1067 htc_job.dagcmds["post"] = {
1068 "defer": "",
1069 "executable": f"{os.path.dirname(__file__)}/subdag_post.sh",
1070 "arguments": f"{jobname} $RETURN",
1071 }
1072 return htc_job
1075def _create_check_job(group_job_name: str, job_label: str, site_values: dict) -> HTCJob:
1076 """Create a job to check status of a group job.
1078 Parameters
1079 ----------
1080 group_job_name : `str`
1081 Name of the group job.
1082 job_label : `str`
1083 Label to use for the check status job.
1084 site_values : `dict`
1085 Site specific values.
1087 Returns
1088 -------
1089 htc_job : `lsst.ctrl.bps.htcondor.HTCJob`
1090 Job description for the job to check group job status.
1091 """
1092 htc_job = HTCJob(name=f"wms_check_status_{group_job_name}", label=job_label)
1093 htc_job.subfile = "${CTRL_BPS_HTCONDOR_DIR}/python/lsst/ctrl/bps/htcondor/check_group_status.sub"
1094 # ADD nodeset to VARS
1095 job_vars = {"group_job_name": group_job_name}
1096 if "nodeset" in site_values and site_values["nodeset"]:
1097 job_vars["job_nodeset"] = site_values["nodeset"]
1098 htc_job.add_dag_cmds({"dir": f"subdags/{group_job_name}", "vars": job_vars})
1100 return htc_job
1103def _generic_workflow_to_htcondor_dag(
1104 config: BpsConfig, generic_workflow: GenericWorkflow, out_prefix: str
1105) -> HTCDag:
1106 """Convert a GenericWorkflow to a HTCDag.
1108 Parameters
1109 ----------
1110 config : `lsst.ctrl.bps.BpsConfig`
1111 Workflow configuration.
1112 generic_workflow : `lsst.ctrl.bps.GenericWorkflow`
1113 The GenericWorkflow to convert.
1114 out_prefix : `str`
1115 Location prefix where the HTCondor files will be written.
1117 Returns
1118 -------
1119 dag : `lsst.ctrl.bps.htcondor.HTCDag`
1120 The HTCDag representation of the given GenericWorkflow.
1121 """
1122 dag = HTCDag(name=generic_workflow.name)
1124 _LOG.debug("htcondor dag attribs %s", generic_workflow.run_attrs)
1125 dag.add_attribs(generic_workflow.run_attrs)
1126 dag.add_attribs(
1127 {
1128 "bps_run_quanta": create_count_summary(generic_workflow.quanta_counts),
1129 "bps_job_summary": create_count_summary(generic_workflow.job_counts),
1130 }
1131 )
1132 _, tmp_template = config.search("subDirTemplate", opt={"replaceVars": False, "default": ""})
1134 _, save_htc_dot = config.search("saveHTCdot", opt={"default": False})
1135 dag.graph["write_dot"] = save_htc_dot
1137 if isinstance(tmp_template, str): 1137 ↛ 1140line 1137 didn't jump to line 1140 because the condition on line 1137 was always true
1138 subdir_template = defaultdict(lambda: tmp_template)
1139 else:
1140 subdir_template = tmp_template
1142 # Save list of lazy group jobs for later extra handling.
1143 lazy_groups = []
1145 # Create all DAG jobs
1146 cached_values = {} # Cache label-specific values to reduce config lookups.
1147 # Note: Can't use get_job_by_label because those only include payload jobs.
1148 for job_name in generic_workflow:
1149 gwjob = generic_workflow.get_job(job_name)
1150 if gwjob.node_type in [ 1150 ↛ 1165line 1150 didn't jump to line 1165 because the condition on line 1150 was always true
1151 GenericWorkflowNodeType.PAYLOAD,
1152 GenericWorkflowNodeType.LAZY_GROUP,
1153 ]:
1154 gwjob = cast(GenericWorkflowJob, gwjob)
1155 if gwjob.label not in cached_values:
1156 cached_values[gwjob.label] = _gather_label_values(config, gwjob.label)
1157 _LOG.debug("cached: %s= %s", gwjob.label, cached_values[gwjob.label])
1158 htc_job = _create_job(
1159 subdir_template[gwjob.label],
1160 cached_values[gwjob.label],
1161 generic_workflow,
1162 gwjob,
1163 out_prefix,
1164 )
1165 elif gwjob.node_type == GenericWorkflowNodeType.NOOP:
1166 gwjob = cast(GenericWorkflowNoopJob, gwjob)
1167 htc_job = HTCJob(f"wms_{gwjob.name}", label=gwjob.label)
1168 htc_job.subfile = "${CTRL_BPS_HTCONDOR_DIR}/python/lsst/ctrl/bps/htcondor/noop.sub"
1169 htc_job.add_job_attrs({"bps_job_name": gwjob.name, "bps_job_label": gwjob.label})
1170 htc_job.add_dag_cmds({"noop": True})
1171 elif gwjob.node_type == GenericWorkflowNodeType.GROUP:
1172 gwjob = cast(GenericWorkflowGroup, gwjob)
1173 cached_values[gwjob.label] = _gather_label_values(config, gwjob.label)
1174 htc_job = _group_to_subdag(config, gwjob, out_prefix)
1175 else:
1176 raise RuntimeError(f"Unsupported generic workflow node type {gwjob.node_type} ({gwjob.name})")
1177 _LOG.debug("Calling adding job %s %s", htc_job.name, htc_job.label)
1178 dag.add_job(htc_job)
1180 # Have to add the placeholder job for the workflow
1181 if gwjob.node_type == GenericWorkflowNodeType.LAZY_GROUP:
1182 lazy_groups.append(gwjob.name)
1184 # Add job dependencies to the DAG (be careful with wms_ jobs)
1185 for job_name in generic_workflow:
1186 gwjob = generic_workflow.get_job(job_name)
1187 parent_name = (
1188 gwjob.name
1189 if gwjob.node_type in [GenericWorkflowNodeType.PAYLOAD, GenericWorkflowNodeType.LAZY_GROUP]
1190 else f"wms_{gwjob.name}"
1191 )
1192 successor_jobs = [generic_workflow.get_job(j) for j in generic_workflow.successors(job_name)]
1193 children_names = []
1194 if gwjob.node_type == GenericWorkflowNodeType.GROUP: 1194 ↛ 1195line 1194 didn't jump to line 1195 because the condition on line 1194 was never true
1195 gwjob = cast(GenericWorkflowGroup, gwjob)
1196 group_children = [] # Dependencies between same group jobs
1197 for sjob in successor_jobs:
1198 if sjob.node_type == GenericWorkflowNodeType.GROUP and sjob.label == gwjob.label:
1199 group_children.append(f"wms_{sjob.name}")
1200 elif sjob.node_type == GenericWorkflowNodeType.PAYLOAD:
1201 children_names.append(sjob.name)
1202 else:
1203 children_names.append(f"wms_{sjob.name}")
1204 if group_children:
1205 dag.add_job_relationships([parent_name], group_children)
1206 if not gwjob.blocking:
1207 # Since subdag will always succeed, need to add a special
1208 # job that fails if group failed to block payload children.
1209 check_job = _create_check_job(
1210 f"wms_{gwjob.name}", gwjob.label, cached_values.get(gwjob.label, {})
1211 )
1212 dag.add_job(check_job)
1213 dag.add_job_relationships([f"wms_{gwjob.name}"], [check_job.name])
1214 parent_name = check_job.name
1215 else:
1216 for sjob in successor_jobs:
1217 if sjob.node_type in [ 1217 ↛ 1223line 1217 didn't jump to line 1223 because the condition on line 1217 was always true
1218 GenericWorkflowNodeType.PAYLOAD,
1219 GenericWorkflowNodeType.LAZY_GROUP,
1220 ]:
1221 children_names.append(sjob.name)
1222 else:
1223 children_names.append(f"wms_{sjob.name}")
1225 dag.add_job_relationships([parent_name], children_names)
1227 # Go back and add placeholder jobs for the lazy group dags
1228 for lazy_group_name in lazy_groups:
1229 _add_lazy_placeholder(lazy_group_name, generic_workflow, dag, out_prefix)
1231 # If final job exists in generic workflow, create DAG final job
1232 final = generic_workflow.get_final()
1233 if final and isinstance(final, GenericWorkflowJob):
1234 if final.label not in cached_values: 1234 ↛ 1236line 1234 didn't jump to line 1236 because the condition on line 1234 was always true
1235 cached_values[final.label] = _gather_label_values(config, final.label)
1236 final_htjob = _create_job(
1237 subdir_template[final.label],
1238 cached_values[final.label],
1239 generic_workflow,
1240 final,
1241 out_prefix,
1242 )
1243 if "post" not in final_htjob.dagcmds: 1243 ↛ 1249line 1243 didn't jump to line 1249 because the condition on line 1243 was always true
1244 final_htjob.dagcmds["post"] = {
1245 "defer": "",
1246 "executable": f"{os.path.dirname(__file__)}/final_post.sh",
1247 "arguments": f"{final.name} $DAG_STATUS $RETURN",
1248 }
1249 dag.add_final_job(final_htjob)
1250 elif final and isinstance(final, GenericWorkflow):
1251 raise NotImplementedError("HTCondor plugin does not support a workflow as the final job")
1252 elif final: 1252 ↛ 1253line 1252 didn't jump to line 1253 because the condition on line 1252 was never true
1253 raise TypeError(f"Invalid type for GenericWorkflow.get_final() results ({type(final)})")
1255 return dag
1258def _add_lazy_placeholder(
1259 prepare_job_name: str,
1260 generic_workflow: GenericWorkflow,
1261 dag: HTCDag,
1262 out_prefix: str,
1263):
1264 _LOG.debug("prepare_job_name = %s", prepare_job_name)
1266 # Make a fake job for the placeholder workflow
1267 job = HTCJob("placeholder")
1268 job.add_job_cmds(
1269 {
1270 "executable": "/usr/bin/echo",
1271 "arguments": '"BPS internal error - placeholder DAG was not replaced."',
1272 }
1273 )
1274 job.add_job_attrs({"bps_job_name": "placeholder", "bps_job_label": "placeholder"})
1275 job.add_job_attrs(generic_workflow.run_attrs)
1277 # Placeholder dag name needs to be the run name because that's
1278 # currently what the bps code will name the workflow.
1279 placeholder_dag_name = generic_workflow.name.replace("_ctrl", "")
1280 placeholder_dag = HTCDag(name=placeholder_dag_name)
1281 _LOG.debug("dag name = %s", placeholder_dag.graph["name"])
1282 placeholder_dag.add_attribs(generic_workflow.run_attrs)
1283 placeholder_dag.add_job(job)
1285 # To help ordering in bps report, save info so can figure out where
1286 # to put jobs in lazy dag in bps_job_label
1287 dag.add_attribs({"bps_lazy_mapping": f"{placeholder_dag.name}:{prepare_job_name}"})
1289 # The subdag job to be added to the control dag
1290 dag_job = HTCJob("wms_lazy_payload", "wms_lazy_payload")
1291 dag_job.subfile = f"{placeholder_dag.name}.condor.sub"
1292 dag_job.subdag = placeholder_dag
1294 dag.add_job(dag_job)
1296 # Update edges inserting between prepare job and any following jobs.
1297 successor_jobs = list(dag.successors(prepare_job_name))
1298 for job in successor_jobs:
1299 dag.add_edge(dag_job.name, job)
1300 dag.remove_edge(prepare_job_name, job)
1302 dag.add_edge(prepare_job_name, dag_job.name)
1304 _LOG.debug("_add_lazy_placeholder: edges = %s", dag.edges)
1307def _update_job_summary(subworkflow_name: str, subworkflow_summary: str, submit_path: str) -> None:
1308 """Add summary for jobs in given subworkflow to parent workflow's job
1309 summary.
1311 Parameters
1312 ----------
1313 subworkflow_name : `str`
1314 Name of subworkflow.
1315 subworkflow_summary : `str`
1316 Job summary for the subworkflow.
1317 submit_path : `str`
1318 Directory in which to find the DAG info file.
1319 """
1320 _LOG.debug("submit_path = %s", submit_path)
1321 filename, dag_info = read_dag_info(submit_path)
1322 _LOG.debug("dag_info = %s", dag_info)
1324 schedd_name = next(iter(dag_info))
1325 dag_values = next(iter(dag_info[schedd_name].values()))
1326 _LOG.debug("dag_values = %s", dag_values)
1328 # Get lazy mapping and the job that generated this dag
1329 generator_name = None
1330 lazy_mapping = dag_values.get("bps_lazy_mapping", None)
1331 if lazy_mapping: # find this workflow's generator job
1332 for part in lazy_mapping.split(";"):
1333 info = part.split(":")
1334 if info[0] == subworkflow_name:
1335 generator_name = info[1]
1336 break
1338 # Update bps_job_summary
1339 _LOG.debug(
1340 "Before replace, name = %s, bps_job_summary = %s, add summary = %s, generator_name = %s",
1341 subworkflow_name,
1342 dag_values["bps_job_summary"],
1343 subworkflow_summary,
1344 generator_name,
1345 )
1346 if generator_name:
1347 generator_summary = f"{generator_name}:1"
1348 dag_values["bps_job_summary"] = dag_values["bps_job_summary"].replace(
1349 generator_summary, f"{generator_summary};{subworkflow_summary}"
1350 )
1351 else:
1352 # just append to end of bps_job_summary
1353 dag_values["bps_job_summary"] += f";{subworkflow_summary}"
1355 _LOG.debug(
1356 "After replace, name = %s, bps_job_summary = %s",
1357 subworkflow_name,
1358 dag_values["bps_job_summary"],
1359 )
1361 # Save updated bps_job_summary
1362 write_dag_info(filename, dag_info)