Coverage for python/lsst/ctrl/bps/htcondor/lssthtc.py: 78%
938 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-16 15:08 -0700
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, subdir: str | os.PathLike | None = None, limit: int = 100
332) -> Path | None:
333 """Backup select HTCondor files in the submit directory.
335 Files will be saved in separate subdirectories which will be created in
336 the submit directory where the files are located. These subdirectories
337 will be consecutive, zero-padded integers. Their values will correspond to
338 the number of HTCondor rescue DAGs in the submit directory.
340 Hence, with the default settings, copies after the initial failed run will
341 be placed in '001' subdirectory, '002' after the first restart, and so on
342 until the limit of backups is reached. If there's no rescue DAG yet, files
343 will be copied to '000' subdirectory.
345 Parameters
346 ----------
347 wms_path : `str` or `os.PathLike`
348 Path to the submit directory either absolute or relative.
349 subdir : `str` or `os.PathLike`, optional
350 A path, relative to the submit directory, where all subdirectories with
351 backup files will be kept. Defaults to None which means that the backup
352 subdirectories will be placed directly in the submit directory.
353 limit : `int`, optional
354 Maximal number of backups. If the number of backups reaches the limit,
355 the last backup files will be overwritten. The default value is 100
356 to match the default value of HTCondor's DAGMAN_MAX_RESCUE_NUM in
357 version 8.8+.
359 Returns
360 -------
361 last_rescue_file : `pathlib.Path` or None
362 Path to the latest rescue file or None if doesn't exist.
364 Raises
365 ------
366 FileNotFoundError
367 If the submit directory or the file that needs to be backed up does not
368 exist.
369 OSError
370 If the submit directory cannot be accessed or backing up a file failed
371 either due to permission or filesystem related issues.
373 Notes
374 -----
375 This is not a generic function for making backups. It is intended to be
376 used once, just before a restart, to make snapshots of files which will be
377 overwritten by HTCondor after during the next run.
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 # also back up any subdag info
418 for subdag_dir in path.glob("subdags/*"):
419 subdag_dest = dest / subdag_dir.relative_to(path)
420 subdag_dest.mkdir(parents=True, exist_ok=False)
421 htc_backup_files_single_path(subdag_dir, subdag_dest)
423 last_rescue_file = rescue_dags[-1] if rescue_dags else None
424 _LOG.debug("last_rescue_file = %s", last_rescue_file)
425 return last_rescue_file
428def htc_backup_files_single_path(src: str | os.PathLike, dest: str | os.PathLike) -> None:
429 """Move particular htc files to a different directory for later debugging.
431 Parameters
432 ----------
433 src : `str` or `os.PathLike`
434 Directory from which to backup particular files.
435 dest : `str` or `os.PathLike`
436 Directory to which particular files are moved.
438 Raises
439 ------
440 RuntimeError
441 If given dest directory matches given src directory.
442 OSError
443 If problems moving file.
444 FileNotFoundError
445 Item matching pattern in src directory isn't a file.
446 """
447 src = Path(src)
448 dest = Path(dest)
449 if dest.samefile(src):
450 raise RuntimeError(f"Destination directory is same as the source directory ({src})")
452 for patt in [
453 "*.info.*",
454 "*.dag.metrics",
455 "*.dag.nodes.log",
456 "*.node_status",
457 "wms_*.dag.post.out",
458 "wms_*.status.txt",
459 ]:
460 for source in src.glob(patt):
461 if source.is_file(): 461 ↛ 468line 461 didn't jump to line 468 because the condition on line 461 was always true
462 target = dest / source.relative_to(src)
463 try:
464 source.rename(target)
465 except OSError as exc:
466 raise type(exc)(f"Backing up '{source}' failed: {exc.strerror}") from None
467 else:
468 raise FileNotFoundError(f"Backing up '{source}' failed: not a file")
471def htc_escape(value):
472 """Escape characters in given value based upon HTCondor syntax.
474 Parameters
475 ----------
476 value : `~typing.Any`
477 Value that needs to have characters escaped if string.
479 Returns
480 -------
481 new_value : `~typing.Any`
482 Given value with characters escaped appropriate for HTCondor if string.
483 """
484 if isinstance(value, str):
485 newval = value.replace('"', '""').replace("'", "''").replace(""", '"')
486 else:
487 newval = value
489 return newval
492def htc_write_attribs(stream, attrs):
493 """Write job attributes in HTCondor format to writeable stream.
495 Parameters
496 ----------
497 stream : `~typing.TextIO`
498 Output text stream (typically an open file).
499 attrs : `dict`
500 HTCondor job attributes (dictionary of attribute key, value).
501 """
502 for key, value in attrs.items():
503 # Make sure strings are syntactically correct for HTCondor.
504 if isinstance(value, str): 504 ↛ 507line 504 didn't jump to line 507 because the condition on line 504 was always true
505 pval = f'"{htc_escape(value)}"'
506 else:
507 pval = value
509 print(f"+{key} = {pval}", file=stream)
512def htc_write_condor_file(
513 filename: str | os.PathLike, job_name: str, job: RestrictedDict, job_attrs: dict[str, Any]
514) -> None:
515 """Write an HTCondor submit file.
517 Parameters
518 ----------
519 filename : `str` or `os.PathLike`
520 Filename for the HTCondor submit file.
521 job_name : `str`
522 Job name to use in submit file.
523 job : `RestrictedDict`
524 Submit script information.
525 job_attrs : `dict`
526 Job attributes.
527 """
528 os.makedirs(os.path.dirname(filename), exist_ok=True)
529 with open(filename, "w") as fh:
530 for key, value in job.items():
531 if value is not None: 531 ↛ 530line 531 didn't jump to line 530 because the condition on line 531 was always true
532 if key in HTC_QUOTE_KEYS: # Assumes internal quotes are already escaped correctly
533 print(f'{key}="{value}"', file=fh)
534 else:
535 print(f"{key}={value}", file=fh)
536 for key in ["output", "error", "log"]:
537 if key not in job:
538 filename = f"{job_name}.$(Cluster).{'out' if key != 'log' else key}"
539 print(f"{key}={filename}", file=fh)
541 if job_attrs is not None:
542 htc_write_attribs(fh, job_attrs)
543 print("queue", file=fh)
546# To avoid doing the version check during every function call select
547# appropriate conversion function at the import time.
548#
549# Make sure that *each* version specific variant of the conversion function(s)
550# has the same signature after applying any changes!
551if HTC_VERSION < version.parse("8.9.8"): 551 ↛ 553line 551 didn't jump to line 553 because the condition on line 551 was never true
553 def htc_tune_schedd_args(**kwargs):
554 """Ensure that arguments for Schedd are version appropriate.
556 The old arguments: 'requirements' and 'attr_list' of
557 'Schedd.history()', 'Schedd.query()', and 'Schedd.xquery()' were
558 deprecated in favor of 'constraint' and 'projection', respectively,
559 starting from version 8.9.8. The function will convert "new" keyword
560 arguments to "old" ones.
562 Parameters
563 ----------
564 **kwargs
565 Any keyword arguments that Schedd.history(), Schedd.query(), and
566 Schedd.xquery() accepts.
568 Returns
569 -------
570 kwargs : `dict` [`str`, `~typing.Any`]
571 Keywords arguments that are guaranteed to work with the Python
572 HTCondor API.
574 Notes
575 -----
576 Function doesn't validate provided keyword arguments beyond converting
577 selected arguments to their version specific form. For example,
578 it won't remove keywords that are not supported by the methods
579 mentioned earlier.
580 """
581 translation_table = {
582 "constraint": "requirements",
583 "projection": "attr_list",
584 }
585 for new, old in translation_table.items():
586 try:
587 kwargs[old] = kwargs.pop(new)
588 except KeyError:
589 pass
590 return kwargs
592else:
594 def htc_tune_schedd_args(**kwargs):
595 """Ensure that arguments for Schedd are version appropriate.
597 This is the fallback function if no version specific alteration are
598 necessary. Effectively, a no-op.
600 Parameters
601 ----------
602 **kwargs
603 Any keyword arguments that Schedd.history(), Schedd.query(), and
604 Schedd.xquery() accepts.
606 Returns
607 -------
608 kwargs : `dict` [`str`, `~typing.Any`]
609 Keywords arguments that were passed to the function.
610 """
611 return kwargs
614def htc_query_history(schedds, **kwargs):
615 """Fetch history records from the condor_schedd daemon.
617 Parameters
618 ----------
619 schedds : `htcondor.Schedd`
620 HTCondor schedulers which to query for job information.
621 **kwargs
622 Any keyword arguments that Schedd.history() accepts.
624 Yields
625 ------
626 schedd_name : `str`
627 Name of the HTCondor scheduler managing the job queue.
628 job_ad : `dict` [`str`, `~typing.Any`]
629 A dictionary representing HTCondor ClassAd describing a job. It maps
630 job attributes names to values of the ClassAd expressions they
631 represent.
632 """
633 # If not set, provide defaults for positional arguments.
634 kwargs.setdefault("constraint", None)
635 kwargs.setdefault("projection", [])
636 kwargs = htc_tune_schedd_args(**kwargs)
637 for schedd_name, schedd in schedds.items():
638 for job_ad in schedd.history(**kwargs):
639 yield schedd_name, dict(job_ad)
642def htc_query_present(schedds, **kwargs):
643 """Query the condor_schedd daemon for job ads.
645 Parameters
646 ----------
647 schedds : `htcondor.Schedd`
648 HTCondor schedulers which to query for job information.
649 **kwargs
650 Any keyword arguments that Schedd.xquery() accepts.
652 Yields
653 ------
654 schedd_name : `str`
655 Name of the HTCondor scheduler managing the job queue.
656 job_ad : `dict` [`str`, `~typing.Any`]
657 A dictionary representing HTCondor ClassAd describing a job. It maps
658 job attributes names to values of the ClassAd expressions they
659 represent.
660 """
661 kwargs = htc_tune_schedd_args(**kwargs)
662 for schedd_name, schedd in schedds.items():
663 for job_ad in schedd.query(**kwargs):
664 yield schedd_name, dict(job_ad)
667def htc_version():
668 """Return the version given by the HTCondor API.
670 Returns
671 -------
672 version : `str`
673 HTCondor version as easily comparable string.
674 """
675 return str(HTC_VERSION)
678def htc_submit_dag(sub):
679 """Submit job for execution.
681 Parameters
682 ----------
683 sub : `htcondor.Submit`
684 An object representing a job submit description.
686 Returns
687 -------
688 schedd_job_info : `dict` [`str`, `dict` [`str`, \
689 `dict` [`str`, `~typing.Any`]]]
690 Information about jobs satisfying the search criteria where for each
691 Scheduler, local HTCondor job ids are mapped to their respective
692 classads.
693 """
694 coll = htcondor.Collector()
695 schedd_ad = coll.locate(htcondor.DaemonTypes.Schedd)
696 schedd = htcondor.Schedd(schedd_ad)
698 # If Schedd.submit() fails, the method will raise an exception. Usually,
699 # that implies issues with the HTCondor pool which BPS can't address.
700 # Hence, no effort is made to handle the exception.
701 submit_result = schedd.submit(sub)
703 # Sadly, the ClassAd from Schedd.submit() (see above) does not have
704 # 'GlobalJobId' so we need to run a regular query to get it anyway.
705 schedd_name = schedd_ad["Name"]
706 schedd_dag_info = condor_q(
707 constraint=f"ClusterId == {submit_result.cluster()}", schedds={schedd_name: schedd}
708 )
709 return schedd_dag_info
712def htc_create_submit_from_dag(
713 dag_filename: str, submit_options: dict[str, Any], dagman_conf_filename: str | os.PathLike | None = None
714) -> htcondor.Submit:
715 """Create a DAGMan job submit description.
717 Parameters
718 ----------
719 dag_filename : `str`
720 Name of file containing HTCondor DAG commands.
721 submit_options : `dict` [`str`, `~typing.Any`], optional
722 Contains extra options for command line (Value of None means flag).
723 dagman_conf_filename : `str` or `os.PathLike`, optional
724 Location of DAGMan configuration file, if Any. Defaults to no
725 file (None).
727 Returns
728 -------
729 sub : `htcondor.Submit`
730 An object representing a job submit description.
732 Notes
733 -----
734 Use with HTCondor versions which support htcondor.Submit.from_dag(),
735 i.e., 8.9.3 or newer.
736 """
737 # Config and environment variables do not seem to override -MaxIdle
738 # on the .dag.condor.sub's command line (broken in some 24.0.x versions).
739 # Explicitly forward them as a submit_option if either exists.
740 # Note: auto generated subdag submit files are still the -MaxIdle=1000
741 # in the broken versions.
742 if "MaxIdle" not in submit_options:
743 max_jobs_idle: int | None = None
744 config_var_name = "DAGMAN_MAX_JOBS_IDLE"
746 if dagman_conf_filename:
747 _LOG.debug("Checking DAGMan config file = %s", dagman_conf_filename)
748 with open(dagman_conf_filename) as fh:
749 for line in fh:
750 _LOG.debug("DAGMan config file line = %s", line)
751 parts = line.split("=")
752 _LOG.debug("DAGMan config file line parts = %s", parts)
753 if len(parts) == 2 and parts[0].strip() == config_var_name:
754 max_jobs_idle = int(parts[1].strip())
755 _LOG.debug("Found %s = %s", config_var_name, max_jobs_idle)
756 break
757 if max_jobs_idle is None:
758 if f"_CONDOR_{config_var_name}" in os.environ:
759 max_jobs_idle = int(os.environ[f"_CONDOR_{config_var_name}"])
760 elif config_var_name in htcondor.param:
761 max_jobs_idle = htcondor.param[config_var_name]
763 if max_jobs_idle:
764 submit_options["MaxIdle"] = max_jobs_idle
765 else:
766 _LOG.debug("MaxIdle already in submit_options: %s", submit_options)
768 _LOG.debug("Using submit_options = %s", submit_options)
769 return htcondor.Submit.from_dag(dag_filename, submit_options)
772def htc_create_submit_from_cmd(dag_filename, submit_options=None):
773 """Create a DAGMan job submit description.
775 Create a DAGMan job submit description by calling ``condor_submit_dag``
776 on given DAG description file.
778 Parameters
779 ----------
780 dag_filename : `str`
781 Name of file containing HTCondor DAG commands.
782 submit_options : `dict` [`str`, `~typing.Any`], optional
783 Contains extra options for command line (Value of None means flag).
785 Returns
786 -------
787 sub : `htcondor.Submit`
788 An object representing a job submit description.
790 Notes
791 -----
792 Use with HTCondor versions which do not support htcondor.Submit.from_dag(),
793 i.e., older than 8.9.3.
794 """
795 # Run command line condor_submit_dag command.
796 cmd = "condor_submit_dag -f -no_submit -notification never -autorescue 1 -UseDagDir -no_recurse "
798 if submit_options is not None:
799 for opt, val in submit_options.items():
800 cmd += f" -{opt} {val or ''}"
801 cmd += f"{dag_filename}"
803 process = subprocess.Popen(
804 cmd.split(), shell=False, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, encoding="utf-8"
805 )
806 process.wait()
808 if process.returncode != 0:
809 print(f"Exit code: {process.returncode}")
810 print(process.communicate()[0])
811 raise RuntimeError("Problems running condor_submit_dag")
813 return htc_create_submit_from_file(f"{dag_filename}.condor.sub")
816def htc_create_submit_from_file(submit_file):
817 """Parse a submission file.
819 Parameters
820 ----------
821 submit_file : `str`
822 Name of the HTCondor submit file.
824 Returns
825 -------
826 sub : `htcondor.Submit`
827 An object representing a job submit description.
828 """
829 descriptors = {}
830 with open(submit_file) as fh:
831 for line in fh:
832 line = line.strip()
833 if not line.startswith("#") and not line == "queue":
834 (key, val) = re.split(r"\s*=\s*", line, maxsplit=1)
835 descriptors[key] = val
837 # Avoid UserWarning: the line 'copy_to_spool = False' was
838 # unused by Submit object. Is it a typo?
839 try:
840 del descriptors["copy_to_spool"]
841 except KeyError:
842 pass
844 return htcondor.Submit(descriptors)
847def _htc_write_job_commands(stream, name, commands, node_type="JOB"):
848 """Output the DAGMan job lines for single job in DAG.
850 Parameters
851 ----------
852 stream : `~typing.TextIO`
853 Writeable text stream (typically an opened file).
854 name : `str`
855 Job name.
856 commands : `RestrictedDict`
857 DAG commands for a job.
858 node_type : `str`, optional
859 Type of DAGMan node (JOB, FINAL, SERVICE). Defaults to "JOB".
860 """
861 # Note: optional pieces of commands include a space at the beginning.
862 # also making sure values aren't empty strings as placeholders.
863 if "pre" in commands and commands["pre"]:
864 defer = ""
865 if "defer" in commands["pre"] and commands["pre"]["defer"]:
866 defer = f" DEFER {commands['pre']['defer']['status']} {commands['pre']['defer']['time']}"
868 debug = ""
869 if "debug" in commands["pre"] and commands["pre"]["debug"]:
870 debug = f" DEBUG {commands['pre']['debug']['filename']} {commands['pre']['debug']['type']}"
872 arguments = ""
873 if "arguments" in commands["pre"] and commands["pre"]["arguments"]:
874 arguments = f" {commands['pre']['arguments']}"
876 executable = commands["pre"]["executable"]
877 print(f"SCRIPT{defer}{debug} PRE {name} {executable}{arguments}", file=stream)
879 if "post" in commands and commands["post"]:
880 defer = ""
881 if "defer" in commands["post"] and commands["post"]["defer"]:
882 defer = f" DEFER {commands['post']['defer']['status']} {commands['post']['defer']['time']}"
884 debug = ""
885 if "debug" in commands["post"] and commands["post"]["debug"]:
886 debug = f" DEBUG {commands['post']['debug']['filename']} {commands['post']['debug']['type']}"
888 arguments = ""
889 if "arguments" in commands["post"] and commands["post"]["arguments"]:
890 arguments = f" {commands['post']['arguments']}"
892 executable = commands["post"]["executable"]
893 print(f"SCRIPT{defer}{debug} POST {name} {executable}{arguments}", file=stream)
895 if "vars" in commands and commands["vars"]:
896 for key, value in commands["vars"].items():
897 print(f'VARS {name} {key}="{htc_escape(value)}"', file=stream)
899 if "pre_skip" in commands and commands["pre_skip"]:
900 print(f"PRE_SKIP {name} {commands['pre_skip']}", file=stream)
902 # FINAL node cannot have a DAGMan retry, abort-dag-on, priority, category
903 if node_type != "FINAL":
904 if "retry" in commands and commands["retry"]:
905 print(f"RETRY {name} {commands['retry']}", end="", file=stream)
906 if "retry_unless_exit" in commands and commands["retry_unless_exit"]:
907 print(f" UNLESS-EXIT {commands['retry_unless_exit']}", end="", file=stream)
908 print("", file=stream) # Since previous prints don't include new line
910 if "abort_dag_on" in commands and commands["abort_dag_on"]:
911 print(
912 f"ABORT-DAG-ON {name} {commands['abort_dag_on']['node_exit']}"
913 f" RETURN {commands['abort_dag_on']['abort_exit']}",
914 file=stream,
915 )
917 if "priority" in commands and commands["priority"]:
918 print(
919 f"PRIORITY {name} {commands['priority']}",
920 file=stream,
921 )
924class HTCJob:
925 """HTCondor job for use in building DAG.
927 Parameters
928 ----------
929 name : `str`
930 Name of the job.
931 label : `str`
932 Label that can used for grouping or lookup.
933 initcmds : `RestrictedDict`
934 Initial job commands for submit file.
935 initdagcmds : `RestrictedDict`
936 Initial commands for job inside DAG.
937 initattrs : `dict`
938 Initial dictionary of job attributes.
939 """
941 def __init__(self, name, label=None, initcmds=(), initdagcmds=(), initattrs=None):
942 self.name = name
943 self.label = label
944 self.cmds = RestrictedDict(HTC_VALID_JOB_KEYS, initcmds)
945 self.dagcmds = RestrictedDict(HTC_VALID_JOB_DAG_KEYS, initdagcmds)
946 self.attrs = initattrs
947 self.subfile = None
948 self.subdir = None
949 self.subdag = None
951 def __str__(self):
952 return self.name
954 def add_job_cmds(self, new_commands):
955 """Add commands to Job (overwrite existing).
957 Parameters
958 ----------
959 new_commands : `dict`
960 Submit file commands to be added to Job.
961 """
962 self.cmds.update(new_commands)
964 def add_dag_cmds(self, new_commands):
965 """Add DAG commands to Job (overwrite existing).
967 Parameters
968 ----------
969 new_commands : `dict`
970 DAG file commands to be added to Job.
971 """
972 self.dagcmds.update(new_commands)
974 def add_job_attrs(self, new_attrs):
975 """Add attributes to Job (overwrite existing).
977 Parameters
978 ----------
979 new_attrs : `dict`
980 Attributes to be added to Job.
981 """
982 if self.attrs is None:
983 self.attrs = {}
984 if new_attrs:
985 self.attrs.update(new_attrs)
987 def write_submit_file(self, submit_path: str | os.PathLike) -> None:
988 """Write job description to submit file.
990 Parameters
991 ----------
992 submit_path : `str` or `os.PathLike`
993 Prefix path for the submit file.
994 """
995 if not self.subfile:
996 self.subfile = f"{self.name}.sub"
998 subfile = self.subfile
999 if self.subdir: 999 ↛ 1000line 999 didn't jump to line 1000 because the condition on line 999 was never true
1000 subfile = Path(self.subdir) / subfile
1002 subfile = Path(os.path.expandvars(subfile))
1003 if not subfile.is_absolute():
1004 subfile = Path(submit_path) / subfile
1005 if not subfile.exists():
1006 htc_write_condor_file(subfile, self.name, self.cmds, self.attrs)
1008 def write_dag_commands(self, stream, dag_rel_path, command_name="JOB"):
1009 """Write DAG commands for single job to output stream.
1011 Parameters
1012 ----------
1013 stream : `~typing.TextIO`
1014 Output Stream.
1015 dag_rel_path : `str`
1016 Relative path of dag to submit directory.
1017 command_name : `str`
1018 Name of the DAG command (e.g., JOB, FINAL).
1019 """
1020 subfile = os.path.expandvars(self.subfile)
1022 # JOB NodeName SubmitDescription [DIR directory] [NOOP] [DONE]
1023 job_line = f'{command_name} {self.name} "{subfile}"'
1024 if "dir" in self.dagcmds:
1025 dir_val = self.dagcmds["dir"]
1026 if dag_rel_path: 1026 ↛ 1028line 1026 didn't jump to line 1028 because the condition on line 1026 was always true
1027 dir_val = os.path.join(dag_rel_path, dir_val)
1028 job_line += f' DIR "{dir_val}"'
1029 if self.dagcmds.get("noop", False):
1030 job_line += " NOOP"
1032 print(job_line, file=stream)
1033 if self.dagcmds:
1034 _htc_write_job_commands(stream, self.name, self.dagcmds, command_name)
1036 def dump(self, fh):
1037 """Dump job information to output stream.
1039 Parameters
1040 ----------
1041 fh : `~typing.TextIO`
1042 Output stream.
1043 """
1044 printer = pprint.PrettyPrinter(indent=4, stream=fh)
1045 printer.pprint(self.name)
1046 printer.pprint(self.cmds)
1047 printer.pprint(self.attrs)
1050class HTCDag(networkx.DiGraph):
1051 """HTCondor DAG.
1053 Parameters
1054 ----------
1055 data : `~typing.Any`
1056 Initial graph data of any format that is supported
1057 by the to_network_graph() function.
1058 name : `str`
1059 Name for DAG.
1060 """
1062 def __init__(self, data=None, name=""):
1063 super().__init__(data=data, name=name)
1065 self.graph["attr"] = {}
1066 self.graph["run_id"] = None
1067 self.graph["submit_path"] = None
1068 self.graph["final_job"] = None
1069 self.graph["service_job"] = None
1070 self.graph["submit_options"] = {}
1072 def __str__(self):
1073 """Represent basic DAG info as string.
1075 Returns
1076 -------
1077 info : `str`
1078 String containing basic DAG info.
1079 """
1080 return f"{self.graph['name']} {len(self)}"
1082 def add_attribs(self, attribs=None):
1083 """Add attributes to the DAG.
1085 Parameters
1086 ----------
1087 attribs : `dict`
1088 DAG attributes.
1089 """
1090 if attribs is not None: 1090 ↛ exitline 1090 didn't return from function 'add_attribs' because the condition on line 1090 was always true
1091 self.graph["attr"].update(attribs)
1093 def add_job(self, job, parent_names=None, child_names=None):
1094 """Add an HTCJob to the HTCDag.
1096 Parameters
1097 ----------
1098 job : `HTCJob`
1099 HTCJob to add to the HTCDag.
1100 parent_names : `~collections.abc.Iterable` [`str`], optional
1101 Names of parent jobs.
1102 child_names : `~collections.abc.Iterable` [`str`], optional
1103 Names of child jobs.
1104 """
1105 assert isinstance(job, HTCJob)
1106 _LOG.debug("Adding job %s to dag", job.name)
1108 # Add dag level attributes to each job
1109 job.add_job_attrs(self.graph["attr"])
1111 self.add_node(job.name, data=job)
1113 if parent_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(parent_names, [job.name])
1116 if child_names is not None: 1116 ↛ 1117line 1116 didn't jump to line 1117 because the condition on line 1116 was never true
1117 self.add_job_relationships(child_names, [job.name])
1119 def add_job_relationships(self, parents, children):
1120 """Add DAG edge between parents and children jobs.
1122 Parameters
1123 ----------
1124 parents : `list` [`str`]
1125 Contains parent job name(s).
1126 children : `list` [`str`]
1127 Contains children job name(s).
1128 """
1129 self.add_edges_from(itertools.product(parents, children))
1131 def add_final_job(self, job):
1132 """Add an HTCJob for the FINAL job in HTCDag.
1134 Parameters
1135 ----------
1136 job : `HTCJob`
1137 HTCJob to add to the HTCDag as a FINAL job.
1138 """
1139 # Add dag level attributes to each job
1140 job.add_job_attrs(self.graph["attr"])
1142 self.graph["final_job"] = job
1144 def add_service_job(self, job):
1145 """Add an HTCJob for the SERVICE job in HTCDag.
1147 Parameters
1148 ----------
1149 job : `HTCJob`
1150 HTCJob to add to the HTCDag as a SERVICE job.
1151 """
1152 # Add dag level attributes to each job
1153 job.add_job_attrs(self.graph["attr"])
1155 self.graph["service_job"] = job
1157 def del_job(self, job_name):
1158 """Delete the job from the DAG.
1160 Parameters
1161 ----------
1162 job_name : `str`
1163 Name of job in DAG to delete.
1164 """
1165 # Reconnect edges around node to delete
1166 parents = self.predecessors(job_name)
1167 children = self.successors(job_name)
1168 self.add_edges_from(itertools.product(parents, children))
1170 # Delete job node (which deletes its edges).
1171 self.remove_node(job_name)
1173 def write(self, submit_path, job_subdir="", dag_subdir="", dag_rel_path=""):
1174 """Write DAG to a file.
1176 Parameters
1177 ----------
1178 submit_path : `str`
1179 Prefix path for all outputs.
1180 job_subdir : `str`, optional
1181 Template for job subdir (submit_path + job_subdir).
1182 dag_subdir : `str`, optional
1183 DAG subdir (submit_path + dag_subdir).
1184 dag_rel_path : `str`, optional
1185 Prefix to job_subdir for jobs inside subdag.
1186 """
1187 self.graph["submit_path"] = submit_path
1188 self.graph["dag_filename"] = os.path.join(dag_subdir, f"{self.graph['name']}.dag")
1189 full_filename = os.path.join(submit_path, self.graph["dag_filename"])
1190 os.makedirs(os.path.dirname(full_filename), exist_ok=True)
1192 try:
1193 dagman_config_path = Path(self.graph["attr"]["bps_wms_config_path"])
1194 except KeyError:
1195 dagman_config_path = None
1196 with open(full_filename, "w") as fh:
1197 if dagman_config_path is not None:
1198 fh.write(f"CONFIG {dag_rel_path / dagman_config_path}\n")
1200 for name, nodeval in self.nodes().items():
1201 try:
1202 job = nodeval["data"]
1203 except KeyError:
1204 _LOG.error("Job %s doesn't have data (keys: %s).", name, nodeval.keys())
1205 raise
1206 if job.subdag: 1206 ↛ 1207line 1206 didn't jump to line 1207 because the condition on line 1206 was never true
1207 dag_subdir = f"subdags/{job.name}"
1208 if "dir" in job.dagcmds:
1209 subdir = job.dagcmds["dir"]
1210 else:
1211 subdir = job_subdir
1212 if dagman_config_path is not None:
1213 job.subdag.add_attribs({"bps_wms_config_path": str(dagman_config_path)})
1214 job.subdag.write(submit_path, subdir, dag_subdir, "../..")
1215 fh.write(
1216 f"SUBDAG EXTERNAL {job.name} {Path(job.subdag.graph['dag_filename']).name} "
1217 f"DIR {dag_subdir}\n"
1218 )
1219 if job.dagcmds:
1220 _htc_write_job_commands(fh, job.name, job.dagcmds)
1221 else:
1222 job.write_submit_file(submit_path)
1223 job.write_dag_commands(fh, dag_rel_path)
1225 for edge in self.edges(): 1225 ↛ 1226line 1225 didn't jump to line 1226 because the loop on line 1225 never started
1226 print(f"PARENT {edge[0]} CHILD {edge[1]}", file=fh)
1227 print(f"DOT {self.name}.dot", file=fh)
1228 print(f"NODE_STATUS_FILE {self.name}.node_status", file=fh)
1230 # Add bps attributes to dag submission
1231 for key, value in self.graph["attr"].items():
1232 print(f'SET_JOB_ATTR {key}= "{htc_escape(value)}"', file=fh)
1234 # Add special nodes if any.
1235 special_jobs = {
1236 "FINAL": self.graph["final_job"],
1237 "SERVICE": self.graph["service_job"],
1238 }
1239 for dagcmd, job in special_jobs.items():
1240 if job is not None: 1240 ↛ 1241line 1240 didn't jump to line 1241 because the condition on line 1240 was never true
1241 job.write_submit_file(submit_path)
1242 job.write_dag_commands(fh, dag_rel_path, dagcmd)
1244 def dump(self, fh):
1245 """Dump DAG info to output stream.
1247 Parameters
1248 ----------
1249 fh : `typing.IO`
1250 Where to dump DAG info as text.
1251 """
1252 for key, value in self.graph:
1253 print(f"{key}={value}", file=fh)
1254 for name, data in self.nodes().items():
1255 print(f"{name}:", file=fh)
1256 data.dump(fh)
1257 for edge in self.edges():
1258 print(f"PARENT {edge[0]} CHILD {edge[1]}", file=fh)
1259 if self.graph["final_job"]:
1260 print(f"FINAL {self.graph['final_job'].name}:", file=fh)
1261 self.graph["final_job"].dump(fh)
1263 def write_dot(self, filename):
1264 """Write a dot version of the DAG.
1266 Parameters
1267 ----------
1268 filename : `str`
1269 Name of the dot file.
1270 """
1271 pos = networkx.nx_agraph.graphviz_layout(self)
1272 networkx.draw(self, pos=pos)
1273 networkx.drawing.nx_pydot.write_dot(self, filename)
1276def condor_q(constraint=None, schedds=None, **kwargs):
1277 """Get information about the jobs in the HTCondor job queue(s).
1279 Parameters
1280 ----------
1281 constraint : `str`, optional
1282 Constraints to be passed to job query.
1283 schedds : `dict` [`str`, `htcondor.Schedd`], optional
1284 HTCondor schedulers which to query for job information. If None
1285 (default), the query will be run against local scheduler only.
1286 **kwargs : `~typing.Any`
1287 Additional keyword arguments that need to be passed to the internal
1288 query method.
1290 Returns
1291 -------
1292 job_info : `dict` [`str`, `dict` [`str`, `dict` [`str`, `~typing.Any`]]]
1293 Information about jobs satisfying the search criteria where for each
1294 Scheduler, local HTCondor job ids are mapped to their respective
1295 classads.
1296 """
1297 return condor_query(constraint, schedds, htc_query_present, **kwargs)
1300def condor_history(constraint=None, schedds=None, **kwargs):
1301 """Get information about the jobs from HTCondor history records.
1303 Parameters
1304 ----------
1305 constraint : `str`, optional
1306 Constraints to be passed to job query.
1307 schedds : `dict` [`str`, `htcondor.Schedd`], optional
1308 HTCondor schedulers which to query for job information. If None
1309 (default), the query will be run against the history file of
1310 the local scheduler only.
1311 **kwargs : `~typing.Any`
1312 Additional keyword arguments that need to be passed to the internal
1313 query method.
1315 Returns
1316 -------
1317 job_info : `dict` [`str`, `dict` [`str`, `dict` [`str`, `~typing.Any`]]]
1318 Information about jobs satisfying the search criteria where for each
1319 Scheduler, local HTCondor job ids are mapped to their respective
1320 classads.
1321 """
1322 return condor_query(constraint, schedds, htc_query_history, **kwargs)
1325def condor_query(constraint=None, schedds=None, query_func=htc_query_present, **kwargs):
1326 """Get information about HTCondor jobs.
1328 Parameters
1329 ----------
1330 constraint : `str`, optional
1331 Constraints to be passed to job query.
1332 schedds : `dict` [`str`, `htcondor.Schedd`], optional
1333 HTCondor schedulers which to query for job information. If None
1334 (default), the query will be run against the history file of
1335 the local scheduler only.
1336 query_func : `~collections.abc.Callable`
1337 An query function which takes following arguments:
1339 - ``schedds``: Schedulers to query (`list` [`htcondor.Schedd`]).
1340 - ``**kwargs``: Keyword arguments that will be passed to the query
1341 function.
1342 **kwargs : `~typing.Any`
1343 Additional keyword arguments that need to be passed to the query
1344 method.
1346 Returns
1347 -------
1348 job_info : `dict` [`str`, `dict` [`str`, `dict` [`str`, `~typing.Any`]]]
1349 Information about jobs satisfying the search criteria where for each
1350 Scheduler, local HTCondor job ids are mapped to their respective
1351 classads.
1352 """
1353 if not schedds:
1354 coll = htcondor.Collector()
1355 schedd_ad = coll.locate(htcondor.DaemonTypes.Schedd)
1356 schedds = {schedd_ad["Name"]: htcondor.Schedd(schedd_ad)}
1358 # Make sure that 'ClusterId' and 'ProcId' attributes are always included
1359 # in the job classad. They are needed to construct the job id.
1360 added_attrs = set()
1361 if "projection" in kwargs and kwargs["projection"]:
1362 requested_attrs = set(kwargs["projection"])
1363 required_attrs = {"ClusterId", "ProcId"}
1364 added_attrs = required_attrs - requested_attrs
1365 for attr in added_attrs:
1366 kwargs["projection"].append(attr)
1368 unwanted_attrs = {"Env", "Environment"} | added_attrs
1369 job_info = defaultdict(dict)
1370 for schedd_name, job_ad in query_func(schedds, constraint=constraint, **kwargs):
1371 id_ = f"{job_ad['ClusterId']}.{job_ad['ProcId']}"
1372 for attr in set(job_ad) & unwanted_attrs:
1373 del job_ad[attr]
1374 job_info[schedd_name][id_] = job_ad
1375 _LOG.debug("query returned %d jobs", sum(len(val) for val in job_info.values()))
1377 # Restore the list of the requested attributes to its original value
1378 # if needed.
1379 if added_attrs:
1380 for attr in added_attrs:
1381 kwargs["projection"].remove(attr)
1383 # When returning the results filter out entries for schedulers with no jobs
1384 # matching the search criteria.
1385 return {key: val for key, val in job_info.items() if val}
1388def condor_search(constraint=None, hist=None, schedds=None):
1389 """Search for running and finished jobs satisfying given criteria.
1391 Parameters
1392 ----------
1393 constraint : `str`, optional
1394 Constraints to be passed to job query.
1395 hist : `float`
1396 Limit history search to this many days.
1397 schedds : `dict` [`str`, `htcondor.Schedd`], optional
1398 The list of the HTCondor schedulers which to query for job information.
1399 If None (default), only the local scheduler will be queried.
1401 Returns
1402 -------
1403 job_info : `dict` [`str`, `dict` [`str`, `dict` [`str` `~typing.Any`]]]
1404 Information about jobs satisfying the search criteria where for each
1405 Scheduler, local HTCondor job ids are mapped to their respective
1406 classads.
1407 """
1408 if not schedds:
1409 coll = htcondor.Collector()
1410 schedd_ad = coll.locate(htcondor.DaemonTypes.Schedd)
1411 schedds = {schedd_ad["Name"]: htcondor.Schedd(locate_ad=schedd_ad)}
1413 job_info = condor_q(constraint=constraint, schedds=schedds)
1414 if hist is not None:
1415 _LOG.debug("Searching history going back %s days", hist)
1416 epoch = (datetime.now() - timedelta(days=hist)).timestamp()
1417 constraint += f" && (CompletionDate >= {epoch} || JobFinishedHookDone >= {epoch})"
1418 hist_info = condor_history(constraint, schedds=schedds)
1419 update_job_info(job_info, hist_info)
1420 return job_info
1423def condor_status(constraint=None, coll=None):
1424 """Get information about HTCondor pool.
1426 Parameters
1427 ----------
1428 constraint : `str`, optional
1429 Constraints to be passed to the query.
1430 coll : `htcondor.Collector`, optional
1431 Object representing HTCondor collector daemon.
1433 Returns
1434 -------
1435 pool_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1436 Mapping between HTCondor slot names and slot information (classAds).
1437 """
1438 if coll is None:
1439 coll = htcondor.Collector()
1440 try:
1441 pool_ads = coll.query(constraint=constraint)
1442 except OSError as ex:
1443 raise RuntimeError(f"Problem querying the Collector. (Constraint='{constraint}')") from ex
1445 pool_info = {}
1446 for slot in pool_ads:
1447 pool_info[slot["name"]] = dict(slot)
1448 _LOG.debug("condor_status returned %d ads", len(pool_info))
1449 return pool_info
1452def update_job_info(job_info, other_info):
1453 """Update results of a job query with results from another query.
1455 Parameters
1456 ----------
1457 job_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1458 Results of the job query that needs to be updated.
1459 other_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1460 Results of the other job query.
1462 Returns
1463 -------
1464 job_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1465 The updated results.
1466 """
1467 for schedd_name, others in other_info.items():
1468 try:
1469 jobs = job_info[schedd_name]
1470 except KeyError:
1471 job_info[schedd_name] = others
1472 else:
1473 for id_, ad in others.items():
1474 jobs.setdefault(id_, {}).update(ad)
1475 return job_info
1478def count_jobs_in_single_dag(
1479 filename: str | os.PathLike,
1480) -> tuple[Counter[str], dict[str, str], dict[str, WmsNodeType]]:
1481 """Build bps_run_summary string from dag file.
1483 Parameters
1484 ----------
1485 filename : `str`
1486 Path that includes dag file for a run.
1488 Returns
1489 -------
1490 counts : `Counter` [`str`]
1491 Semi-colon separated list of job labels and counts.
1492 (Same format as saved in dag classad).
1493 job_name_to_label : `dict` [`str`, `str`]
1494 Mapping of job names to job labels.
1495 job_name_to_type : `dict` [`str`, `lsst.ctrl.bps.htcondor.WmsNodeType`]
1496 Mapping of job names to job types
1497 (e.g., payload, final, service).
1498 """
1499 # Later code depends upon insertion order
1500 counts: Counter = Counter() # counts of payload jobs per label
1501 job_name_to_label: dict[str, str] = {}
1502 job_name_to_type: dict[str, WmsNodeType] = {}
1503 with open(filename) as fh:
1504 for line in fh:
1505 # Skip any line that contains commands irrelevant to job counting.
1506 if not line.startswith(
1507 (
1508 "JOB",
1509 "FINAL",
1510 "SERVICE",
1511 "SUBDAG EXTERNAL",
1512 )
1513 ):
1514 continue
1516 m = re.match(
1517 r"(?P<command>JOB|FINAL|SERVICE|SUBDAG EXTERNAL)\s+"
1518 r'(?P<jobname>(?P<wms>wms_)?\S+)\s+"?(?P<subfile>\S+)"?\s*'
1519 r'(DIR "?(?P<dir>[^\s"]+)"?)?\s*(?P<noop>NOOP)?',
1520 line,
1521 )
1522 if m: 1522 ↛ 1574line 1522 didn't jump to line 1574 because the condition on line 1522 was always true
1523 job_name = m.group("jobname")
1524 name_parts = job_name.split("_")
1526 label = ""
1527 if m.group("dir"):
1528 dir_match = re.search(r"jobs/([^\s/]+)", m.group("dir"))
1529 if dir_match:
1530 label = dir_match.group(1)
1531 else:
1532 _LOG.debug("Parse DAG: unparsed dir = %s", line)
1533 elif m.group("subfile"): 1533 ↛ 1540line 1533 didn't jump to line 1540 because the condition on line 1533 was always true
1534 subfile_match = re.search(r"jobs/([^\s/]+)", m.group("subfile"))
1535 if subfile_match:
1536 label = m.group("subfile").split("/")[1]
1537 else:
1538 label = pegasus_name_to_label(job_name)
1540 match m.group("command"):
1541 case "JOB":
1542 if m.group("noop"):
1543 job_type = WmsNodeType.NOOP
1544 # wms_noop_label
1545 label = name_parts[2]
1546 elif m.group("wms"):
1547 if name_parts[1] == "check": 1547 ↛ 1552line 1547 didn't jump to line 1552 because the condition on line 1547 was always true
1548 job_type = WmsNodeType.SUBDAG_CHECK
1549 # wms_check_status_wms_group_label
1550 label = name_parts[5]
1551 else:
1552 _LOG.warning(
1553 "Unexpected skipping of dag line due to unknown wms job: %s", line
1554 )
1555 else:
1556 job_type = WmsNodeType.PAYLOAD
1557 if label == "init": 1557 ↛ 1558line 1557 didn't jump to line 1558 because the condition on line 1557 was never true
1558 label = "pipetaskInit"
1559 counts[label] += 1
1560 case "FINAL":
1561 job_type = WmsNodeType.FINAL
1562 counts[label] += 1 # final counts a payload job.
1563 case "SERVICE":
1564 job_type = WmsNodeType.SERVICE
1565 case "SUBDAG EXTERNAL": 1565 ↛ 1569line 1565 didn't jump to line 1569 because the pattern on line 1565 always matched
1566 job_type = WmsNodeType.SUBDAG
1567 label = name_parts[2]
1569 job_name_to_label[job_name] = label
1570 job_name_to_type[job_name] = job_type
1571 else:
1572 # The line should, but didn't match the pattern above. Probably
1573 # problems with regex.
1574 _LOG.warning("Unexpected skipping of dag line: %s", line)
1576 return counts, job_name_to_label, job_name_to_type
1579def summarize_dag(dir_name: str) -> tuple[str, dict[str, str], dict[str, WmsNodeType]]:
1580 """Build bps_run_summary string from dag file.
1582 Parameters
1583 ----------
1584 dir_name : `str`
1585 Path that includes dag file for a run.
1587 Returns
1588 -------
1589 summary : `str`
1590 Semi-colon separated list of job labels and counts
1591 (Same format as saved in dag classad).
1592 job_name_to_label : `dict` [`str`, `str`]
1593 Mapping of job names to job labels.
1594 job_name_to_type : `dict` [`str`, `lsst.ctrl.bps.htcondor.WmsNodeType`]
1595 Mapping of job names to job types
1596 (e.g., payload, final, service).
1597 """
1598 # Later code depends upon insertion order
1599 counts: Counter[str] = Counter() # counts of payload jobs per label
1600 job_name_to_label: dict[str, str] = {}
1601 job_name_to_type: dict[str, WmsNodeType] = {}
1602 for filename in Path(dir_name).glob("*.dag"):
1603 single_counts, single_job_name_to_label, single_job_name_to_type = count_jobs_in_single_dag(filename)
1604 counts += single_counts
1605 _update_dicts(job_name_to_label, single_job_name_to_label)
1606 _update_dicts(job_name_to_type, single_job_name_to_type)
1608 for filename in Path(dir_name).glob("subdags/*/*.dag"):
1609 single_counts, single_job_name_to_label, single_job_name_to_type = count_jobs_in_single_dag(filename)
1610 counts += single_counts
1611 _update_dicts(job_name_to_label, single_job_name_to_label)
1612 _update_dicts(job_name_to_type, single_job_name_to_type)
1614 summary = ";".join([f"{name}:{counts[name]}" for name in counts])
1615 _LOG.debug("summarize_dag: %s %s %s", summary, job_name_to_label, job_name_to_type)
1616 return summary, job_name_to_label, job_name_to_type
1619def pegasus_name_to_label(name):
1620 """Convert pegasus job name to a label for the report.
1622 Parameters
1623 ----------
1624 name : `str`
1625 Name of job.
1627 Returns
1628 -------
1629 label : `str`
1630 Label for job.
1631 """
1632 label = "UNK"
1633 if name.startswith("create_dir") or name.startswith("stage_in") or name.startswith("stage_out"): 1633 ↛ 1634line 1633 didn't jump to line 1634 because the condition on line 1633 was never true
1634 label = "pegasus"
1635 else:
1636 m = re.match(r"pipetask_(\d+_)?([^_]+)", name)
1637 if m: 1637 ↛ 1638line 1637 didn't jump to line 1638 because the condition on line 1637 was never true
1638 label = m.group(2)
1639 if label == "init":
1640 label = "pipetaskInit"
1642 return label
1645def read_single_dag_status(filename: str | os.PathLike) -> dict[str, Any]:
1646 """Read the node status file for DAG summary information.
1648 Parameters
1649 ----------
1650 filename : `str` or `Path.pathlib`
1651 Node status filename.
1653 Returns
1654 -------
1655 dag_ad : `dict` [`str`, `~typing.Any`]
1656 DAG summary information.
1657 """
1658 dag_ad: dict[str, Any] = {}
1660 # While this is probably more up to date than dag classad, only read from
1661 # file if need to.
1662 try:
1663 node_stat_file = Path(filename)
1664 _LOG.debug("Reading Node Status File %s", node_stat_file)
1665 with open(node_stat_file) as infh:
1666 dag_ad = dict(classad.parseNext(infh)) # pylint: disable=E1101
1668 if not dag_ad: 1668 ↛ 1670line 1668 didn't jump to line 1670 because the condition on line 1668 was never true
1669 # Pegasus check here
1670 metrics_file = node_stat_file.with_suffix(".dag.metrics")
1671 if metrics_file.exists():
1672 with open(metrics_file) as infh:
1673 metrics = json.load(infh)
1674 dag_ad["NodesTotal"] = metrics.get("jobs", 0)
1675 dag_ad["NodesFailed"] = metrics.get("jobs_failed", 0)
1676 dag_ad["NodesDone"] = metrics.get("jobs_succeeded", 0)
1677 metrics_file = node_stat_file.with_suffix(".metrics")
1678 with open(metrics_file) as infh:
1679 metrics = json.load(infh)
1680 dag_ad["NodesTotal"] = metrics["wf_metrics"]["total_jobs"]
1681 except (OSError, PermissionError):
1682 pass
1684 _LOG.debug("read_dag_status: %s", dag_ad)
1685 return dag_ad
1688def read_dag_status(wms_path: str | os.PathLike) -> dict[str, Any]:
1689 """Read the node status file for DAG summary information.
1691 Parameters
1692 ----------
1693 wms_path : `str` or `os.PathLike`
1694 Path that includes node status file for a run.
1696 Returns
1697 -------
1698 dag_ad : `dict` [`str`, `~typing.Any`]
1699 DAG summary information, counts summed across any subdags.
1700 """
1701 dag_ads: dict[str, Any] = {}
1702 path = Path(wms_path)
1703 try:
1704 node_stat_file = next(path.glob("*.node_status"))
1705 except StopIteration as exc:
1706 raise FileNotFoundError(f"DAGMan node status not found in {wms_path}") from exc
1708 dag_ads = read_single_dag_status(node_stat_file)
1710 for node_stat_file in path.glob("subdags/*/*.node_status"):
1711 dag_ad = read_single_dag_status(node_stat_file)
1712 dag_ads["JobProcsHeld"] += dag_ad.get("JobProcsHeld", 0)
1713 dag_ads["NodesPost"] += dag_ad.get("NodesPost", 0)
1714 dag_ads["JobProcsIdle"] += dag_ad.get("JobProcsIdle", 0)
1715 dag_ads["NodesTotal"] += dag_ad.get("NodesTotal", 0)
1716 dag_ads["NodesFailed"] += dag_ad.get("NodesFailed", 0)
1717 dag_ads["NodesDone"] += dag_ad.get("NodesDone", 0)
1718 dag_ads["NodesQueued"] += dag_ad.get("NodesQueued", 0)
1719 dag_ads["NodesPre"] += dag_ad.get("NodesReady", 0)
1720 dag_ads["NodesFutile"] += dag_ad.get("NodesFutile", 0)
1721 dag_ads["NodesUnready"] += dag_ad.get("NodesUnready", 0)
1723 return dag_ads
1726def read_single_node_status(filename: str | os.PathLike, init_fake_id: int) -> dict[str, Any]:
1727 """Read entire node status file.
1729 Parameters
1730 ----------
1731 filename : `str` or `pathlib.Path`
1732 Node status filename.
1733 init_fake_id : `int`
1734 Initial fake id value.
1736 Returns
1737 -------
1738 jobs : `dict` [`str`, `~typing.Any`]
1739 DAG summary information compiled from the node status file combined
1740 with the information found in the node event log.
1742 Currently, if the same job attribute is found in both files, its value
1743 from the event log takes precedence over the value from the node status
1744 file.
1745 """
1746 filename = Path(filename)
1748 # Get jobid info from other places to fill in gaps in info from node_status
1749 _, job_name_to_label, job_name_to_type = count_jobs_in_single_dag(filename.with_suffix(".dag"))
1750 loginfo: dict[str, dict[str, Any]] = {}
1751 try:
1752 wms_workflow_id, loginfo = read_single_dag_log(filename.with_suffix(".dag.dagman.log"))
1753 loginfo = read_single_dag_nodes_log(filename.with_suffix(".dag.nodes.log"))
1754 except (OSError, PermissionError):
1755 pass
1757 job_name_to_id: dict[str, str] = {}
1758 _LOG.debug("loginfo = %s", loginfo)
1759 log_job_name_to_id: dict[str, str] = {}
1760 for job_id, job_info in loginfo.items():
1761 if "LogNotes" in job_info: 1761 ↛ 1760line 1761 didn't jump to line 1760 because the condition on line 1761 was always true
1762 m = re.match(r"DAG Node: (\S+)", job_info["LogNotes"])
1763 if m: 1763 ↛ 1760line 1763 didn't jump to line 1760 because the condition on line 1763 was always true
1764 job_name = m.group(1)
1765 log_job_name_to_id[job_name] = job_id
1766 job_info["DAGNodeName"] = job_name
1767 job_info["wms_node_type"] = job_name_to_type[job_name]
1768 job_info["bps_job_label"] = job_name_to_label[job_name]
1770 jobs = {}
1771 fake_id = init_fake_id # For nodes that do not yet have a job id, give fake one
1772 try:
1773 with open(filename) as fh:
1774 for ad in classad.parseAds(fh):
1775 match ad["Type"]:
1776 case "DagStatus":
1777 # Skip DAG summary.
1778 pass
1779 case "NodeStatus":
1780 job_name = ad["Node"]
1781 if job_name in job_name_to_label: 1781 ↛ 1783line 1781 didn't jump to line 1783 because the condition on line 1781 was always true
1782 job_label = job_name_to_label[job_name]
1783 elif "_" in job_name:
1784 job_label = job_name.split("_")[1]
1785 else:
1786 job_label = job_name
1788 job = dict(ad)
1789 if job_name in log_job_name_to_id:
1790 job_id = str(log_job_name_to_id[job_name])
1791 _update_dicts(job, loginfo[job_id])
1792 else:
1793 job_id = str(fake_id)
1794 job = dict(ad)
1795 fake_id -= 1
1796 jobs[job_id] = job
1797 job_name_to_id[job_name] = job_id
1799 # Make job info as if came from condor_q.
1800 job["ClusterId"] = int(float(job_id))
1801 job["DAGManJobID"] = wms_workflow_id
1802 job["DAGNodeName"] = job_name
1803 job["bps_job_label"] = job_label
1804 job["wms_node_type"] = job_name_to_type[job_name]
1806 case "StatusEnd": 1806 ↛ 1809line 1806 didn't jump to line 1809 because the pattern on line 1806 always matched
1807 # Skip node status file "epilog".
1808 pass
1809 case _:
1810 _LOG.debug(
1811 "Ignoring unknown classad type '%s' in the node status file '%s'",
1812 ad["Type"],
1813 filename,
1814 )
1815 except (OSError, PermissionError):
1816 pass
1818 # Check for missing jobs (e.g., submission failure or not submitted yet)
1819 # Use dag info to create job placeholders
1820 for name in set(job_name_to_label) - set(job_name_to_id):
1821 if name in log_job_name_to_id: # job was in nodes.log, but not node_status
1822 job_id = str(log_job_name_to_id[name])
1823 job = dict(loginfo[job_id])
1824 else:
1825 job_id = str(fake_id)
1826 fake_id -= 1
1827 job = {}
1828 job["NodeStatus"] = NodeStatus.NOT_READY
1830 job["ClusterId"] = int(float(job_id))
1831 job["ProcId"] = 0
1832 job["DAGManJobID"] = wms_workflow_id
1833 job["DAGNodeName"] = name
1834 job["bps_job_label"] = job_name_to_label[name]
1835 job["wms_node_type"] = job_name_to_type[name]
1836 jobs[f"{job['ClusterId']}.{job['ProcId']}"] = job
1838 for job_info in jobs.values():
1839 job_info["from_dag_job"] = f"wms_{filename.stem}"
1841 return jobs
1844def read_node_status(wms_path: str | os.PathLike) -> dict[str, dict[str, Any]]:
1845 """Read entire node status file.
1847 Parameters
1848 ----------
1849 wms_path : `str` or `os.PathLike`
1850 Path that includes node status file for a run.
1852 Returns
1853 -------
1854 jobs : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1855 DAG summary information compiled from the node status file combined
1856 with the information found in the node event log.
1858 Currently, if the same job attribute is found in both files, its value
1859 from the event log takes precedence over the value from the node status
1860 file.
1861 """
1862 jobs: dict[str, dict[str, Any]] = {}
1863 init_fake_id = -1
1865 # subdags may not have run so wouldn't have node_status file
1866 # use dag files and let read_single_node_status handle missing
1867 # node_status file.
1868 for dag_filename in Path(wms_path).glob("*.dag"):
1869 filename = dag_filename.with_suffix(".node_status")
1870 info = read_single_node_status(filename, init_fake_id)
1871 init_fake_id -= len(info)
1872 _update_dicts(jobs, info)
1874 for dag_filename in Path(wms_path).glob("subdags/*/*.dag"):
1875 filename = dag_filename.with_suffix(".node_status")
1876 info = read_single_node_status(filename, init_fake_id)
1877 init_fake_id -= len(info)
1878 _update_dicts(jobs, info)
1880 # Propagate pruned from subdags to jobs
1881 name_to_id: dict[str, str] = {}
1882 missing_status: dict[str, list[str]] = {}
1883 for id_, job in jobs.items():
1884 if job["DAGNodeName"].startswith("wms_"):
1885 name_to_id[job["DAGNodeName"]] = id_
1886 if "NodeStatus" not in job or job["NodeStatus"] == NodeStatus.NOT_READY:
1887 missing_status.setdefault(job["from_dag_job"], []).append(id_)
1889 for name, dag_id in name_to_id.items():
1890 dag_status = jobs[dag_id].get("NodeStatus", NodeStatus.NOT_READY)
1891 if dag_status in {NodeStatus.NOT_READY, NodeStatus.FUTILE}:
1892 for id_ in missing_status.get(name, []):
1893 jobs[id_]["NodeStatus"] = dag_status
1895 return jobs
1898def read_single_dag_log(log_filename: str | os.PathLike) -> tuple[str, dict[str, dict[str, Any]]]:
1899 """Read job information from the DAGMan log file.
1901 Parameters
1902 ----------
1903 log_filename : `str` or `os.PathLike`
1904 DAGMan log filename.
1906 Returns
1907 -------
1908 wms_workflow_id : `str`
1909 HTCondor job id (i.e., <ClusterId>.<ProcId>) of the DAGMan job.
1910 dag_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1911 HTCondor job information read from the log file mapped to HTCondor
1912 job id.
1914 Raises
1915 ------
1916 FileNotFoundError
1917 If cannot find DAGMan log in given wms_path.
1918 """
1919 wms_workflow_id = "0"
1920 dag_info: dict[str, dict[str, Any]] = {}
1922 filename = Path(log_filename)
1923 if filename.exists():
1924 _LOG.debug("dag node log filename: %s", filename)
1926 info: dict[str, Any] = {}
1927 job_event_log = htcondor.JobEventLog(str(filename))
1928 for event in job_event_log.events(stop_after=0):
1929 id_ = f"{event['Cluster']}.{event['Proc']}"
1930 if id_ not in info:
1931 info[id_] = {}
1932 wms_workflow_id = id_ # taking last job id in case of restarts
1933 _update_dicts(info[id_], event)
1934 info[id_][f"{event.type.name.lower()}_time"] = event["EventTime"]
1936 # only save latest DAG job
1937 dag_info = {wms_workflow_id: info[wms_workflow_id]}
1939 return wms_workflow_id, dag_info
1942def read_dag_log(wms_path: str | os.PathLike) -> tuple[str, dict[str, Any]]:
1943 """Read job information from the DAGMan log file.
1945 Parameters
1946 ----------
1947 wms_path : `str` or `os.PathLike`
1948 Path containing the DAGMan log file.
1950 Returns
1951 -------
1952 wms_workflow_id : `str`
1953 HTCondor job id (i.e., <ClusterId>.<ProcId>) of the DAGMan job.
1954 dag_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1955 HTCondor job information read from the log file mapped to HTCondor
1956 job id.
1958 Raises
1959 ------
1960 FileNotFoundError
1961 If cannot find DAGMan log in given wms_path.
1962 """
1963 wms_workflow_id = MISSING_ID
1964 dag_info: dict[str, dict[str, Any]] = {}
1966 path = Path(wms_path)
1967 if path.exists(): 1967 ↛ 1975line 1967 didn't jump to line 1975 because the condition on line 1967 was always true
1968 try:
1969 filename = next(path.glob("*.dag.dagman.log"))
1970 except StopIteration as exc:
1971 raise FileNotFoundError(f"DAGMan log not found in {wms_path}") from exc
1972 _LOG.debug("dag node log filename: %s", filename)
1973 wms_workflow_id, dag_info = read_single_dag_log(filename)
1975 return wms_workflow_id, dag_info
1978def read_single_dag_nodes_log(filename: str | os.PathLike) -> dict[str, dict[str, Any]]:
1979 """Read job information from the DAGMan nodes log file.
1981 Parameters
1982 ----------
1983 filename : `str` or `os.PathLike`
1984 Path containing the DAGMan nodes log file.
1986 Returns
1987 -------
1988 info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1989 HTCondor job information read from the log file mapped to HTCondor
1990 job id.
1992 Raises
1993 ------
1994 FileNotFoundError
1995 If cannot find DAGMan node log in given wms_path.
1996 """
1997 _LOG.debug("dag node log filename: %s", filename)
1998 filename = Path(filename)
2000 info: dict[str, dict[str, Any]] = {}
2001 if not filename.exists():
2002 raise FileNotFoundError(f"{filename} does not exist")
2004 try:
2005 job_event_log = htcondor.JobEventLog(str(filename))
2006 except htcondor.HTCondorIOError as ex:
2007 _LOG.error("Problem reading nodes log file (%s): %s", filename, ex)
2008 import traceback
2010 traceback.print_stack()
2011 raise
2012 for event in job_event_log.events(stop_after=0):
2013 _LOG.debug("log event type = %s, keys = %s", event["EventTypeNumber"], event.keys())
2015 try:
2016 id_ = f"{event['Cluster']}.{event['Proc']}"
2017 except KeyError:
2018 _LOG.warn(
2019 "Log event missing ids (DAGNodeName=%s, EventTime=%s, EventTypeNumber=%s)",
2020 event.get("DAGNodeName", "UNK"),
2021 event.get("EventTime", "UNK"),
2022 event.get("EventTypeNumber", "UNK"),
2023 )
2024 else:
2025 if id_ not in info:
2026 info[id_] = {}
2027 # Workaround: Please check to see if still problem in
2028 # future HTCondor versions. Sometimes get a
2029 # JobAbortedEvent for a subdag job after it already
2030 # terminated normally. Seems to happen when using job
2031 # plus subdags.
2032 if event["EventTypeNumber"] == 9 and info[id_].get("EventTypeNumber", -1) == 5: 2032 ↛ 2033line 2032 didn't jump to line 2033 because the condition on line 2032 was never true
2033 _LOG.debug("Skipping spurious JobAbortedEvent: %s", dict(event))
2034 else:
2035 _update_dicts(info[id_], event)
2036 info[id_][f"{event.type.name.lower()}_time"] = event["EventTime"]
2038 return info
2041def read_dag_nodes_log(wms_path: str | os.PathLike) -> dict[str, dict[str, Any]]:
2042 """Read job information from the DAGMan nodes log file.
2044 Parameters
2045 ----------
2046 wms_path : `str` or `os.PathLike`
2047 Path containing the DAGMan nodes log file.
2049 Returns
2050 -------
2051 info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
2052 HTCondor job information read from the log file mapped to HTCondor
2053 job id.
2055 Raises
2056 ------
2057 FileNotFoundError
2058 If cannot find DAGMan node log in given wms_path.
2059 """
2060 info: dict[str, dict[str, Any]] = {}
2061 for filename in Path(wms_path).glob("*.dag.nodes.log"):
2062 _LOG.debug("dag node log filename: %s", filename)
2063 _update_dicts(info, read_single_dag_nodes_log(filename))
2065 # If submitted, the main nodes log file should exist
2066 if not info:
2067 raise FileNotFoundError(f"DAGMan node log not found in {wms_path}")
2069 # Subdags will not have dag nodes log files if they haven't
2070 # started running yet (so missing is not an error).
2071 for filename in Path(wms_path).glob("subdags/*/*.dag.nodes.log"):
2072 _LOG.debug("dag node log filename: %s", filename)
2073 _update_dicts(info, read_single_dag_nodes_log(filename))
2075 return info
2078def read_dag_info(wms_path: str | os.PathLike) -> dict[str, dict[str, Any]]:
2079 """Read custom DAGMan job information from the file.
2081 Parameters
2082 ----------
2083 wms_path : `str` or `os.PathLike`
2084 Path containing the file with the DAGMan job info.
2086 Returns
2087 -------
2088 dag_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
2089 HTCondor job information.
2091 Raises
2092 ------
2093 FileNotFoundError
2094 If cannot find DAGMan job info file in the given location.
2095 """
2096 dag_info: dict[str, dict[str, Any]] = {}
2097 try:
2098 filename = next(Path(wms_path).glob("*.info.json"))
2099 except StopIteration as exc:
2100 raise FileNotFoundError(f"File with DAGMan job information not found in {wms_path}") from exc
2101 _LOG.debug("DAGMan job information filename: %s", filename)
2102 try:
2103 with open(filename) as fh:
2104 dag_info = json.load(fh)
2105 except (OSError, PermissionError) as exc:
2106 _LOG.debug("Retrieving DAGMan job information failed: %s", exc)
2107 return dag_info
2110def write_dag_info(filename, dag_info):
2111 """Write custom job information about DAGMan job.
2113 Parameters
2114 ----------
2115 filename : `str`
2116 Name of the file where the information will be stored.
2117 dag_info : `dict` [`str` `dict` [`str`, `~typing.Any`]]
2118 Information about the DAGMan job.
2119 """
2120 schedd_name = next(iter(dag_info))
2121 dag_id = next(iter(dag_info[schedd_name]))
2122 dag_ad = dag_info[schedd_name][dag_id]
2123 ad = {"ClusterId": dag_ad["ClusterId"], "GlobalJobId": dag_ad["GlobalJobId"]}
2124 ad.update({key: val for key, val in dag_ad.items() if key.startswith("bps")})
2125 try:
2126 with open(filename, "w") as fh:
2127 info = {schedd_name: {dag_id: ad}}
2128 json.dump(info, fh)
2129 except (KeyError, OSError, PermissionError) as exc:
2130 _LOG.debug("Persisting DAGMan job information failed: %s", exc)
2133def htc_tweak_log_info(wms_path: str | Path, job: dict[str, Any]) -> None:
2134 """Massage the given job info has same structure as if came from condor_q.
2136 Parameters
2137 ----------
2138 wms_path : `str` | `os.PathLike`
2139 Path containing an HTCondor event log file.
2140 job : `dict` [ `str`, `~typing.Any` ]
2141 A mapping between HTCondor job id and job information read from
2142 the log.
2143 """
2144 _LOG.debug("htc_tweak_log_info: %s %s", wms_path, job)
2146 # Use the presence of 'MyType' key as a proxy to determine if the job ad
2147 # contains the info extracted from the event log. Exit early if it doesn't
2148 # (e.g. it is a job ad for a pruned job).
2149 if "MyType" not in job:
2150 return
2152 try:
2153 job["ClusterId"] = job["Cluster"]
2154 job["ProcId"] = job["Proc"]
2155 except KeyError as e:
2156 _LOG.error("Missing key %s in job: %s", str(e), job)
2157 raise
2158 job["Iwd"] = str(wms_path)
2159 job["Owner"] = Path(wms_path).owner()
2161 if "LogNotes" in job:
2162 m = re.match(r"DAG Node: (\S+)", job["LogNotes"])
2163 if m: 2163 ↛ 2166line 2163 didn't jump to line 2166 because the condition on line 2163 was always true
2164 job["DAGNodeName"] = m.group(1)
2166 match job["MyType"]:
2167 case "ExecuteEvent":
2168 job["JobStatus"] = htcondor.JobStatus.RUNNING
2169 case "JobTerminatedEvent" | "PostScriptTerminatedEvent":
2170 job["JobStatus"] = htcondor.JobStatus.COMPLETED
2171 case "SubmitEvent":
2172 job["JobStatus"] = htcondor.JobStatus.IDLE
2173 case "JobAbortedEvent":
2174 job["JobStatus"] = htcondor.JobStatus.REMOVED
2175 case "JobHeldEvent":
2176 job["JobStatus"] = htcondor.JobStatus.HELD
2177 case "JobReleaseEvent":
2178 # If the job managing the execution of the root DAG is held and
2179 # released this will be the last event showing up in its
2180 # job event log even if the job is still running. If this is
2181 # the last event for a job corresponding to the workflow node
2182 # (either a normal payload job or the job managing the execution
2183 # of an inner DAG), its final status will be determined later
2184 # using node status log (see _htc_status_to_wms_state()).
2185 job["JobStatus"] = htcondor.JobStatus.RUNNING if "DAGNodeName" not in job else None
2186 case _:
2187 _LOG.debug("Unknown log event type: %s", job["MyType"])
2188 job["JobStatus"] = None
2190 # Use available information to add either "ExitCode" or "ExitSignal"
2191 # attribute that captures respectively job's exit status (if it finished
2192 # on its own accord) or its exit signal (if it was terminated by
2193 # a signal). Also, include a flag "ExitBySignal" to make distinguishing
2194 # between these two cases easy later on.
2195 if job["JobStatus"] in {
2196 htcondor.JobStatus.COMPLETED,
2197 htcondor.JobStatus.HELD,
2198 htcondor.JobStatus.REMOVED,
2199 }:
2200 new_job = HTC_JOB_AD_HANDLERS.handle(job)
2201 if new_job is not None:
2202 job = new_job
2203 else:
2204 _LOG.error("Could not determine exit status for job '%s.%s'", job["ClusterId"], job["ProcId"])
2207def htc_check_dagman_output(wms_path: str | os.PathLike) -> str:
2208 """Check the DAGMan output for error messages.
2210 Parameters
2211 ----------
2212 wms_path : `str` or `os.PathLike`
2213 Directory containing the DAGman output file.
2215 Returns
2216 -------
2217 message : `str`
2218 Message containing error messages from the DAGMan output. Empty
2219 string if no messages.
2221 Raises
2222 ------
2223 FileNotFoundError
2224 If cannot find DAGMan standard output file in given wms_path.
2225 """
2226 try:
2227 filename = next(Path(wms_path).glob("*.dag.dagman.out"))
2228 except StopIteration as exc:
2229 raise FileNotFoundError(f"DAGMan standard output file not found in {wms_path}") from exc
2230 _LOG.debug("dag output filename: %s", filename)
2232 p = re.compile(r"^(\d\d/\d\d/\d\d \d\d:\d\d:\d\d) (Job submit try \d+/\d+ failed|Warning:.*$|ERROR:.*$)")
2234 message = ""
2235 try:
2236 with open(filename) as fh:
2237 last_submit_failed = "" # Since submit retries multiple times only report last one
2238 for line in fh:
2239 m = p.match(line)
2240 if m:
2241 if m.group(2).startswith("Job submit try"):
2242 last_submit_failed = m.group(1)
2243 elif m.group(2).startswith("ERROR: submit attempt failed"):
2244 pass # Should be handled by Job submit try
2245 elif m.group(2).startswith("Warning"):
2246 if ".dag.nodes.log is in /tmp" in m.group(2):
2247 last_warning = "Cannot submit from /tmp."
2248 else:
2249 last_warning = m.group(2)
2250 elif m.group(2) == "ERROR: Warning is fatal error because of DAGMAN_USE_STRICT setting":
2251 message += "ERROR: "
2252 message += last_warning
2253 message += "\n"
2254 elif m.group(2) in [ 2254 ↛ 2260line 2254 didn't jump to line 2260 because the condition on line 2254 was always true
2255 "ERROR: the following job(s) failed:",
2256 "ERROR: the following Node(s) failed:",
2257 ]:
2258 pass
2259 else:
2260 message += m.group(2)
2261 message += "\n"
2263 if last_submit_failed:
2264 message += f"Warn: Job submission issues (last: {last_submit_failed})"
2265 except (OSError, PermissionError):
2266 message = f"Warn: Could not read dagman output file from {wms_path}."
2267 _LOG.debug("dag output file message: %s", message)
2268 return message
2271def _read_rescue_headers(infh: TextIO) -> tuple[list[str], list[str]]:
2272 """Read header lines from a rescue file.
2274 Parameters
2275 ----------
2276 infh : `TextIO`
2277 The rescue file from which to read the header lines.
2279 Returns
2280 -------
2281 header_lines : `list` [`str`]
2282 Header lines read from the rescue file.
2283 failed_subdags : `list` [`str`]
2284 Names of failed subdag jobs.
2285 """
2286 header_lines: list[str] = []
2287 failed = False
2288 failed_subdags: list[str] = []
2290 for line in infh: 2290 ↛ 2313line 2290 didn't jump to line 2313 because the loop on line 2290 didn't complete
2291 line = line.strip()
2292 if line.startswith("#"):
2293 if line.startswith("# Nodes that failed:"):
2294 failed = True
2295 header_lines.append(line)
2296 elif failed:
2297 orig_failed_nodes = line[1:].strip().split(",")
2298 new_failed_nodes = []
2299 for node in orig_failed_nodes:
2300 if node.startswith("wms_check_status"):
2301 group_node = node[17:]
2302 failed_subdags.append(group_node)
2303 new_failed_nodes.append(group_node)
2304 else:
2305 new_failed_nodes.append(node)
2306 header_lines.append(f"# {','.join(new_failed_nodes)}")
2307 if orig_failed_nodes[-1] == "<ENDLIST>": 2307 ↛ 2290line 2307 didn't jump to line 2290 because the condition on line 2307 was always true
2308 failed = False
2309 else:
2310 header_lines.append(line)
2311 elif line.strip() == "": # end of headers 2311 ↛ 2290line 2311 didn't jump to line 2290 because the condition on line 2311 was always true
2312 break
2313 return header_lines, failed_subdags
2316def _write_rescue_headers(header_lines: list[str], failed_subdags: list[str], outfh: TextIO) -> None:
2317 """Write the header lines to the new rescue file.
2319 Parameters
2320 ----------
2321 header_lines : `list` [`str`]
2322 Header lines to write to the new rescue file.
2323 failed_subdags : `list` [`str`]
2324 Job names of the failed subdags.
2325 outfh : `TextIO`
2326 New rescue file.
2327 """
2328 done_str = "# Nodes premarked DONE"
2329 pattern = f"^{done_str}:\\s+(\\d+)"
2330 for header_line in header_lines:
2331 m = re.match(pattern, header_line)
2332 if m:
2333 print(f"{done_str}: {int(m.group(1)) - len(failed_subdags)}", file=outfh)
2334 else:
2335 print(header_line, file=outfh)
2337 print("", file=outfh)
2340def _copy_done_lines(failed_subdags: list[str], infh: TextIO, outfh: TextIO) -> None:
2341 """Copy the DONE lines from the original rescue file skipping
2342 the failed group jobs.
2344 Parameters
2345 ----------
2346 failed_subdags : `list` [`str`]
2347 List of job names for the failed subdags
2348 infh : `TextIO`
2349 Original rescue file to copy from.
2350 outfh : `TextIO`
2351 New rescue file to copy to.
2352 """
2353 for line in infh:
2354 line = line.strip()
2355 try:
2356 _, node_name = line.split()
2357 except ValueError:
2358 _LOG.error(f"Unexpected line in rescue file = '{line}'")
2359 raise
2360 if node_name not in failed_subdags:
2361 print(line, file=outfh)
2364def _update_rescue_file(rescue_file: Path) -> None:
2365 """Update the subdag failures in the main rescue file
2366 and backup the failed subdag dirs.
2368 Parameters
2369 ----------
2370 rescue_file : `pathlib.Path`
2371 The main rescue file that needs to be updated.
2372 """
2373 # To reduce memory requirements, not reading entire file into memory.
2374 rescue_tmp = rescue_file.with_suffix(rescue_file.suffix + ".tmp")
2375 with open(rescue_file) as infh:
2376 header_lines, failed_subdags = _read_rescue_headers(infh)
2377 with open(rescue_tmp, "w") as outfh:
2378 _write_rescue_headers(header_lines, failed_subdags, outfh)
2379 _copy_done_lines(failed_subdags, infh, outfh)
2380 rescue_file.unlink()
2381 rescue_tmp.rename(rescue_file)
2382 for failed_subdag in failed_subdags:
2383 htc_backup_files(
2384 rescue_file.parent / "subdags" / failed_subdag, subdir=f"backups/subdags/{failed_subdag}"
2385 )
2388def _update_dicts(dict1, dict2):
2389 """Update dict1 with info in dict2.
2391 (Basically an update for nested dictionaries.)
2393 Parameters
2394 ----------
2395 dict1 : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
2396 HTCondor job information to be updated.
2397 dict2 : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
2398 Additional HTCondor job information.
2399 """
2400 for key, value in dict2.items():
2401 if key in dict1 and isinstance(dict1[key], dict) and isinstance(value, dict):
2402 _update_dicts(dict1[key], value)
2403 else:
2404 dict1[key] = value
2407def _locate_schedds(locate_all=False):
2408 """Find out Scheduler daemons in an HTCondor pool.
2410 Parameters
2411 ----------
2412 locate_all : `bool`, optional
2413 If True, all available schedulers in the HTCondor pool will be located.
2414 False by default which means that the search will be limited to looking
2415 for the Scheduler running on a local host.
2417 Returns
2418 -------
2419 schedds : `dict` [`str`, `htcondor.Schedd`]
2420 A mapping between Scheduler names and Python objects allowing for
2421 interacting with them.
2422 """
2423 coll = htcondor.Collector()
2425 schedd_ads = []
2426 if locate_all:
2427 schedd_ads.extend(coll.locateAll(htcondor.DaemonTypes.Schedd))
2428 else:
2429 schedd_ads.append(coll.locate(htcondor.DaemonTypes.Schedd))
2430 return {ad["Name"]: htcondor.Schedd(ad) for ad in schedd_ads}