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