Coverage for python/lsst/ctrl/bps/htcondor/lssthtc.py: 78%
938 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-15 03:14 +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"""Placeholder HTCondor DAGMan API.
30There is new work on a python DAGMan API from HTCondor. However, at this
31time, it tries to make things easier by assuming DAG is easily broken into
32levels where there are 1-1 or all-to-all relationships to nodes in next
33level. LSST workflows are more complicated.
34"""
36__all__ = [
37 "MISSING_ID",
38 "DagStatus",
39 "HTCDag",
40 "HTCJob",
41 "NodeStatus",
42 "RestrictedDict",
43 "WmsNodeType",
44 "condor_history",
45 "condor_q",
46 "condor_search",
47 "condor_status",
48 "htc_backup_files",
49 "htc_check_dagman_output",
50 "htc_create_submit_from_cmd",
51 "htc_create_submit_from_dag",
52 "htc_create_submit_from_file",
53 "htc_escape",
54 "htc_query_history",
55 "htc_query_present",
56 "htc_submit_dag",
57 "htc_tweak_log_info",
58 "htc_version",
59 "htc_write_attribs",
60 "htc_write_condor_file",
61 "pegasus_name_to_label",
62 "read_dag_info",
63 "read_dag_log",
64 "read_dag_nodes_log",
65 "read_dag_status",
66 "read_node_status",
67 "summarize_dag",
68 "update_job_info",
69 "write_dag_info",
70]
73import itertools
74import json
75import logging
76import os
77import pprint
78import re
79import subprocess
80from collections import Counter, defaultdict
81from collections.abc import MutableMapping
82from datetime import datetime, timedelta
83from enum import IntEnum, auto
84from pathlib import Path
85from typing import Any, TextIO
87import classad
88import htcondor
89import networkx
90from deprecated.sphinx import deprecated
91from packaging import version
93from .handlers import HTC_JOB_AD_HANDLERS
95_LOG = logging.getLogger(__name__)
97MISSING_ID = "-99999"
100class DagStatus(IntEnum):
101 """HTCondor DAGMan's statuses for a DAG."""
103 OK = 0
104 ERROR = 1 # an error condition different than those listed here
105 FAILED = 2 # one or more nodes in the DAG have failed
106 ABORTED = 3 # the DAG has been aborted by an ABORT-DAG-ON specification
107 REMOVED = 4 # the DAG has been removed by condor_rm
108 CYCLE = 5 # a cycle was found in the DAG
109 SUSPENDED = 6 # the DAG has been suspended (see section 2.10.8)
112@deprecated(
113 reason="The JobStatus is internally replaced by htcondor.JobStatus. "
114 "External reporting code should be using ctrl_bps.WmsStates. "
115 "This class will be removed after v30.",
116 version="v30.0",
117 category=FutureWarning,
118)
119class JobStatus(IntEnum):
120 """HTCondor's statuses for jobs."""
122 UNEXPANDED = 0 # Unexpanded
123 IDLE = 1 # Idle
124 RUNNING = 2 # Running
125 REMOVED = 3 # Removed
126 COMPLETED = 4 # Completed
127 HELD = 5 # Held
128 TRANSFERRING_OUTPUT = 6 # Transferring_Output
129 SUSPENDED = 7 # Suspended
132class NodeStatus(IntEnum):
133 """HTCondor's statuses for DAGman nodes."""
135 # (STATUS_NOT_READY): At least one parent has not yet finished or the node
136 # is a FINAL node.
137 NOT_READY = 0
139 # (STATUS_READY): All parents have finished, but the node is not yet
140 # running.
141 READY = 1
143 # (STATUS_PRERUN): The node’s PRE script is running.
144 PRERUN = 2
146 # (STATUS_SUBMITTED): The node’s HTCondor job(s) are in the queue.
147 # StatusDetails = "not_idle" -> running.
148 # JobProcsHeld = 1-> hold.
149 # JobProcsQueued = 1 -> idle.
150 SUBMITTED = 3
152 # (STATUS_POSTRUN): The node’s POST script is running.
153 POSTRUN = 4
155 # (STATUS_DONE): The node has completed successfully.
156 DONE = 5
158 # (STATUS_ERROR): The node has failed. StatusDetails has info (e.g.,
159 # ULOG_JOB_ABORTED for deleted job).
160 ERROR = 6
162 # (STATUS_FUTILE): The node will never run because ancestor node failed.
163 FUTILE = 7
166class WmsNodeType(IntEnum):
167 """HTCondor plugin node types to help with payload reporting."""
169 UNKNOWN = auto()
170 """Dummy value when missing."""
172 PAYLOAD = auto()
173 """Payload job."""
175 FINAL = auto()
176 """Final job."""
178 SERVICE = auto()
179 """Service job."""
181 NOOP = auto()
182 """NOOP job used for ordering jobs."""
184 SUBDAG = auto()
185 """SUBDAG job used for ordering jobs."""
187 SUBDAG_CHECK = auto()
188 """Job used to correctly prune jobs after a subdag."""
191HTC_QUOTE_KEYS = {"environment"}
192HTC_VALID_JOB_KEYS = {
193 "universe",
194 "executable",
195 "arguments",
196 "environment",
197 "log",
198 "error",
199 "output",
200 "should_transfer_files",
201 "when_to_transfer_output",
202 "getenv",
203 "notification",
204 "notify_user",
205 "concurrency_limit",
206 "transfer_executable",
207 "transfer_input_files",
208 "transfer_output_files",
209 "transfer_output_remaps",
210 "request_cpus",
211 "request_memory",
212 "request_disk",
213 "priority",
214 "category",
215 "requirements",
216 "on_exit_hold",
217 "on_exit_hold_reason",
218 "on_exit_hold_subcode",
219 "max_retries",
220 "retry_until",
221 "periodic_release",
222 "periodic_remove",
223 "accounting_group",
224 "accounting_group_user",
225}
226HTC_VALID_JOB_DAG_KEYS = {
227 "dir",
228 "noop",
229 "done",
230 "vars",
231 "pre",
232 "post",
233 "retry",
234 "retry_unless_exit",
235 "abort_dag_on",
236 "abort_exit",
237 "priority",
238}
239HTC_VERSION = version.parse(htcondor.__version__)
242class RestrictedDict(MutableMapping):
243 """A dictionary that only allows certain keys.
245 Parameters
246 ----------
247 valid_keys : `~collections.abc.Container`
248 Strings that are valid keys.
249 init_data : `dict` or `RestrictedDict`, optional
250 Initial data.
252 Raises
253 ------
254 KeyError
255 If invalid key(s) in init_data.
256 """
258 def __init__(self, valid_keys, init_data=()):
259 self.valid_keys = valid_keys
260 self.data = {}
261 self.update(init_data)
263 def __getitem__(self, key):
264 """Return value for given key if exists.
266 Parameters
267 ----------
268 key : `str`
269 Identifier for value to return.
271 Returns
272 -------
273 value : `~typing.Any`
274 Value associated with given key.
276 Raises
277 ------
278 KeyError
279 If key doesn't exist.
280 """
281 return self.data[key]
283 def __delitem__(self, key):
284 """Delete value for given key if exists.
286 Parameters
287 ----------
288 key : `str`
289 Identifier for value to delete.
291 Raises
292 ------
293 KeyError
294 If key doesn't exist.
295 """
296 del self.data[key]
298 def __setitem__(self, key, value):
299 """Store key,value in internal dict only if key is valid.
301 Parameters
302 ----------
303 key : `str`
304 Identifier to associate with given value.
305 value : `~typing.Any`
306 Value to store.
308 Raises
309 ------
310 KeyError
311 If key is invalid.
312 """
313 if key not in self.valid_keys: 313 ↛ 314line 313 didn't jump to line 314 because the condition on line 313 was never true
314 raise KeyError(f"Invalid key {key}")
315 self.data[key] = value
317 def __iter__(self):
318 return self.data.__iter__()
320 def __len__(self):
321 return len(self.data)
323 def __str__(self):
324 return str(self.data)
327def htc_backup_files(
328 wms_path: str | os.PathLike, subdir: str | os.PathLike | None = None, limit: int = 100
329) -> Path | None:
330 """Backup select HTCondor files in the submit directory.
332 Files will be saved in separate subdirectories which will be created in
333 the submit directory where the files are located. These subdirectories
334 will be consecutive, zero-padded integers. Their values will correspond to
335 the number of HTCondor rescue DAGs in the submit directory.
337 Hence, with the default settings, copies after the initial failed run will
338 be placed in '001' subdirectory, '002' after the first restart, and so on
339 until the limit of backups is reached. If there's no rescue DAG yet, files
340 will be copied to '000' subdirectory.
342 Parameters
343 ----------
344 wms_path : `str` or `os.PathLike`
345 Path to the submit directory either absolute or relative.
346 subdir : `str` or `os.PathLike`, optional
347 A path, relative to the submit directory, where all subdirectories with
348 backup files will be kept. Defaults to None which means that the backup
349 subdirectories will be placed directly in the submit directory.
350 limit : `int`, optional
351 Maximal number of backups. If the number of backups reaches the limit,
352 the last backup files will be overwritten. The default value is 100
353 to match the default value of HTCondor's DAGMAN_MAX_RESCUE_NUM in
354 version 8.8+.
356 Returns
357 -------
358 last_rescue_file : `pathlib.Path` or None
359 Path to the latest rescue file or None if doesn't exist.
361 Raises
362 ------
363 FileNotFoundError
364 If the submit directory or the file that needs to be backed up does not
365 exist.
366 OSError
367 If the submit directory cannot be accessed or backing up a file failed
368 either due to permission or filesystem related issues.
370 Notes
371 -----
372 This is not a generic function for making backups. It is intended to be
373 used once, just before a restart, to make snapshots of files which will be
374 overwritten by HTCondor after during the next run.
375 """
376 width = len(str(limit))
378 path = Path(wms_path).resolve()
379 if not path.is_dir():
380 raise FileNotFoundError(f"Directory {path} not found")
382 # Initialize the backup counter.
383 rescue_dags = list(path.glob("*.rescue[0-9][0-9][0-9]"))
384 counter = min(len(rescue_dags), limit)
386 # Create the backup directory and move select files there.
387 dest = path
388 if subdir:
389 # PurePath.is_relative_to() is not available before Python 3.9. Hence
390 # we need to check is 'subdir' is in the submit directory in some other
391 # way if it is an absolute path.
392 subdir = Path(subdir)
393 if subdir.is_absolute():
394 subdir = subdir.resolve() # Since resolve was run on path, must run it here
395 if dest not in subdir.parents:
396 _LOG.warning(
397 "Invalid backup location: '%s' not in the submit directory, will use '%s' instead.",
398 subdir,
399 wms_path,
400 )
401 else:
402 dest /= subdir
403 else:
404 dest /= subdir
405 dest /= f"{counter:0{width}}"
406 _LOG.debug("dest = %s", dest)
407 try:
408 dest.mkdir(parents=True, exist_ok=False if counter < limit else True)
409 except FileExistsError:
410 _LOG.warning("Refusing to do backups: target directory '%s' already exists", dest)
411 else:
412 htc_backup_files_single_path(path, dest)
414 # also back up any subdag info
415 for subdag_dir in path.glob("subdags/*"):
416 subdag_dest = dest / subdag_dir.relative_to(path)
417 subdag_dest.mkdir(parents=True, exist_ok=False)
418 htc_backup_files_single_path(subdag_dir, subdag_dest)
420 last_rescue_file = rescue_dags[-1] if rescue_dags else None
421 _LOG.debug("last_rescue_file = %s", last_rescue_file)
422 return last_rescue_file
425def htc_backup_files_single_path(src: str | os.PathLike, dest: str | os.PathLike) -> None:
426 """Move particular htc files to a different directory for later debugging.
428 Parameters
429 ----------
430 src : `str` or `os.PathLike`
431 Directory from which to backup particular files.
432 dest : `str` or `os.PathLike`
433 Directory to which particular files are moved.
435 Raises
436 ------
437 RuntimeError
438 If given dest directory matches given src directory.
439 OSError
440 If problems moving file.
441 FileNotFoundError
442 Item matching pattern in src directory isn't a file.
443 """
444 src = Path(src)
445 dest = Path(dest)
446 if dest.samefile(src):
447 raise RuntimeError(f"Destination directory is same as the source directory ({src})")
449 for patt in [
450 "*.info.*",
451 "*.dag.metrics",
452 "*.dag.nodes.log",
453 "*.node_status",
454 "wms_*.dag.post.out",
455 "wms_*.status.txt",
456 ]:
457 for source in src.glob(patt):
458 if source.is_file(): 458 ↛ 465line 458 didn't jump to line 465 because the condition on line 458 was always true
459 target = dest / source.relative_to(src)
460 try:
461 source.rename(target)
462 except OSError as exc:
463 raise type(exc)(f"Backing up '{source}' failed: {exc.strerror}") from None
464 else:
465 raise FileNotFoundError(f"Backing up '{source}' failed: not a file")
468def htc_escape(value):
469 """Escape characters in given value based upon HTCondor syntax.
471 Parameters
472 ----------
473 value : `~typing.Any`
474 Value that needs to have characters escaped if string.
476 Returns
477 -------
478 new_value : `~typing.Any`
479 Given value with characters escaped appropriate for HTCondor if string.
480 """
481 if isinstance(value, str):
482 newval = value.replace('"', '""').replace("'", "''").replace(""", '"')
483 else:
484 newval = value
486 return newval
489def htc_write_attribs(stream, attrs):
490 """Write job attributes in HTCondor format to writeable stream.
492 Parameters
493 ----------
494 stream : `~typing.TextIO`
495 Output text stream (typically an open file).
496 attrs : `dict`
497 HTCondor job attributes (dictionary of attribute key, value).
498 """
499 for key, value in attrs.items():
500 # Make sure strings are syntactically correct for HTCondor.
501 if isinstance(value, str): 501 ↛ 504line 501 didn't jump to line 504 because the condition on line 501 was always true
502 pval = f'"{htc_escape(value)}"'
503 else:
504 pval = value
506 print(f"+{key} = {pval}", file=stream)
509def htc_write_condor_file(
510 filename: str | os.PathLike, job_name: str, job: RestrictedDict, job_attrs: dict[str, Any]
511) -> None:
512 """Write an HTCondor submit file.
514 Parameters
515 ----------
516 filename : `str` or `os.PathLike`
517 Filename for the HTCondor submit file.
518 job_name : `str`
519 Job name to use in submit file.
520 job : `RestrictedDict`
521 Submit script information.
522 job_attrs : `dict`
523 Job attributes.
524 """
525 os.makedirs(os.path.dirname(filename), exist_ok=True)
526 with open(filename, "w") as fh:
527 for key, value in job.items():
528 if value is not None: 528 ↛ 527line 528 didn't jump to line 527 because the condition on line 528 was always true
529 if key in HTC_QUOTE_KEYS: # Assumes internal quotes are already escaped correctly
530 print(f'{key}="{value}"', file=fh)
531 else:
532 print(f"{key}={value}", file=fh)
533 for key in ["output", "error", "log"]:
534 if key not in job:
535 filename = f"{job_name}.$(Cluster).{'out' if key != 'log' else key}"
536 print(f"{key}={filename}", file=fh)
538 if job_attrs is not None:
539 htc_write_attribs(fh, job_attrs)
540 print("queue", file=fh)
543# To avoid doing the version check during every function call select
544# appropriate conversion function at the import time.
545#
546# Make sure that *each* version specific variant of the conversion function(s)
547# has the same signature after applying any changes!
548if HTC_VERSION < version.parse("8.9.8"): 548 ↛ 550line 548 didn't jump to line 550 because the condition on line 548 was never true
550 def htc_tune_schedd_args(**kwargs):
551 """Ensure that arguments for Schedd are version appropriate.
553 The old arguments: 'requirements' and 'attr_list' of
554 'Schedd.history()', 'Schedd.query()', and 'Schedd.xquery()' were
555 deprecated in favor of 'constraint' and 'projection', respectively,
556 starting from version 8.9.8. The function will convert "new" keyword
557 arguments to "old" ones.
559 Parameters
560 ----------
561 **kwargs
562 Any keyword arguments that Schedd.history(), Schedd.query(), and
563 Schedd.xquery() accepts.
565 Returns
566 -------
567 kwargs : `dict` [`str`, `~typing.Any`]
568 Keywords arguments that are guaranteed to work with the Python
569 HTCondor API.
571 Notes
572 -----
573 Function doesn't validate provided keyword arguments beyond converting
574 selected arguments to their version specific form. For example,
575 it won't remove keywords that are not supported by the methods
576 mentioned earlier.
577 """
578 translation_table = {
579 "constraint": "requirements",
580 "projection": "attr_list",
581 }
582 for new, old in translation_table.items():
583 try:
584 kwargs[old] = kwargs.pop(new)
585 except KeyError:
586 pass
587 return kwargs
589else:
591 def htc_tune_schedd_args(**kwargs):
592 """Ensure that arguments for Schedd are version appropriate.
594 This is the fallback function if no version specific alteration are
595 necessary. Effectively, a no-op.
597 Parameters
598 ----------
599 **kwargs
600 Any keyword arguments that Schedd.history(), Schedd.query(), and
601 Schedd.xquery() accepts.
603 Returns
604 -------
605 kwargs : `dict` [`str`, `~typing.Any`]
606 Keywords arguments that were passed to the function.
607 """
608 return kwargs
611def htc_query_history(schedds, **kwargs):
612 """Fetch history records from the condor_schedd daemon.
614 Parameters
615 ----------
616 schedds : `htcondor.Schedd`
617 HTCondor schedulers which to query for job information.
618 **kwargs
619 Any keyword arguments that Schedd.history() accepts.
621 Yields
622 ------
623 schedd_name : `str`
624 Name of the HTCondor scheduler managing the job queue.
625 job_ad : `dict` [`str`, `~typing.Any`]
626 A dictionary representing HTCondor ClassAd describing a job. It maps
627 job attributes names to values of the ClassAd expressions they
628 represent.
629 """
630 # If not set, provide defaults for positional arguments.
631 kwargs.setdefault("constraint", None)
632 kwargs.setdefault("projection", [])
633 kwargs = htc_tune_schedd_args(**kwargs)
634 for schedd_name, schedd in schedds.items():
635 for job_ad in schedd.history(**kwargs):
636 yield schedd_name, dict(job_ad)
639def htc_query_present(schedds, **kwargs):
640 """Query the condor_schedd daemon for job ads.
642 Parameters
643 ----------
644 schedds : `htcondor.Schedd`
645 HTCondor schedulers which to query for job information.
646 **kwargs
647 Any keyword arguments that Schedd.xquery() accepts.
649 Yields
650 ------
651 schedd_name : `str`
652 Name of the HTCondor scheduler managing the job queue.
653 job_ad : `dict` [`str`, `~typing.Any`]
654 A dictionary representing HTCondor ClassAd describing a job. It maps
655 job attributes names to values of the ClassAd expressions they
656 represent.
657 """
658 kwargs = htc_tune_schedd_args(**kwargs)
659 for schedd_name, schedd in schedds.items():
660 for job_ad in schedd.query(**kwargs):
661 yield schedd_name, dict(job_ad)
664def htc_version():
665 """Return the version given by the HTCondor API.
667 Returns
668 -------
669 version : `str`
670 HTCondor version as easily comparable string.
671 """
672 return str(HTC_VERSION)
675def htc_submit_dag(sub):
676 """Submit job for execution.
678 Parameters
679 ----------
680 sub : `htcondor.Submit`
681 An object representing a job submit description.
683 Returns
684 -------
685 schedd_job_info : `dict` [`str`, `dict` [`str`, \
686 `dict` [`str`, `~typing.Any`]]]
687 Information about jobs satisfying the search criteria where for each
688 Scheduler, local HTCondor job ids are mapped to their respective
689 classads.
690 """
691 coll = htcondor.Collector()
692 schedd_ad = coll.locate(htcondor.DaemonTypes.Schedd)
693 schedd = htcondor.Schedd(schedd_ad)
695 # If Schedd.submit() fails, the method will raise an exception. Usually,
696 # that implies issues with the HTCondor pool which BPS can't address.
697 # Hence, no effort is made to handle the exception.
698 submit_result = schedd.submit(sub)
700 # Sadly, the ClassAd from Schedd.submit() (see above) does not have
701 # 'GlobalJobId' so we need to run a regular query to get it anyway.
702 schedd_name = schedd_ad["Name"]
703 schedd_dag_info = condor_q(
704 constraint=f"ClusterId == {submit_result.cluster()}", schedds={schedd_name: schedd}
705 )
706 return schedd_dag_info
709def htc_create_submit_from_dag(
710 dag_filename: str, submit_options: dict[str, Any], dagman_conf_filename: str | os.PathLike | None = None
711) -> htcondor.Submit:
712 """Create a DAGMan job submit description.
714 Parameters
715 ----------
716 dag_filename : `str`
717 Name of file containing HTCondor DAG commands.
718 submit_options : `dict` [`str`, `~typing.Any`], optional
719 Contains extra options for command line (Value of None means flag).
720 dagman_conf_filename : `str` or `os.PathLike`, optional
721 Location of DAGMan configuration file, if Any. Defaults to no
722 file (None).
724 Returns
725 -------
726 sub : `htcondor.Submit`
727 An object representing a job submit description.
729 Notes
730 -----
731 Use with HTCondor versions which support htcondor.Submit.from_dag(),
732 i.e., 8.9.3 or newer.
733 """
734 # Config and environment variables do not seem to override -MaxIdle
735 # on the .dag.condor.sub's command line (broken in some 24.0.x versions).
736 # Explicitly forward them as a submit_option if either exists.
737 # Note: auto generated subdag submit files are still the -MaxIdle=1000
738 # in the broken versions.
739 if "MaxIdle" not in submit_options:
740 max_jobs_idle: int | None = None
741 config_var_name = "DAGMAN_MAX_JOBS_IDLE"
743 if dagman_conf_filename:
744 _LOG.debug("Checking DAGMan config file = %s", dagman_conf_filename)
745 with open(dagman_conf_filename) as fh:
746 for line in fh:
747 _LOG.debug("DAGMan config file line = %s", line)
748 parts = line.split("=")
749 _LOG.debug("DAGMan config file line parts = %s", parts)
750 if len(parts) == 2 and parts[0].strip() == config_var_name:
751 max_jobs_idle = int(parts[1].strip())
752 _LOG.debug("Found %s = %s", config_var_name, max_jobs_idle)
753 break
754 if max_jobs_idle is None:
755 if f"_CONDOR_{config_var_name}" in os.environ:
756 max_jobs_idle = int(os.environ[f"_CONDOR_{config_var_name}"])
757 elif config_var_name in htcondor.param:
758 max_jobs_idle = htcondor.param[config_var_name]
760 if max_jobs_idle:
761 submit_options["MaxIdle"] = max_jobs_idle
762 else:
763 _LOG.debug("MaxIdle already in submit_options: %s", submit_options)
765 _LOG.debug("Using submit_options = %s", submit_options)
766 return htcondor.Submit.from_dag(dag_filename, submit_options)
769def htc_create_submit_from_cmd(dag_filename, submit_options=None):
770 """Create a DAGMan job submit description.
772 Create a DAGMan job submit description by calling ``condor_submit_dag``
773 on given DAG description file.
775 Parameters
776 ----------
777 dag_filename : `str`
778 Name of file containing HTCondor DAG commands.
779 submit_options : `dict` [`str`, `~typing.Any`], optional
780 Contains extra options for command line (Value of None means flag).
782 Returns
783 -------
784 sub : `htcondor.Submit`
785 An object representing a job submit description.
787 Notes
788 -----
789 Use with HTCondor versions which do not support htcondor.Submit.from_dag(),
790 i.e., older than 8.9.3.
791 """
792 # Run command line condor_submit_dag command.
793 cmd = "condor_submit_dag -f -no_submit -notification never -autorescue 1 -UseDagDir -no_recurse "
795 if submit_options is not None:
796 for opt, val in submit_options.items():
797 cmd += f" -{opt} {val or ''}"
798 cmd += f"{dag_filename}"
800 process = subprocess.Popen(
801 cmd.split(), shell=False, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, encoding="utf-8"
802 )
803 process.wait()
805 if process.returncode != 0:
806 print(f"Exit code: {process.returncode}")
807 print(process.communicate()[0])
808 raise RuntimeError("Problems running condor_submit_dag")
810 return htc_create_submit_from_file(f"{dag_filename}.condor.sub")
813def htc_create_submit_from_file(submit_file):
814 """Parse a submission file.
816 Parameters
817 ----------
818 submit_file : `str`
819 Name of the HTCondor submit file.
821 Returns
822 -------
823 sub : `htcondor.Submit`
824 An object representing a job submit description.
825 """
826 descriptors = {}
827 with open(submit_file) as fh:
828 for line in fh:
829 line = line.strip()
830 if not line.startswith("#") and not line == "queue":
831 (key, val) = re.split(r"\s*=\s*", line, maxsplit=1)
832 descriptors[key] = val
834 # Avoid UserWarning: the line 'copy_to_spool = False' was
835 # unused by Submit object. Is it a typo?
836 try:
837 del descriptors["copy_to_spool"]
838 except KeyError:
839 pass
841 return htcondor.Submit(descriptors)
844def _htc_write_job_commands(stream, name, commands, node_type="JOB"):
845 """Output the DAGMan job lines for single job in DAG.
847 Parameters
848 ----------
849 stream : `~typing.TextIO`
850 Writeable text stream (typically an opened file).
851 name : `str`
852 Job name.
853 commands : `RestrictedDict`
854 DAG commands for a job.
855 node_type : `str`, optional
856 Type of DAGMan node (JOB, FINAL, SERVICE). Defaults to "JOB".
857 """
858 # Note: optional pieces of commands include a space at the beginning.
859 # also making sure values aren't empty strings as placeholders.
860 if "pre" in commands and commands["pre"]:
861 defer = ""
862 if "defer" in commands["pre"] and commands["pre"]["defer"]:
863 defer = f" DEFER {commands['pre']['defer']['status']} {commands['pre']['defer']['time']}"
865 debug = ""
866 if "debug" in commands["pre"] and commands["pre"]["debug"]:
867 debug = f" DEBUG {commands['pre']['debug']['filename']} {commands['pre']['debug']['type']}"
869 arguments = ""
870 if "arguments" in commands["pre"] and commands["pre"]["arguments"]:
871 arguments = f" {commands['pre']['arguments']}"
873 executable = commands["pre"]["executable"]
874 print(f"SCRIPT{defer}{debug} PRE {name} {executable}{arguments}", file=stream)
876 if "post" in commands and commands["post"]:
877 defer = ""
878 if "defer" in commands["post"] and commands["post"]["defer"]:
879 defer = f" DEFER {commands['post']['defer']['status']} {commands['post']['defer']['time']}"
881 debug = ""
882 if "debug" in commands["post"] and commands["post"]["debug"]:
883 debug = f" DEBUG {commands['post']['debug']['filename']} {commands['post']['debug']['type']}"
885 arguments = ""
886 if "arguments" in commands["post"] and commands["post"]["arguments"]:
887 arguments = f" {commands['post']['arguments']}"
889 executable = commands["post"]["executable"]
890 print(f"SCRIPT{defer}{debug} POST {name} {executable}{arguments}", file=stream)
892 if "vars" in commands and commands["vars"]:
893 for key, value in commands["vars"].items():
894 print(f'VARS {name} {key}="{htc_escape(value)}"', file=stream)
896 if "pre_skip" in commands and commands["pre_skip"]:
897 print(f"PRE_SKIP {name} {commands['pre_skip']}", file=stream)
899 # FINAL node cannot have a DAGMan retry, abort-dag-on, priority, category
900 if node_type != "FINAL":
901 if "retry" in commands and commands["retry"]:
902 print(f"RETRY {name} {commands['retry']}", end="", file=stream)
903 if "retry_unless_exit" in commands and commands["retry_unless_exit"]:
904 print(f" UNLESS-EXIT {commands['retry_unless_exit']}", end="", file=stream)
905 print("", file=stream) # Since previous prints don't include new line
907 if "abort_dag_on" in commands and commands["abort_dag_on"]:
908 print(
909 f"ABORT-DAG-ON {name} {commands['abort_dag_on']['node_exit']}"
910 f" RETURN {commands['abort_dag_on']['abort_exit']}",
911 file=stream,
912 )
914 if "priority" in commands and commands["priority"]:
915 print(
916 f"PRIORITY {name} {commands['priority']}",
917 file=stream,
918 )
921class HTCJob:
922 """HTCondor job for use in building DAG.
924 Parameters
925 ----------
926 name : `str`
927 Name of the job.
928 label : `str`
929 Label that can used for grouping or lookup.
930 initcmds : `RestrictedDict`
931 Initial job commands for submit file.
932 initdagcmds : `RestrictedDict`
933 Initial commands for job inside DAG.
934 initattrs : `dict`
935 Initial dictionary of job attributes.
936 """
938 def __init__(self, name, label=None, initcmds=(), initdagcmds=(), initattrs=None):
939 self.name = name
940 self.label = label
941 self.cmds = RestrictedDict(HTC_VALID_JOB_KEYS, initcmds)
942 self.dagcmds = RestrictedDict(HTC_VALID_JOB_DAG_KEYS, initdagcmds)
943 self.attrs = initattrs
944 self.subfile = None
945 self.subdir = None
946 self.subdag = None
948 def __str__(self):
949 return self.name
951 def add_job_cmds(self, new_commands):
952 """Add commands to Job (overwrite existing).
954 Parameters
955 ----------
956 new_commands : `dict`
957 Submit file commands to be added to Job.
958 """
959 self.cmds.update(new_commands)
961 def add_dag_cmds(self, new_commands):
962 """Add DAG commands to Job (overwrite existing).
964 Parameters
965 ----------
966 new_commands : `dict`
967 DAG file commands to be added to Job.
968 """
969 self.dagcmds.update(new_commands)
971 def add_job_attrs(self, new_attrs):
972 """Add attributes to Job (overwrite existing).
974 Parameters
975 ----------
976 new_attrs : `dict`
977 Attributes to be added to Job.
978 """
979 if self.attrs is None:
980 self.attrs = {}
981 if new_attrs:
982 self.attrs.update(new_attrs)
984 def write_submit_file(self, submit_path: str | os.PathLike) -> None:
985 """Write job description to submit file.
987 Parameters
988 ----------
989 submit_path : `str` or `os.PathLike`
990 Prefix path for the submit file.
991 """
992 if not self.subfile:
993 self.subfile = f"{self.name}.sub"
995 subfile = self.subfile
996 if self.subdir: 996 ↛ 997line 996 didn't jump to line 997 because the condition on line 996 was never true
997 subfile = Path(self.subdir) / subfile
999 subfile = Path(os.path.expandvars(subfile))
1000 if not subfile.is_absolute():
1001 subfile = Path(submit_path) / subfile
1002 if not subfile.exists():
1003 htc_write_condor_file(subfile, self.name, self.cmds, self.attrs)
1005 def write_dag_commands(self, stream, dag_rel_path, command_name="JOB"):
1006 """Write DAG commands for single job to output stream.
1008 Parameters
1009 ----------
1010 stream : `~typing.TextIO`
1011 Output Stream.
1012 dag_rel_path : `str`
1013 Relative path of dag to submit directory.
1014 command_name : `str`
1015 Name of the DAG command (e.g., JOB, FINAL).
1016 """
1017 subfile = os.path.expandvars(self.subfile)
1019 # JOB NodeName SubmitDescription [DIR directory] [NOOP] [DONE]
1020 job_line = f'{command_name} {self.name} "{subfile}"'
1021 if "dir" in self.dagcmds:
1022 dir_val = self.dagcmds["dir"]
1023 if dag_rel_path: 1023 ↛ 1025line 1023 didn't jump to line 1025 because the condition on line 1023 was always true
1024 dir_val = os.path.join(dag_rel_path, dir_val)
1025 job_line += f' DIR "{dir_val}"'
1026 if self.dagcmds.get("noop", False):
1027 job_line += " NOOP"
1029 print(job_line, file=stream)
1030 if self.dagcmds:
1031 _htc_write_job_commands(stream, self.name, self.dagcmds, command_name)
1033 def dump(self, fh):
1034 """Dump job information to output stream.
1036 Parameters
1037 ----------
1038 fh : `~typing.TextIO`
1039 Output stream.
1040 """
1041 printer = pprint.PrettyPrinter(indent=4, stream=fh)
1042 printer.pprint(self.name)
1043 printer.pprint(self.cmds)
1044 printer.pprint(self.attrs)
1047class HTCDag(networkx.DiGraph):
1048 """HTCondor DAG.
1050 Parameters
1051 ----------
1052 data : `~typing.Any`
1053 Initial graph data of any format that is supported
1054 by the to_network_graph() function.
1055 name : `str`
1056 Name for DAG.
1057 """
1059 def __init__(self, data=None, name=""):
1060 super().__init__(data=data, name=name)
1062 self.graph["attr"] = {}
1063 self.graph["run_id"] = None
1064 self.graph["submit_path"] = None
1065 self.graph["final_job"] = None
1066 self.graph["service_job"] = None
1067 self.graph["submit_options"] = {}
1069 def __str__(self):
1070 """Represent basic DAG info as string.
1072 Returns
1073 -------
1074 info : `str`
1075 String containing basic DAG info.
1076 """
1077 return f"{self.graph['name']} {len(self)}"
1079 def add_attribs(self, attribs=None):
1080 """Add attributes to the DAG.
1082 Parameters
1083 ----------
1084 attribs : `dict`
1085 DAG attributes.
1086 """
1087 if attribs is not None: 1087 ↛ exitline 1087 didn't return from function 'add_attribs' because the condition on line 1087 was always true
1088 self.graph["attr"].update(attribs)
1090 def add_job(self, job, parent_names=None, child_names=None):
1091 """Add an HTCJob to the HTCDag.
1093 Parameters
1094 ----------
1095 job : `HTCJob`
1096 HTCJob to add to the HTCDag.
1097 parent_names : `~collections.abc.Iterable` [`str`], optional
1098 Names of parent jobs.
1099 child_names : `~collections.abc.Iterable` [`str`], optional
1100 Names of child jobs.
1101 """
1102 assert isinstance(job, HTCJob)
1103 _LOG.debug("Adding job %s to dag", job.name)
1105 # Add dag level attributes to each job
1106 job.add_job_attrs(self.graph["attr"])
1108 self.add_node(job.name, data=job)
1110 if parent_names is not None: 1110 ↛ 1111line 1110 didn't jump to line 1111 because the condition on line 1110 was never true
1111 self.add_job_relationships(parent_names, [job.name])
1113 if child_names is not None: 1113 ↛ 1114line 1113 didn't jump to line 1114 because the condition on line 1113 was never true
1114 self.add_job_relationships(child_names, [job.name])
1116 def add_job_relationships(self, parents, children):
1117 """Add DAG edge between parents and children jobs.
1119 Parameters
1120 ----------
1121 parents : `list` [`str`]
1122 Contains parent job name(s).
1123 children : `list` [`str`]
1124 Contains children job name(s).
1125 """
1126 self.add_edges_from(itertools.product(parents, children))
1128 def add_final_job(self, job):
1129 """Add an HTCJob for the FINAL job in HTCDag.
1131 Parameters
1132 ----------
1133 job : `HTCJob`
1134 HTCJob to add to the HTCDag as a FINAL job.
1135 """
1136 # Add dag level attributes to each job
1137 job.add_job_attrs(self.graph["attr"])
1139 self.graph["final_job"] = job
1141 def add_service_job(self, job):
1142 """Add an HTCJob for the SERVICE job in HTCDag.
1144 Parameters
1145 ----------
1146 job : `HTCJob`
1147 HTCJob to add to the HTCDag as a FINAL job.
1148 """
1149 # Add dag level attributes to each job
1150 job.add_job_attrs(self.graph["attr"])
1152 self.graph["service_job"] = job
1154 def del_job(self, job_name):
1155 """Delete the job from the DAG.
1157 Parameters
1158 ----------
1159 job_name : `str`
1160 Name of job in DAG to delete.
1161 """
1162 # Reconnect edges around node to delete
1163 parents = self.predecessors(job_name)
1164 children = self.successors(job_name)
1165 self.add_edges_from(itertools.product(parents, children))
1167 # Delete job node (which deletes its edges).
1168 self.remove_node(job_name)
1170 def write(self, submit_path, job_subdir="", dag_subdir="", dag_rel_path=""):
1171 """Write DAG to a file.
1173 Parameters
1174 ----------
1175 submit_path : `str`
1176 Prefix path for all outputs.
1177 job_subdir : `str`, optional
1178 Template for job subdir (submit_path + job_subdir).
1179 dag_subdir : `str`, optional
1180 DAG subdir (submit_path + dag_subdir).
1181 dag_rel_path : `str`, optional
1182 Prefix to job_subdir for jobs inside subdag.
1183 """
1184 self.graph["submit_path"] = submit_path
1185 self.graph["dag_filename"] = os.path.join(dag_subdir, f"{self.graph['name']}.dag")
1186 full_filename = os.path.join(submit_path, self.graph["dag_filename"])
1187 os.makedirs(os.path.dirname(full_filename), exist_ok=True)
1189 try:
1190 dagman_config_path = Path(self.graph["attr"]["bps_wms_config_path"])
1191 except KeyError:
1192 dagman_config_path = None
1193 with open(full_filename, "w") as fh:
1194 if dagman_config_path is not None:
1195 fh.write(f"CONFIG {dag_rel_path / dagman_config_path}\n")
1197 for name, nodeval in self.nodes().items():
1198 try:
1199 job = nodeval["data"]
1200 except KeyError:
1201 _LOG.error("Job %s doesn't have data (keys: %s).", name, nodeval.keys())
1202 raise
1203 if job.subdag: 1203 ↛ 1204line 1203 didn't jump to line 1204 because the condition on line 1203 was never true
1204 dag_subdir = f"subdags/{job.name}"
1205 if "dir" in job.dagcmds:
1206 subdir = job.dagcmds["dir"]
1207 else:
1208 subdir = job_subdir
1209 if dagman_config_path is not None:
1210 job.subdag.add_attribs({"bps_wms_config_path": str(dagman_config_path)})
1211 job.subdag.write(submit_path, subdir, dag_subdir, "../..")
1212 fh.write(
1213 f"SUBDAG EXTERNAL {job.name} {Path(job.subdag.graph['dag_filename']).name} "
1214 f"DIR {dag_subdir}\n"
1215 )
1216 if job.dagcmds:
1217 _htc_write_job_commands(fh, job.name, job.dagcmds)
1218 else:
1219 job.write_submit_file(submit_path)
1220 job.write_dag_commands(fh, dag_rel_path)
1222 for edge in self.edges(): 1222 ↛ 1223line 1222 didn't jump to line 1223 because the loop on line 1222 never started
1223 print(f"PARENT {edge[0]} CHILD {edge[1]}", file=fh)
1224 print(f"DOT {self.name}.dot", file=fh)
1225 print(f"NODE_STATUS_FILE {self.name}.node_status", file=fh)
1227 # Add bps attributes to dag submission
1228 for key, value in self.graph["attr"].items():
1229 print(f'SET_JOB_ATTR {key}= "{htc_escape(value)}"', file=fh)
1231 # Add special nodes if any.
1232 special_jobs = {
1233 "FINAL": self.graph["final_job"],
1234 "SERVICE": self.graph["service_job"],
1235 }
1236 for dagcmd, job in special_jobs.items():
1237 if job is not None: 1237 ↛ 1238line 1237 didn't jump to line 1238 because the condition on line 1237 was never true
1238 job.write_submit_file(submit_path)
1239 job.write_dag_commands(fh, dag_rel_path, dagcmd)
1241 def dump(self, fh):
1242 """Dump DAG info to output stream.
1244 Parameters
1245 ----------
1246 fh : `typing.IO`
1247 Where to dump DAG info as text.
1248 """
1249 for key, value in self.graph:
1250 print(f"{key}={value}", file=fh)
1251 for name, data in self.nodes().items():
1252 print(f"{name}:", file=fh)
1253 data.dump(fh)
1254 for edge in self.edges():
1255 print(f"PARENT {edge[0]} CHILD {edge[1]}", file=fh)
1256 if self.graph["final_job"]:
1257 print(f"FINAL {self.graph['final_job'].name}:", file=fh)
1258 self.graph["final_job"].dump(fh)
1260 def write_dot(self, filename):
1261 """Write a dot version of the DAG.
1263 Parameters
1264 ----------
1265 filename : `str`
1266 Name of the dot file.
1267 """
1268 pos = networkx.nx_agraph.graphviz_layout(self)
1269 networkx.draw(self, pos=pos)
1270 networkx.drawing.nx_pydot.write_dot(self, filename)
1273def condor_q(constraint=None, schedds=None, **kwargs):
1274 """Get information about the jobs in the HTCondor job queue(s).
1276 Parameters
1277 ----------
1278 constraint : `str`, optional
1279 Constraints to be passed to job query.
1280 schedds : `dict` [`str`, `htcondor.Schedd`], optional
1281 HTCondor schedulers which to query for job information. If None
1282 (default), the query will be run against local scheduler only.
1283 **kwargs : `~typing.Any`
1284 Additional keyword arguments that need to be passed to the internal
1285 query method.
1287 Returns
1288 -------
1289 job_info : `dict` [`str`, `dict` [`str`, `dict` [`str`, `~typing.Any`]]]
1290 Information about jobs satisfying the search criteria where for each
1291 Scheduler, local HTCondor job ids are mapped to their respective
1292 classads.
1293 """
1294 return condor_query(constraint, schedds, htc_query_present, **kwargs)
1297def condor_history(constraint=None, schedds=None, **kwargs):
1298 """Get information about the jobs from HTCondor history records.
1300 Parameters
1301 ----------
1302 constraint : `str`, optional
1303 Constraints to be passed to job query.
1304 schedds : `dict` [`str`, `htcondor.Schedd`], optional
1305 HTCondor schedulers which to query for job information. If None
1306 (default), the query will be run against the history file of
1307 the local scheduler only.
1308 **kwargs : `~typing.Any`
1309 Additional keyword arguments that need to be passed to the internal
1310 query method.
1312 Returns
1313 -------
1314 job_info : `dict` [`str`, `dict` [`str`, `dict` [`str`, `~typing.Any`]]]
1315 Information about jobs satisfying the search criteria where for each
1316 Scheduler, local HTCondor job ids are mapped to their respective
1317 classads.
1318 """
1319 return condor_query(constraint, schedds, htc_query_history, **kwargs)
1322def condor_query(constraint=None, schedds=None, query_func=htc_query_present, **kwargs):
1323 """Get information about HTCondor jobs.
1325 Parameters
1326 ----------
1327 constraint : `str`, optional
1328 Constraints to be passed to job query.
1329 schedds : `dict` [`str`, `htcondor.Schedd`], optional
1330 HTCondor schedulers which to query for job information. If None
1331 (default), the query will be run against the history file of
1332 the local scheduler only.
1333 query_func : `~collections.abc.Callable`
1334 An query function which takes following arguments:
1336 - ``schedds``: Schedulers to query (`list` [`htcondor.Schedd`]).
1337 - ``**kwargs``: Keyword arguments that will be passed to the query
1338 function.
1339 **kwargs : `~typing.Any`
1340 Additional keyword arguments that need to be passed to the query
1341 method.
1343 Returns
1344 -------
1345 job_info : `dict` [`str`, `dict` [`str`, `dict` [`str`, `~typing.Any`]]]
1346 Information about jobs satisfying the search criteria where for each
1347 Scheduler, local HTCondor job ids are mapped to their respective
1348 classads.
1349 """
1350 if not schedds:
1351 coll = htcondor.Collector()
1352 schedd_ad = coll.locate(htcondor.DaemonTypes.Schedd)
1353 schedds = {schedd_ad["Name"]: htcondor.Schedd(schedd_ad)}
1355 # Make sure that 'ClusterId' and 'ProcId' attributes are always included
1356 # in the job classad. They are needed to construct the job id.
1357 added_attrs = set()
1358 if "projection" in kwargs and kwargs["projection"]:
1359 requested_attrs = set(kwargs["projection"])
1360 required_attrs = {"ClusterId", "ProcId"}
1361 added_attrs = required_attrs - requested_attrs
1362 for attr in added_attrs:
1363 kwargs["projection"].append(attr)
1365 unwanted_attrs = {"Env", "Environment"} | added_attrs
1366 job_info = defaultdict(dict)
1367 for schedd_name, job_ad in query_func(schedds, constraint=constraint, **kwargs):
1368 id_ = f"{job_ad['ClusterId']}.{job_ad['ProcId']}"
1369 for attr in set(job_ad) & unwanted_attrs:
1370 del job_ad[attr]
1371 job_info[schedd_name][id_] = job_ad
1372 _LOG.debug("query returned %d jobs", sum(len(val) for val in job_info.values()))
1374 # Restore the list of the requested attributes to its original value
1375 # if needed.
1376 if added_attrs:
1377 for attr in added_attrs:
1378 kwargs["projection"].remove(attr)
1380 # When returning the results filter out entries for schedulers with no jobs
1381 # matching the search criteria.
1382 return {key: val for key, val in job_info.items() if val}
1385def condor_search(constraint=None, hist=None, schedds=None):
1386 """Search for running and finished jobs satisfying given criteria.
1388 Parameters
1389 ----------
1390 constraint : `str`, optional
1391 Constraints to be passed to job query.
1392 hist : `float`
1393 Limit history search to this many days.
1394 schedds : `dict` [`str`, `htcondor.Schedd`], optional
1395 The list of the HTCondor schedulers which to query for job information.
1396 If None (default), only the local scheduler will be queried.
1398 Returns
1399 -------
1400 job_info : `dict` [`str`, `dict` [`str`, `dict` [`str` `~typing.Any`]]]
1401 Information about jobs satisfying the search criteria where for each
1402 Scheduler, local HTCondor job ids are mapped to their respective
1403 classads.
1404 """
1405 if not schedds:
1406 coll = htcondor.Collector()
1407 schedd_ad = coll.locate(htcondor.DaemonTypes.Schedd)
1408 schedds = {schedd_ad["Name"]: htcondor.Schedd(locate_ad=schedd_ad)}
1410 job_info = condor_q(constraint=constraint, schedds=schedds)
1411 if hist is not None:
1412 _LOG.debug("Searching history going back %s days", hist)
1413 epoch = (datetime.now() - timedelta(days=hist)).timestamp()
1414 constraint += f" && (CompletionDate >= {epoch} || JobFinishedHookDone >= {epoch})"
1415 hist_info = condor_history(constraint, schedds=schedds)
1416 update_job_info(job_info, hist_info)
1417 return job_info
1420def condor_status(constraint=None, coll=None):
1421 """Get information about HTCondor pool.
1423 Parameters
1424 ----------
1425 constraint : `str`, optional
1426 Constraints to be passed to the query.
1427 coll : `htcondor.Collector`, optional
1428 Object representing HTCondor collector daemon.
1430 Returns
1431 -------
1432 pool_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1433 Mapping between HTCondor slot names and slot information (classAds).
1434 """
1435 if coll is None:
1436 coll = htcondor.Collector()
1437 try:
1438 pool_ads = coll.query(constraint=constraint)
1439 except OSError as ex:
1440 raise RuntimeError(f"Problem querying the Collector. (Constraint='{constraint}')") from ex
1442 pool_info = {}
1443 for slot in pool_ads:
1444 pool_info[slot["name"]] = dict(slot)
1445 _LOG.debug("condor_status returned %d ads", len(pool_info))
1446 return pool_info
1449def update_job_info(job_info, other_info):
1450 """Update results of a job query with results from another query.
1452 Parameters
1453 ----------
1454 job_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1455 Results of the job query that needs to be updated.
1456 other_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1457 Results of the other job query.
1459 Returns
1460 -------
1461 job_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1462 The updated results.
1463 """
1464 for schedd_name, others in other_info.items():
1465 try:
1466 jobs = job_info[schedd_name]
1467 except KeyError:
1468 job_info[schedd_name] = others
1469 else:
1470 for id_, ad in others.items():
1471 jobs.setdefault(id_, {}).update(ad)
1472 return job_info
1475def count_jobs_in_single_dag(
1476 filename: str | os.PathLike,
1477) -> tuple[Counter[str], dict[str, str], dict[str, WmsNodeType]]:
1478 """Build bps_run_summary string from dag file.
1480 Parameters
1481 ----------
1482 filename : `str`
1483 Path that includes dag file for a run.
1485 Returns
1486 -------
1487 counts : `Counter` [`str`]
1488 Semi-colon separated list of job labels and counts.
1489 (Same format as saved in dag classad).
1490 job_name_to_label : `dict` [`str`, `str`]
1491 Mapping of job names to job labels.
1492 job_name_to_type : `dict` [`str`, `lsst.ctrl.bps.htcondor.WmsNodeType`]
1493 Mapping of job names to job types
1494 (e.g., payload, final, service).
1495 """
1496 # Later code depends upon insertion order
1497 counts: Counter = Counter() # counts of payload jobs per label
1498 job_name_to_label: dict[str, str] = {}
1499 job_name_to_type: dict[str, WmsNodeType] = {}
1500 with open(filename) as fh:
1501 for line in fh:
1502 # Skip any line that contains commands irrelevant to job counting.
1503 if not line.startswith(
1504 (
1505 "JOB",
1506 "FINAL",
1507 "SERVICE",
1508 "SUBDAG EXTERNAL",
1509 )
1510 ):
1511 continue
1513 m = re.match(
1514 r"(?P<command>JOB|FINAL|SERVICE|SUBDAG EXTERNAL)\s+"
1515 r'(?P<jobname>(?P<wms>wms_)?\S+)\s+"?(?P<subfile>\S+)"?\s*'
1516 r'(DIR "?(?P<dir>[^\s"]+)"?)?\s*(?P<noop>NOOP)?',
1517 line,
1518 )
1519 if m: 1519 ↛ 1571line 1519 didn't jump to line 1571 because the condition on line 1519 was always true
1520 job_name = m.group("jobname")
1521 name_parts = job_name.split("_")
1523 label = ""
1524 if m.group("dir"):
1525 dir_match = re.search(r"jobs/([^\s/]+)", m.group("dir"))
1526 if dir_match:
1527 label = dir_match.group(1)
1528 else:
1529 _LOG.debug("Parse DAG: unparsed dir = %s", line)
1530 elif m.group("subfile"): 1530 ↛ 1537line 1530 didn't jump to line 1537 because the condition on line 1530 was always true
1531 subfile_match = re.search(r"jobs/([^\s/]+)", m.group("subfile"))
1532 if subfile_match:
1533 label = m.group("subfile").split("/")[1]
1534 else:
1535 label = pegasus_name_to_label(job_name)
1537 match m.group("command"):
1538 case "JOB":
1539 if m.group("noop"):
1540 job_type = WmsNodeType.NOOP
1541 # wms_noop_label
1542 label = name_parts[2]
1543 elif m.group("wms"):
1544 if name_parts[1] == "check": 1544 ↛ 1549line 1544 didn't jump to line 1549 because the condition on line 1544 was always true
1545 job_type = WmsNodeType.SUBDAG_CHECK
1546 # wms_check_status_wms_group_label
1547 label = name_parts[5]
1548 else:
1549 _LOG.warning(
1550 "Unexpected skipping of dag line due to unknown wms job: %s", line
1551 )
1552 else:
1553 job_type = WmsNodeType.PAYLOAD
1554 if label == "init": 1554 ↛ 1555line 1554 didn't jump to line 1555 because the condition on line 1554 was never true
1555 label = "pipetaskInit"
1556 counts[label] += 1
1557 case "FINAL":
1558 job_type = WmsNodeType.FINAL
1559 counts[label] += 1 # final counts a payload job.
1560 case "SERVICE":
1561 job_type = WmsNodeType.SERVICE
1562 case "SUBDAG EXTERNAL": 1562 ↛ 1566line 1562 didn't jump to line 1566 because the pattern on line 1562 always matched
1563 job_type = WmsNodeType.SUBDAG
1564 label = name_parts[2]
1566 job_name_to_label[job_name] = label
1567 job_name_to_type[job_name] = job_type
1568 else:
1569 # The line should, but didn't match the pattern above. Probably
1570 # problems with regex.
1571 _LOG.warning("Unexpected skipping of dag line: %s", line)
1573 return counts, job_name_to_label, job_name_to_type
1576def summarize_dag(dir_name: str) -> tuple[str, dict[str, str], dict[str, WmsNodeType]]:
1577 """Build bps_run_summary string from dag file.
1579 Parameters
1580 ----------
1581 dir_name : `str`
1582 Path that includes dag file for a run.
1584 Returns
1585 -------
1586 summary : `str`
1587 Semi-colon separated list of job labels and counts
1588 (Same format as saved in dag classad).
1589 job_name_to_label : `dict` [`str`, `str`]
1590 Mapping of job names to job labels.
1591 job_name_to_type : `dict` [`str`, `lsst.ctrl.bps.htcondor.WmsNodeType`]
1592 Mapping of job names to job types
1593 (e.g., payload, final, service).
1594 """
1595 # Later code depends upon insertion order
1596 counts: Counter[str] = Counter() # counts of payload jobs per label
1597 job_name_to_label: dict[str, str] = {}
1598 job_name_to_type: dict[str, WmsNodeType] = {}
1599 for filename in Path(dir_name).glob("*.dag"):
1600 single_counts, single_job_name_to_label, single_job_name_to_type = count_jobs_in_single_dag(filename)
1601 counts += single_counts
1602 _update_dicts(job_name_to_label, single_job_name_to_label)
1603 _update_dicts(job_name_to_type, single_job_name_to_type)
1605 for filename in Path(dir_name).glob("subdags/*/*.dag"):
1606 single_counts, single_job_name_to_label, single_job_name_to_type = count_jobs_in_single_dag(filename)
1607 counts += single_counts
1608 _update_dicts(job_name_to_label, single_job_name_to_label)
1609 _update_dicts(job_name_to_type, single_job_name_to_type)
1611 summary = ";".join([f"{name}:{counts[name]}" for name in counts])
1612 _LOG.debug("summarize_dag: %s %s %s", summary, job_name_to_label, job_name_to_type)
1613 return summary, job_name_to_label, job_name_to_type
1616def pegasus_name_to_label(name):
1617 """Convert pegasus job name to a label for the report.
1619 Parameters
1620 ----------
1621 name : `str`
1622 Name of job.
1624 Returns
1625 -------
1626 label : `str`
1627 Label for job.
1628 """
1629 label = "UNK"
1630 if name.startswith("create_dir") or name.startswith("stage_in") or name.startswith("stage_out"): 1630 ↛ 1631line 1630 didn't jump to line 1631 because the condition on line 1630 was never true
1631 label = "pegasus"
1632 else:
1633 m = re.match(r"pipetask_(\d+_)?([^_]+)", name)
1634 if m: 1634 ↛ 1635line 1634 didn't jump to line 1635 because the condition on line 1634 was never true
1635 label = m.group(2)
1636 if label == "init":
1637 label = "pipetaskInit"
1639 return label
1642def read_single_dag_status(filename: str | os.PathLike) -> dict[str, Any]:
1643 """Read the node status file for DAG summary information.
1645 Parameters
1646 ----------
1647 filename : `str` or `Path.pathlib`
1648 Node status filename.
1650 Returns
1651 -------
1652 dag_ad : `dict` [`str`, `~typing.Any`]
1653 DAG summary information.
1654 """
1655 dag_ad: dict[str, Any] = {}
1657 # While this is probably more up to date than dag classad, only read from
1658 # file if need to.
1659 try:
1660 node_stat_file = Path(filename)
1661 _LOG.debug("Reading Node Status File %s", node_stat_file)
1662 with open(node_stat_file) as infh:
1663 dag_ad = dict(classad.parseNext(infh)) # pylint: disable=E1101
1665 if not dag_ad: 1665 ↛ 1667line 1665 didn't jump to line 1667 because the condition on line 1665 was never true
1666 # Pegasus check here
1667 metrics_file = node_stat_file.with_suffix(".dag.metrics")
1668 if metrics_file.exists():
1669 with open(metrics_file) as infh:
1670 metrics = json.load(infh)
1671 dag_ad["NodesTotal"] = metrics.get("jobs", 0)
1672 dag_ad["NodesFailed"] = metrics.get("jobs_failed", 0)
1673 dag_ad["NodesDone"] = metrics.get("jobs_succeeded", 0)
1674 metrics_file = node_stat_file.with_suffix(".metrics")
1675 with open(metrics_file) as infh:
1676 metrics = json.load(infh)
1677 dag_ad["NodesTotal"] = metrics["wf_metrics"]["total_jobs"]
1678 except (OSError, PermissionError):
1679 pass
1681 _LOG.debug("read_dag_status: %s", dag_ad)
1682 return dag_ad
1685def read_dag_status(wms_path: str | os.PathLike) -> dict[str, Any]:
1686 """Read the node status file for DAG summary information.
1688 Parameters
1689 ----------
1690 wms_path : `str` or `os.PathLike`
1691 Path that includes node status file for a run.
1693 Returns
1694 -------
1695 dag_ad : `dict` [`str`, `~typing.Any`]
1696 DAG summary information, counts summed across any subdags.
1697 """
1698 dag_ads: dict[str, Any] = {}
1699 path = Path(wms_path)
1700 try:
1701 node_stat_file = next(path.glob("*.node_status"))
1702 except StopIteration as exc:
1703 raise FileNotFoundError(f"DAGMan node status not found in {wms_path}") from exc
1705 dag_ads = read_single_dag_status(node_stat_file)
1707 for node_stat_file in path.glob("subdags/*/*.node_status"):
1708 dag_ad = read_single_dag_status(node_stat_file)
1709 dag_ads["JobProcsHeld"] += dag_ad.get("JobProcsHeld", 0)
1710 dag_ads["NodesPost"] += dag_ad.get("NodesPost", 0)
1711 dag_ads["JobProcsIdle"] += dag_ad.get("JobProcsIdle", 0)
1712 dag_ads["NodesTotal"] += dag_ad.get("NodesTotal", 0)
1713 dag_ads["NodesFailed"] += dag_ad.get("NodesFailed", 0)
1714 dag_ads["NodesDone"] += dag_ad.get("NodesDone", 0)
1715 dag_ads["NodesQueued"] += dag_ad.get("NodesQueued", 0)
1716 dag_ads["NodesPre"] += dag_ad.get("NodesReady", 0)
1717 dag_ads["NodesFutile"] += dag_ad.get("NodesFutile", 0)
1718 dag_ads["NodesUnready"] += dag_ad.get("NodesUnready", 0)
1720 return dag_ads
1723def read_single_node_status(filename: str | os.PathLike, init_fake_id: int) -> dict[str, Any]:
1724 """Read entire node status file.
1726 Parameters
1727 ----------
1728 filename : `str` or `pathlib.Path`
1729 Node status filename.
1730 init_fake_id : `int`
1731 Initial fake id value.
1733 Returns
1734 -------
1735 jobs : `dict` [`str`, `~typing.Any`]
1736 DAG summary information compiled from the node status file combined
1737 with the information found in the node event log.
1739 Currently, if the same job attribute is found in both files, its value
1740 from the event log takes precedence over the value from the node status
1741 file.
1742 """
1743 filename = Path(filename)
1745 # Get jobid info from other places to fill in gaps in info from node_status
1746 _, job_name_to_label, job_name_to_type = count_jobs_in_single_dag(filename.with_suffix(".dag"))
1747 loginfo: dict[str, dict[str, Any]] = {}
1748 try:
1749 wms_workflow_id, loginfo = read_single_dag_log(filename.with_suffix(".dag.dagman.log"))
1750 loginfo = read_single_dag_nodes_log(filename.with_suffix(".dag.nodes.log"))
1751 except (OSError, PermissionError):
1752 pass
1754 job_name_to_id: dict[str, str] = {}
1755 _LOG.debug("loginfo = %s", loginfo)
1756 log_job_name_to_id: dict[str, str] = {}
1757 for job_id, job_info in loginfo.items():
1758 if "LogNotes" in job_info: 1758 ↛ 1757line 1758 didn't jump to line 1757 because the condition on line 1758 was always true
1759 m = re.match(r"DAG Node: (\S+)", job_info["LogNotes"])
1760 if m: 1760 ↛ 1757line 1760 didn't jump to line 1757 because the condition on line 1760 was always true
1761 job_name = m.group(1)
1762 log_job_name_to_id[job_name] = job_id
1763 job_info["DAGNodeName"] = job_name
1764 job_info["wms_node_type"] = job_name_to_type[job_name]
1765 job_info["bps_job_label"] = job_name_to_label[job_name]
1767 jobs = {}
1768 fake_id = init_fake_id # For nodes that do not yet have a job id, give fake one
1769 try:
1770 with open(filename) as fh:
1771 for ad in classad.parseAds(fh):
1772 match ad["Type"]:
1773 case "DagStatus":
1774 # Skip DAG summary.
1775 pass
1776 case "NodeStatus":
1777 job_name = ad["Node"]
1778 if job_name in job_name_to_label: 1778 ↛ 1780line 1778 didn't jump to line 1780 because the condition on line 1778 was always true
1779 job_label = job_name_to_label[job_name]
1780 elif "_" in job_name:
1781 job_label = job_name.split("_")[1]
1782 else:
1783 job_label = job_name
1785 job = dict(ad)
1786 if job_name in log_job_name_to_id:
1787 job_id = str(log_job_name_to_id[job_name])
1788 _update_dicts(job, loginfo[job_id])
1789 else:
1790 job_id = str(fake_id)
1791 job = dict(ad)
1792 fake_id -= 1
1793 jobs[job_id] = job
1794 job_name_to_id[job_name] = job_id
1796 # Make job info as if came from condor_q.
1797 job["ClusterId"] = int(float(job_id))
1798 job["DAGManJobID"] = wms_workflow_id
1799 job["DAGNodeName"] = job_name
1800 job["bps_job_label"] = job_label
1801 job["wms_node_type"] = job_name_to_type[job_name]
1803 case "StatusEnd": 1803 ↛ 1806line 1803 didn't jump to line 1806 because the pattern on line 1803 always matched
1804 # Skip node status file "epilog".
1805 pass
1806 case _:
1807 _LOG.debug(
1808 "Ignoring unknown classad type '%s' in the node status file '%s'",
1809 ad["Type"],
1810 filename,
1811 )
1812 except (OSError, PermissionError):
1813 pass
1815 # Check for missing jobs (e.g., submission failure or not submitted yet)
1816 # Use dag info to create job placeholders
1817 for name in set(job_name_to_label) - set(job_name_to_id):
1818 if name in log_job_name_to_id: # job was in nodes.log, but not node_status
1819 job_id = str(log_job_name_to_id[name])
1820 job = dict(loginfo[job_id])
1821 else:
1822 job_id = str(fake_id)
1823 fake_id -= 1
1824 job = {}
1825 job["NodeStatus"] = NodeStatus.NOT_READY
1827 job["ClusterId"] = int(float(job_id))
1828 job["ProcId"] = 0
1829 job["DAGManJobID"] = wms_workflow_id
1830 job["DAGNodeName"] = name
1831 job["bps_job_label"] = job_name_to_label[name]
1832 job["wms_node_type"] = job_name_to_type[name]
1833 jobs[f"{job['ClusterId']}.{job['ProcId']}"] = job
1835 for job_info in jobs.values():
1836 job_info["from_dag_job"] = f"wms_{filename.stem}"
1838 return jobs
1841def read_node_status(wms_path: str | os.PathLike) -> dict[str, dict[str, Any]]:
1842 """Read entire node status file.
1844 Parameters
1845 ----------
1846 wms_path : `str` or `os.PathLike`
1847 Path that includes node status file for a run.
1849 Returns
1850 -------
1851 jobs : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1852 DAG summary information compiled from the node status file combined
1853 with the information found in the node event log.
1855 Currently, if the same job attribute is found in both files, its value
1856 from the event log takes precedence over the value from the node status
1857 file.
1858 """
1859 jobs: dict[str, dict[str, Any]] = {}
1860 init_fake_id = -1
1862 # subdags may not have run so wouldn't have node_status file
1863 # use dag files and let read_single_node_status handle missing
1864 # node_status file.
1865 for dag_filename in Path(wms_path).glob("*.dag"):
1866 filename = dag_filename.with_suffix(".node_status")
1867 info = read_single_node_status(filename, init_fake_id)
1868 init_fake_id -= len(info)
1869 _update_dicts(jobs, info)
1871 for dag_filename in Path(wms_path).glob("subdags/*/*.dag"):
1872 filename = dag_filename.with_suffix(".node_status")
1873 info = read_single_node_status(filename, init_fake_id)
1874 init_fake_id -= len(info)
1875 _update_dicts(jobs, info)
1877 # Propagate pruned from subdags to jobs
1878 name_to_id: dict[str, str] = {}
1879 missing_status: dict[str, list[str]] = {}
1880 for id_, job in jobs.items():
1881 if job["DAGNodeName"].startswith("wms_"):
1882 name_to_id[job["DAGNodeName"]] = id_
1883 if "NodeStatus" not in job or job["NodeStatus"] == NodeStatus.NOT_READY:
1884 missing_status.setdefault(job["from_dag_job"], []).append(id_)
1886 for name, dag_id in name_to_id.items():
1887 dag_status = jobs[dag_id].get("NodeStatus", NodeStatus.NOT_READY)
1888 if dag_status in {NodeStatus.NOT_READY, NodeStatus.FUTILE}:
1889 for id_ in missing_status.get(name, []):
1890 jobs[id_]["NodeStatus"] = dag_status
1892 return jobs
1895def read_single_dag_log(log_filename: str | os.PathLike) -> tuple[str, dict[str, dict[str, Any]]]:
1896 """Read job information from the DAGMan log file.
1898 Parameters
1899 ----------
1900 log_filename : `str` or `os.PathLike`
1901 DAGMan log filename.
1903 Returns
1904 -------
1905 wms_workflow_id : `str`
1906 HTCondor job id (i.e., <ClusterId>.<ProcId>) of the DAGMan job.
1907 dag_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1908 HTCondor job information read from the log file mapped to HTCondor
1909 job id.
1911 Raises
1912 ------
1913 FileNotFoundError
1914 If cannot find DAGMan log in given wms_path.
1915 """
1916 wms_workflow_id = "0"
1917 dag_info: dict[str, dict[str, Any]] = {}
1919 filename = Path(log_filename)
1920 if filename.exists():
1921 _LOG.debug("dag node log filename: %s", filename)
1923 info: dict[str, Any] = {}
1924 job_event_log = htcondor.JobEventLog(str(filename))
1925 for event in job_event_log.events(stop_after=0):
1926 id_ = f"{event['Cluster']}.{event['Proc']}"
1927 if id_ not in info:
1928 info[id_] = {}
1929 wms_workflow_id = id_ # taking last job id in case of restarts
1930 _update_dicts(info[id_], event)
1931 info[id_][f"{event.type.name.lower()}_time"] = event["EventTime"]
1933 # only save latest DAG job
1934 dag_info = {wms_workflow_id: info[wms_workflow_id]}
1936 return wms_workflow_id, dag_info
1939def read_dag_log(wms_path: str | os.PathLike) -> tuple[str, dict[str, Any]]:
1940 """Read job information from the DAGMan log file.
1942 Parameters
1943 ----------
1944 wms_path : `str` or `os.PathLike`
1945 Path containing the DAGMan log file.
1947 Returns
1948 -------
1949 wms_workflow_id : `str`
1950 HTCondor job id (i.e., <ClusterId>.<ProcId>) of the DAGMan job.
1951 dag_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1952 HTCondor job information read from the log file mapped to HTCondor
1953 job id.
1955 Raises
1956 ------
1957 FileNotFoundError
1958 If cannot find DAGMan log in given wms_path.
1959 """
1960 wms_workflow_id = MISSING_ID
1961 dag_info: dict[str, dict[str, Any]] = {}
1963 path = Path(wms_path)
1964 if path.exists(): 1964 ↛ 1972line 1964 didn't jump to line 1972 because the condition on line 1964 was always true
1965 try:
1966 filename = next(path.glob("*.dag.dagman.log"))
1967 except StopIteration as exc:
1968 raise FileNotFoundError(f"DAGMan log not found in {wms_path}") from exc
1969 _LOG.debug("dag node log filename: %s", filename)
1970 wms_workflow_id, dag_info = read_single_dag_log(filename)
1972 return wms_workflow_id, dag_info
1975def read_single_dag_nodes_log(filename: str | os.PathLike) -> dict[str, dict[str, Any]]:
1976 """Read job information from the DAGMan nodes log file.
1978 Parameters
1979 ----------
1980 filename : `str` or `os.PathLike`
1981 Path containing the DAGMan nodes log file.
1983 Returns
1984 -------
1985 info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1986 HTCondor job information read from the log file mapped to HTCondor
1987 job id.
1989 Raises
1990 ------
1991 FileNotFoundError
1992 If cannot find DAGMan node log in given wms_path.
1993 """
1994 _LOG.debug("dag node log filename: %s", filename)
1995 filename = Path(filename)
1997 info: dict[str, dict[str, Any]] = {}
1998 if not filename.exists():
1999 raise FileNotFoundError(f"{filename} does not exist")
2001 try:
2002 job_event_log = htcondor.JobEventLog(str(filename))
2003 except htcondor.HTCondorIOError as ex:
2004 _LOG.error("Problem reading nodes log file (%s): %s", filename, ex)
2005 import traceback
2007 traceback.print_stack()
2008 raise
2009 for event in job_event_log.events(stop_after=0):
2010 _LOG.debug("log event type = %s, keys = %s", event["EventTypeNumber"], event.keys())
2012 try:
2013 id_ = f"{event['Cluster']}.{event['Proc']}"
2014 except KeyError:
2015 _LOG.warn(
2016 "Log event missing ids (DAGNodeName=%s, EventTime=%s, EventTypeNumber=%s)",
2017 event.get("DAGNodeName", "UNK"),
2018 event.get("EventTime", "UNK"),
2019 event.get("EventTypeNumber", "UNK"),
2020 )
2021 else:
2022 if id_ not in info:
2023 info[id_] = {}
2024 # Workaround: Please check to see if still problem in
2025 # future HTCondor versions. Sometimes get a
2026 # JobAbortedEvent for a subdag job after it already
2027 # terminated normally. Seems to happen when using job
2028 # plus subdags.
2029 if event["EventTypeNumber"] == 9 and info[id_].get("EventTypeNumber", -1) == 5: 2029 ↛ 2030line 2029 didn't jump to line 2030 because the condition on line 2029 was never true
2030 _LOG.debug("Skipping spurious JobAbortedEvent: %s", dict(event))
2031 else:
2032 _update_dicts(info[id_], event)
2033 info[id_][f"{event.type.name.lower()}_time"] = event["EventTime"]
2035 return info
2038def read_dag_nodes_log(wms_path: str | os.PathLike) -> dict[str, dict[str, Any]]:
2039 """Read job information from the DAGMan nodes log file.
2041 Parameters
2042 ----------
2043 wms_path : `str` or `os.PathLike`
2044 Path containing the DAGMan nodes log file.
2046 Returns
2047 -------
2048 info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
2049 HTCondor job information read from the log file mapped to HTCondor
2050 job id.
2052 Raises
2053 ------
2054 FileNotFoundError
2055 If cannot find DAGMan node log in given wms_path.
2056 """
2057 info: dict[str, dict[str, Any]] = {}
2058 for filename in Path(wms_path).glob("*.dag.nodes.log"):
2059 _LOG.debug("dag node log filename: %s", filename)
2060 _update_dicts(info, read_single_dag_nodes_log(filename))
2062 # If submitted, the main nodes log file should exist
2063 if not info:
2064 raise FileNotFoundError(f"DAGMan node log not found in {wms_path}")
2066 # Subdags will not have dag nodes log files if they haven't
2067 # started running yet (so missing is not an error).
2068 for filename in Path(wms_path).glob("subdags/*/*.dag.nodes.log"):
2069 _LOG.debug("dag node log filename: %s", filename)
2070 _update_dicts(info, read_single_dag_nodes_log(filename))
2072 return info
2075def read_dag_info(wms_path: str | os.PathLike) -> dict[str, dict[str, Any]]:
2076 """Read custom DAGMan job information from the file.
2078 Parameters
2079 ----------
2080 wms_path : `str` or `os.PathLike`
2081 Path containing the file with the DAGMan job info.
2083 Returns
2084 -------
2085 dag_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
2086 HTCondor job information.
2088 Raises
2089 ------
2090 FileNotFoundError
2091 If cannot find DAGMan job info file in the given location.
2092 """
2093 dag_info: dict[str, dict[str, Any]] = {}
2094 try:
2095 filename = next(Path(wms_path).glob("*.info.json"))
2096 except StopIteration as exc:
2097 raise FileNotFoundError(f"File with DAGMan job information not found in {wms_path}") from exc
2098 _LOG.debug("DAGMan job information filename: %s", filename)
2099 try:
2100 with open(filename) as fh:
2101 dag_info = json.load(fh)
2102 except (OSError, PermissionError) as exc:
2103 _LOG.debug("Retrieving DAGMan job information failed: %s", exc)
2104 return dag_info
2107def write_dag_info(filename, dag_info):
2108 """Write custom job information about DAGMan job.
2110 Parameters
2111 ----------
2112 filename : `str`
2113 Name of the file where the information will be stored.
2114 dag_info : `dict` [`str` `dict` [`str`, `~typing.Any`]]
2115 Information about the DAGMan job.
2116 """
2117 schedd_name = next(iter(dag_info))
2118 dag_id = next(iter(dag_info[schedd_name]))
2119 dag_ad = dag_info[schedd_name][dag_id]
2120 ad = {"ClusterId": dag_ad["ClusterId"], "GlobalJobId": dag_ad["GlobalJobId"]}
2121 ad.update({key: val for key, val in dag_ad.items() if key.startswith("bps")})
2122 try:
2123 with open(filename, "w") as fh:
2124 info = {schedd_name: {dag_id: ad}}
2125 json.dump(info, fh)
2126 except (KeyError, OSError, PermissionError) as exc:
2127 _LOG.debug("Persisting DAGMan job information failed: %s", exc)
2130def htc_tweak_log_info(wms_path: str | Path, job: dict[str, Any]) -> None:
2131 """Massage the given job info has same structure as if came from condor_q.
2133 Parameters
2134 ----------
2135 wms_path : `str` | `os.PathLike`
2136 Path containing an HTCondor event log file.
2137 job : `dict` [ `str`, `~typing.Any` ]
2138 A mapping between HTCondor job id and job information read from
2139 the log.
2140 """
2141 _LOG.debug("htc_tweak_log_info: %s %s", wms_path, job)
2143 # Use the presence of 'MyType' key as a proxy to determine if the job ad
2144 # contains the info extracted from the event log. Exit early if it doesn't
2145 # (e.g. it is a job ad for a pruned job).
2146 if "MyType" not in job:
2147 return
2149 try:
2150 job["ClusterId"] = job["Cluster"]
2151 job["ProcId"] = job["Proc"]
2152 except KeyError as e:
2153 _LOG.error("Missing key %s in job: %s", str(e), job)
2154 raise
2155 job["Iwd"] = str(wms_path)
2156 job["Owner"] = Path(wms_path).owner()
2158 if "LogNotes" in job:
2159 m = re.match(r"DAG Node: (\S+)", job["LogNotes"])
2160 if m: 2160 ↛ 2163line 2160 didn't jump to line 2163 because the condition on line 2160 was always true
2161 job["DAGNodeName"] = m.group(1)
2163 match job["MyType"]:
2164 case "ExecuteEvent":
2165 job["JobStatus"] = htcondor.JobStatus.RUNNING
2166 case "JobTerminatedEvent" | "PostScriptTerminatedEvent":
2167 job["JobStatus"] = htcondor.JobStatus.COMPLETED
2168 case "SubmitEvent":
2169 job["JobStatus"] = htcondor.JobStatus.IDLE
2170 case "JobAbortedEvent":
2171 job["JobStatus"] = htcondor.JobStatus.REMOVED
2172 case "JobHeldEvent":
2173 job["JobStatus"] = htcondor.JobStatus.HELD
2174 case "JobReleaseEvent":
2175 # If the job managing the execution of the root DAG is held and
2176 # released this will be the last event showing up in its
2177 # job event log even if the job is still running. If this is
2178 # the last event for a job corresponding to the workflow node
2179 # (either a normal payload job or the job managing the execution
2180 # of an inner DAG), its final status will be determined later
2181 # using node status log (see _htc_status_to_wms_state()).
2182 job["JobStatus"] = htcondor.JobStatus.RUNNING if "DAGNodeName" not in job else None
2183 case _:
2184 _LOG.debug("Unknown log event type: %s", job["MyType"])
2185 job["JobStatus"] = None
2187 # Use available information to add either "ExitCode" or "ExitSignal"
2188 # attribute that captures respectively job's exit status (if it finished
2189 # on its own accord) or its exit signal (if it was terminated by
2190 # a signal). Also, include a flag "ExitBySignal" to make distinguishing
2191 # between these two cases easy later on.
2192 if job["JobStatus"] in {
2193 htcondor.JobStatus.COMPLETED,
2194 htcondor.JobStatus.HELD,
2195 htcondor.JobStatus.REMOVED,
2196 }:
2197 new_job = HTC_JOB_AD_HANDLERS.handle(job)
2198 if new_job is not None:
2199 job = new_job
2200 else:
2201 _LOG.error("Could not determine exit status for job '%s.%s'", job["ClusterId"], job["ProcId"])
2204def htc_check_dagman_output(wms_path: str | os.PathLike) -> str:
2205 """Check the DAGMan output for error messages.
2207 Parameters
2208 ----------
2209 wms_path : `str` or `os.PathLike`
2210 Directory containing the DAGman output file.
2212 Returns
2213 -------
2214 message : `str`
2215 Message containing error messages from the DAGMan output. Empty
2216 string if no messages.
2218 Raises
2219 ------
2220 FileNotFoundError
2221 If cannot find DAGMan standard output file in given wms_path.
2222 """
2223 try:
2224 filename = next(Path(wms_path).glob("*.dag.dagman.out"))
2225 except StopIteration as exc:
2226 raise FileNotFoundError(f"DAGMan standard output file not found in {wms_path}") from exc
2227 _LOG.debug("dag output filename: %s", filename)
2229 p = re.compile(r"^(\d\d/\d\d/\d\d \d\d:\d\d:\d\d) (Job submit try \d+/\d+ failed|Warning:.*$|ERROR:.*$)")
2231 message = ""
2232 try:
2233 with open(filename) as fh:
2234 last_submit_failed = "" # Since submit retries multiple times only report last one
2235 for line in fh:
2236 m = p.match(line)
2237 if m:
2238 if m.group(2).startswith("Job submit try"):
2239 last_submit_failed = m.group(1)
2240 elif m.group(2).startswith("ERROR: submit attempt failed"):
2241 pass # Should be handled by Job submit try
2242 elif m.group(2).startswith("Warning"):
2243 if ".dag.nodes.log is in /tmp" in m.group(2):
2244 last_warning = "Cannot submit from /tmp."
2245 else:
2246 last_warning = m.group(2)
2247 elif m.group(2) == "ERROR: Warning is fatal error because of DAGMAN_USE_STRICT setting":
2248 message += "ERROR: "
2249 message += last_warning
2250 message += "\n"
2251 elif m.group(2) in [ 2251 ↛ 2257line 2251 didn't jump to line 2257 because the condition on line 2251 was always true
2252 "ERROR: the following job(s) failed:",
2253 "ERROR: the following Node(s) failed:",
2254 ]:
2255 pass
2256 else:
2257 message += m.group(2)
2258 message += "\n"
2260 if last_submit_failed:
2261 message += f"Warn: Job submission issues (last: {last_submit_failed})"
2262 except (OSError, PermissionError):
2263 message = f"Warn: Could not read dagman output file from {wms_path}."
2264 _LOG.debug("dag output file message: %s", message)
2265 return message
2268def _read_rescue_headers(infh: TextIO) -> tuple[list[str], list[str]]:
2269 """Read header lines from a rescue file.
2271 Parameters
2272 ----------
2273 infh : `TextIO`
2274 The rescue file from which to read the header lines.
2276 Returns
2277 -------
2278 header_lines : `list` [`str`]
2279 Header lines read from the rescue file.
2280 failed_subdags : `list` [`str`]
2281 Names of failed subdag jobs.
2282 """
2283 header_lines: list[str] = []
2284 failed = False
2285 failed_subdags: list[str] = []
2287 for line in infh: 2287 ↛ 2310line 2287 didn't jump to line 2310 because the loop on line 2287 didn't complete
2288 line = line.strip()
2289 if line.startswith("#"):
2290 if line.startswith("# Nodes that failed:"):
2291 failed = True
2292 header_lines.append(line)
2293 elif failed:
2294 orig_failed_nodes = line[1:].strip().split(",")
2295 new_failed_nodes = []
2296 for node in orig_failed_nodes:
2297 if node.startswith("wms_check_status"):
2298 group_node = node[17:]
2299 failed_subdags.append(group_node)
2300 new_failed_nodes.append(group_node)
2301 else:
2302 new_failed_nodes.append(node)
2303 header_lines.append(f"# {','.join(new_failed_nodes)}")
2304 if orig_failed_nodes[-1] == "<ENDLIST>": 2304 ↛ 2287line 2304 didn't jump to line 2287 because the condition on line 2304 was always true
2305 failed = False
2306 else:
2307 header_lines.append(line)
2308 elif line.strip() == "": # end of headers 2308 ↛ 2287line 2308 didn't jump to line 2287 because the condition on line 2308 was always true
2309 break
2310 return header_lines, failed_subdags
2313def _write_rescue_headers(header_lines: list[str], failed_subdags: list[str], outfh: TextIO) -> None:
2314 """Write the header lines to the new rescue file.
2316 Parameters
2317 ----------
2318 header_lines : `list` [`str`]
2319 Header lines to write to the new rescue file.
2320 failed_subdags : `list` [`str`]
2321 Job names of the failed subdags.
2322 outfh : `TextIO`
2323 New rescue file.
2324 """
2325 done_str = "# Nodes premarked DONE"
2326 pattern = f"^{done_str}:\\s+(\\d+)"
2327 for header_line in header_lines:
2328 m = re.match(pattern, header_line)
2329 if m:
2330 print(f"{done_str}: {int(m.group(1)) - len(failed_subdags)}", file=outfh)
2331 else:
2332 print(header_line, file=outfh)
2334 print("", file=outfh)
2337def _copy_done_lines(failed_subdags: list[str], infh: TextIO, outfh: TextIO) -> None:
2338 """Copy the DONE lines from the original rescue file skipping
2339 the failed group jobs.
2341 Parameters
2342 ----------
2343 failed_subdags : `list` [`str`]
2344 List of job names for the failed subdags
2345 infh : `TextIO`
2346 Original rescue file to copy from.
2347 outfh : `TextIO`
2348 New rescue file to copy to.
2349 """
2350 for line in infh:
2351 line = line.strip()
2352 try:
2353 _, node_name = line.split()
2354 except ValueError:
2355 _LOG.error(f"Unexpected line in rescue file = '{line}'")
2356 raise
2357 if node_name not in failed_subdags:
2358 print(line, file=outfh)
2361def _update_rescue_file(rescue_file: Path) -> None:
2362 """Update the subdag failures in the main rescue file
2363 and backup the failed subdag dirs.
2365 Parameters
2366 ----------
2367 rescue_file : `pathlib.Path`
2368 The main rescue file that needs to be updated.
2369 """
2370 # To reduce memory requirements, not reading entire file into memory.
2371 rescue_tmp = rescue_file.with_suffix(rescue_file.suffix + ".tmp")
2372 with open(rescue_file) as infh:
2373 header_lines, failed_subdags = _read_rescue_headers(infh)
2374 with open(rescue_tmp, "w") as outfh:
2375 _write_rescue_headers(header_lines, failed_subdags, outfh)
2376 _copy_done_lines(failed_subdags, infh, outfh)
2377 rescue_file.unlink()
2378 rescue_tmp.rename(rescue_file)
2379 for failed_subdag in failed_subdags:
2380 htc_backup_files(
2381 rescue_file.parent / "subdags" / failed_subdag, subdir=f"backups/subdags/{failed_subdag}"
2382 )
2385def _update_dicts(dict1, dict2):
2386 """Update dict1 with info in dict2.
2388 (Basically an update for nested dictionaries.)
2390 Parameters
2391 ----------
2392 dict1 : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
2393 HTCondor job information to be updated.
2394 dict2 : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
2395 Additional HTCondor job information.
2396 """
2397 for key, value in dict2.items():
2398 if key in dict1 and isinstance(dict1[key], dict) and isinstance(value, dict):
2399 _update_dicts(dict1[key], value)
2400 else:
2401 dict1[key] = value
2404def _locate_schedds(locate_all=False):
2405 """Find out Scheduler daemons in an HTCondor pool.
2407 Parameters
2408 ----------
2409 locate_all : `bool`, optional
2410 If True, all available schedulers in the HTCondor pool will be located.
2411 False by default which means that the search will be limited to looking
2412 for the Scheduler running on a local host.
2414 Returns
2415 -------
2416 schedds : `dict` [`str`, `htcondor.Schedd`]
2417 A mapping between Scheduler names and Python objects allowing for
2418 interacting with them.
2419 """
2420 coll = htcondor.Collector()
2422 schedd_ads = []
2423 if locate_all:
2424 schedd_ads.extend(coll.locateAll(htcondor.DaemonTypes.Schedd))
2425 else:
2426 schedd_ads.append(coll.locate(htcondor.DaemonTypes.Schedd))
2427 return {ad["Name"]: htcondor.Schedd(ad) for ad in schedd_ads}