Coverage for python/lsst/ctrl/bps/htcondor/lssthtc.py: 77%
933 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-22 17:25 +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 "kill_sig",
226 "want_graceful_removal",
227 "job_max_vacate_time",
228}
229HTC_VALID_JOB_DAG_KEYS = {
230 "dir",
231 "noop",
232 "done",
233 "vars",
234 "pre",
235 "post",
236 "retry",
237 "retry_unless_exit",
238 "abort_dag_on",
239 "abort_exit",
240 "priority",
241}
242HTC_VERSION = version.parse(htcondor.__version__)
245class RestrictedDict(MutableMapping):
246 """A dictionary that only allows certain keys.
248 Parameters
249 ----------
250 valid_keys : `~collections.abc.Container`
251 Strings that are valid keys.
252 init_data : `dict` or `RestrictedDict`, optional
253 Initial data.
255 Raises
256 ------
257 KeyError
258 If invalid key(s) in init_data.
259 """
261 def __init__(self, valid_keys, init_data=()):
262 self.valid_keys = valid_keys
263 self.data = {}
264 self.update(init_data)
266 def __getitem__(self, key):
267 """Return value for given key if exists.
269 Parameters
270 ----------
271 key : `str`
272 Identifier for value to return.
274 Returns
275 -------
276 value : `~typing.Any`
277 Value associated with given key.
279 Raises
280 ------
281 KeyError
282 If key doesn't exist.
283 """
284 return self.data[key]
286 def __delitem__(self, key):
287 """Delete value for given key if exists.
289 Parameters
290 ----------
291 key : `str`
292 Identifier for value to delete.
294 Raises
295 ------
296 KeyError
297 If key doesn't exist.
298 """
299 del self.data[key]
301 def __setitem__(self, key, value):
302 """Store key,value in internal dict only if key is valid.
304 Parameters
305 ----------
306 key : `str`
307 Identifier to associate with given value.
308 value : `~typing.Any`
309 Value to store.
311 Raises
312 ------
313 KeyError
314 If key is invalid.
315 """
316 if key not in self.valid_keys: 316 ↛ 317line 316 didn't jump to line 317 because the condition on line 316 was never true
317 raise KeyError(f"Invalid key {key}")
318 self.data[key] = value
320 def __iter__(self):
321 return self.data.__iter__()
323 def __len__(self):
324 return len(self.data)
326 def __str__(self):
327 return str(self.data)
330def htc_backup_files(
331 wms_path: str | os.PathLike,
332 subdir: str | os.PathLike | None = None,
333 limit: int = 100,
334 failed_subdags: set[str] | None = None,
335) -> None:
336 """Backup select HTCondor files in the submit directory.
338 Files will be saved in separate subdirectories which will be created in
339 the submit directory where the files are located. These subdirectories
340 will be consecutive, zero-padded integers. Their values will correspond to
341 the number of HTCondor rescue DAGs in the submit directory.
343 Hence, with the default settings, copies after the initial failed run will
344 be placed in '001' subdirectory, '002' after the first restart, and so on
345 until the limit of backups is reached. If there's no rescue DAG yet, files
346 will be copied to '000' subdirectory.
348 This is not a generic function for making backups. It is intended to be
349 used once, just before a restart, to make snapshots of files which will be
350 overwritten by HTCondor after during the next run.
352 Parameters
353 ----------
354 wms_path : `str` or `os.PathLike`
355 Path to the submit directory either absolute or relative.
356 subdir : `str` or `os.PathLike`, optional
357 A path, relative to the submit directory, where all subdirectories with
358 backup files will be kept. Defaults to None which means that the backup
359 subdirectories will be placed directly in the submit directory.
360 limit : `int`, optional
361 Maximal number of backups. If the number of backups reaches the limit,
362 the last backup files will be overwritten. The default value is 100
363 to match the default value of HTCondor's DAGMAN_MAX_RESCUE_NUM in
364 version 8.8+.
365 failed_subdags : `set` [`str`], optional
366 Names of subdag jobs that failed. Only files of these subdags will be
367 backed up. If None (the default), files of all subdags with rescue
368 files are backed up.
370 Raises
371 ------
372 FileNotFoundError
373 If the submit directory or the file that needs to be backed up does not
374 exist.
375 OSError
376 If the submit directory cannot be accessed or backing up a file failed
377 either due to permission or filesystem related issues.
378 """
379 width = len(str(limit))
381 path = Path(wms_path).resolve()
382 if not path.is_dir():
383 raise FileNotFoundError(f"Directory {path} not found")
385 # Initialize the backup counter.
386 rescue_dags = list(path.glob("*.rescue[0-9][0-9][0-9]"))
387 counter = min(len(rescue_dags), limit)
389 # Create the backup directory and move select files there.
390 dest = path
391 if subdir:
392 # PurePath.is_relative_to() is not available before Python 3.9. Hence
393 # we need to check is 'subdir' is in the submit directory in some other
394 # way if it is an absolute path.
395 subdir = Path(subdir)
396 if subdir.is_absolute():
397 subdir = subdir.resolve() # Since resolve was run on path, must run it here
398 if dest not in subdir.parents:
399 _LOG.warning(
400 "Invalid backup location: '%s' not in the submit directory, will use '%s' instead.",
401 subdir,
402 wms_path,
403 )
404 else:
405 dest /= subdir
406 else:
407 dest /= subdir
408 dest /= f"{counter:0{width}}"
409 _LOG.debug("dest = %s", dest)
410 try:
411 dest.mkdir(parents=True, exist_ok=False if counter < limit else True)
412 except FileExistsError:
413 _LOG.warning("Refusing to do backups: target directory '%s' already exists", dest)
414 else:
415 htc_backup_files_single_path(path, dest)
417 # Back up selected files for failed subdags as well.
418 #
419 # Do NOT back up files of subdags that succeeded! These files need to stay
420 # in their respective directories as they will not be recreated by HTCondor
421 # after the run is restarted. As HTCondorService.report() uses information
422 # in these files to determine job statuses, their absence may lead to
423 # reporting incorrect job status counts.
424 for subdag_dir in {file.parent for file in path.glob("subdags/*/*.rescue*")}:
425 if failed_subdags is not None and subdag_dir.name not in failed_subdags: 425 ↛ 426line 425 didn't jump to line 426 because the condition on line 425 was never true
426 continue
427 subdag_dest = dest / subdag_dir.relative_to(path)
428 subdag_dest.mkdir(parents=True, exist_ok=False)
429 htc_backup_files_single_path(subdag_dir, subdag_dest)
432def htc_backup_files_single_path(src: str | os.PathLike, dest: str | os.PathLike) -> None:
433 """Move particular htc files to a different directory for later debugging.
435 Parameters
436 ----------
437 src : `str` or `os.PathLike`
438 Directory from which to back up particular files.
439 dest : `str` or `os.PathLike`
440 Directory to which particular files are moved.
442 Raises
443 ------
444 RuntimeError
445 If given dest directory matches given src directory.
446 OSError
447 If problems moving file.
448 FileNotFoundError
449 Item matching pattern in src directory isn't a file.
450 """
451 src = Path(src)
452 dest = Path(dest)
453 if dest.samefile(src):
454 raise RuntimeError(f"Destination directory is same as the source directory ({src})")
456 for patt in [
457 "*.info.*",
458 "*.dag.metrics",
459 "*.dag.nodes.log",
460 "*.node_status",
461 "wms_*.dag.post.out",
462 "wms_*.status.txt",
463 ]:
464 for source in src.glob(patt):
465 if source.is_file(): 465 ↛ 472line 465 didn't jump to line 472 because the condition on line 465 was always true
466 target = dest / source.relative_to(src)
467 try:
468 source.rename(target)
469 except OSError as exc:
470 raise type(exc)(f"Backing up '{source}' failed: {exc.strerror}") from None
471 else:
472 raise FileNotFoundError(f"Backing up '{source}' failed: not a file")
475def htc_escape(value):
476 """Escape characters in given value based upon HTCondor syntax.
478 Parameters
479 ----------
480 value : `~typing.Any`
481 Value that needs to have characters escaped if string.
483 Returns
484 -------
485 new_value : `~typing.Any`
486 Given value with characters escaped appropriate for HTCondor if string.
487 """
488 if isinstance(value, str):
489 newval = value.replace('"', '""').replace("'", "''").replace(""", '"')
490 else:
491 newval = value
493 return newval
496def htc_write_attribs(stream, attrs):
497 """Write job attributes in HTCondor format to writeable stream.
499 Parameters
500 ----------
501 stream : `~typing.TextIO`
502 Output text stream (typically an open file).
503 attrs : `dict`
504 HTCondor job attributes (dictionary of attribute key, value).
505 """
506 for key, value in attrs.items():
507 # Make sure strings are syntactically correct for HTCondor.
508 if isinstance(value, str): 508 ↛ 511line 508 didn't jump to line 511 because the condition on line 508 was always true
509 pval = f'"{htc_escape(value)}"'
510 else:
511 pval = value
513 print(f"+{key} = {pval}", file=stream)
516def htc_write_condor_file(
517 filename: str | os.PathLike, job_name: str, job: RestrictedDict, job_attrs: dict[str, Any]
518) -> None:
519 """Write an HTCondor submit file.
521 Parameters
522 ----------
523 filename : `str` or `os.PathLike`
524 Filename for the HTCondor submit file.
525 job_name : `str`
526 Job name to use in submit file.
527 job : `RestrictedDict`
528 Submit script information.
529 job_attrs : `dict`
530 Job attributes.
531 """
532 os.makedirs(os.path.dirname(filename), exist_ok=True)
533 with open(filename, "w") as fh:
534 for key, value in job.items():
535 if value is not None: 535 ↛ 534line 535 didn't jump to line 534 because the condition on line 535 was always true
536 if key in HTC_QUOTE_KEYS: # Assumes internal quotes are already escaped correctly
537 print(f'{key}="{value}"', file=fh)
538 else:
539 print(f"{key}={value}", file=fh)
540 for key in ["output", "error", "log"]:
541 if key not in job:
542 filename = f"{job_name}.$(Cluster).{'out' if key != 'log' else key}"
543 print(f"{key}={filename}", file=fh)
545 if job_attrs is not None:
546 htc_write_attribs(fh, job_attrs)
547 print("queue", file=fh)
550# To avoid doing the version check during every function call select
551# appropriate conversion function at the import time.
552#
553# Make sure that *each* version specific variant of the conversion function(s)
554# has the same signature after applying any changes!
555if HTC_VERSION < version.parse("8.9.8"): 555 ↛ 557line 555 didn't jump to line 557 because the condition on line 555 was never true
557 def htc_tune_schedd_args(**kwargs):
558 """Ensure that arguments for Schedd are version appropriate.
560 The old arguments: 'requirements' and 'attr_list' of
561 'Schedd.history()', 'Schedd.query()', and 'Schedd.xquery()' were
562 deprecated in favor of 'constraint' and 'projection', respectively,
563 starting from version 8.9.8. The function will convert "new" keyword
564 arguments to "old" ones.
566 Parameters
567 ----------
568 **kwargs
569 Any keyword arguments that Schedd.history(), Schedd.query(), and
570 Schedd.xquery() accepts.
572 Returns
573 -------
574 kwargs : `dict` [`str`, `~typing.Any`]
575 Keywords arguments that are guaranteed to work with the Python
576 HTCondor API.
578 Notes
579 -----
580 Function doesn't validate provided keyword arguments beyond converting
581 selected arguments to their version specific form. For example,
582 it won't remove keywords that are not supported by the methods
583 mentioned earlier.
584 """
585 translation_table = {
586 "constraint": "requirements",
587 "projection": "attr_list",
588 }
589 for new, old in translation_table.items():
590 try:
591 kwargs[old] = kwargs.pop(new)
592 except KeyError:
593 pass
594 return kwargs
596else:
598 def htc_tune_schedd_args(**kwargs):
599 """Ensure that arguments for Schedd are version appropriate.
601 This is the fallback function if no version specific alteration are
602 necessary. Effectively, a no-op.
604 Parameters
605 ----------
606 **kwargs
607 Any keyword arguments that Schedd.history(), Schedd.query(), and
608 Schedd.xquery() accepts.
610 Returns
611 -------
612 kwargs : `dict` [`str`, `~typing.Any`]
613 Keywords arguments that were passed to the function.
614 """
615 return kwargs
618def htc_query_history(schedds, **kwargs):
619 """Fetch history records from the condor_schedd daemon.
621 Parameters
622 ----------
623 schedds : `htcondor.Schedd`
624 HTCondor schedulers which to query for job information.
625 **kwargs
626 Any keyword arguments that Schedd.history() accepts.
628 Yields
629 ------
630 schedd_name : `str`
631 Name of the HTCondor scheduler managing the job queue.
632 job_ad : `dict` [`str`, `~typing.Any`]
633 A dictionary representing HTCondor ClassAd describing a job. It maps
634 job attributes names to values of the ClassAd expressions they
635 represent.
636 """
637 # If not set, provide defaults for positional arguments.
638 kwargs.setdefault("constraint", None)
639 kwargs.setdefault("projection", [])
640 kwargs = htc_tune_schedd_args(**kwargs)
641 for schedd_name, schedd in schedds.items():
642 for job_ad in schedd.history(**kwargs):
643 yield schedd_name, dict(job_ad)
646def htc_query_present(schedds, **kwargs):
647 """Query the condor_schedd daemon for job ads.
649 Parameters
650 ----------
651 schedds : `htcondor.Schedd`
652 HTCondor schedulers which to query for job information.
653 **kwargs
654 Any keyword arguments that Schedd.xquery() accepts.
656 Yields
657 ------
658 schedd_name : `str`
659 Name of the HTCondor scheduler managing the job queue.
660 job_ad : `dict` [`str`, `~typing.Any`]
661 A dictionary representing HTCondor ClassAd describing a job. It maps
662 job attributes names to values of the ClassAd expressions they
663 represent.
664 """
665 kwargs = htc_tune_schedd_args(**kwargs)
666 for schedd_name, schedd in schedds.items():
667 for job_ad in schedd.query(**kwargs):
668 yield schedd_name, dict(job_ad)
671def htc_version():
672 """Return the version given by the HTCondor API.
674 Returns
675 -------
676 version : `str`
677 HTCondor version as easily comparable string.
678 """
679 return str(HTC_VERSION)
682def htc_submit_dag(sub):
683 """Submit job for execution.
685 Parameters
686 ----------
687 sub : `htcondor.Submit`
688 An object representing a job submit description.
690 Returns
691 -------
692 schedd_job_info : `dict` [`str`, `dict` [`str`, \
693 `dict` [`str`, `~typing.Any`]]]
694 Information about jobs satisfying the search criteria where for each
695 Scheduler, local HTCondor job ids are mapped to their respective
696 classads.
697 """
698 coll = htcondor.Collector()
699 schedd_ad = coll.locate(htcondor.DaemonTypes.Schedd)
700 schedd = htcondor.Schedd(schedd_ad)
702 # If Schedd.submit() fails, the method will raise an exception. Usually,
703 # that implies issues with the HTCondor pool which BPS can't address.
704 # Hence, no effort is made to handle the exception.
705 submit_result = schedd.submit(sub)
707 # Sadly, the ClassAd from Schedd.submit() (see above) does not have
708 # 'GlobalJobId' so we need to run a regular query to get it anyway.
709 schedd_name = schedd_ad["Name"]
710 schedd_dag_info = condor_q(
711 constraint=f"ClusterId == {submit_result.cluster()}", schedds={schedd_name: schedd}
712 )
713 return schedd_dag_info
716def htc_create_submit_from_dag(
717 dag_filename: str, submit_options: dict[str, Any], dagman_conf_filename: str | os.PathLike | None = None
718) -> htcondor.Submit:
719 """Create a DAGMan job submit description.
721 Parameters
722 ----------
723 dag_filename : `str`
724 Name of file containing HTCondor DAG commands.
725 submit_options : `dict` [`str`, `~typing.Any`], optional
726 Contains extra options for command line (Value of None means flag).
727 dagman_conf_filename : `str` or `os.PathLike`, optional
728 Location of DAGMan configuration file, if Any. Defaults to no
729 file (None).
731 Returns
732 -------
733 sub : `htcondor.Submit`
734 An object representing a job submit description.
736 Notes
737 -----
738 Use with HTCondor versions which support htcondor.Submit.from_dag(),
739 i.e., 8.9.3 or newer.
740 """
741 # Config and environment variables do not seem to override -MaxIdle
742 # on the .dag.condor.sub's command line (broken in some 24.0.x versions).
743 # Explicitly forward them as a submit_option if either exists.
744 # Note: auto generated subdag submit files are still the -MaxIdle=1000
745 # in the broken versions.
746 if "MaxIdle" not in submit_options:
747 max_jobs_idle: int | None = None
748 config_var_name = "DAGMAN_MAX_JOBS_IDLE"
750 if dagman_conf_filename:
751 _LOG.debug("Checking DAGMan config file = %s", dagman_conf_filename)
752 with open(dagman_conf_filename) as fh:
753 for line in fh:
754 _LOG.debug("DAGMan config file line = %s", line)
755 parts = line.split("=")
756 _LOG.debug("DAGMan config file line parts = %s", parts)
757 if len(parts) == 2 and parts[0].strip() == config_var_name:
758 max_jobs_idle = int(parts[1].strip())
759 _LOG.debug("Found %s = %s", config_var_name, max_jobs_idle)
760 break
761 if max_jobs_idle is None:
762 if f"_CONDOR_{config_var_name}" in os.environ:
763 max_jobs_idle = int(os.environ[f"_CONDOR_{config_var_name}"])
764 elif config_var_name in htcondor.param:
765 max_jobs_idle = htcondor.param[config_var_name]
767 if max_jobs_idle:
768 submit_options["MaxIdle"] = max_jobs_idle
769 else:
770 _LOG.debug("MaxIdle already in submit_options: %s", submit_options)
772 _LOG.debug("Using submit_options = %s", submit_options)
773 return htcondor.Submit.from_dag(dag_filename, submit_options)
776def htc_create_submit_from_cmd(dag_filename, submit_options=None):
777 """Create a DAGMan job submit description.
779 Create a DAGMan job submit description by calling ``condor_submit_dag``
780 on given DAG description file.
782 Parameters
783 ----------
784 dag_filename : `str`
785 Name of file containing HTCondor DAG commands.
786 submit_options : `dict` [`str`, `~typing.Any`], optional
787 Contains extra options for command line (Value of None means flag).
789 Returns
790 -------
791 sub : `htcondor.Submit`
792 An object representing a job submit description.
794 Notes
795 -----
796 Use with HTCondor versions which do not support htcondor.Submit.from_dag(),
797 i.e., older than 8.9.3.
798 """
799 # Run command line condor_submit_dag command.
800 cmd = "condor_submit_dag -f -no_submit -notification never -autorescue 1 -UseDagDir -no_recurse "
802 if submit_options is not None:
803 for opt, val in submit_options.items():
804 cmd += f" -{opt} {val or ''}"
805 cmd += f"{dag_filename}"
807 process = subprocess.Popen(
808 cmd.split(), shell=False, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, encoding="utf-8"
809 )
810 process.wait()
812 if process.returncode != 0:
813 print(f"Exit code: {process.returncode}")
814 print(process.communicate()[0])
815 raise RuntimeError("Problems running condor_submit_dag")
817 return htc_create_submit_from_file(f"{dag_filename}.condor.sub")
820def htc_create_submit_from_file(submit_file):
821 """Parse a submission file.
823 Parameters
824 ----------
825 submit_file : `str`
826 Name of the HTCondor submit file.
828 Returns
829 -------
830 sub : `htcondor.Submit`
831 An object representing a job submit description.
832 """
833 descriptors = {}
834 with open(submit_file) as fh:
835 for line in fh:
836 line = line.strip()
837 if not line.startswith("#") and not line == "queue":
838 (key, val) = re.split(r"\s*=\s*", line, maxsplit=1)
839 descriptors[key] = val
841 # Avoid UserWarning: the line 'copy_to_spool = False' was
842 # unused by Submit object. Is it a typo?
843 try:
844 del descriptors["copy_to_spool"]
845 except KeyError:
846 pass
848 return htcondor.Submit(descriptors)
851def _htc_write_job_commands(stream, name, commands, node_type="JOB"):
852 """Output the DAGMan job lines for single job in DAG.
854 Parameters
855 ----------
856 stream : `~typing.TextIO`
857 Writeable text stream (typically an opened file).
858 name : `str`
859 Job name.
860 commands : `RestrictedDict`
861 DAG commands for a job.
862 node_type : `str`, optional
863 Type of DAGMan node (JOB, FINAL, SERVICE). Defaults to "JOB".
864 """
865 # Note: optional pieces of commands include a space at the beginning.
866 # also making sure values aren't empty strings as placeholders.
867 if "pre" in commands and commands["pre"]:
868 defer = ""
869 if "defer" in commands["pre"] and commands["pre"]["defer"]:
870 defer = f" DEFER {commands['pre']['defer']['status']} {commands['pre']['defer']['time']}"
872 debug = ""
873 if "debug" in commands["pre"] and commands["pre"]["debug"]:
874 debug = f" DEBUG {commands['pre']['debug']['filename']} {commands['pre']['debug']['type']}"
876 arguments = ""
877 if "arguments" in commands["pre"] and commands["pre"]["arguments"]:
878 arguments = f" {commands['pre']['arguments']}"
880 executable = commands["pre"]["executable"]
881 print(f"SCRIPT{defer}{debug} PRE {name} {executable}{arguments}", file=stream)
883 if "post" in commands and commands["post"]:
884 defer = ""
885 if "defer" in commands["post"] and commands["post"]["defer"]:
886 defer = f" DEFER {commands['post']['defer']['status']} {commands['post']['defer']['time']}"
888 debug = ""
889 if "debug" in commands["post"] and commands["post"]["debug"]:
890 debug = f" DEBUG {commands['post']['debug']['filename']} {commands['post']['debug']['type']}"
892 arguments = ""
893 if "arguments" in commands["post"] and commands["post"]["arguments"]:
894 arguments = f" {commands['post']['arguments']}"
896 executable = commands["post"]["executable"]
897 print(f"SCRIPT{defer}{debug} POST {name} {executable}{arguments}", file=stream)
899 if "vars" in commands and commands["vars"]:
900 for key, value in commands["vars"].items():
901 print(f'VARS {name} {key}="{htc_escape(value)}"', file=stream)
903 if "pre_skip" in commands and commands["pre_skip"]:
904 print(f"PRE_SKIP {name} {commands['pre_skip']}", file=stream)
906 # FINAL node cannot have a DAGMan retry, abort-dag-on, priority, category
907 if node_type != "FINAL":
908 if "retry" in commands and commands["retry"]:
909 print(f"RETRY {name} {commands['retry']}", end="", file=stream)
910 if "retry_unless_exit" in commands and commands["retry_unless_exit"]:
911 print(f" UNLESS-EXIT {commands['retry_unless_exit']}", end="", file=stream)
912 print("", file=stream) # Since previous prints don't include new line
914 if "abort_dag_on" in commands and commands["abort_dag_on"]:
915 print(
916 f"ABORT-DAG-ON {name} {commands['abort_dag_on']['node_exit']}"
917 f" RETURN {commands['abort_dag_on']['abort_exit']}",
918 file=stream,
919 )
921 if "priority" in commands and commands["priority"]:
922 print(
923 f"PRIORITY {name} {commands['priority']}",
924 file=stream,
925 )
928class HTCJob:
929 """HTCondor job for use in building DAG.
931 Parameters
932 ----------
933 name : `str`
934 Name of the job.
935 label : `str`
936 Label that can used for grouping or lookup.
937 initcmds : `RestrictedDict`
938 Initial job commands for submit file.
939 initdagcmds : `RestrictedDict`
940 Initial commands for job inside DAG.
941 initattrs : `dict`
942 Initial dictionary of job attributes.
943 """
945 def __init__(self, name, label=None, initcmds=(), initdagcmds=(), initattrs=None):
946 self.name = name
947 self.label = label
948 self.cmds = RestrictedDict(HTC_VALID_JOB_KEYS, initcmds)
949 self.dagcmds = RestrictedDict(HTC_VALID_JOB_DAG_KEYS, initdagcmds)
950 self.attrs = initattrs
951 self.subfile = None
952 self.subdir = None
953 self.subdag = None
955 def __str__(self):
956 return self.name
958 def add_job_cmds(self, new_commands):
959 """Add commands to Job (overwrite existing).
961 Parameters
962 ----------
963 new_commands : `dict`
964 Submit file commands to be added to Job.
965 """
966 self.cmds.update(new_commands)
968 def add_dag_cmds(self, new_commands):
969 """Add DAG commands to Job (overwrite existing).
971 Parameters
972 ----------
973 new_commands : `dict`
974 DAG file commands to be added to Job.
975 """
976 self.dagcmds.update(new_commands)
978 def add_job_attrs(self, new_attrs):
979 """Add attributes to Job (overwrite existing).
981 Parameters
982 ----------
983 new_attrs : `dict`
984 Attributes to be added to Job.
985 """
986 if self.attrs is None:
987 self.attrs = {}
988 if new_attrs:
989 self.attrs.update(new_attrs)
991 def write_submit_file(self, submit_path: str | os.PathLike) -> None:
992 """Write job description to submit file.
994 Parameters
995 ----------
996 submit_path : `str` or `os.PathLike`
997 Prefix path for the submit file.
998 """
999 if not self.subfile:
1000 self.subfile = f"{self.name}.sub"
1002 subfile = self.subfile
1003 if self.subdir: 1003 ↛ 1004line 1003 didn't jump to line 1004 because the condition on line 1003 was never true
1004 subfile = Path(self.subdir) / subfile
1006 subfile = Path(os.path.expandvars(subfile))
1007 if not subfile.is_absolute():
1008 subfile = Path(submit_path) / subfile
1009 if not subfile.exists():
1010 htc_write_condor_file(subfile, self.name, self.cmds, self.attrs)
1012 def write_dag_commands(self, stream, dag_rel_path, command_name="JOB"):
1013 """Write DAG commands for single job to output stream.
1015 Parameters
1016 ----------
1017 stream : `~typing.TextIO`
1018 Output Stream.
1019 dag_rel_path : `str`
1020 Relative path of dag to submit directory.
1021 command_name : `str`
1022 Name of the DAG command (e.g., JOB, FINAL).
1023 """
1024 subfile = os.path.expandvars(self.subfile)
1026 # JOB NodeName SubmitDescription [DIR directory] [NOOP] [DONE]
1027 job_line = f'{command_name} {self.name} "{subfile}"'
1028 if "dir" in self.dagcmds:
1029 dir_val = self.dagcmds["dir"]
1030 if dag_rel_path: 1030 ↛ 1032line 1030 didn't jump to line 1032 because the condition on line 1030 was always true
1031 dir_val = os.path.join(dag_rel_path, dir_val)
1032 job_line += f' DIR "{dir_val}"'
1033 if self.dagcmds.get("noop", False):
1034 job_line += " NOOP"
1036 print(job_line, file=stream)
1037 if self.dagcmds:
1038 _htc_write_job_commands(stream, self.name, self.dagcmds, command_name)
1040 def dump(self, fh):
1041 """Dump job information to output stream.
1043 Parameters
1044 ----------
1045 fh : `~typing.TextIO`
1046 Output stream.
1047 """
1048 printer = pprint.PrettyPrinter(indent=4, stream=fh)
1049 printer.pprint(self.name)
1050 printer.pprint(self.cmds)
1051 printer.pprint(self.attrs)
1054class HTCDag(networkx.DiGraph):
1055 """HTCondor DAG.
1057 Parameters
1058 ----------
1059 data : `~typing.Any`
1060 Initial graph data of any format that is supported
1061 by the to_network_graph() function.
1062 name : `str`
1063 Name for DAG.
1064 """
1066 def __init__(self, data=None, name=""):
1067 super().__init__(data=data, name=name)
1069 self.graph["attr"] = {}
1070 self.graph["run_id"] = None
1071 self.graph["submit_path"] = None
1072 self.graph["final_job"] = None
1073 self.graph["service_job"] = None
1074 self.graph["submit_options"] = {}
1076 def __str__(self):
1077 """Represent basic DAG info as string.
1079 Returns
1080 -------
1081 info : `str`
1082 String containing basic DAG info.
1083 """
1084 return f"{self.graph['name']} {len(self)}"
1086 def add_attribs(self, attribs=None):
1087 """Add attributes to the DAG.
1089 Parameters
1090 ----------
1091 attribs : `dict`
1092 DAG attributes.
1093 """
1094 if attribs is not None: 1094 ↛ exitline 1094 didn't return from function 'add_attribs' because the condition on line 1094 was always true
1095 self.graph["attr"].update(attribs)
1097 def add_job(self, job, parent_names=None, child_names=None):
1098 """Add an HTCJob to the HTCDag.
1100 Parameters
1101 ----------
1102 job : `HTCJob`
1103 HTCJob to add to the HTCDag.
1104 parent_names : `~collections.abc.Iterable` [`str`], optional
1105 Names of parent jobs.
1106 child_names : `~collections.abc.Iterable` [`str`], optional
1107 Names of child jobs.
1108 """
1109 assert isinstance(job, HTCJob)
1110 _LOG.debug("Adding job %s to dag", job.name)
1112 # Add dag level attributes to each job
1113 job.add_job_attrs(self.graph["attr"])
1115 self.add_node(job.name, data=job)
1117 if parent_names is not None: 1117 ↛ 1118line 1117 didn't jump to line 1118 because the condition on line 1117 was never true
1118 self.add_job_relationships(parent_names, [job.name])
1120 if child_names is not None: 1120 ↛ 1121line 1120 didn't jump to line 1121 because the condition on line 1120 was never true
1121 self.add_job_relationships(child_names, [job.name])
1123 def add_job_relationships(self, parents, children):
1124 """Add DAG edge between parents and children jobs.
1126 Parameters
1127 ----------
1128 parents : `list` [`str`]
1129 Contains parent job name(s).
1130 children : `list` [`str`]
1131 Contains children job name(s).
1132 """
1133 self.add_edges_from(itertools.product(parents, children))
1135 def add_final_job(self, job):
1136 """Add an HTCJob for the FINAL job in HTCDag.
1138 Parameters
1139 ----------
1140 job : `HTCJob`
1141 HTCJob to add to the HTCDag as a FINAL job.
1142 """
1143 # Add dag level attributes to each job
1144 job.add_job_attrs(self.graph["attr"])
1146 self.graph["final_job"] = job
1148 def add_service_job(self, job):
1149 """Add an HTCJob for the SERVICE job in HTCDag.
1151 Parameters
1152 ----------
1153 job : `HTCJob`
1154 HTCJob to add to the HTCDag as a SERVICE job.
1155 """
1156 # Add dag level attributes to each job
1157 job.add_job_attrs(self.graph["attr"])
1159 self.graph["service_job"] = job
1161 def del_job(self, job_name):
1162 """Delete the job from the DAG.
1164 Parameters
1165 ----------
1166 job_name : `str`
1167 Name of job in DAG to delete.
1168 """
1169 # Reconnect edges around node to delete
1170 parents = self.predecessors(job_name)
1171 children = self.successors(job_name)
1172 self.add_edges_from(itertools.product(parents, children))
1174 # Delete job node (which deletes its edges).
1175 self.remove_node(job_name)
1177 def write(self, submit_path, job_subdir="", dag_subdir="", dag_rel_path=""):
1178 """Write DAG to a file.
1180 Parameters
1181 ----------
1182 submit_path : `str`
1183 Prefix path for all outputs.
1184 job_subdir : `str`, optional
1185 Template for job subdir (submit_path + job_subdir).
1186 dag_subdir : `str`, optional
1187 DAG subdir (submit_path + dag_subdir).
1188 dag_rel_path : `str`, optional
1189 Prefix to job_subdir for jobs inside subdag.
1190 """
1191 self.graph["submit_path"] = submit_path
1192 self.graph["dag_filename"] = os.path.join(dag_subdir, f"{self.graph['name']}.dag")
1193 full_filename = os.path.join(submit_path, self.graph["dag_filename"])
1194 os.makedirs(os.path.dirname(full_filename), exist_ok=True)
1196 try:
1197 dagman_config_path = Path(self.graph["attr"]["bps_wms_config_path"])
1198 except KeyError:
1199 dagman_config_path = None
1200 with open(full_filename, "w") as fh:
1201 if dagman_config_path is not None:
1202 fh.write(f"CONFIG {dag_rel_path / dagman_config_path}\n")
1204 for name, nodeval in self.nodes().items():
1205 try:
1206 job = nodeval["data"]
1207 except KeyError:
1208 _LOG.error("Job %s doesn't have data (keys: %s).", name, nodeval.keys())
1209 raise
1210 if job.subdag: 1210 ↛ 1211line 1210 didn't jump to line 1211 because the condition on line 1210 was never true
1211 dag_subdir = f"subdags/{job.name}"
1212 if "dir" in job.dagcmds:
1213 subdir = job.dagcmds["dir"]
1214 else:
1215 subdir = job_subdir
1216 if dagman_config_path is not None:
1217 job.subdag.add_attribs({"bps_wms_config_path": str(dagman_config_path)})
1218 job.subdag.write(submit_path, subdir, dag_subdir, "../..")
1219 fh.write(
1220 f"SUBDAG EXTERNAL {job.name} {Path(job.subdag.graph['dag_filename']).name} "
1221 f"DIR {dag_subdir}\n"
1222 )
1223 if job.dagcmds:
1224 _htc_write_job_commands(fh, job.name, job.dagcmds)
1225 else:
1226 job.write_submit_file(submit_path)
1227 job.write_dag_commands(fh, dag_rel_path)
1229 for edge in self.edges(): 1229 ↛ 1230line 1229 didn't jump to line 1230 because the loop on line 1229 never started
1230 print(f"PARENT {edge[0]} CHILD {edge[1]}", file=fh)
1231 print(f"DOT {self.name}.dot", file=fh)
1232 print(f"NODE_STATUS_FILE {self.name}.node_status", file=fh)
1234 # Add bps attributes to dag submission
1235 for key, value in self.graph["attr"].items():
1236 print(f'SET_JOB_ATTR {key}= "{htc_escape(value)}"', file=fh)
1238 # Add special nodes if any.
1239 special_jobs = {
1240 "FINAL": self.graph["final_job"],
1241 "SERVICE": self.graph["service_job"],
1242 }
1243 for dagcmd, job in special_jobs.items():
1244 if job is not None: 1244 ↛ 1245line 1244 didn't jump to line 1245 because the condition on line 1244 was never true
1245 job.write_submit_file(submit_path)
1246 job.write_dag_commands(fh, dag_rel_path, dagcmd)
1248 def dump(self, fh):
1249 """Dump DAG info to output stream.
1251 Parameters
1252 ----------
1253 fh : `typing.IO`
1254 Where to dump DAG info as text.
1255 """
1256 for key, value in self.graph:
1257 print(f"{key}={value}", file=fh)
1258 for name, data in self.nodes().items():
1259 print(f"{name}:", file=fh)
1260 data.dump(fh)
1261 for edge in self.edges():
1262 print(f"PARENT {edge[0]} CHILD {edge[1]}", file=fh)
1263 if self.graph["final_job"]:
1264 print(f"FINAL {self.graph['final_job'].name}:", file=fh)
1265 self.graph["final_job"].dump(fh)
1267 def write_dot(self, filename):
1268 """Write a dot version of the DAG.
1270 Parameters
1271 ----------
1272 filename : `str`
1273 Name of the dot file.
1274 """
1275 pos = networkx.nx_agraph.graphviz_layout(self)
1276 networkx.draw(self, pos=pos)
1277 networkx.drawing.nx_pydot.write_dot(self, filename)
1280def condor_q(constraint=None, schedds=None, **kwargs):
1281 """Get information about the jobs in the HTCondor job queue(s).
1283 Parameters
1284 ----------
1285 constraint : `str`, optional
1286 Constraints to be passed to job query.
1287 schedds : `dict` [`str`, `htcondor.Schedd`], optional
1288 HTCondor schedulers which to query for job information. If None
1289 (default), the query will be run against local scheduler only.
1290 **kwargs : `~typing.Any`
1291 Additional keyword arguments that need to be passed to the internal
1292 query method.
1294 Returns
1295 -------
1296 job_info : `dict` [`str`, `dict` [`str`, `dict` [`str`, `~typing.Any`]]]
1297 Information about jobs satisfying the search criteria where for each
1298 Scheduler, local HTCondor job ids are mapped to their respective
1299 classads.
1300 """
1301 return condor_query(constraint, schedds, htc_query_present, **kwargs)
1304def condor_history(constraint=None, schedds=None, **kwargs):
1305 """Get information about the jobs from HTCondor history records.
1307 Parameters
1308 ----------
1309 constraint : `str`, optional
1310 Constraints to be passed to job query.
1311 schedds : `dict` [`str`, `htcondor.Schedd`], optional
1312 HTCondor schedulers which to query for job information. If None
1313 (default), the query will be run against the history file of
1314 the local scheduler only.
1315 **kwargs : `~typing.Any`
1316 Additional keyword arguments that need to be passed to the internal
1317 query method.
1319 Returns
1320 -------
1321 job_info : `dict` [`str`, `dict` [`str`, `dict` [`str`, `~typing.Any`]]]
1322 Information about jobs satisfying the search criteria where for each
1323 Scheduler, local HTCondor job ids are mapped to their respective
1324 classads.
1325 """
1326 return condor_query(constraint, schedds, htc_query_history, **kwargs)
1329def condor_query(constraint=None, schedds=None, query_func=htc_query_present, **kwargs):
1330 """Get information about HTCondor jobs.
1332 Parameters
1333 ----------
1334 constraint : `str`, optional
1335 Constraints to be passed to job query.
1336 schedds : `dict` [`str`, `htcondor.Schedd`], optional
1337 HTCondor schedulers which to query for job information. If None
1338 (default), the query will be run against the history file of
1339 the local scheduler only.
1340 query_func : `~collections.abc.Callable`
1341 An query function which takes following arguments:
1343 - ``schedds``: Schedulers to query (`list` [`htcondor.Schedd`]).
1344 - ``**kwargs``: Keyword arguments that will be passed to the query
1345 function.
1346 **kwargs : `~typing.Any`
1347 Additional keyword arguments that need to be passed to the query
1348 method.
1350 Returns
1351 -------
1352 job_info : `dict` [`str`, `dict` [`str`, `dict` [`str`, `~typing.Any`]]]
1353 Information about jobs satisfying the search criteria where for each
1354 Scheduler, local HTCondor job ids are mapped to their respective
1355 classads.
1356 """
1357 if not schedds:
1358 coll = htcondor.Collector()
1359 schedd_ad = coll.locate(htcondor.DaemonTypes.Schedd)
1360 schedds = {schedd_ad["Name"]: htcondor.Schedd(schedd_ad)}
1362 # Make sure that 'ClusterId' and 'ProcId' attributes are always included
1363 # in the job classad. They are needed to construct the job id.
1364 added_attrs = set()
1365 if "projection" in kwargs and kwargs["projection"]:
1366 requested_attrs = set(kwargs["projection"])
1367 required_attrs = {"ClusterId", "ProcId"}
1368 added_attrs = required_attrs - requested_attrs
1369 for attr in added_attrs:
1370 kwargs["projection"].append(attr)
1372 unwanted_attrs = {"Env", "Environment"} | added_attrs
1373 job_info = defaultdict(dict)
1374 for schedd_name, job_ad in query_func(schedds, constraint=constraint, **kwargs):
1375 id_ = f"{job_ad['ClusterId']}.{job_ad['ProcId']}"
1376 for attr in set(job_ad) & unwanted_attrs:
1377 del job_ad[attr]
1378 job_info[schedd_name][id_] = job_ad
1379 _LOG.debug("query returned %d jobs", sum(len(val) for val in job_info.values()))
1381 # Restore the list of the requested attributes to its original value
1382 # if needed.
1383 if added_attrs:
1384 for attr in added_attrs:
1385 kwargs["projection"].remove(attr)
1387 # When returning the results filter out entries for schedulers with no jobs
1388 # matching the search criteria.
1389 return {key: val for key, val in job_info.items() if val}
1392def condor_search(constraint=None, hist=None, schedds=None):
1393 """Search for running and finished jobs satisfying given criteria.
1395 Parameters
1396 ----------
1397 constraint : `str`, optional
1398 Constraints to be passed to job query.
1399 hist : `float`
1400 Limit history search to this many days.
1401 schedds : `dict` [`str`, `htcondor.Schedd`], optional
1402 The list of the HTCondor schedulers which to query for job information.
1403 If None (default), only the local scheduler will be queried.
1405 Returns
1406 -------
1407 job_info : `dict` [`str`, `dict` [`str`, `dict` [`str` `~typing.Any`]]]
1408 Information about jobs satisfying the search criteria where for each
1409 Scheduler, local HTCondor job ids are mapped to their respective
1410 classads.
1411 """
1412 if not schedds:
1413 coll = htcondor.Collector()
1414 schedd_ad = coll.locate(htcondor.DaemonTypes.Schedd)
1415 schedds = {schedd_ad["Name"]: htcondor.Schedd(locate_ad=schedd_ad)}
1417 job_info = condor_q(constraint=constraint, schedds=schedds)
1418 if hist is not None:
1419 _LOG.debug("Searching history going back %s days", hist)
1420 epoch = (datetime.now() - timedelta(days=hist)).timestamp()
1421 constraint += f" && (CompletionDate >= {epoch} || JobFinishedHookDone >= {epoch})"
1422 hist_info = condor_history(constraint, schedds=schedds)
1423 update_job_info(job_info, hist_info)
1424 return job_info
1427def condor_status(constraint=None, coll=None):
1428 """Get information about HTCondor pool.
1430 Parameters
1431 ----------
1432 constraint : `str`, optional
1433 Constraints to be passed to the query.
1434 coll : `htcondor.Collector`, optional
1435 Object representing HTCondor collector daemon.
1437 Returns
1438 -------
1439 pool_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1440 Mapping between HTCondor slot names and slot information (classAds).
1441 """
1442 if coll is None:
1443 coll = htcondor.Collector()
1444 try:
1445 pool_ads = coll.query(constraint=constraint)
1446 except OSError as ex:
1447 raise RuntimeError(f"Problem querying the Collector. (Constraint='{constraint}')") from ex
1449 pool_info = {}
1450 for slot in pool_ads:
1451 pool_info[slot["name"]] = dict(slot)
1452 _LOG.debug("condor_status returned %d ads", len(pool_info))
1453 return pool_info
1456def update_job_info(job_info, other_info):
1457 """Update results of a job query with results from another query.
1459 Parameters
1460 ----------
1461 job_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1462 Results of the job query that needs to be updated.
1463 other_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1464 Results of the other job query.
1466 Returns
1467 -------
1468 job_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1469 The updated results.
1470 """
1471 for schedd_name, others in other_info.items():
1472 try:
1473 jobs = job_info[schedd_name]
1474 except KeyError:
1475 job_info[schedd_name] = others
1476 else:
1477 for id_, ad in others.items():
1478 jobs.setdefault(id_, {}).update(ad)
1479 return job_info
1482def count_jobs_in_single_dag(
1483 filename: str | os.PathLike,
1484) -> tuple[Counter[str], dict[str, str], dict[str, WmsNodeType]]:
1485 """Build bps_run_summary string from dag file.
1487 Parameters
1488 ----------
1489 filename : `str`
1490 Path that includes dag file for a run.
1492 Returns
1493 -------
1494 counts : `Counter` [`str`]
1495 Semi-colon separated list of job labels and counts.
1496 (Same format as saved in dag classad).
1497 job_name_to_label : `dict` [`str`, `str`]
1498 Mapping of job names to job labels.
1499 job_name_to_type : `dict` [`str`, `lsst.ctrl.bps.htcondor.WmsNodeType`]
1500 Mapping of job names to job types
1501 (e.g., payload, final, service).
1502 """
1503 # Later code depends upon insertion order
1504 counts: Counter = Counter() # counts of payload jobs per label
1505 job_name_to_label: dict[str, str] = {}
1506 job_name_to_type: dict[str, WmsNodeType] = {}
1507 with open(filename) as fh:
1508 for line in fh:
1509 # Skip any line that contains commands irrelevant to job counting.
1510 if not line.startswith(
1511 (
1512 "JOB",
1513 "FINAL",
1514 "SERVICE",
1515 "SUBDAG EXTERNAL",
1516 )
1517 ):
1518 continue
1520 m = re.match(
1521 r"(?P<command>JOB|FINAL|SERVICE|SUBDAG EXTERNAL)\s+"
1522 r'(?P<jobname>(?P<wms>wms_)?\S+)\s+"?(?P<subfile>\S+)"?\s*'
1523 r'(DIR "?(?P<dir>[^\s"]+)"?)?\s*(?P<noop>NOOP)?',
1524 line,
1525 )
1526 if m: 1526 ↛ 1578line 1526 didn't jump to line 1578 because the condition on line 1526 was always true
1527 job_name = m.group("jobname")
1528 name_parts = job_name.split("_")
1530 label = ""
1531 if m.group("dir"):
1532 dir_match = re.search(r"jobs/([^\s/]+)", m.group("dir"))
1533 if dir_match:
1534 label = dir_match.group(1)
1535 else:
1536 _LOG.debug("Parse DAG: unparsed dir = %s", line)
1537 elif m.group("subfile"): 1537 ↛ 1544line 1537 didn't jump to line 1544 because the condition on line 1537 was always true
1538 subfile_match = re.search(r"jobs/([^\s/]+)", m.group("subfile"))
1539 if subfile_match:
1540 label = m.group("subfile").split("/")[1]
1541 else:
1542 label = pegasus_name_to_label(job_name)
1544 match m.group("command"):
1545 case "JOB":
1546 if m.group("noop"):
1547 job_type = WmsNodeType.NOOP
1548 # wms_noop_label
1549 label = name_parts[2]
1550 elif m.group("wms"):
1551 if name_parts[1] == "check": 1551 ↛ 1556line 1551 didn't jump to line 1556 because the condition on line 1551 was always true
1552 job_type = WmsNodeType.SUBDAG_CHECK
1553 # wms_check_status_wms_group_label
1554 label = name_parts[5]
1555 else:
1556 _LOG.warning(
1557 "Unexpected skipping of dag line due to unknown wms job: %s", line
1558 )
1559 else:
1560 job_type = WmsNodeType.PAYLOAD
1561 if label == "init": 1561 ↛ 1562line 1561 didn't jump to line 1562 because the condition on line 1561 was never true
1562 label = "pipetaskInit"
1563 counts[label] += 1
1564 case "FINAL":
1565 job_type = WmsNodeType.FINAL
1566 counts[label] += 1 # final counts a payload job.
1567 case "SERVICE":
1568 job_type = WmsNodeType.SERVICE
1569 case "SUBDAG EXTERNAL": 1569 ↛ 1573line 1569 didn't jump to line 1573 because the pattern on line 1569 always matched
1570 job_type = WmsNodeType.SUBDAG
1571 label = name_parts[2]
1573 job_name_to_label[job_name] = label
1574 job_name_to_type[job_name] = job_type
1575 else:
1576 # The line should, but didn't match the pattern above. Probably
1577 # problems with regex.
1578 _LOG.warning("Unexpected skipping of dag line: %s", line)
1580 return counts, job_name_to_label, job_name_to_type
1583def summarize_dag(dir_name: str) -> tuple[str, dict[str, str], dict[str, WmsNodeType]]:
1584 """Build bps_run_summary string from dag file.
1586 Parameters
1587 ----------
1588 dir_name : `str`
1589 Path that includes dag file for a run.
1591 Returns
1592 -------
1593 summary : `str`
1594 Semi-colon separated list of job labels and counts
1595 (Same format as saved in dag classad).
1596 job_name_to_label : `dict` [`str`, `str`]
1597 Mapping of job names to job labels.
1598 job_name_to_type : `dict` [`str`, `lsst.ctrl.bps.htcondor.WmsNodeType`]
1599 Mapping of job names to job types
1600 (e.g., payload, final, service).
1601 """
1602 # Later code depends upon insertion order
1603 counts: Counter[str] = Counter() # counts of payload jobs per label
1604 job_name_to_label: dict[str, str] = {}
1605 job_name_to_type: dict[str, WmsNodeType] = {}
1606 for filename in Path(dir_name).glob("*.dag"):
1607 single_counts, single_job_name_to_label, single_job_name_to_type = count_jobs_in_single_dag(filename)
1608 counts += single_counts
1609 _update_dicts(job_name_to_label, single_job_name_to_label)
1610 _update_dicts(job_name_to_type, single_job_name_to_type)
1612 for filename in Path(dir_name).glob("subdags/*/*.dag"):
1613 single_counts, single_job_name_to_label, single_job_name_to_type = count_jobs_in_single_dag(filename)
1614 counts += single_counts
1615 _update_dicts(job_name_to_label, single_job_name_to_label)
1616 _update_dicts(job_name_to_type, single_job_name_to_type)
1618 summary = ";".join([f"{name}:{counts[name]}" for name in counts])
1619 _LOG.debug("summarize_dag: %s %s %s", summary, job_name_to_label, job_name_to_type)
1620 return summary, job_name_to_label, job_name_to_type
1623def pegasus_name_to_label(name):
1624 """Convert pegasus job name to a label for the report.
1626 Parameters
1627 ----------
1628 name : `str`
1629 Name of job.
1631 Returns
1632 -------
1633 label : `str`
1634 Label for job.
1635 """
1636 label = "UNK"
1637 if name.startswith("create_dir") or name.startswith("stage_in") or name.startswith("stage_out"): 1637 ↛ 1638line 1637 didn't jump to line 1638 because the condition on line 1637 was never true
1638 label = "pegasus"
1639 else:
1640 m = re.match(r"pipetask_(\d+_)?([^_]+)", name)
1641 if m: 1641 ↛ 1642line 1641 didn't jump to line 1642 because the condition on line 1641 was never true
1642 label = m.group(2)
1643 if label == "init":
1644 label = "pipetaskInit"
1646 return label
1649def read_single_dag_status(filename: str | os.PathLike) -> dict[str, Any]:
1650 """Read the node status file for DAG summary information.
1652 Parameters
1653 ----------
1654 filename : `str` or `Path.pathlib`
1655 Node status filename.
1657 Returns
1658 -------
1659 dag_ad : `dict` [`str`, `~typing.Any`]
1660 DAG summary information.
1661 """
1662 dag_ad: dict[str, Any] = {}
1664 # While this is probably more up to date than dag classad, only read from
1665 # file if need to.
1666 try:
1667 node_stat_file = Path(filename)
1668 _LOG.debug("Reading Node Status File %s", node_stat_file)
1669 with open(node_stat_file) as infh:
1670 dag_ad = dict(classad.parseNext(infh)) # pylint: disable=E1101
1672 if not dag_ad: 1672 ↛ 1674line 1672 didn't jump to line 1674 because the condition on line 1672 was never true
1673 # Pegasus check here
1674 metrics_file = node_stat_file.with_suffix(".dag.metrics")
1675 if metrics_file.exists():
1676 with open(metrics_file) as infh:
1677 metrics = json.load(infh)
1678 dag_ad["NodesTotal"] = metrics.get("jobs", 0)
1679 dag_ad["NodesFailed"] = metrics.get("jobs_failed", 0)
1680 dag_ad["NodesDone"] = metrics.get("jobs_succeeded", 0)
1681 metrics_file = node_stat_file.with_suffix(".metrics")
1682 with open(metrics_file) as infh:
1683 metrics = json.load(infh)
1684 dag_ad["NodesTotal"] = metrics["wf_metrics"]["total_jobs"]
1685 except (OSError, PermissionError):
1686 pass
1688 _LOG.debug("read_dag_status: %s", dag_ad)
1689 return dag_ad
1692def read_dag_status(wms_path: str | os.PathLike) -> dict[str, Any]:
1693 """Read the node status file for DAG summary information.
1695 Parameters
1696 ----------
1697 wms_path : `str` or `os.PathLike`
1698 Path that includes node status file for a run.
1700 Returns
1701 -------
1702 dag_ad : `dict` [`str`, `~typing.Any`]
1703 DAG summary information, counts summed across any subdags.
1704 """
1705 dag_ads: dict[str, Any] = {}
1706 path = Path(wms_path)
1707 try:
1708 node_stat_file = next(path.glob("*.node_status"))
1709 except StopIteration as exc:
1710 raise FileNotFoundError(f"DAGMan node status not found in {wms_path}") from exc
1712 dag_ads = read_single_dag_status(node_stat_file)
1714 for node_stat_file in path.glob("subdags/*/*.node_status"):
1715 dag_ad = read_single_dag_status(node_stat_file)
1716 dag_ads["JobProcsHeld"] += dag_ad.get("JobProcsHeld", 0)
1717 dag_ads["NodesPost"] += dag_ad.get("NodesPost", 0)
1718 dag_ads["JobProcsIdle"] += dag_ad.get("JobProcsIdle", 0)
1719 dag_ads["NodesTotal"] += dag_ad.get("NodesTotal", 0)
1720 dag_ads["NodesFailed"] += dag_ad.get("NodesFailed", 0)
1721 dag_ads["NodesDone"] += dag_ad.get("NodesDone", 0)
1722 dag_ads["NodesQueued"] += dag_ad.get("NodesQueued", 0)
1723 dag_ads["NodesPre"] += dag_ad.get("NodesReady", 0)
1724 dag_ads["NodesFutile"] += dag_ad.get("NodesFutile", 0)
1725 dag_ads["NodesUnready"] += dag_ad.get("NodesUnready", 0)
1727 return dag_ads
1730def read_single_node_status(filename: str | os.PathLike, init_fake_id: int) -> dict[str, Any]:
1731 """Read entire node status file.
1733 Parameters
1734 ----------
1735 filename : `str` or `pathlib.Path`
1736 Node status filename.
1737 init_fake_id : `int`
1738 Initial fake id value.
1740 Returns
1741 -------
1742 jobs : `dict` [`str`, `~typing.Any`]
1743 DAG summary information compiled from the node status file combined
1744 with the information found in the node event log.
1746 Currently, if the same job attribute is found in both files, its value
1747 from the event log takes precedence over the value from the node status
1748 file.
1749 """
1750 filename = Path(filename)
1752 # Get jobid info from other places to fill in gaps in info from node_status
1753 _, job_name_to_label, job_name_to_type = count_jobs_in_single_dag(filename.with_suffix(".dag"))
1754 loginfo: dict[str, dict[str, Any]] = {}
1755 try:
1756 wms_workflow_id, _ = read_single_dag_log(filename.with_suffix(".dag.dagman.log"))
1757 loginfo = read_single_dag_nodes_log(filename.with_suffix(".dag.nodes.log"))
1758 except (OSError, PermissionError):
1759 pass
1761 job_name_to_id: dict[str, str] = {}
1762 _LOG.debug("loginfo = %s", loginfo)
1763 log_job_name_to_id: dict[str, str] = {}
1764 for job_id, job_info in loginfo.items():
1765 if "LogNotes" in job_info: 1765 ↛ 1764line 1765 didn't jump to line 1764 because the condition on line 1765 was always true
1766 m = re.match(r"DAG Node: (\S+)", job_info["LogNotes"])
1767 if m: 1767 ↛ 1764line 1767 didn't jump to line 1764 because the condition on line 1767 was always true
1768 job_name = m.group(1)
1769 log_job_name_to_id[job_name] = job_id
1770 job_info["DAGNodeName"] = job_name
1771 job_info["wms_node_type"] = job_name_to_type[job_name]
1772 job_info["bps_job_label"] = job_name_to_label[job_name]
1774 jobs = {}
1775 fake_id = init_fake_id # For nodes that do not yet have a job id, give fake one
1776 try:
1777 with open(filename) as fh:
1778 for ad in classad.parseAds(fh):
1779 match ad["Type"]:
1780 case "DagStatus":
1781 # Skip DAG summary.
1782 pass
1783 case "NodeStatus":
1784 job_name = ad["Node"]
1785 if job_name in job_name_to_label: 1785 ↛ 1787line 1785 didn't jump to line 1787 because the condition on line 1785 was always true
1786 job_label = job_name_to_label[job_name]
1787 elif "_" in job_name:
1788 job_label = job_name.split("_")[1]
1789 else:
1790 job_label = job_name
1792 job = dict(ad)
1793 if job_name in log_job_name_to_id:
1794 job_id = str(log_job_name_to_id[job_name])
1795 _update_dicts(job, loginfo[job_id])
1796 else:
1797 job_id = str(fake_id)
1798 job = dict(ad)
1799 fake_id -= 1
1800 jobs[job_id] = job
1801 job_name_to_id[job_name] = job_id
1803 # Make job info as if came from condor_q.
1804 job["ClusterId"] = int(float(job_id))
1805 job["DAGManJobID"] = wms_workflow_id
1806 job["DAGNodeName"] = job_name
1807 job["bps_job_label"] = job_label
1808 job["wms_node_type"] = job_name_to_type[job_name]
1810 case "StatusEnd": 1810 ↛ 1813line 1810 didn't jump to line 1813 because the pattern on line 1810 always matched
1811 # Skip node status file "epilog".
1812 pass
1813 case _:
1814 _LOG.debug(
1815 "Ignoring unknown classad type '%s' in the node status file '%s'",
1816 ad["Type"],
1817 filename,
1818 )
1819 except (OSError, PermissionError):
1820 pass
1822 # Check for missing jobs (e.g., submission failure or not submitted yet)
1823 # Use dag info to create job placeholders
1824 for name in set(job_name_to_label) - set(job_name_to_id):
1825 if name in log_job_name_to_id: # job was in nodes.log, but not node_status
1826 job_id = str(log_job_name_to_id[name])
1827 job = dict(loginfo[job_id])
1828 else:
1829 job_id = str(fake_id)
1830 fake_id -= 1
1831 job = {}
1832 job["NodeStatus"] = NodeStatus.NOT_READY
1834 job["ClusterId"] = int(float(job_id))
1835 job["ProcId"] = 0
1836 job["DAGManJobID"] = wms_workflow_id
1837 job["DAGNodeName"] = name
1838 job["bps_job_label"] = job_name_to_label[name]
1839 job["wms_node_type"] = job_name_to_type[name]
1840 jobs[f"{job['ClusterId']}.{job['ProcId']}"] = job
1842 for job_info in jobs.values():
1843 job_info["from_dag_job"] = f"wms_{filename.stem}"
1845 return jobs
1848def read_node_status(wms_path: str | os.PathLike) -> dict[str, dict[str, Any]]:
1849 """Read entire node status file.
1851 Parameters
1852 ----------
1853 wms_path : `str` or `os.PathLike`
1854 Path that includes node status file for a run.
1856 Returns
1857 -------
1858 jobs : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1859 DAG summary information compiled from the node status file combined
1860 with the information found in the node event log.
1862 Currently, if the same job attribute is found in both files, its value
1863 from the event log takes precedence over the value from the node status
1864 file.
1865 """
1866 jobs: dict[str, dict[str, Any]] = {}
1867 init_fake_id = -1
1869 # subdags may not have run so wouldn't have node_status file
1870 # use dag files and let read_single_node_status handle missing
1871 # node_status file.
1872 for dag_filename in Path(wms_path).glob("*.dag"):
1873 filename = dag_filename.with_suffix(".node_status")
1874 info = read_single_node_status(filename, init_fake_id)
1875 init_fake_id -= len(info)
1876 _update_dicts(jobs, info)
1878 for dag_filename in Path(wms_path).glob("subdags/*/*.dag"):
1879 filename = dag_filename.with_suffix(".node_status")
1880 info = read_single_node_status(filename, init_fake_id)
1881 init_fake_id -= len(info)
1882 _update_dicts(jobs, info)
1884 # Propagate pruned from subdags to jobs
1885 name_to_id: dict[str, str] = {}
1886 missing_status: dict[str, list[str]] = {}
1887 for id_, job in jobs.items():
1888 if job["DAGNodeName"].startswith("wms_"):
1889 name_to_id[job["DAGNodeName"]] = id_
1890 if "NodeStatus" not in job or job["NodeStatus"] == NodeStatus.NOT_READY:
1891 missing_status.setdefault(job["from_dag_job"], []).append(id_)
1893 for name, dag_id in name_to_id.items():
1894 dag_status = jobs[dag_id].get("NodeStatus", NodeStatus.NOT_READY)
1895 if dag_status in {NodeStatus.NOT_READY, NodeStatus.FUTILE}:
1896 for id_ in missing_status.get(name, []):
1897 jobs[id_]["NodeStatus"] = dag_status
1899 return jobs
1902def read_single_dag_log(log_filename: str | os.PathLike) -> tuple[str, dict[str, dict[str, Any]]]:
1903 """Read job information from the DAGMan log file.
1905 Parameters
1906 ----------
1907 log_filename : `str` or `os.PathLike`
1908 DAGMan log filename.
1910 Returns
1911 -------
1912 wms_workflow_id : `str`
1913 HTCondor job id (i.e., <ClusterId>.<ProcId>) of the DAGMan job.
1914 dag_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1915 HTCondor job information read from the log file mapped to HTCondor
1916 job id.
1918 Raises
1919 ------
1920 FileNotFoundError
1921 If cannot find DAGMan log in given wms_path.
1922 """
1923 wms_workflow_id = "0"
1924 dag_info: dict[str, dict[str, Any]] = {}
1926 filename = Path(log_filename)
1927 if filename.exists():
1928 _LOG.debug("dag node log filename: %s", filename)
1930 info: dict[str, Any] = {}
1931 job_event_log = htcondor.JobEventLog(str(filename))
1932 for event in job_event_log.events(stop_after=0):
1933 id_ = f"{event['Cluster']}.{event['Proc']}"
1934 if id_ not in info:
1935 info[id_] = {}
1936 wms_workflow_id = id_ # taking last job id in case of restarts
1937 _update_dicts(info[id_], event)
1938 info[id_][f"{event.type.name.lower()}_time"] = event["EventTime"]
1940 # only save latest DAG job
1941 dag_info = {wms_workflow_id: info[wms_workflow_id]}
1943 return wms_workflow_id, dag_info
1946def read_dag_log(wms_path: str | os.PathLike) -> tuple[str, dict[str, Any]]:
1947 """Read job information from the DAGMan log file.
1949 Parameters
1950 ----------
1951 wms_path : `str` or `os.PathLike`
1952 Path containing the DAGMan log file.
1954 Returns
1955 -------
1956 wms_workflow_id : `str`
1957 HTCondor job id (i.e., <ClusterId>.<ProcId>) of the DAGMan job.
1958 dag_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1959 HTCondor job information read from the log file mapped to HTCondor
1960 job id.
1962 Raises
1963 ------
1964 FileNotFoundError
1965 If cannot find DAGMan log in given wms_path.
1966 """
1967 wms_workflow_id = MISSING_ID
1968 dag_info: dict[str, dict[str, Any]] = {}
1970 path = Path(wms_path)
1971 if path.exists(): 1971 ↛ 1979line 1971 didn't jump to line 1979 because the condition on line 1971 was always true
1972 try:
1973 filename = next(path.glob("*.dag.dagman.log"))
1974 except StopIteration as exc:
1975 raise FileNotFoundError(f"DAGMan log not found in {wms_path}") from exc
1976 _LOG.debug("dag node log filename: %s", filename)
1977 wms_workflow_id, dag_info = read_single_dag_log(filename)
1979 return wms_workflow_id, dag_info
1982def read_single_dag_nodes_log(filename: str | os.PathLike) -> dict[str, dict[str, Any]]:
1983 """Read job information from the DAGMan nodes log file.
1985 Parameters
1986 ----------
1987 filename : `str` or `os.PathLike`
1988 Path containing the DAGMan nodes log file.
1990 Returns
1991 -------
1992 info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1993 HTCondor job information read from the log file mapped to HTCondor
1994 job id.
1996 Raises
1997 ------
1998 FileNotFoundError
1999 If cannot find DAGMan node log in given wms_path.
2000 """
2001 _LOG.debug("dag node log filename: %s", filename)
2002 filename = Path(filename)
2004 info: dict[str, dict[str, Any]] = {}
2005 if not filename.exists():
2006 raise FileNotFoundError(f"{filename} does not exist")
2008 try:
2009 job_event_log = htcondor.JobEventLog(str(filename))
2010 except htcondor.HTCondorIOError as ex:
2011 _LOG.error("Problem reading nodes log file (%s): %s", filename, ex)
2012 import traceback
2014 traceback.print_stack()
2015 raise
2016 for event in job_event_log.events(stop_after=0):
2017 _LOG.debug("log event type = %s, keys = %s", event["EventTypeNumber"], event.keys())
2019 try:
2020 id_ = f"{event['Cluster']}.{event['Proc']}"
2021 except KeyError:
2022 _LOG.warn(
2023 "Log event missing ids (DAGNodeName=%s, EventTime=%s, EventTypeNumber=%s)",
2024 event.get("DAGNodeName", "UNK"),
2025 event.get("EventTime", "UNK"),
2026 event.get("EventTypeNumber", "UNK"),
2027 )
2028 else:
2029 if id_ not in info:
2030 info[id_] = {}
2031 # Workaround: Please check to see if still problem in
2032 # future HTCondor versions. Sometimes get a
2033 # JobAbortedEvent for a subdag job after it already
2034 # terminated normally. Seems to happen when using job
2035 # plus subdags.
2036 if event["EventTypeNumber"] == 9 and info[id_].get("EventTypeNumber", -1) == 5: 2036 ↛ 2037line 2036 didn't jump to line 2037 because the condition on line 2036 was never true
2037 _LOG.debug("Skipping spurious JobAbortedEvent: %s", dict(event))
2038 else:
2039 _update_dicts(info[id_], event)
2040 info[id_][f"{event.type.name.lower()}_time"] = event["EventTime"]
2042 return info
2045def read_dag_nodes_log(wms_path: str | os.PathLike) -> dict[str, dict[str, Any]]:
2046 """Read job information from the DAGMan nodes log file.
2048 Parameters
2049 ----------
2050 wms_path : `str` or `os.PathLike`
2051 Path containing the DAGMan nodes log file.
2053 Returns
2054 -------
2055 info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
2056 HTCondor job information read from the log file mapped to HTCondor
2057 job id.
2059 Raises
2060 ------
2061 FileNotFoundError
2062 If cannot find DAGMan node log in given wms_path.
2063 """
2064 info: dict[str, dict[str, Any]] = {}
2065 for filename in Path(wms_path).glob("*.dag.nodes.log"):
2066 _LOG.debug("dag node log filename: %s", filename)
2067 _update_dicts(info, read_single_dag_nodes_log(filename))
2069 # If submitted, the main nodes log file should exist
2070 if not info:
2071 raise FileNotFoundError(f"DAGMan node log not found in {wms_path}")
2073 # Subdags will not have dag nodes log files if they haven't
2074 # started running yet (so missing is not an error).
2075 for filename in Path(wms_path).glob("subdags/*/*.dag.nodes.log"):
2076 _LOG.debug("dag node log filename: %s", filename)
2077 _update_dicts(info, read_single_dag_nodes_log(filename))
2079 return info
2082def read_dag_info(wms_path: str | os.PathLike) -> dict[str, dict[str, Any]]:
2083 """Read custom DAGMan job information from the file.
2085 Parameters
2086 ----------
2087 wms_path : `str` or `os.PathLike`
2088 Path containing the file with the DAGMan job info.
2090 Returns
2091 -------
2092 dag_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
2093 HTCondor job information.
2095 Raises
2096 ------
2097 FileNotFoundError
2098 If cannot find DAGMan job info file in the given location.
2099 """
2100 dag_info: dict[str, dict[str, Any]] = {}
2101 try:
2102 filename = next(Path(wms_path).glob("*.info.json"))
2103 except StopIteration as exc:
2104 raise FileNotFoundError(f"File with DAGMan job information not found in {wms_path}") from exc
2105 _LOG.debug("DAGMan job information filename: %s", filename)
2106 try:
2107 with open(filename) as fh:
2108 dag_info = json.load(fh)
2109 except (OSError, PermissionError) as exc:
2110 _LOG.debug("Retrieving DAGMan job information failed: %s", exc)
2111 return dag_info
2114def write_dag_info(filename, dag_info):
2115 """Write custom job information about DAGMan job.
2117 Parameters
2118 ----------
2119 filename : `str`
2120 Name of the file where the information will be stored.
2121 dag_info : `dict` [`str` `dict` [`str`, `~typing.Any`]]
2122 Information about the DAGMan job.
2123 """
2124 schedd_name = next(iter(dag_info))
2125 dag_id = next(iter(dag_info[schedd_name]))
2126 dag_ad = dag_info[schedd_name][dag_id]
2127 ad = {"ClusterId": dag_ad["ClusterId"], "GlobalJobId": dag_ad["GlobalJobId"]}
2128 ad.update({key: val for key, val in dag_ad.items() if key.startswith("bps")})
2129 try:
2130 with open(filename, "w") as fh:
2131 info = {schedd_name: {dag_id: ad}}
2132 json.dump(info, fh)
2133 except (KeyError, OSError, PermissionError) as exc:
2134 _LOG.debug("Persisting DAGMan job information failed: %s", exc)
2137def htc_tweak_log_info(wms_path: str | Path, job: dict[str, Any]) -> None:
2138 """Massage the given job info has same structure as if came from condor_q.
2140 Parameters
2141 ----------
2142 wms_path : `str` | `os.PathLike`
2143 Path containing an HTCondor event log file.
2144 job : `dict` [ `str`, `~typing.Any` ]
2145 A mapping between HTCondor job id and job information read from
2146 the log.
2147 """
2148 _LOG.debug("htc_tweak_log_info: %s %s", wms_path, job)
2150 # Use the presence of 'MyType' key as a proxy to determine if the job ad
2151 # contains the info extracted from the event log. Exit early if it doesn't
2152 # (e.g. it is a job ad for a pruned job).
2153 if "MyType" not in job:
2154 return
2156 try:
2157 job["ClusterId"] = job["Cluster"]
2158 job["ProcId"] = job["Proc"]
2159 except KeyError as e:
2160 _LOG.error("Missing key %s in job: %s", str(e), job)
2161 raise
2162 job["Iwd"] = str(wms_path)
2163 job["Owner"] = Path(wms_path).owner()
2165 if "LogNotes" in job:
2166 m = re.match(r"DAG Node: (\S+)", job["LogNotes"])
2167 if m: 2167 ↛ 2170line 2167 didn't jump to line 2170 because the condition on line 2167 was always true
2168 job["DAGNodeName"] = m.group(1)
2170 match job["MyType"]:
2171 case "ExecuteEvent":
2172 job["JobStatus"] = htcondor.JobStatus.RUNNING
2173 case "JobTerminatedEvent" | "PostScriptTerminatedEvent":
2174 job["JobStatus"] = htcondor.JobStatus.COMPLETED
2175 case "SubmitEvent":
2176 job["JobStatus"] = htcondor.JobStatus.IDLE
2177 case "JobAbortedEvent":
2178 job["JobStatus"] = htcondor.JobStatus.REMOVED
2179 case "JobHeldEvent":
2180 job["JobStatus"] = htcondor.JobStatus.HELD
2181 case "JobReleaseEvent":
2182 # If the job managing the execution of the root DAG is held and
2183 # released this will be the last event showing up in its
2184 # job event log even if the job is still running. If this is
2185 # the last event for a job corresponding to the workflow node
2186 # (either a normal payload job or the job managing the execution
2187 # of an inner DAG), its final status will be determined later
2188 # using node status log (see _htc_status_to_wms_state()).
2189 job["JobStatus"] = htcondor.JobStatus.RUNNING if "DAGNodeName" not in job else None
2190 case _:
2191 _LOG.debug("Unknown log event type: %s", job["MyType"])
2192 job["JobStatus"] = None
2194 # Use available information to add either "ExitCode" or "ExitSignal"
2195 # attribute that captures respectively job's exit status (if it finished
2196 # on its own accord) or its exit signal (if it was terminated by
2197 # a signal). Also, include a flag "ExitBySignal" to make distinguishing
2198 # between these two cases easy later on.
2199 if job["JobStatus"] in {
2200 htcondor.JobStatus.COMPLETED,
2201 htcondor.JobStatus.HELD,
2202 htcondor.JobStatus.REMOVED,
2203 }:
2204 new_job = HTC_JOB_AD_HANDLERS.handle(job)
2205 if new_job is not None:
2206 job = new_job
2207 else:
2208 _LOG.error("Could not determine exit status for job '%s.%s'", job["ClusterId"], job["ProcId"])
2211def htc_check_dagman_output(wms_path: str | os.PathLike) -> str:
2212 """Check the DAGMan output for error messages.
2214 Parameters
2215 ----------
2216 wms_path : `str` or `os.PathLike`
2217 Directory containing the DAGman output file.
2219 Returns
2220 -------
2221 message : `str`
2222 Message containing error messages from the DAGMan output. Empty
2223 string if no messages.
2225 Raises
2226 ------
2227 FileNotFoundError
2228 If cannot find DAGMan standard output file in given wms_path.
2229 """
2230 try:
2231 filename = next(Path(wms_path).glob("*.dag.dagman.out"))
2232 except StopIteration as exc:
2233 raise FileNotFoundError(f"DAGMan standard output file not found in {wms_path}") from exc
2234 _LOG.debug("dag output filename: %s", filename)
2236 p = re.compile(r"^(\d\d/\d\d/\d\d \d\d:\d\d:\d\d) (Job submit try \d+/\d+ failed|Warning:.*$|ERROR:.*$)")
2238 message = ""
2239 try:
2240 with open(filename) as fh:
2241 last_submit_failed = "" # Since submit retries multiple times only report last one
2242 for line in fh:
2243 m = p.match(line)
2244 if m:
2245 if m.group(2).startswith("Job submit try"):
2246 last_submit_failed = m.group(1)
2247 elif m.group(2).startswith("ERROR: submit attempt failed"):
2248 pass # Should be handled by Job submit try
2249 elif m.group(2).startswith("Warning"):
2250 if ".dag.nodes.log is in /tmp" in m.group(2):
2251 last_warning = "Cannot submit from /tmp."
2252 else:
2253 last_warning = m.group(2)
2254 elif m.group(2) == "ERROR: Warning is fatal error because of DAGMAN_USE_STRICT setting":
2255 message += "ERROR: "
2256 message += last_warning
2257 message += "\n"
2258 elif m.group(2) in [ 2258 ↛ 2264line 2258 didn't jump to line 2264 because the condition on line 2258 was always true
2259 "ERROR: the following job(s) failed:",
2260 "ERROR: the following Node(s) failed:",
2261 ]:
2262 pass
2263 else:
2264 message += m.group(2)
2265 message += "\n"
2267 if last_submit_failed:
2268 message += f"Warn: Job submission issues (last: {last_submit_failed})"
2269 except (OSError, PermissionError):
2270 message = f"Warn: Could not read dagman output file from {wms_path}."
2271 _LOG.debug("dag output file message: %s", message)
2272 return message
2275def _read_rescue_headers(infh: TextIO) -> list[str]:
2276 """Read header lines from a rescue file.
2278 Parameters
2279 ----------
2280 infh : `TextIO`
2281 The rescue file from which to read the header lines.
2283 Returns
2284 -------
2285 header_lines : `list` [`str`]
2286 Header lines read from the rescue file.
2287 """
2288 header_lines = []
2289 for line in infh:
2290 line = line.strip()
2291 if not line.startswith("#"):
2292 break
2293 header_lines.append(line)
2294 return header_lines
2297def _update_rescue_headers(header_lines: list[str]) -> list[str]:
2298 """Update rescue header lines in place.
2300 Replaces ``wms_check_status`` node name prefixes with the corresponding
2301 group node names and adjusts the count of nodes premarked DONE.
2303 Parameters
2304 ----------
2305 header_lines : `list` [`str`]
2306 Header lines to update in place.
2308 Returns
2309 -------
2310 failed_subdags : `list` [`str`]
2311 Names of failed subdag jobs.
2312 """
2313 failed_subdags = []
2315 # Relies on HTCondor writing all failed node names on a single line
2316 # (see src/condor_dagman/dag.cpp in https://github.com/htcondor/htcondor).
2317 for i, line in enumerate(header_lines):
2318 if line.startswith("# Nodes that failed:"):
2319 nodes = header_lines[i + 1][1:].strip().split(",")
2320 new_nodes = []
2321 for node in nodes:
2322 if node.startswith("wms_check_status"):
2323 node = node[17:]
2324 failed_subdags.append(node)
2325 new_nodes.append(node)
2326 header_lines[i + 1] = f"# {','.join(new_nodes)}"
2327 break
2329 for i, line in enumerate(header_lines):
2330 m = re.match(r"^(# Nodes premarked DONE:)\s+(\d+)", line)
2331 if m:
2332 header_lines[i] = f"{m.group(1)} {int(m.group(2)) - len(failed_subdags)}"
2333 break
2335 return failed_subdags
2338def _write_rescue_headers(header_lines: list[str], outfh: TextIO) -> None:
2339 """Write the header lines to the new rescue file.
2341 Parameters
2342 ----------
2343 header_lines : `list` [`str`]
2344 Header lines to write to the new rescue file.
2345 outfh : `TextIO`
2346 New rescue file.
2347 """
2348 for line in header_lines:
2349 print(line, file=outfh)
2350 print("", file=outfh)
2353def _copy_done_lines(failed_subdags: list[str], infh: TextIO, outfh: TextIO) -> None:
2354 """Copy the DONE lines from the original rescue file skipping
2355 the failed group jobs.
2357 Parameters
2358 ----------
2359 failed_subdags : `list` [`str`]
2360 List of job names for the failed subdags
2361 infh : `TextIO`
2362 Original rescue file to copy from.
2363 outfh : `TextIO`
2364 New rescue file to copy to.
2365 """
2366 for line in infh:
2367 line = line.strip()
2368 try:
2369 _, node_name = line.split()
2370 except ValueError:
2371 _LOG.error(f"Unexpected line in rescue file = '{line}'")
2372 raise
2373 if node_name not in failed_subdags:
2374 print(line, file=outfh)
2377def _update_rescue_file(rescue_file: Path) -> list[str]:
2378 """Update the subdag failures in the main rescue file.
2380 Parameters
2381 ----------
2382 rescue_file : `pathlib.Path`
2383 The main rescue file that needs to be updated.
2385 Returns
2386 -------
2387 failed_subdags : `list` [`str`]
2388 Names of failed subdag jobs found in the DAGMan rescue file.
2389 """
2390 # To reduce memory requirements, not reading entire file into memory.
2391 rescue_tmp = rescue_file.with_suffix(rescue_file.suffix + ".tmp")
2392 with open(rescue_file) as infh:
2393 header_lines = _read_rescue_headers(infh)
2394 failed_subdags = _update_rescue_headers(header_lines)
2395 with open(rescue_tmp, "w") as outfh:
2396 _write_rescue_headers(header_lines, outfh)
2397 _copy_done_lines(failed_subdags, infh, outfh)
2398 rescue_file.unlink()
2399 rescue_tmp.rename(rescue_file)
2400 return failed_subdags
2403def _update_dicts(dict1, dict2):
2404 """Update dict1 with info in dict2.
2406 (Basically an update for nested dictionaries.)
2408 Parameters
2409 ----------
2410 dict1 : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
2411 HTCondor job information to be updated.
2412 dict2 : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
2413 Additional HTCondor job information.
2414 """
2415 for key, value in dict2.items():
2416 if key in dict1 and isinstance(dict1[key], dict) and isinstance(value, dict):
2417 _update_dicts(dict1[key], value)
2418 else:
2419 dict1[key] = value
2422def _locate_schedds(locate_all=False):
2423 """Find out Scheduler daemons in an HTCondor pool.
2425 Parameters
2426 ----------
2427 locate_all : `bool`, optional
2428 If True, all available schedulers in the HTCondor pool will be located.
2429 False by default which means that the search will be limited to looking
2430 for the Scheduler running on a local host.
2432 Returns
2433 -------
2434 schedds : `dict` [`str`, `htcondor.Schedd`]
2435 A mapping between Scheduler names and Python objects allowing for
2436 interacting with them.
2437 """
2438 coll = htcondor.Collector()
2440 schedd_ads = []
2441 if locate_all:
2442 schedd_ads.extend(coll.locateAll(htcondor.DaemonTypes.Schedd))
2443 else:
2444 schedd_ads.append(coll.locate(htcondor.DaemonTypes.Schedd))
2445 return {ad["Name"]: htcondor.Schedd(ad) for ad in schedd_ads}