Coverage for python / lsst / ctrl / bps / htcondor / lssthtc.py: 11%
923 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-31 01:12 -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 "priority",
238}
239HTC_VERSION = version.parse(htcondor.__version__)
242class RestrictedDict(MutableMapping):
243 """A dictionary that only allows certain keys.
245 Parameters
246 ----------
247 valid_keys : `Container`
248 Strings that are valid keys.
249 init_data : `dict` or `RestrictedDict`, optional
250 Initial data.
252 Raises
253 ------
254 KeyError
255 If invalid key(s) in init_data.
256 """
258 def __init__(self, valid_keys, init_data=()):
259 self.valid_keys = valid_keys
260 self.data = {}
261 self.update(init_data)
263 def __getitem__(self, key):
264 """Return value for given key if exists.
266 Parameters
267 ----------
268 key : `str`
269 Identifier for value to return.
271 Returns
272 -------
273 value : `~typing.Any`
274 Value associated with given key.
276 Raises
277 ------
278 KeyError
279 If key doesn't exist.
280 """
281 return self.data[key]
283 def __delitem__(self, key):
284 """Delete value for given key if exists.
286 Parameters
287 ----------
288 key : `str`
289 Identifier for value to delete.
291 Raises
292 ------
293 KeyError
294 If key doesn't exist.
295 """
296 del self.data[key]
298 def __setitem__(self, key, value):
299 """Store key,value in internal dict only if key is valid.
301 Parameters
302 ----------
303 key : `str`
304 Identifier to associate with given value.
305 value : `~typing.Any`
306 Value to store.
308 Raises
309 ------
310 KeyError
311 If key is invalid.
312 """
313 if key not in self.valid_keys:
314 raise KeyError(f"Invalid key {key}")
315 self.data[key] = value
317 def __iter__(self):
318 return self.data.__iter__()
320 def __len__(self):
321 return len(self.data)
323 def __str__(self):
324 return str(self.data)
327def htc_backup_files(
328 wms_path: str | os.PathLike, subdir: str | os.PathLike | None = None, limit: int = 100
329) -> Path | None:
330 """Backup select HTCondor files in the submit directory.
332 Files will be saved in separate subdirectories which will be created in
333 the submit directory where the files are located. These subdirectories
334 will be consecutive, zero-padded integers. Their values will correspond to
335 the number of HTCondor rescue DAGs in the submit directory.
337 Hence, with the default settings, copies after the initial failed run will
338 be placed in '001' subdirectory, '002' after the first restart, and so on
339 until the limit of backups is reached. If there's no rescue DAG yet, files
340 will be copied to '000' subdirectory.
342 Parameters
343 ----------
344 wms_path : `str` or `os.PathLike`
345 Path to the submit directory either absolute or relative.
346 subdir : `str` or `os.PathLike`, optional
347 A path, relative to the submit directory, where all subdirectories with
348 backup files will be kept. Defaults to None which means that the backup
349 subdirectories will be placed directly in the submit directory.
350 limit : `int`, optional
351 Maximal number of backups. If the number of backups reaches the limit,
352 the last backup files will be overwritten. The default value is 100
353 to match the default value of HTCondor's DAGMAN_MAX_RESCUE_NUM in
354 version 8.8+.
356 Returns
357 -------
358 last_rescue_file : `pathlib.Path` or None
359 Path to the latest rescue file or None if doesn't exist.
361 Raises
362 ------
363 FileNotFoundError
364 If the submit directory or the file that needs to be backed up does not
365 exist.
366 OSError
367 If the submit directory cannot be accessed or backing up a file failed
368 either due to permission or filesystem related issues.
370 Notes
371 -----
372 This is not a generic function for making backups. It is intended to be
373 used once, just before a restart, to make snapshots of files which will be
374 overwritten by HTCondor after during the next run.
375 """
376 width = len(str(limit))
378 path = Path(wms_path).resolve()
379 if not path.is_dir():
380 raise FileNotFoundError(f"Directory {path} not found")
382 # Initialize the backup counter.
383 rescue_dags = list(path.glob("*.rescue[0-9][0-9][0-9]"))
384 counter = min(len(rescue_dags), limit)
386 # Create the backup directory and move select files there.
387 dest = path
388 if subdir:
389 # PurePath.is_relative_to() is not available before Python 3.9. Hence
390 # we need to check is 'subdir' is in the submit directory in some other
391 # way if it is an absolute path.
392 subdir = Path(subdir)
393 if subdir.is_absolute():
394 subdir = subdir.resolve() # Since resolve was run on path, must run it here
395 if dest not in subdir.parents:
396 _LOG.warning(
397 "Invalid backup location: '%s' not in the submit directory, will use '%s' instead.",
398 subdir,
399 wms_path,
400 )
401 else:
402 dest /= subdir
403 else:
404 dest /= subdir
405 dest /= f"{counter:0{width}}"
406 _LOG.debug("dest = %s", dest)
407 try:
408 dest.mkdir(parents=True, exist_ok=False if counter < limit else True)
409 except FileExistsError:
410 _LOG.warning("Refusing to do backups: target directory '%s' already exists", dest)
411 else:
412 htc_backup_files_single_path(path, dest)
414 # also back up any subdag info
415 for subdag_dir in path.glob("subdags/*"):
416 subdag_dest = dest / subdag_dir.relative_to(path)
417 subdag_dest.mkdir(parents=True, exist_ok=False)
418 htc_backup_files_single_path(subdag_dir, subdag_dest)
420 last_rescue_file = rescue_dags[-1] if rescue_dags else None
421 _LOG.debug("last_rescue_file = %s", last_rescue_file)
422 return last_rescue_file
425def htc_backup_files_single_path(src: str | os.PathLike, dest: str | os.PathLike) -> None:
426 """Move particular htc files to a different directory for later debugging.
428 Parameters
429 ----------
430 src : `str` or `os.PathLike`
431 Directory from which to backup particular files.
432 dest : `str` or `os.PathLike`
433 Directory to which particular files are moved.
435 Raises
436 ------
437 RuntimeError
438 If given dest directory matches given src directory.
439 OSError
440 If problems moving file.
441 FileNotFoundError
442 Item matching pattern in src directory isn't a file.
443 """
444 src = Path(src)
445 dest = Path(dest)
446 if dest.samefile(src):
447 raise RuntimeError(f"Destination directory is same as the source directory ({src})")
449 for patt in [
450 "*.info.*",
451 "*.dag.metrics",
452 "*.dag.nodes.log",
453 "*.node_status",
454 "wms_*.dag.post.out",
455 "wms_*.status.txt",
456 ]:
457 for source in src.glob(patt):
458 if source.is_file():
459 target = dest / source.relative_to(src)
460 try:
461 source.rename(target)
462 except OSError as exc:
463 raise type(exc)(f"Backing up '{source}' failed: {exc.strerror}") from None
464 else:
465 raise FileNotFoundError(f"Backing up '{source}' failed: not a file")
468def htc_escape(value):
469 """Escape characters in given value based upon HTCondor syntax.
471 Parameters
472 ----------
473 value : `~typing.Any`
474 Value that needs to have characters escaped if string.
476 Returns
477 -------
478 new_value : `~typing.Any`
479 Given value with characters escaped appropriate for HTCondor if string.
480 """
481 if isinstance(value, str):
482 newval = value.replace('"', '""').replace("'", "''").replace(""", '"')
483 else:
484 newval = value
486 return newval
489def htc_write_attribs(stream, attrs):
490 """Write job attributes in HTCondor format to writeable stream.
492 Parameters
493 ----------
494 stream : `~io.TextIOBase`
495 Output text stream (typically an open file).
496 attrs : `dict`
497 HTCondor job attributes (dictionary of attribute key, value).
498 """
499 for key, value in attrs.items():
500 # Make sure strings are syntactically correct for HTCondor.
501 if isinstance(value, str):
502 pval = f'"{htc_escape(value)}"'
503 else:
504 pval = value
506 print(f"+{key} = {pval}", file=stream)
509def htc_write_condor_file(
510 filename: str | os.PathLike, job_name: str, job: RestrictedDict, job_attrs: dict[str, Any]
511) -> None:
512 """Write an HTCondor submit file.
514 Parameters
515 ----------
516 filename : `str` or os.PathLike
517 Filename for the HTCondor submit file.
518 job_name : `str`
519 Job name to use in submit file.
520 job : `RestrictedDict`
521 Submit script information.
522 job_attrs : `dict`
523 Job attributes.
524 """
525 os.makedirs(os.path.dirname(filename), exist_ok=True)
526 with open(filename, "w") as fh:
527 for key, value in job.items():
528 if value is not None:
529 if key in HTC_QUOTE_KEYS: # Assumes internal quotes are already escaped correctly
530 print(f'{key}="{value}"', file=fh)
531 else:
532 print(f"{key}={value}", file=fh)
533 for key in ["output", "error", "log"]:
534 if key not in job:
535 filename = f"{job_name}.$(Cluster).{'out' if key != 'log' else key}"
536 print(f"{key}={filename}", file=fh)
538 if job_attrs is not None:
539 htc_write_attribs(fh, job_attrs)
540 print("queue", file=fh)
543# To avoid doing the version check during every function call select
544# appropriate conversion function at the import time.
545#
546# Make sure that *each* version specific variant of the conversion function(s)
547# has the same signature after applying any changes!
548if HTC_VERSION < version.parse("8.9.8"): 548 ↛ 550line 548 didn't jump to line 550 because the condition on line 548 was never true
550 def htc_tune_schedd_args(**kwargs):
551 """Ensure that arguments for Schedd are version appropriate.
553 The old arguments: 'requirements' and 'attr_list' of
554 'Schedd.history()', 'Schedd.query()', and 'Schedd.xquery()' were
555 deprecated in favor of 'constraint' and 'projection', respectively,
556 starting from version 8.9.8. The function will convert "new" keyword
557 arguments to "old" ones.
559 Parameters
560 ----------
561 **kwargs
562 Any keyword arguments that Schedd.history(), Schedd.query(), and
563 Schedd.xquery() accepts.
565 Returns
566 -------
567 kwargs : `dict` [`str`, `~typing.Any`]
568 Keywords arguments that are guaranteed to work with the Python
569 HTCondor API.
571 Notes
572 -----
573 Function doesn't validate provided keyword arguments beyond converting
574 selected arguments to their version specific form. For example,
575 it won't remove keywords that are not supported by the methods
576 mentioned earlier.
577 """
578 translation_table = {
579 "constraint": "requirements",
580 "projection": "attr_list",
581 }
582 for new, old in translation_table.items():
583 try:
584 kwargs[old] = kwargs.pop(new)
585 except KeyError:
586 pass
587 return kwargs
589else:
591 def htc_tune_schedd_args(**kwargs):
592 """Ensure that arguments for Schedd are version appropriate.
594 This is the fallback function if no version specific alteration are
595 necessary. Effectively, a no-op.
597 Parameters
598 ----------
599 **kwargs
600 Any keyword arguments that Schedd.history(), Schedd.query(), and
601 Schedd.xquery() accepts.
603 Returns
604 -------
605 kwargs : `dict` [`str`, `~typing.Any`]
606 Keywords arguments that were passed to the function.
607 """
608 return kwargs
611def htc_query_history(schedds, **kwargs):
612 """Fetch history records from the condor_schedd daemon.
614 Parameters
615 ----------
616 schedds : `htcondor.Schedd`
617 HTCondor schedulers which to query for job information.
618 **kwargs
619 Any keyword arguments that Schedd.history() accepts.
621 Yields
622 ------
623 schedd_name : `str`
624 Name of the HTCondor scheduler managing the job queue.
625 job_ad : `dict` [`str`, `~typing.Any`]
626 A dictionary representing HTCondor ClassAd describing a job. It maps
627 job attributes names to values of the ClassAd expressions they
628 represent.
629 """
630 # If not set, provide defaults for positional arguments.
631 kwargs.setdefault("constraint", None)
632 kwargs.setdefault("projection", [])
633 kwargs = htc_tune_schedd_args(**kwargs)
634 for schedd_name, schedd in schedds.items():
635 for job_ad in schedd.history(**kwargs):
636 yield schedd_name, dict(job_ad)
639def htc_query_present(schedds, **kwargs):
640 """Query the condor_schedd daemon for job ads.
642 Parameters
643 ----------
644 schedds : `htcondor.Schedd`
645 HTCondor schedulers which to query for job information.
646 **kwargs
647 Any keyword arguments that Schedd.xquery() accepts.
649 Yields
650 ------
651 schedd_name : `str`
652 Name of the HTCondor scheduler managing the job queue.
653 job_ad : `dict` [`str`, `~typing.Any`]
654 A dictionary representing HTCondor ClassAd describing a job. It maps
655 job attributes names to values of the ClassAd expressions they
656 represent.
657 """
658 kwargs = htc_tune_schedd_args(**kwargs)
659 for schedd_name, schedd in schedds.items():
660 for job_ad in schedd.query(**kwargs):
661 yield schedd_name, dict(job_ad)
664def htc_version():
665 """Return the version given by the HTCondor API.
667 Returns
668 -------
669 version : `str`
670 HTCondor version as easily comparable string.
671 """
672 return str(HTC_VERSION)
675def htc_submit_dag(sub):
676 """Submit job for execution.
678 Parameters
679 ----------
680 sub : `htcondor.Submit`
681 An object representing a job submit description.
683 Returns
684 -------
685 schedd_job_info : `dict` [`str`, `dict` [`str`, \
686 `dict` [`str`, `~typing.Any`]]]
687 Information about jobs satisfying the search criteria where for each
688 Scheduler, local HTCondor job ids are mapped to their respective
689 classads.
690 """
691 coll = htcondor.Collector()
692 schedd_ad = coll.locate(htcondor.DaemonTypes.Schedd)
693 schedd = htcondor.Schedd(schedd_ad)
695 # If Schedd.submit() fails, the method will raise an exception. Usually,
696 # that implies issues with the HTCondor pool which BPS can't address.
697 # Hence, no effort is made to handle the exception.
698 submit_result = schedd.submit(sub)
700 # Sadly, the ClassAd from Schedd.submit() (see above) does not have
701 # 'GlobalJobId' so we need to run a regular query to get it anyway.
702 schedd_name = schedd_ad["Name"]
703 schedd_dag_info = condor_q(
704 constraint=f"ClusterId == {submit_result.cluster()}", schedds={schedd_name: schedd}
705 )
706 return schedd_dag_info
709def htc_create_submit_from_dag(dag_filename: str, submit_options: dict[str, Any]) -> htcondor.Submit:
710 """Create a DAGMan job submit description.
712 Parameters
713 ----------
714 dag_filename : `str`
715 Name of file containing HTCondor DAG commands.
716 submit_options : `dict` [`str`, `~typing.Any`], optional
717 Contains extra options for command line (Value of None means flag).
719 Returns
720 -------
721 sub : `htcondor.Submit`
722 An object representing a job submit description.
724 Notes
725 -----
726 Use with HTCondor versions which support htcondor.Submit.from_dag(),
727 i.e., 8.9.3 or newer.
728 """
729 # Config and environment variables do not seem to override -MaxIdle
730 # on the .dag.condor.sub's command line (broken in some 24.0.x versions).
731 # Explicitly forward them as a submit_option if either exists.
732 # Note: auto generated subdag submit files are still the -MaxIdle=1000
733 # in the broken versions.
734 if "MaxIdle" not in submit_options:
735 max_jobs_idle: int | None = None
736 config_var_name = "DAGMAN_MAX_JOBS_IDLE"
737 if f"_CONDOR_{config_var_name}" in os.environ:
738 max_jobs_idle = int(os.environ[f"_CONDOR_{config_var_name}"])
739 elif config_var_name in htcondor.param:
740 max_jobs_idle = htcondor.param[config_var_name]
741 if max_jobs_idle:
742 submit_options["MaxIdle"] = max_jobs_idle
744 return htcondor.Submit.from_dag(dag_filename, submit_options)
747def htc_create_submit_from_cmd(dag_filename, submit_options=None):
748 """Create a DAGMan job submit description.
750 Create a DAGMan job submit description by calling ``condor_submit_dag``
751 on given DAG description file.
753 Parameters
754 ----------
755 dag_filename : `str`
756 Name of file containing HTCondor DAG commands.
757 submit_options : `dict` [`str`, `~typing.Any`], optional
758 Contains extra options for command line (Value of None means flag).
760 Returns
761 -------
762 sub : `htcondor.Submit`
763 An object representing a job submit description.
765 Notes
766 -----
767 Use with HTCondor versions which do not support htcondor.Submit.from_dag(),
768 i.e., older than 8.9.3.
769 """
770 # Run command line condor_submit_dag command.
771 cmd = "condor_submit_dag -f -no_submit -notification never -autorescue 1 -UseDagDir -no_recurse "
773 if submit_options is not None:
774 for opt, val in submit_options.items():
775 cmd += f" -{opt} {val or ''}"
776 cmd += f"{dag_filename}"
778 process = subprocess.Popen(
779 cmd.split(), shell=False, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, encoding="utf-8"
780 )
781 process.wait()
783 if process.returncode != 0:
784 print(f"Exit code: {process.returncode}")
785 print(process.communicate()[0])
786 raise RuntimeError("Problems running condor_submit_dag")
788 return htc_create_submit_from_file(f"{dag_filename}.condor.sub")
791def htc_create_submit_from_file(submit_file):
792 """Parse a submission file.
794 Parameters
795 ----------
796 submit_file : `str`
797 Name of the HTCondor submit file.
799 Returns
800 -------
801 sub : `htcondor.Submit`
802 An object representing a job submit description.
803 """
804 descriptors = {}
805 with open(submit_file) as fh:
806 for line in fh:
807 line = line.strip()
808 if not line.startswith("#") and not line == "queue":
809 (key, val) = re.split(r"\s*=\s*", line, maxsplit=1)
810 descriptors[key] = val
812 # Avoid UserWarning: the line 'copy_to_spool = False' was
813 # unused by Submit object. Is it a typo?
814 try:
815 del descriptors["copy_to_spool"]
816 except KeyError:
817 pass
819 return htcondor.Submit(descriptors)
822def _htc_write_job_commands(stream, name, commands):
823 """Output the DAGMan job lines for single job in DAG.
825 Parameters
826 ----------
827 stream : `~io.TextIOBase`
828 Writeable text stream (typically an opened file).
829 name : `str`
830 Job name.
831 commands : `RestrictedDict`
832 DAG commands for a job.
833 """
834 # Note: optional pieces of commands include a space at the beginning.
835 # also making sure values aren't empty strings as placeholders.
836 if "pre" in commands and commands["pre"]:
837 defer = ""
838 if "defer" in commands["pre"] and commands["pre"]["defer"]:
839 defer = f" DEFER {commands['pre']['defer']['status']} {commands['pre']['defer']['time']}"
841 debug = ""
842 if "debug" in commands["pre"] and commands["pre"]["debug"]:
843 debug = f" DEBUG {commands['pre']['debug']['filename']} {commands['pre']['debug']['type']}"
845 arguments = ""
846 if "arguments" in commands["pre"] and commands["pre"]["arguments"]:
847 arguments = f" {commands['pre']['arguments']}"
849 executable = commands["pre"]["executable"]
850 print(f"SCRIPT{defer}{debug} PRE {name} {executable}{arguments}", file=stream)
852 if "post" in commands and commands["post"]:
853 defer = ""
854 if "defer" in commands["post"] and commands["post"]["defer"]:
855 defer = f" DEFER {commands['post']['defer']['status']} {commands['post']['defer']['time']}"
857 debug = ""
858 if "debug" in commands["post"] and commands["post"]["debug"]:
859 debug = f" DEBUG {commands['post']['debug']['filename']} {commands['post']['debug']['type']}"
861 arguments = ""
862 if "arguments" in commands["post"] and commands["post"]["arguments"]:
863 arguments = f" {commands['post']['arguments']}"
865 executable = commands["post"]["executable"]
866 print(f"SCRIPT{defer}{debug} POST {name} {executable}{arguments}", file=stream)
868 if "vars" in commands and commands["vars"]:
869 for key, value in commands["vars"].items():
870 print(f'VARS {name} {key}="{htc_escape(value)}"', file=stream)
872 if "pre_skip" in commands and commands["pre_skip"]:
873 print(f"PRE_SKIP {name} {commands['pre_skip']}", file=stream)
875 if "retry" in commands and commands["retry"]:
876 print(f"RETRY {name} {commands['retry']}", end="", file=stream)
877 if "retry_unless_exit" in commands:
878 print(f" UNLESS-EXIT {commands['retry_unless_exit']}", end="", file=stream)
879 print("", file=stream) # Since previous prints don't include new line
881 if "abort_dag_on" in commands and commands["abort_dag_on"]:
882 print(
883 f"ABORT-DAG-ON {name} {commands['abort_dag_on']['node_exit']}"
884 f" RETURN {commands['abort_dag_on']['abort_exit']}",
885 file=stream,
886 )
888 if "priority" in commands and commands["priority"]:
889 print(
890 f"PRIORITY {name} {commands['priority']}",
891 file=stream,
892 )
895class HTCJob:
896 """HTCondor job for use in building DAG.
898 Parameters
899 ----------
900 name : `str`
901 Name of the job.
902 label : `str`
903 Label that can used for grouping or lookup.
904 initcmds : `RestrictedDict`
905 Initial job commands for submit file.
906 initdagcmds : `RestrictedDict`
907 Initial commands for job inside DAG.
908 initattrs : `dict`
909 Initial dictionary of job attributes.
910 """
912 def __init__(self, name, label=None, initcmds=(), initdagcmds=(), initattrs=None):
913 self.name = name
914 self.label = label
915 self.cmds = RestrictedDict(HTC_VALID_JOB_KEYS, initcmds)
916 self.dagcmds = RestrictedDict(HTC_VALID_JOB_DAG_KEYS, initdagcmds)
917 self.attrs = initattrs
918 self.subfile = None
919 self.subdir = None
920 self.subdag = None
922 def __str__(self):
923 return self.name
925 def add_job_cmds(self, new_commands):
926 """Add commands to Job (overwrite existing).
928 Parameters
929 ----------
930 new_commands : `dict`
931 Submit file commands to be added to Job.
932 """
933 self.cmds.update(new_commands)
935 def add_dag_cmds(self, new_commands):
936 """Add DAG commands to Job (overwrite existing).
938 Parameters
939 ----------
940 new_commands : `dict`
941 DAG file commands to be added to Job.
942 """
943 self.dagcmds.update(new_commands)
945 def add_job_attrs(self, new_attrs):
946 """Add attributes to Job (overwrite existing).
948 Parameters
949 ----------
950 new_attrs : `dict`
951 Attributes to be added to Job.
952 """
953 if self.attrs is None:
954 self.attrs = {}
955 if new_attrs:
956 self.attrs.update(new_attrs)
958 def write_submit_file(self, submit_path: str | os.PathLike) -> None:
959 """Write job description to submit file.
961 Parameters
962 ----------
963 submit_path : `str` or `os.PathLike`
964 Prefix path for the submit file.
965 """
966 if not self.subfile:
967 self.subfile = f"{self.name}.sub"
969 subfile = self.subfile
970 if self.subdir:
971 subfile = Path(self.subdir) / subfile
973 subfile = Path(os.path.expandvars(subfile))
974 if not subfile.is_absolute():
975 subfile = Path(submit_path) / subfile
976 if not subfile.exists():
977 htc_write_condor_file(subfile, self.name, self.cmds, self.attrs)
979 def write_dag_commands(self, stream, dag_rel_path, command_name="JOB"):
980 """Write DAG commands for single job to output stream.
982 Parameters
983 ----------
984 stream : `IO` or `str`
985 Output Stream.
986 dag_rel_path : `str`
987 Relative path of dag to submit directory.
988 command_name : `str`
989 Name of the DAG command (e.g., JOB, FINAL).
990 """
991 subfile = os.path.expandvars(self.subfile)
993 # JOB NodeName SubmitDescription [DIR directory] [NOOP] [DONE]
994 job_line = f'{command_name} {self.name} "{subfile}"'
995 if "dir" in self.dagcmds:
996 dir_val = self.dagcmds["dir"]
997 if dag_rel_path:
998 dir_val = os.path.join(dag_rel_path, dir_val)
999 job_line += f' DIR "{dir_val}"'
1000 if self.dagcmds.get("noop", False):
1001 job_line += " NOOP"
1003 print(job_line, file=stream)
1004 if self.dagcmds:
1005 _htc_write_job_commands(stream, self.name, self.dagcmds)
1007 def dump(self, fh):
1008 """Dump job information to output stream.
1010 Parameters
1011 ----------
1012 fh : `~io.TextIOBase`
1013 Output stream.
1014 """
1015 printer = pprint.PrettyPrinter(indent=4, stream=fh)
1016 printer.pprint(self.name)
1017 printer.pprint(self.cmds)
1018 printer.pprint(self.attrs)
1021class HTCDag(networkx.DiGraph):
1022 """HTCondor DAG.
1024 Parameters
1025 ----------
1026 data : networkx.DiGraph.data
1027 Initial graph.
1028 name : `str`
1029 Name for DAG.
1030 """
1032 def __init__(self, data=None, name=""):
1033 super().__init__(data=data, name=name)
1035 self.graph["attr"] = {}
1036 self.graph["run_id"] = None
1037 self.graph["submit_path"] = None
1038 self.graph["final_job"] = None
1039 self.graph["service_job"] = None
1040 self.graph["submit_options"] = {}
1042 def __str__(self):
1043 """Represent basic DAG info as string.
1045 Returns
1046 -------
1047 info : `str`
1048 String containing basic DAG info.
1049 """
1050 return f"{self.graph['name']} {len(self)}"
1052 def add_attribs(self, attribs=None):
1053 """Add attributes to the DAG.
1055 Parameters
1056 ----------
1057 attribs : `dict`
1058 DAG attributes.
1059 """
1060 if attribs is not None:
1061 self.graph["attr"].update(attribs)
1063 def add_job(self, job, parent_names=None, child_names=None):
1064 """Add an HTCJob to the HTCDag.
1066 Parameters
1067 ----------
1068 job : `HTCJob`
1069 HTCJob to add to the HTCDag.
1070 parent_names : `~collections.abc.Iterable` [`str`], optional
1071 Names of parent jobs.
1072 child_names : `~collections.abc.Iterable` [`str`], optional
1073 Names of child jobs.
1074 """
1075 assert isinstance(job, HTCJob)
1076 _LOG.debug("Adding job %s to dag", job.name)
1078 # Add dag level attributes to each job
1079 job.add_job_attrs(self.graph["attr"])
1081 self.add_node(job.name, data=job)
1083 if parent_names is not None:
1084 self.add_job_relationships(parent_names, [job.name])
1086 if child_names is not None:
1087 self.add_job_relationships(child_names, [job.name])
1089 def add_job_relationships(self, parents, children):
1090 """Add DAG edge between parents and children jobs.
1092 Parameters
1093 ----------
1094 parents : `list` [`str`]
1095 Contains parent job name(s).
1096 children : `list` [`str`]
1097 Contains children job name(s).
1098 """
1099 self.add_edges_from(itertools.product(parents, children))
1101 def add_final_job(self, job):
1102 """Add an HTCJob for the FINAL job in HTCDag.
1104 Parameters
1105 ----------
1106 job : `HTCJob`
1107 HTCJob to add to the HTCDag as a FINAL job.
1108 """
1109 # Add dag level attributes to each job
1110 job.add_job_attrs(self.graph["attr"])
1112 self.graph["final_job"] = job
1114 def add_service_job(self, job):
1115 """Add an HTCJob for the SERVICE job in HTCDag.
1117 Parameters
1118 ----------
1119 job : `HTCJob`
1120 HTCJob to add to the HTCDag as a FINAL job.
1121 """
1122 # Add dag level attributes to each job
1123 job.add_job_attrs(self.graph["attr"])
1125 self.graph["service_job"] = job
1127 def del_job(self, job_name):
1128 """Delete the job from the DAG.
1130 Parameters
1131 ----------
1132 job_name : `str`
1133 Name of job in DAG to delete.
1134 """
1135 # Reconnect edges around node to delete
1136 parents = self.predecessors(job_name)
1137 children = self.successors(job_name)
1138 self.add_edges_from(itertools.product(parents, children))
1140 # Delete job node (which deletes its edges).
1141 self.remove_node(job_name)
1143 def write(self, submit_path, job_subdir="", dag_subdir="", dag_rel_path=""):
1144 """Write DAG to a file.
1146 Parameters
1147 ----------
1148 submit_path : `str`
1149 Prefix path for all outputs.
1150 job_subdir : `str`, optional
1151 Template for job subdir (submit_path + job_subdir).
1152 dag_subdir : `str`, optional
1153 DAG subdir (submit_path + dag_subdir).
1154 dag_rel_path : `str`, optional
1155 Prefix to job_subdir for jobs inside subdag.
1156 """
1157 self.graph["submit_path"] = submit_path
1158 self.graph["dag_filename"] = os.path.join(dag_subdir, f"{self.graph['name']}.dag")
1159 full_filename = os.path.join(submit_path, self.graph["dag_filename"])
1160 os.makedirs(os.path.dirname(full_filename), exist_ok=True)
1162 try:
1163 dagman_config_path = Path(self.graph["attr"]["bps_wms_config_path"])
1164 except KeyError:
1165 dagman_config_path = None
1166 with open(full_filename, "w") as fh:
1167 if dagman_config_path is not None:
1168 fh.write(f"CONFIG {dag_rel_path / dagman_config_path}\n")
1170 for name, nodeval in self.nodes().items():
1171 try:
1172 job = nodeval["data"]
1173 except KeyError:
1174 _LOG.error("Job %s doesn't have data (keys: %s).", name, nodeval.keys())
1175 raise
1176 if job.subdag:
1177 dag_subdir = f"subdags/{job.name}"
1178 if "dir" in job.dagcmds:
1179 subdir = job.dagcmds["dir"]
1180 else:
1181 subdir = job_subdir
1182 if dagman_config_path is not None:
1183 job.subdag.add_attribs({"bps_wms_config_path": str(dagman_config_path)})
1184 job.subdag.write(submit_path, subdir, dag_subdir, "../..")
1185 fh.write(
1186 f"SUBDAG EXTERNAL {job.name} {Path(job.subdag.graph['dag_filename']).name} "
1187 f"DIR {dag_subdir}\n"
1188 )
1189 if job.dagcmds:
1190 _htc_write_job_commands(fh, job.name, job.dagcmds)
1191 else:
1192 job.write_submit_file(submit_path)
1193 job.write_dag_commands(fh, dag_rel_path)
1195 for edge in self.edges():
1196 print(f"PARENT {edge[0]} CHILD {edge[1]}", file=fh)
1197 print(f"DOT {self.name}.dot", file=fh)
1198 print(f"NODE_STATUS_FILE {self.name}.node_status", file=fh)
1200 # Add bps attributes to dag submission
1201 for key, value in self.graph["attr"].items():
1202 print(f'SET_JOB_ATTR {key}= "{htc_escape(value)}"', file=fh)
1204 # Add special nodes if any.
1205 special_jobs = {
1206 "FINAL": self.graph["final_job"],
1207 "SERVICE": self.graph["service_job"],
1208 }
1209 for dagcmd, job in special_jobs.items():
1210 if job is not None:
1211 job.write_submit_file(submit_path)
1212 job.write_dag_commands(fh, dag_rel_path, dagcmd)
1214 def dump(self, fh):
1215 """Dump DAG info to output stream.
1217 Parameters
1218 ----------
1219 fh : `io.IO` or `str`
1220 Where to dump DAG info as text.
1221 """
1222 for key, value in self.graph:
1223 print(f"{key}={value}", file=fh)
1224 for name, data in self.nodes().items():
1225 print(f"{name}:", file=fh)
1226 data.dump(fh)
1227 for edge in self.edges():
1228 print(f"PARENT {edge[0]} CHILD {edge[1]}", file=fh)
1229 if self.graph["final_job"]:
1230 print(f"FINAL {self.graph['final_job'].name}:", file=fh)
1231 self.graph["final_job"].dump(fh)
1233 def write_dot(self, filename):
1234 """Write a dot version of the DAG.
1236 Parameters
1237 ----------
1238 filename : `str`
1239 Name of the dot file.
1240 """
1241 pos = networkx.nx_agraph.graphviz_layout(self)
1242 networkx.draw(self, pos=pos)
1243 networkx.drawing.nx_pydot.write_dot(self, filename)
1246def condor_q(constraint=None, schedds=None, **kwargs):
1247 """Get information about the jobs in the HTCondor job queue(s).
1249 Parameters
1250 ----------
1251 constraint : `str`, optional
1252 Constraints to be passed to job query.
1253 schedds : `dict` [`str`, `htcondor.Schedd`], optional
1254 HTCondor schedulers which to query for job information. If None
1255 (default), the query will be run against local scheduler only.
1256 **kwargs : `~typing.Any`
1257 Additional keyword arguments that need to be passed to the internal
1258 query method.
1260 Returns
1261 -------
1262 job_info : `dict` [`str`, `dict` [`str`, `dict` [`str`, `~typing.Any`]]]
1263 Information about jobs satisfying the search criteria where for each
1264 Scheduler, local HTCondor job ids are mapped to their respective
1265 classads.
1266 """
1267 return condor_query(constraint, schedds, htc_query_present, **kwargs)
1270def condor_history(constraint=None, schedds=None, **kwargs):
1271 """Get information about the jobs from HTCondor history records.
1273 Parameters
1274 ----------
1275 constraint : `str`, optional
1276 Constraints to be passed to job query.
1277 schedds : `dict` [`str`, `htcondor.Schedd`], optional
1278 HTCondor schedulers which to query for job information. If None
1279 (default), the query will be run against the history file of
1280 the local scheduler only.
1281 **kwargs : `~typing.Any`
1282 Additional keyword arguments that need to be passed to the internal
1283 query method.
1285 Returns
1286 -------
1287 job_info : `dict` [`str`, `dict` [`str`, `dict` [`str`, `~typing.Any`]]]
1288 Information about jobs satisfying the search criteria where for each
1289 Scheduler, local HTCondor job ids are mapped to their respective
1290 classads.
1291 """
1292 return condor_query(constraint, schedds, htc_query_history, **kwargs)
1295def condor_query(constraint=None, schedds=None, query_func=htc_query_present, **kwargs):
1296 """Get information about HTCondor jobs.
1298 Parameters
1299 ----------
1300 constraint : `str`, optional
1301 Constraints to be passed to job query.
1302 schedds : `dict` [`str`, `htcondor.Schedd`], optional
1303 HTCondor schedulers which to query for job information. If None
1304 (default), the query will be run against the history file of
1305 the local scheduler only.
1306 query_func : callable
1307 An query function which takes following arguments:
1309 - ``schedds``: Schedulers to query (`list` [`htcondor.Schedd`]).
1310 - ``**kwargs``: Keyword arguments that will be passed to the query
1311 function.
1312 **kwargs : `~typing.Any`
1313 Additional keyword arguments that need to be passed to the query
1314 method.
1316 Returns
1317 -------
1318 job_info : `dict` [`str`, `dict` [`str`, `dict` [`str`, `~typing.Any`]]]
1319 Information about jobs satisfying the search criteria where for each
1320 Scheduler, local HTCondor job ids are mapped to their respective
1321 classads.
1322 """
1323 if not schedds:
1324 coll = htcondor.Collector()
1325 schedd_ad = coll.locate(htcondor.DaemonTypes.Schedd)
1326 schedds = {schedd_ad["Name"]: htcondor.Schedd(schedd_ad)}
1328 # Make sure that 'ClusterId' and 'ProcId' attributes are always included
1329 # in the job classad. They are needed to construct the job id.
1330 added_attrs = set()
1331 if "projection" in kwargs and kwargs["projection"]:
1332 requested_attrs = set(kwargs["projection"])
1333 required_attrs = {"ClusterId", "ProcId"}
1334 added_attrs = required_attrs - requested_attrs
1335 for attr in added_attrs:
1336 kwargs["projection"].append(attr)
1338 unwanted_attrs = {"Env", "Environment"} | added_attrs
1339 job_info = defaultdict(dict)
1340 for schedd_name, job_ad in query_func(schedds, constraint=constraint, **kwargs):
1341 id_ = f"{job_ad['ClusterId']}.{job_ad['ProcId']}"
1342 for attr in set(job_ad) & unwanted_attrs:
1343 del job_ad[attr]
1344 job_info[schedd_name][id_] = job_ad
1345 _LOG.debug("query returned %d jobs", sum(len(val) for val in job_info.values()))
1347 # Restore the list of the requested attributes to its original value
1348 # if needed.
1349 if added_attrs:
1350 for attr in added_attrs:
1351 kwargs["projection"].remove(attr)
1353 # When returning the results filter out entries for schedulers with no jobs
1354 # matching the search criteria.
1355 return {key: val for key, val in job_info.items() if val}
1358def condor_search(constraint=None, hist=None, schedds=None):
1359 """Search for running and finished jobs satisfying given criteria.
1361 Parameters
1362 ----------
1363 constraint : `str`, optional
1364 Constraints to be passed to job query.
1365 hist : `float`
1366 Limit history search to this many days.
1367 schedds : `dict` [`str`, `htcondor.Schedd`], optional
1368 The list of the HTCondor schedulers which to query for job information.
1369 If None (default), only the local scheduler will be queried.
1371 Returns
1372 -------
1373 job_info : `dict` [`str`, `dict` [`str`, `dict` [`str` `~typing.Any`]]]
1374 Information about jobs satisfying the search criteria where for each
1375 Scheduler, local HTCondor job ids are mapped to their respective
1376 classads.
1377 """
1378 if not schedds:
1379 coll = htcondor.Collector()
1380 schedd_ad = coll.locate(htcondor.DaemonTypes.Schedd)
1381 schedds = {schedd_ad["Name"]: htcondor.Schedd(locate_ad=schedd_ad)}
1383 job_info = condor_q(constraint=constraint, schedds=schedds)
1384 if hist is not None:
1385 _LOG.debug("Searching history going back %s days", hist)
1386 epoch = (datetime.now() - timedelta(days=hist)).timestamp()
1387 constraint += f" && (CompletionDate >= {epoch} || JobFinishedHookDone >= {epoch})"
1388 hist_info = condor_history(constraint, schedds=schedds)
1389 update_job_info(job_info, hist_info)
1390 return job_info
1393def condor_status(constraint=None, coll=None):
1394 """Get information about HTCondor pool.
1396 Parameters
1397 ----------
1398 constraint : `str`, optional
1399 Constraints to be passed to the query.
1400 coll : `htcondor.Collector`, optional
1401 Object representing HTCondor collector daemon.
1403 Returns
1404 -------
1405 pool_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1406 Mapping between HTCondor slot names and slot information (classAds).
1407 """
1408 if coll is None:
1409 coll = htcondor.Collector()
1410 try:
1411 pool_ads = coll.query(constraint=constraint)
1412 except OSError as ex:
1413 raise RuntimeError(f"Problem querying the Collector. (Constraint='{constraint}')") from ex
1415 pool_info = {}
1416 for slot in pool_ads:
1417 pool_info[slot["name"]] = dict(slot)
1418 _LOG.debug("condor_status returned %d ads", len(pool_info))
1419 return pool_info
1422def update_job_info(job_info, other_info):
1423 """Update results of a job query with results from another query.
1425 Parameters
1426 ----------
1427 job_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1428 Results of the job query that needs to be updated.
1429 other_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1430 Results of the other job query.
1432 Returns
1433 -------
1434 job_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1435 The updated results.
1436 """
1437 for schedd_name, others in other_info.items():
1438 try:
1439 jobs = job_info[schedd_name]
1440 except KeyError:
1441 job_info[schedd_name] = others
1442 else:
1443 for id_, ad in others.items():
1444 jobs.setdefault(id_, {}).update(ad)
1445 return job_info
1448def count_jobs_in_single_dag(
1449 filename: str | os.PathLike,
1450) -> tuple[Counter[str], dict[str, str], dict[str, WmsNodeType]]:
1451 """Build bps_run_summary string from dag file.
1453 Parameters
1454 ----------
1455 filename : `str`
1456 Path that includes dag file for a run.
1458 Returns
1459 -------
1460 counts : `Counter` [`str`]
1461 Semi-colon separated list of job labels and counts.
1462 (Same format as saved in dag classad).
1463 job_name_to_label : `dict` [`str`, `str`]
1464 Mapping of job names to job labels.
1465 job_name_to_type : `dict` [`str`, `lsst.ctrl.bps.htcondor.WmsNodeType`]
1466 Mapping of job names to job types
1467 (e.g., payload, final, service).
1468 """
1469 # Later code depends upon insertion order
1470 counts: Counter = Counter() # counts of payload jobs per label
1471 job_name_to_label: dict[str, str] = {}
1472 job_name_to_type: dict[str, WmsNodeType] = {}
1473 with open(filename) as fh:
1474 for line in fh:
1475 # Skip any line that contains commands irrelevant to job counting.
1476 if not line.startswith(
1477 (
1478 "JOB",
1479 "FINAL",
1480 "SERVICE",
1481 "SUBDAG EXTERNAL",
1482 )
1483 ):
1484 continue
1486 m = re.match(
1487 r"(?P<command>JOB|FINAL|SERVICE|SUBDAG EXTERNAL)\s+"
1488 r'(?P<jobname>(?P<wms>wms_)?\S+)\s+"?(?P<subfile>\S+)"?\s*'
1489 r'(DIR "?(?P<dir>[^\s"]+)"?)?\s*(?P<noop>NOOP)?',
1490 line,
1491 )
1492 if m:
1493 job_name = m.group("jobname")
1494 name_parts = job_name.split("_")
1496 label = ""
1497 if m.group("dir"):
1498 dir_match = re.search(r"jobs/([^\s/]+)", m.group("dir"))
1499 if dir_match:
1500 label = dir_match.group(1)
1501 else:
1502 _LOG.debug("Parse DAG: unparsed dir = %s", line)
1503 elif m.group("subfile"):
1504 subfile_match = re.search(r"jobs/([^\s/]+)", m.group("subfile"))
1505 if subfile_match:
1506 label = m.group("subfile").split("/")[1]
1507 else:
1508 label = pegasus_name_to_label(job_name)
1510 match m.group("command"):
1511 case "JOB":
1512 if m.group("noop"):
1513 job_type = WmsNodeType.NOOP
1514 # wms_noop_label
1515 label = name_parts[2]
1516 elif m.group("wms"):
1517 if name_parts[1] == "check":
1518 job_type = WmsNodeType.SUBDAG_CHECK
1519 # wms_check_status_wms_group_label
1520 label = name_parts[5]
1521 else:
1522 _LOG.warning(
1523 "Unexpected skipping of dag line due to unknown wms job: %s", line
1524 )
1525 else:
1526 job_type = WmsNodeType.PAYLOAD
1527 if label == "init":
1528 label = "pipetaskInit"
1529 counts[label] += 1
1530 case "FINAL":
1531 job_type = WmsNodeType.FINAL
1532 counts[label] += 1 # final counts a payload job.
1533 case "SERVICE":
1534 job_type = WmsNodeType.SERVICE
1535 case "SUBDAG EXTERNAL":
1536 job_type = WmsNodeType.SUBDAG
1537 label = name_parts[2]
1539 job_name_to_label[job_name] = label
1540 job_name_to_type[job_name] = job_type
1541 else:
1542 # The line should, but didn't match the pattern above. Probably
1543 # problems with regex.
1544 _LOG.warning("Unexpected skipping of dag line: %s", line)
1546 return counts, job_name_to_label, job_name_to_type
1549def summarize_dag(dir_name: str) -> tuple[str, dict[str, str], dict[str, WmsNodeType]]:
1550 """Build bps_run_summary string from dag file.
1552 Parameters
1553 ----------
1554 dir_name : `str`
1555 Path that includes dag file for a run.
1557 Returns
1558 -------
1559 summary : `str`
1560 Semi-colon separated list of job labels and counts
1561 (Same format as saved in dag classad).
1562 job_name_to_label : `dict` [`str`, `str`]
1563 Mapping of job names to job labels.
1564 job_name_to_type : `dict` [`str`, `lsst.ctrl.bps.htcondor.WmsNodeType`]
1565 Mapping of job names to job types
1566 (e.g., payload, final, service).
1567 """
1568 # Later code depends upon insertion order
1569 counts: Counter[str] = Counter() # counts of payload jobs per label
1570 job_name_to_label: dict[str, str] = {}
1571 job_name_to_type: dict[str, WmsNodeType] = {}
1572 for filename in Path(dir_name).glob("*.dag"):
1573 single_counts, single_job_name_to_label, single_job_name_to_type = count_jobs_in_single_dag(filename)
1574 counts += single_counts
1575 _update_dicts(job_name_to_label, single_job_name_to_label)
1576 _update_dicts(job_name_to_type, single_job_name_to_type)
1578 for filename in Path(dir_name).glob("subdags/*/*.dag"):
1579 single_counts, single_job_name_to_label, single_job_name_to_type = count_jobs_in_single_dag(filename)
1580 counts += single_counts
1581 _update_dicts(job_name_to_label, single_job_name_to_label)
1582 _update_dicts(job_name_to_type, single_job_name_to_type)
1584 summary = ";".join([f"{name}:{counts[name]}" for name in counts])
1585 _LOG.debug("summarize_dag: %s %s %s", summary, job_name_to_label, job_name_to_type)
1586 return summary, job_name_to_label, job_name_to_type
1589def pegasus_name_to_label(name):
1590 """Convert pegasus job name to a label for the report.
1592 Parameters
1593 ----------
1594 name : `str`
1595 Name of job.
1597 Returns
1598 -------
1599 label : `str`
1600 Label for job.
1601 """
1602 label = "UNK"
1603 if name.startswith("create_dir") or name.startswith("stage_in") or name.startswith("stage_out"):
1604 label = "pegasus"
1605 else:
1606 m = re.match(r"pipetask_(\d+_)?([^_]+)", name)
1607 if m:
1608 label = m.group(2)
1609 if label == "init":
1610 label = "pipetaskInit"
1612 return label
1615def read_single_dag_status(filename: str | os.PathLike) -> dict[str, Any]:
1616 """Read the node status file for DAG summary information.
1618 Parameters
1619 ----------
1620 filename : `str` or `Path.pathlib`
1621 Node status filename.
1623 Returns
1624 -------
1625 dag_ad : `dict` [`str`, `~typing.Any`]
1626 DAG summary information.
1627 """
1628 dag_ad: dict[str, Any] = {}
1630 # While this is probably more up to date than dag classad, only read from
1631 # file if need to.
1632 try:
1633 node_stat_file = Path(filename)
1634 _LOG.debug("Reading Node Status File %s", node_stat_file)
1635 with open(node_stat_file) as infh:
1636 dag_ad = dict(classad.parseNext(infh)) # pylint: disable=E1101
1638 if not dag_ad:
1639 # Pegasus check here
1640 metrics_file = node_stat_file.with_suffix(".dag.metrics")
1641 if metrics_file.exists():
1642 with open(metrics_file) as infh:
1643 metrics = json.load(infh)
1644 dag_ad["NodesTotal"] = metrics.get("jobs", 0)
1645 dag_ad["NodesFailed"] = metrics.get("jobs_failed", 0)
1646 dag_ad["NodesDone"] = metrics.get("jobs_succeeded", 0)
1647 metrics_file = node_stat_file.with_suffix(".metrics")
1648 with open(metrics_file) as infh:
1649 metrics = json.load(infh)
1650 dag_ad["NodesTotal"] = metrics["wf_metrics"]["total_jobs"]
1651 except (OSError, PermissionError):
1652 pass
1654 _LOG.debug("read_dag_status: %s", dag_ad)
1655 return dag_ad
1658def read_dag_status(wms_path: str | os.PathLike) -> dict[str, Any]:
1659 """Read the node status file for DAG summary information.
1661 Parameters
1662 ----------
1663 wms_path : `str` or `os.PathLike
1664 Path that includes node status file for a run.
1666 Returns
1667 -------
1668 dag_ad : `dict` [`str`, `~typing.Any`]
1669 DAG summary information, counts summed across any subdags.
1670 """
1671 dag_ads: dict[str, Any] = {}
1672 path = Path(wms_path)
1673 try:
1674 node_stat_file = next(path.glob("*.node_status"))
1675 except StopIteration as exc:
1676 raise FileNotFoundError(f"DAGMan node status not found in {wms_path}") from exc
1678 dag_ads = read_single_dag_status(node_stat_file)
1680 for node_stat_file in path.glob("subdags/*/*.node_status"):
1681 dag_ad = read_single_dag_status(node_stat_file)
1682 dag_ads["JobProcsHeld"] += dag_ad.get("JobProcsHeld", 0)
1683 dag_ads["NodesPost"] += dag_ad.get("NodesPost", 0)
1684 dag_ads["JobProcsIdle"] += dag_ad.get("JobProcsIdle", 0)
1685 dag_ads["NodesTotal"] += dag_ad.get("NodesTotal", 0)
1686 dag_ads["NodesFailed"] += dag_ad.get("NodesFailed", 0)
1687 dag_ads["NodesDone"] += dag_ad.get("NodesDone", 0)
1688 dag_ads["NodesQueued"] += dag_ad.get("NodesQueued", 0)
1689 dag_ads["NodesPre"] += dag_ad.get("NodesReady", 0)
1690 dag_ads["NodesFutile"] += dag_ad.get("NodesFutile", 0)
1691 dag_ads["NodesUnready"] += dag_ad.get("NodesUnready", 0)
1693 return dag_ads
1696def read_single_node_status(filename: str | os.PathLike, init_fake_id: int) -> dict[str, Any]:
1697 """Read entire node status file.
1699 Parameters
1700 ----------
1701 filename : `str` or `pathlib.Path`
1702 Node status filename.
1703 init_fake_id : `int`
1704 Initial fake id value.
1706 Returns
1707 -------
1708 jobs : `dict` [`str`, `~typing.Any`]
1709 DAG summary information compiled from the node status file combined
1710 with the information found in the node event log.
1712 Currently, if the same job attribute is found in both files, its value
1713 from the event log takes precedence over the value from the node status
1714 file.
1715 """
1716 filename = Path(filename)
1718 # Get jobid info from other places to fill in gaps in info from node_status
1719 _, job_name_to_label, job_name_to_type = count_jobs_in_single_dag(filename.with_suffix(".dag"))
1720 loginfo: dict[str, dict[str, Any]] = {}
1721 try:
1722 wms_workflow_id, loginfo = read_single_dag_log(filename.with_suffix(".dag.dagman.log"))
1723 loginfo = read_single_dag_nodes_log(filename.with_suffix(".dag.nodes.log"))
1724 except (OSError, PermissionError):
1725 pass
1727 job_name_to_id: dict[str, str] = {}
1728 _LOG.debug("loginfo = %s", loginfo)
1729 log_job_name_to_id: dict[str, str] = {}
1730 for job_id, job_info in loginfo.items():
1731 if "LogNotes" in job_info:
1732 m = re.match(r"DAG Node: (\S+)", job_info["LogNotes"])
1733 if m:
1734 job_name = m.group(1)
1735 log_job_name_to_id[job_name] = job_id
1736 job_info["DAGNodeName"] = job_name
1737 job_info["wms_node_type"] = job_name_to_type[job_name]
1738 job_info["bps_job_label"] = job_name_to_label[job_name]
1740 jobs = {}
1741 fake_id = init_fake_id # For nodes that do not yet have a job id, give fake one
1742 try:
1743 with open(filename) as fh:
1744 for ad in classad.parseAds(fh):
1745 match ad["Type"]:
1746 case "DagStatus":
1747 # Skip DAG summary.
1748 pass
1749 case "NodeStatus":
1750 job_name = ad["Node"]
1751 if job_name in job_name_to_label:
1752 job_label = job_name_to_label[job_name]
1753 elif "_" in job_name:
1754 job_label = job_name.split("_")[1]
1755 else:
1756 job_label = job_name
1758 job = dict(ad)
1759 if job_name in log_job_name_to_id:
1760 job_id = str(log_job_name_to_id[job_name])
1761 _update_dicts(job, loginfo[job_id])
1762 else:
1763 job_id = str(fake_id)
1764 job = dict(ad)
1765 fake_id -= 1
1766 jobs[job_id] = job
1767 job_name_to_id[job_name] = job_id
1769 # Make job info as if came from condor_q.
1770 job["ClusterId"] = int(float(job_id))
1771 job["DAGManJobID"] = wms_workflow_id
1772 job["DAGNodeName"] = job_name
1773 job["bps_job_label"] = job_label
1774 job["wms_node_type"] = job_name_to_type[job_name]
1776 case "StatusEnd":
1777 # Skip node status file "epilog".
1778 pass
1779 case _:
1780 _LOG.debug(
1781 "Ignoring unknown classad type '%s' in the node status file '%s'",
1782 ad["Type"],
1783 filename,
1784 )
1785 except (OSError, PermissionError):
1786 pass
1788 # Check for missing jobs (e.g., submission failure or not submitted yet)
1789 # Use dag info to create job placeholders
1790 for name in set(job_name_to_label) - set(job_name_to_id):
1791 if name in log_job_name_to_id: # job was in nodes.log, but not node_status
1792 job_id = str(log_job_name_to_id[name])
1793 job = dict(loginfo[job_id])
1794 else:
1795 job_id = str(fake_id)
1796 fake_id -= 1
1797 job = {}
1798 job["NodeStatus"] = NodeStatus.NOT_READY
1800 job["ClusterId"] = int(float(job_id))
1801 job["ProcId"] = 0
1802 job["DAGManJobID"] = wms_workflow_id
1803 job["DAGNodeName"] = name
1804 job["bps_job_label"] = job_name_to_label[name]
1805 job["wms_node_type"] = job_name_to_type[name]
1806 jobs[f"{job['ClusterId']}.{job['ProcId']}"] = job
1808 for job_info in jobs.values():
1809 job_info["from_dag_job"] = f"wms_{filename.stem}"
1811 return jobs
1814def read_node_status(wms_path: str | os.PathLike) -> dict[str, dict[str, Any]]:
1815 """Read entire node status file.
1817 Parameters
1818 ----------
1819 wms_path : `str` or `os.PathLike`
1820 Path that includes node status file for a run.
1822 Returns
1823 -------
1824 jobs : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1825 DAG summary information compiled from the node status file combined
1826 with the information found in the node event log.
1828 Currently, if the same job attribute is found in both files, its value
1829 from the event log takes precedence over the value from the node status
1830 file.
1831 """
1832 jobs: dict[str, dict[str, Any]] = {}
1833 init_fake_id = -1
1835 # subdags may not have run so wouldn't have node_status file
1836 # use dag files and let read_single_node_status handle missing
1837 # node_status file.
1838 for dag_filename in Path(wms_path).glob("*.dag"):
1839 filename = dag_filename.with_suffix(".node_status")
1840 info = read_single_node_status(filename, init_fake_id)
1841 init_fake_id -= len(info)
1842 _update_dicts(jobs, info)
1844 for dag_filename in Path(wms_path).glob("subdags/*/*.dag"):
1845 filename = dag_filename.with_suffix(".node_status")
1846 info = read_single_node_status(filename, init_fake_id)
1847 init_fake_id -= len(info)
1848 _update_dicts(jobs, info)
1850 # Propagate pruned from subdags to jobs
1851 name_to_id: dict[str, str] = {}
1852 missing_status: dict[str, list[str]] = {}
1853 for id_, job in jobs.items():
1854 if job["DAGNodeName"].startswith("wms_"):
1855 name_to_id[job["DAGNodeName"]] = id_
1856 if "NodeStatus" not in job or job["NodeStatus"] == NodeStatus.NOT_READY:
1857 missing_status.setdefault(job["from_dag_job"], []).append(id_)
1859 for name, dag_id in name_to_id.items():
1860 dag_status = jobs[dag_id].get("NodeStatus", NodeStatus.NOT_READY)
1861 if dag_status in {NodeStatus.NOT_READY, NodeStatus.FUTILE}:
1862 for id_ in missing_status.get(name, []):
1863 jobs[id_]["NodeStatus"] = dag_status
1865 return jobs
1868def read_single_dag_log(log_filename: str | os.PathLike) -> tuple[str, dict[str, dict[str, Any]]]:
1869 """Read job information from the DAGMan log file.
1871 Parameters
1872 ----------
1873 log_filename : `str` or `os.PathLike`
1874 DAGMan log filename.
1876 Returns
1877 -------
1878 wms_workflow_id : `str`
1879 HTCondor job id (i.e., <ClusterId>.<ProcId>) of the DAGMan job.
1880 dag_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1881 HTCondor job information read from the log file mapped to HTCondor
1882 job id.
1884 Raises
1885 ------
1886 FileNotFoundError
1887 If cannot find DAGMan log in given wms_path.
1888 """
1889 wms_workflow_id = "0"
1890 dag_info: dict[str, dict[str, Any]] = {}
1892 filename = Path(log_filename)
1893 if filename.exists():
1894 _LOG.debug("dag node log filename: %s", filename)
1896 info: dict[str, Any] = {}
1897 job_event_log = htcondor.JobEventLog(str(filename))
1898 for event in job_event_log.events(stop_after=0):
1899 id_ = f"{event['Cluster']}.{event['Proc']}"
1900 if id_ not in info:
1901 info[id_] = {}
1902 wms_workflow_id = id_ # taking last job id in case of restarts
1903 _update_dicts(info[id_], event)
1904 info[id_][f"{event.type.name.lower()}_time"] = event["EventTime"]
1906 # only save latest DAG job
1907 dag_info = {wms_workflow_id: info[wms_workflow_id]}
1909 return wms_workflow_id, dag_info
1912def read_dag_log(wms_path: str | os.PathLike) -> tuple[str, dict[str, Any]]:
1913 """Read job information from the DAGMan log file.
1915 Parameters
1916 ----------
1917 wms_path : `str` or `os.PathLike`
1918 Path containing the DAGMan log file.
1920 Returns
1921 -------
1922 wms_workflow_id : `str`
1923 HTCondor job id (i.e., <ClusterId>.<ProcId>) of the DAGMan job.
1924 dag_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1925 HTCondor job information read from the log file mapped to HTCondor
1926 job id.
1928 Raises
1929 ------
1930 FileNotFoundError
1931 If cannot find DAGMan log in given wms_path.
1932 """
1933 wms_workflow_id = MISSING_ID
1934 dag_info: dict[str, dict[str, Any]] = {}
1936 path = Path(wms_path)
1937 if path.exists():
1938 try:
1939 filename = next(path.glob("*.dag.dagman.log"))
1940 except StopIteration as exc:
1941 raise FileNotFoundError(f"DAGMan log not found in {wms_path}") from exc
1942 _LOG.debug("dag node log filename: %s", filename)
1943 wms_workflow_id, dag_info = read_single_dag_log(filename)
1945 return wms_workflow_id, dag_info
1948def read_single_dag_nodes_log(filename: str | os.PathLike) -> dict[str, dict[str, Any]]:
1949 """Read job information from the DAGMan nodes log file.
1951 Parameters
1952 ----------
1953 filename : `str` or `os.PathLike`
1954 Path containing the DAGMan nodes log file.
1956 Returns
1957 -------
1958 info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1959 HTCondor job information read from the log file mapped to HTCondor
1960 job id.
1962 Raises
1963 ------
1964 FileNotFoundError
1965 If cannot find DAGMan node log in given wms_path.
1966 """
1967 _LOG.debug("dag node log filename: %s", filename)
1968 filename = Path(filename)
1970 info: dict[str, dict[str, Any]] = {}
1971 if not filename.exists():
1972 raise FileNotFoundError(f"{filename} does not exist")
1974 try:
1975 job_event_log = htcondor.JobEventLog(str(filename))
1976 except htcondor.HTCondorIOError as ex:
1977 _LOG.error("Problem reading nodes log file (%s): %s", filename, ex)
1978 import traceback
1980 traceback.print_stack()
1981 raise
1982 for event in job_event_log.events(stop_after=0):
1983 _LOG.debug("log event type = %s, keys = %s", event["EventTypeNumber"], event.keys())
1985 try:
1986 id_ = f"{event['Cluster']}.{event['Proc']}"
1987 except KeyError:
1988 _LOG.warn(
1989 "Log event missing ids (DAGNodeName=%s, EventTime=%s, EventTypeNumber=%s)",
1990 event.get("DAGNodeName", "UNK"),
1991 event.get("EventTime", "UNK"),
1992 event.get("EventTypeNumber", "UNK"),
1993 )
1994 else:
1995 if id_ not in info:
1996 info[id_] = {}
1997 # Workaround: Please check to see if still problem in
1998 # future HTCondor versions. Sometimes get a
1999 # JobAbortedEvent for a subdag job after it already
2000 # terminated normally. Seems to happen when using job
2001 # plus subdags.
2002 if event["EventTypeNumber"] == 9 and info[id_].get("EventTypeNumber", -1) == 5:
2003 _LOG.debug("Skipping spurious JobAbortedEvent: %s", dict(event))
2004 else:
2005 _update_dicts(info[id_], event)
2006 info[id_][f"{event.type.name.lower()}_time"] = event["EventTime"]
2008 return info
2011def read_dag_nodes_log(wms_path: str | os.PathLike) -> dict[str, dict[str, Any]]:
2012 """Read job information from the DAGMan nodes log file.
2014 Parameters
2015 ----------
2016 wms_path : `str` or `os.PathLike`
2017 Path containing the DAGMan nodes log file.
2019 Returns
2020 -------
2021 info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
2022 HTCondor job information read from the log file mapped to HTCondor
2023 job id.
2025 Raises
2026 ------
2027 FileNotFoundError
2028 If cannot find DAGMan node log in given wms_path.
2029 """
2030 info: dict[str, dict[str, Any]] = {}
2031 for filename in Path(wms_path).glob("*.dag.nodes.log"):
2032 _LOG.debug("dag node log filename: %s", filename)
2033 _update_dicts(info, read_single_dag_nodes_log(filename))
2035 # If submitted, the main nodes log file should exist
2036 if not info:
2037 raise FileNotFoundError(f"DAGMan node log not found in {wms_path}")
2039 # Subdags will not have dag nodes log files if they haven't
2040 # started running yet (so missing is not an error).
2041 for filename in Path(wms_path).glob("subdags/*/*.dag.nodes.log"):
2042 _LOG.debug("dag node log filename: %s", filename)
2043 _update_dicts(info, read_single_dag_nodes_log(filename))
2045 return info
2048def read_dag_info(wms_path: str | os.PathLike) -> dict[str, dict[str, Any]]:
2049 """Read custom DAGMan job information from the file.
2051 Parameters
2052 ----------
2053 wms_path : `str` or `os.PathLike`
2054 Path containing the file with the DAGMan job info.
2056 Returns
2057 -------
2058 dag_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
2059 HTCondor job information.
2061 Raises
2062 ------
2063 FileNotFoundError
2064 If cannot find DAGMan job info file in the given location.
2065 """
2066 dag_info: dict[str, dict[str, Any]] = {}
2067 try:
2068 filename = next(Path(wms_path).glob("*.info.json"))
2069 except StopIteration as exc:
2070 raise FileNotFoundError(f"File with DAGMan job information not found in {wms_path}") from exc
2071 _LOG.debug("DAGMan job information filename: %s", filename)
2072 try:
2073 with open(filename) as fh:
2074 dag_info = json.load(fh)
2075 except (OSError, PermissionError) as exc:
2076 _LOG.debug("Retrieving DAGMan job information failed: %s", exc)
2077 return dag_info
2080def write_dag_info(filename, dag_info):
2081 """Write custom job information about DAGMan job.
2083 Parameters
2084 ----------
2085 filename : `str`
2086 Name of the file where the information will be stored.
2087 dag_info : `dict` [`str` `dict` [`str`, `~typing.Any`]]
2088 Information about the DAGMan job.
2089 """
2090 schedd_name = next(iter(dag_info))
2091 dag_id = next(iter(dag_info[schedd_name]))
2092 dag_ad = dag_info[schedd_name][dag_id]
2093 ad = {"ClusterId": dag_ad["ClusterId"], "GlobalJobId": dag_ad["GlobalJobId"]}
2094 ad.update({key: val for key, val in dag_ad.items() if key.startswith("bps")})
2095 try:
2096 with open(filename, "w") as fh:
2097 info = {schedd_name: {dag_id: ad}}
2098 json.dump(info, fh)
2099 except (KeyError, OSError, PermissionError) as exc:
2100 _LOG.debug("Persisting DAGMan job information failed: %s", exc)
2103def htc_tweak_log_info(wms_path: str | Path, job: dict[str, Any]) -> None:
2104 """Massage the given job info has same structure as if came from condor_q.
2106 Parameters
2107 ----------
2108 wms_path : `str` | `os.PathLike`
2109 Path containing an HTCondor event log file.
2110 job : `dict` [ `str`, `~typing.Any` ]
2111 A mapping between HTCondor job id and job information read from
2112 the log.
2113 """
2114 _LOG.debug("htc_tweak_log_info: %s %s", wms_path, job)
2116 # Use the presence of 'MyType' key as a proxy to determine if the job ad
2117 # contains the info extracted from the event log. Exit early if it doesn't
2118 # (e.g. it is a job ad for a pruned job).
2119 if "MyType" not in job:
2120 return
2122 try:
2123 job["ClusterId"] = job["Cluster"]
2124 job["ProcId"] = job["Proc"]
2125 except KeyError as e:
2126 _LOG.error("Missing key %s in job: %s", str(e), job)
2127 raise
2128 job["Iwd"] = str(wms_path)
2129 job["Owner"] = Path(wms_path).owner()
2131 match job["MyType"]:
2132 case "ExecuteEvent":
2133 job["JobStatus"] = htcondor.JobStatus.RUNNING
2134 case "JobTerminatedEvent" | "PostScriptTerminatedEvent":
2135 job["JobStatus"] = htcondor.JobStatus.COMPLETED
2136 case "SubmitEvent":
2137 job["JobStatus"] = htcondor.JobStatus.IDLE
2138 case "JobAbortedEvent":
2139 job["JobStatus"] = htcondor.JobStatus.REMOVED
2140 case "JobHeldEvent":
2141 job["JobStatus"] = htcondor.JobStatus.HELD
2142 case "JobReleaseEvent":
2143 # Shows up as the last event if a DAG job was held and released,
2144 # so assume the job is running. If a regular job is released,
2145 # there will be other events, so JobReleaseEvent won't be the last.
2146 job["JobStatus"] = htcondor.JobStatus.RUNNING
2147 case _:
2148 _LOG.debug("Unknown log event type: %s", job["MyType"])
2149 job["JobStatus"] = None
2151 if job["JobStatus"] in {
2152 htcondor.JobStatus.COMPLETED,
2153 htcondor.JobStatus.HELD,
2154 htcondor.JobStatus.REMOVED,
2155 }:
2156 new_job = HTC_JOB_AD_HANDLERS.handle(job)
2157 if new_job is not None:
2158 job = new_job
2159 else:
2160 _LOG.error("Could not determine exit status for job '%s.%s'", job["ClusterId"], job["ProcId"])
2162 if "LogNotes" in job:
2163 m = re.match(r"DAG Node: (\S+)", job["LogNotes"])
2164 if m:
2165 job["DAGNodeName"] = m.group(1)
2168def htc_check_dagman_output(wms_path: str | os.PathLike) -> str:
2169 """Check the DAGMan output for error messages.
2171 Parameters
2172 ----------
2173 wms_path : `str` or `os.PathLike`
2174 Directory containing the DAGman output file.
2176 Returns
2177 -------
2178 message : `str`
2179 Message containing error messages from the DAGMan output. Empty
2180 string if no messages.
2182 Raises
2183 ------
2184 FileNotFoundError
2185 If cannot find DAGMan standard output file in given wms_path.
2186 """
2187 try:
2188 filename = next(Path(wms_path).glob("*.dag.dagman.out"))
2189 except StopIteration as exc:
2190 raise FileNotFoundError(f"DAGMan standard output file not found in {wms_path}") from exc
2191 _LOG.debug("dag output filename: %s", filename)
2193 p = re.compile(r"^(\d\d/\d\d/\d\d \d\d:\d\d:\d\d) (Job submit try \d+/\d+ failed|Warning:.*$|ERROR:.*$)")
2195 message = ""
2196 try:
2197 with open(filename) as fh:
2198 last_submit_failed = "" # Since submit retries multiple times only report last one
2199 for line in fh:
2200 m = p.match(line)
2201 if m:
2202 if m.group(2).startswith("Job submit try"):
2203 last_submit_failed = m.group(1)
2204 elif m.group(2).startswith("ERROR: submit attempt failed"):
2205 pass # Should be handled by Job submit try
2206 elif m.group(2).startswith("Warning"):
2207 if ".dag.nodes.log is in /tmp" in m.group(2):
2208 last_warning = "Cannot submit from /tmp."
2209 else:
2210 last_warning = m.group(2)
2211 elif m.group(2) == "ERROR: Warning is fatal error because of DAGMAN_USE_STRICT setting":
2212 message += "ERROR: "
2213 message += last_warning
2214 message += "\n"
2215 elif m.group(2) in [
2216 "ERROR: the following job(s) failed:",
2217 "ERROR: the following Node(s) failed:",
2218 ]:
2219 pass
2220 else:
2221 message += m.group(2)
2222 message += "\n"
2224 if last_submit_failed:
2225 message += f"Warn: Job submission issues (last: {last_submit_failed})"
2226 except (OSError, PermissionError):
2227 message = f"Warn: Could not read dagman output file from {wms_path}."
2228 _LOG.debug("dag output file message: %s", message)
2229 return message
2232def _read_rescue_headers(infh: TextIO) -> tuple[list[str], list[str]]:
2233 """Read header lines from a rescue file.
2235 Parameters
2236 ----------
2237 infh : `TextIO`
2238 The rescue file from which to read the header lines.
2240 Returns
2241 -------
2242 header_lines : `list` [`str`]
2243 Header lines read from the rescue file.
2244 failed_subdags : `list` [`str`]
2245 Names of failed subdag jobs.
2246 """
2247 header_lines: list[str] = []
2248 failed = False
2249 failed_subdags: list[str] = []
2251 for line in infh:
2252 line = line.strip()
2253 if line.startswith("#"):
2254 if line.startswith("# Nodes that failed:"):
2255 failed = True
2256 header_lines.append(line)
2257 elif failed:
2258 orig_failed_nodes = line[1:].strip().split(",")
2259 new_failed_nodes = []
2260 for node in orig_failed_nodes:
2261 if node.startswith("wms_check_status"):
2262 group_node = node[17:]
2263 failed_subdags.append(group_node)
2264 new_failed_nodes.append(group_node)
2265 else:
2266 new_failed_nodes.append(node)
2267 header_lines.append(f"# {','.join(new_failed_nodes)}")
2268 if orig_failed_nodes[-1] == "<ENDLIST>":
2269 failed = False
2270 else:
2271 header_lines.append(line)
2272 elif line.strip() == "": # end of headers
2273 break
2274 return header_lines, failed_subdags
2277def _write_rescue_headers(header_lines: list[str], failed_subdags: list[str], outfh: TextIO) -> None:
2278 """Write the header lines to the new rescue file.
2280 Parameters
2281 ----------
2282 header_lines : `list` [`str`]
2283 Header lines to write to the new rescue file.
2284 failed_subdags : `list` [`str`]
2285 Job names of the failed subdags.
2286 outfh : `TextIO`
2287 New rescue file.
2288 """
2289 done_str = "# Nodes premarked DONE"
2290 pattern = f"^{done_str}:\\s+(\\d+)"
2291 for header_line in header_lines:
2292 m = re.match(pattern, header_line)
2293 if m:
2294 print(f"{done_str}: {int(m.group(1)) - len(failed_subdags)}", file=outfh)
2295 else:
2296 print(header_line, file=outfh)
2298 print("", file=outfh)
2301def _copy_done_lines(failed_subdags: list[str], infh: TextIO, outfh: TextIO) -> None:
2302 """Copy the DONE lines from the original rescue file skipping
2303 the failed group jobs.
2305 Parameters
2306 ----------
2307 failed_subdags : `list` [`str`]
2308 List of job names for the failed subdags
2309 infh : `TextIO`
2310 Original rescue file to copy from.
2311 outfh : `TextIO`
2312 New rescue file to copy to.
2313 """
2314 for line in infh:
2315 line = line.strip()
2316 try:
2317 _, node_name = line.split()
2318 except ValueError:
2319 _LOG.error(f"Unexpected line in rescue file = '{line}'")
2320 raise
2321 if node_name not in failed_subdags:
2322 print(line, file=outfh)
2325def _update_rescue_file(rescue_file: Path) -> None:
2326 """Update the subdag failures in the main rescue file
2327 and backup the failed subdag dirs.
2329 Parameters
2330 ----------
2331 rescue_file : `pathlib.Path`
2332 The main rescue file that needs to be updated.
2333 """
2334 # To reduce memory requirements, not reading entire file into memory.
2335 rescue_tmp = rescue_file.with_suffix(rescue_file.suffix + ".tmp")
2336 with open(rescue_file) as infh:
2337 header_lines, failed_subdags = _read_rescue_headers(infh)
2338 with open(rescue_tmp, "w") as outfh:
2339 _write_rescue_headers(header_lines, failed_subdags, outfh)
2340 _copy_done_lines(failed_subdags, infh, outfh)
2341 rescue_file.unlink()
2342 rescue_tmp.rename(rescue_file)
2343 for failed_subdag in failed_subdags:
2344 htc_backup_files(
2345 rescue_file.parent / "subdags" / failed_subdag, subdir=f"backups/subdags/{failed_subdag}"
2346 )
2349def _update_dicts(dict1, dict2):
2350 """Update dict1 with info in dict2.
2352 (Basically an update for nested dictionaries.)
2354 Parameters
2355 ----------
2356 dict1 : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
2357 HTCondor job information to be updated.
2358 dict2 : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
2359 Additional HTCondor job information.
2360 """
2361 for key, value in dict2.items():
2362 if key in dict1 and isinstance(dict1[key], dict) and isinstance(value, dict):
2363 _update_dicts(dict1[key], value)
2364 else:
2365 dict1[key] = value
2368def _locate_schedds(locate_all=False):
2369 """Find out Scheduler daemons in an HTCondor pool.
2371 Parameters
2372 ----------
2373 locate_all : `bool`, optional
2374 If True, all available schedulers in the HTCondor pool will be located.
2375 False by default which means that the search will be limited to looking
2376 for the Scheduler running on a local host.
2378 Returns
2379 -------
2380 schedds : `dict` [`str`, `htcondor.Schedd`]
2381 A mapping between Scheduler names and Python objects allowing for
2382 interacting with them.
2383 """
2384 coll = htcondor.Collector()
2386 schedd_ads = []
2387 if locate_all:
2388 schedd_ads.extend(coll.locateAll(htcondor.DaemonTypes.Schedd))
2389 else:
2390 schedd_ads.append(coll.locate(htcondor.DaemonTypes.Schedd))
2391 return {ad["Name"]: htcondor.Schedd(ad) for ad in schedd_ads}