Coverage for python / lsst / ctrl / bps / htcondor / lssthtc.py: 11%
923 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-02-13 01:21 -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 : `~collections.abc.Container`
248 Strings that are valid keys.
249 init_data : `dict` or `RestrictedDict`, optional
250 Initial data.
252 Raises
253 ------
254 KeyError
255 If invalid key(s) in init_data.
256 """
258 def __init__(self, valid_keys, init_data=()):
259 self.valid_keys = valid_keys
260 self.data = {}
261 self.update(init_data)
263 def __getitem__(self, key):
264 """Return value for given key if exists.
266 Parameters
267 ----------
268 key : `str`
269 Identifier for value to return.
271 Returns
272 -------
273 value : `~typing.Any`
274 Value associated with given key.
276 Raises
277 ------
278 KeyError
279 If key doesn't exist.
280 """
281 return self.data[key]
283 def __delitem__(self, key):
284 """Delete value for given key if exists.
286 Parameters
287 ----------
288 key : `str`
289 Identifier for value to delete.
291 Raises
292 ------
293 KeyError
294 If key doesn't exist.
295 """
296 del self.data[key]
298 def __setitem__(self, key, value):
299 """Store key,value in internal dict only if key is valid.
301 Parameters
302 ----------
303 key : `str`
304 Identifier to associate with given value.
305 value : `~typing.Any`
306 Value to store.
308 Raises
309 ------
310 KeyError
311 If key is invalid.
312 """
313 if key not in self.valid_keys:
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 : `~typing.TextIO`
495 Output text stream (typically an open file).
496 attrs : `dict`
497 HTCondor job attributes (dictionary of attribute key, value).
498 """
499 for key, value in attrs.items():
500 # Make sure strings are syntactically correct for HTCondor.
501 if isinstance(value, str):
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 : `~typing.TextIO`
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 : `~typing.TextIO`
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 : `~typing.TextIO`
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 : `~typing.Any`
1027 Initial graph data of any format that is supported
1028 by the to_network_graph() function.
1029 name : `str`
1030 Name for DAG.
1031 """
1033 def __init__(self, data=None, name=""):
1034 super().__init__(data=data, name=name)
1036 self.graph["attr"] = {}
1037 self.graph["run_id"] = None
1038 self.graph["submit_path"] = None
1039 self.graph["final_job"] = None
1040 self.graph["service_job"] = None
1041 self.graph["submit_options"] = {}
1043 def __str__(self):
1044 """Represent basic DAG info as string.
1046 Returns
1047 -------
1048 info : `str`
1049 String containing basic DAG info.
1050 """
1051 return f"{self.graph['name']} {len(self)}"
1053 def add_attribs(self, attribs=None):
1054 """Add attributes to the DAG.
1056 Parameters
1057 ----------
1058 attribs : `dict`
1059 DAG attributes.
1060 """
1061 if attribs is not None:
1062 self.graph["attr"].update(attribs)
1064 def add_job(self, job, parent_names=None, child_names=None):
1065 """Add an HTCJob to the HTCDag.
1067 Parameters
1068 ----------
1069 job : `HTCJob`
1070 HTCJob to add to the HTCDag.
1071 parent_names : `~collections.abc.Iterable` [`str`], optional
1072 Names of parent jobs.
1073 child_names : `~collections.abc.Iterable` [`str`], optional
1074 Names of child jobs.
1075 """
1076 assert isinstance(job, HTCJob)
1077 _LOG.debug("Adding job %s to dag", job.name)
1079 # Add dag level attributes to each job
1080 job.add_job_attrs(self.graph["attr"])
1082 self.add_node(job.name, data=job)
1084 if parent_names is not None:
1085 self.add_job_relationships(parent_names, [job.name])
1087 if child_names is not None:
1088 self.add_job_relationships(child_names, [job.name])
1090 def add_job_relationships(self, parents, children):
1091 """Add DAG edge between parents and children jobs.
1093 Parameters
1094 ----------
1095 parents : `list` [`str`]
1096 Contains parent job name(s).
1097 children : `list` [`str`]
1098 Contains children job name(s).
1099 """
1100 self.add_edges_from(itertools.product(parents, children))
1102 def add_final_job(self, job):
1103 """Add an HTCJob for the FINAL job in HTCDag.
1105 Parameters
1106 ----------
1107 job : `HTCJob`
1108 HTCJob to add to the HTCDag as a FINAL job.
1109 """
1110 # Add dag level attributes to each job
1111 job.add_job_attrs(self.graph["attr"])
1113 self.graph["final_job"] = job
1115 def add_service_job(self, job):
1116 """Add an HTCJob for the SERVICE job in HTCDag.
1118 Parameters
1119 ----------
1120 job : `HTCJob`
1121 HTCJob to add to the HTCDag as a FINAL job.
1122 """
1123 # Add dag level attributes to each job
1124 job.add_job_attrs(self.graph["attr"])
1126 self.graph["service_job"] = job
1128 def del_job(self, job_name):
1129 """Delete the job from the DAG.
1131 Parameters
1132 ----------
1133 job_name : `str`
1134 Name of job in DAG to delete.
1135 """
1136 # Reconnect edges around node to delete
1137 parents = self.predecessors(job_name)
1138 children = self.successors(job_name)
1139 self.add_edges_from(itertools.product(parents, children))
1141 # Delete job node (which deletes its edges).
1142 self.remove_node(job_name)
1144 def write(self, submit_path, job_subdir="", dag_subdir="", dag_rel_path=""):
1145 """Write DAG to a file.
1147 Parameters
1148 ----------
1149 submit_path : `str`
1150 Prefix path for all outputs.
1151 job_subdir : `str`, optional
1152 Template for job subdir (submit_path + job_subdir).
1153 dag_subdir : `str`, optional
1154 DAG subdir (submit_path + dag_subdir).
1155 dag_rel_path : `str`, optional
1156 Prefix to job_subdir for jobs inside subdag.
1157 """
1158 self.graph["submit_path"] = submit_path
1159 self.graph["dag_filename"] = os.path.join(dag_subdir, f"{self.graph['name']}.dag")
1160 full_filename = os.path.join(submit_path, self.graph["dag_filename"])
1161 os.makedirs(os.path.dirname(full_filename), exist_ok=True)
1163 try:
1164 dagman_config_path = Path(self.graph["attr"]["bps_wms_config_path"])
1165 except KeyError:
1166 dagman_config_path = None
1167 with open(full_filename, "w") as fh:
1168 if dagman_config_path is not None:
1169 fh.write(f"CONFIG {dag_rel_path / dagman_config_path}\n")
1171 for name, nodeval in self.nodes().items():
1172 try:
1173 job = nodeval["data"]
1174 except KeyError:
1175 _LOG.error("Job %s doesn't have data (keys: %s).", name, nodeval.keys())
1176 raise
1177 if job.subdag:
1178 dag_subdir = f"subdags/{job.name}"
1179 if "dir" in job.dagcmds:
1180 subdir = job.dagcmds["dir"]
1181 else:
1182 subdir = job_subdir
1183 if dagman_config_path is not None:
1184 job.subdag.add_attribs({"bps_wms_config_path": str(dagman_config_path)})
1185 job.subdag.write(submit_path, subdir, dag_subdir, "../..")
1186 fh.write(
1187 f"SUBDAG EXTERNAL {job.name} {Path(job.subdag.graph['dag_filename']).name} "
1188 f"DIR {dag_subdir}\n"
1189 )
1190 if job.dagcmds:
1191 _htc_write_job_commands(fh, job.name, job.dagcmds)
1192 else:
1193 job.write_submit_file(submit_path)
1194 job.write_dag_commands(fh, dag_rel_path)
1196 for edge in self.edges():
1197 print(f"PARENT {edge[0]} CHILD {edge[1]}", file=fh)
1198 print(f"DOT {self.name}.dot", file=fh)
1199 print(f"NODE_STATUS_FILE {self.name}.node_status", file=fh)
1201 # Add bps attributes to dag submission
1202 for key, value in self.graph["attr"].items():
1203 print(f'SET_JOB_ATTR {key}= "{htc_escape(value)}"', file=fh)
1205 # Add special nodes if any.
1206 special_jobs = {
1207 "FINAL": self.graph["final_job"],
1208 "SERVICE": self.graph["service_job"],
1209 }
1210 for dagcmd, job in special_jobs.items():
1211 if job is not None:
1212 job.write_submit_file(submit_path)
1213 job.write_dag_commands(fh, dag_rel_path, dagcmd)
1215 def dump(self, fh):
1216 """Dump DAG info to output stream.
1218 Parameters
1219 ----------
1220 fh : `typing.IO`
1221 Where to dump DAG info as text.
1222 """
1223 for key, value in self.graph:
1224 print(f"{key}={value}", file=fh)
1225 for name, data in self.nodes().items():
1226 print(f"{name}:", file=fh)
1227 data.dump(fh)
1228 for edge in self.edges():
1229 print(f"PARENT {edge[0]} CHILD {edge[1]}", file=fh)
1230 if self.graph["final_job"]:
1231 print(f"FINAL {self.graph['final_job'].name}:", file=fh)
1232 self.graph["final_job"].dump(fh)
1234 def write_dot(self, filename):
1235 """Write a dot version of the DAG.
1237 Parameters
1238 ----------
1239 filename : `str`
1240 Name of the dot file.
1241 """
1242 pos = networkx.nx_agraph.graphviz_layout(self)
1243 networkx.draw(self, pos=pos)
1244 networkx.drawing.nx_pydot.write_dot(self, filename)
1247def condor_q(constraint=None, schedds=None, **kwargs):
1248 """Get information about the jobs in the HTCondor job queue(s).
1250 Parameters
1251 ----------
1252 constraint : `str`, optional
1253 Constraints to be passed to job query.
1254 schedds : `dict` [`str`, `htcondor.Schedd`], optional
1255 HTCondor schedulers which to query for job information. If None
1256 (default), the query will be run against local scheduler only.
1257 **kwargs : `~typing.Any`
1258 Additional keyword arguments that need to be passed to the internal
1259 query method.
1261 Returns
1262 -------
1263 job_info : `dict` [`str`, `dict` [`str`, `dict` [`str`, `~typing.Any`]]]
1264 Information about jobs satisfying the search criteria where for each
1265 Scheduler, local HTCondor job ids are mapped to their respective
1266 classads.
1267 """
1268 return condor_query(constraint, schedds, htc_query_present, **kwargs)
1271def condor_history(constraint=None, schedds=None, **kwargs):
1272 """Get information about the jobs from HTCondor history records.
1274 Parameters
1275 ----------
1276 constraint : `str`, optional
1277 Constraints to be passed to job query.
1278 schedds : `dict` [`str`, `htcondor.Schedd`], optional
1279 HTCondor schedulers which to query for job information. If None
1280 (default), the query will be run against the history file of
1281 the local scheduler only.
1282 **kwargs : `~typing.Any`
1283 Additional keyword arguments that need to be passed to the internal
1284 query method.
1286 Returns
1287 -------
1288 job_info : `dict` [`str`, `dict` [`str`, `dict` [`str`, `~typing.Any`]]]
1289 Information about jobs satisfying the search criteria where for each
1290 Scheduler, local HTCondor job ids are mapped to their respective
1291 classads.
1292 """
1293 return condor_query(constraint, schedds, htc_query_history, **kwargs)
1296def condor_query(constraint=None, schedds=None, query_func=htc_query_present, **kwargs):
1297 """Get information about HTCondor jobs.
1299 Parameters
1300 ----------
1301 constraint : `str`, optional
1302 Constraints to be passed to job query.
1303 schedds : `dict` [`str`, `htcondor.Schedd`], optional
1304 HTCondor schedulers which to query for job information. If None
1305 (default), the query will be run against the history file of
1306 the local scheduler only.
1307 query_func : `~collections.abc.Callable`
1308 An query function which takes following arguments:
1310 - ``schedds``: Schedulers to query (`list` [`htcondor.Schedd`]).
1311 - ``**kwargs``: Keyword arguments that will be passed to the query
1312 function.
1313 **kwargs : `~typing.Any`
1314 Additional keyword arguments that need to be passed to the query
1315 method.
1317 Returns
1318 -------
1319 job_info : `dict` [`str`, `dict` [`str`, `dict` [`str`, `~typing.Any`]]]
1320 Information about jobs satisfying the search criteria where for each
1321 Scheduler, local HTCondor job ids are mapped to their respective
1322 classads.
1323 """
1324 if not schedds:
1325 coll = htcondor.Collector()
1326 schedd_ad = coll.locate(htcondor.DaemonTypes.Schedd)
1327 schedds = {schedd_ad["Name"]: htcondor.Schedd(schedd_ad)}
1329 # Make sure that 'ClusterId' and 'ProcId' attributes are always included
1330 # in the job classad. They are needed to construct the job id.
1331 added_attrs = set()
1332 if "projection" in kwargs and kwargs["projection"]:
1333 requested_attrs = set(kwargs["projection"])
1334 required_attrs = {"ClusterId", "ProcId"}
1335 added_attrs = required_attrs - requested_attrs
1336 for attr in added_attrs:
1337 kwargs["projection"].append(attr)
1339 unwanted_attrs = {"Env", "Environment"} | added_attrs
1340 job_info = defaultdict(dict)
1341 for schedd_name, job_ad in query_func(schedds, constraint=constraint, **kwargs):
1342 id_ = f"{job_ad['ClusterId']}.{job_ad['ProcId']}"
1343 for attr in set(job_ad) & unwanted_attrs:
1344 del job_ad[attr]
1345 job_info[schedd_name][id_] = job_ad
1346 _LOG.debug("query returned %d jobs", sum(len(val) for val in job_info.values()))
1348 # Restore the list of the requested attributes to its original value
1349 # if needed.
1350 if added_attrs:
1351 for attr in added_attrs:
1352 kwargs["projection"].remove(attr)
1354 # When returning the results filter out entries for schedulers with no jobs
1355 # matching the search criteria.
1356 return {key: val for key, val in job_info.items() if val}
1359def condor_search(constraint=None, hist=None, schedds=None):
1360 """Search for running and finished jobs satisfying given criteria.
1362 Parameters
1363 ----------
1364 constraint : `str`, optional
1365 Constraints to be passed to job query.
1366 hist : `float`
1367 Limit history search to this many days.
1368 schedds : `dict` [`str`, `htcondor.Schedd`], optional
1369 The list of the HTCondor schedulers which to query for job information.
1370 If None (default), only the local scheduler will be queried.
1372 Returns
1373 -------
1374 job_info : `dict` [`str`, `dict` [`str`, `dict` [`str` `~typing.Any`]]]
1375 Information about jobs satisfying the search criteria where for each
1376 Scheduler, local HTCondor job ids are mapped to their respective
1377 classads.
1378 """
1379 if not schedds:
1380 coll = htcondor.Collector()
1381 schedd_ad = coll.locate(htcondor.DaemonTypes.Schedd)
1382 schedds = {schedd_ad["Name"]: htcondor.Schedd(locate_ad=schedd_ad)}
1384 job_info = condor_q(constraint=constraint, schedds=schedds)
1385 if hist is not None:
1386 _LOG.debug("Searching history going back %s days", hist)
1387 epoch = (datetime.now() - timedelta(days=hist)).timestamp()
1388 constraint += f" && (CompletionDate >= {epoch} || JobFinishedHookDone >= {epoch})"
1389 hist_info = condor_history(constraint, schedds=schedds)
1390 update_job_info(job_info, hist_info)
1391 return job_info
1394def condor_status(constraint=None, coll=None):
1395 """Get information about HTCondor pool.
1397 Parameters
1398 ----------
1399 constraint : `str`, optional
1400 Constraints to be passed to the query.
1401 coll : `htcondor.Collector`, optional
1402 Object representing HTCondor collector daemon.
1404 Returns
1405 -------
1406 pool_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1407 Mapping between HTCondor slot names and slot information (classAds).
1408 """
1409 if coll is None:
1410 coll = htcondor.Collector()
1411 try:
1412 pool_ads = coll.query(constraint=constraint)
1413 except OSError as ex:
1414 raise RuntimeError(f"Problem querying the Collector. (Constraint='{constraint}')") from ex
1416 pool_info = {}
1417 for slot in pool_ads:
1418 pool_info[slot["name"]] = dict(slot)
1419 _LOG.debug("condor_status returned %d ads", len(pool_info))
1420 return pool_info
1423def update_job_info(job_info, other_info):
1424 """Update results of a job query with results from another query.
1426 Parameters
1427 ----------
1428 job_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1429 Results of the job query that needs to be updated.
1430 other_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1431 Results of the other job query.
1433 Returns
1434 -------
1435 job_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1436 The updated results.
1437 """
1438 for schedd_name, others in other_info.items():
1439 try:
1440 jobs = job_info[schedd_name]
1441 except KeyError:
1442 job_info[schedd_name] = others
1443 else:
1444 for id_, ad in others.items():
1445 jobs.setdefault(id_, {}).update(ad)
1446 return job_info
1449def count_jobs_in_single_dag(
1450 filename: str | os.PathLike,
1451) -> tuple[Counter[str], dict[str, str], dict[str, WmsNodeType]]:
1452 """Build bps_run_summary string from dag file.
1454 Parameters
1455 ----------
1456 filename : `str`
1457 Path that includes dag file for a run.
1459 Returns
1460 -------
1461 counts : `Counter` [`str`]
1462 Semi-colon separated list of job labels and counts.
1463 (Same format as saved in dag classad).
1464 job_name_to_label : `dict` [`str`, `str`]
1465 Mapping of job names to job labels.
1466 job_name_to_type : `dict` [`str`, `lsst.ctrl.bps.htcondor.WmsNodeType`]
1467 Mapping of job names to job types
1468 (e.g., payload, final, service).
1469 """
1470 # Later code depends upon insertion order
1471 counts: Counter = Counter() # counts of payload jobs per label
1472 job_name_to_label: dict[str, str] = {}
1473 job_name_to_type: dict[str, WmsNodeType] = {}
1474 with open(filename) as fh:
1475 for line in fh:
1476 # Skip any line that contains commands irrelevant to job counting.
1477 if not line.startswith(
1478 (
1479 "JOB",
1480 "FINAL",
1481 "SERVICE",
1482 "SUBDAG EXTERNAL",
1483 )
1484 ):
1485 continue
1487 m = re.match(
1488 r"(?P<command>JOB|FINAL|SERVICE|SUBDAG EXTERNAL)\s+"
1489 r'(?P<jobname>(?P<wms>wms_)?\S+)\s+"?(?P<subfile>\S+)"?\s*'
1490 r'(DIR "?(?P<dir>[^\s"]+)"?)?\s*(?P<noop>NOOP)?',
1491 line,
1492 )
1493 if m:
1494 job_name = m.group("jobname")
1495 name_parts = job_name.split("_")
1497 label = ""
1498 if m.group("dir"):
1499 dir_match = re.search(r"jobs/([^\s/]+)", m.group("dir"))
1500 if dir_match:
1501 label = dir_match.group(1)
1502 else:
1503 _LOG.debug("Parse DAG: unparsed dir = %s", line)
1504 elif m.group("subfile"):
1505 subfile_match = re.search(r"jobs/([^\s/]+)", m.group("subfile"))
1506 if subfile_match:
1507 label = m.group("subfile").split("/")[1]
1508 else:
1509 label = pegasus_name_to_label(job_name)
1511 match m.group("command"):
1512 case "JOB":
1513 if m.group("noop"):
1514 job_type = WmsNodeType.NOOP
1515 # wms_noop_label
1516 label = name_parts[2]
1517 elif m.group("wms"):
1518 if name_parts[1] == "check":
1519 job_type = WmsNodeType.SUBDAG_CHECK
1520 # wms_check_status_wms_group_label
1521 label = name_parts[5]
1522 else:
1523 _LOG.warning(
1524 "Unexpected skipping of dag line due to unknown wms job: %s", line
1525 )
1526 else:
1527 job_type = WmsNodeType.PAYLOAD
1528 if label == "init":
1529 label = "pipetaskInit"
1530 counts[label] += 1
1531 case "FINAL":
1532 job_type = WmsNodeType.FINAL
1533 counts[label] += 1 # final counts a payload job.
1534 case "SERVICE":
1535 job_type = WmsNodeType.SERVICE
1536 case "SUBDAG EXTERNAL":
1537 job_type = WmsNodeType.SUBDAG
1538 label = name_parts[2]
1540 job_name_to_label[job_name] = label
1541 job_name_to_type[job_name] = job_type
1542 else:
1543 # The line should, but didn't match the pattern above. Probably
1544 # problems with regex.
1545 _LOG.warning("Unexpected skipping of dag line: %s", line)
1547 return counts, job_name_to_label, job_name_to_type
1550def summarize_dag(dir_name: str) -> tuple[str, dict[str, str], dict[str, WmsNodeType]]:
1551 """Build bps_run_summary string from dag file.
1553 Parameters
1554 ----------
1555 dir_name : `str`
1556 Path that includes dag file for a run.
1558 Returns
1559 -------
1560 summary : `str`
1561 Semi-colon separated list of job labels and counts
1562 (Same format as saved in dag classad).
1563 job_name_to_label : `dict` [`str`, `str`]
1564 Mapping of job names to job labels.
1565 job_name_to_type : `dict` [`str`, `lsst.ctrl.bps.htcondor.WmsNodeType`]
1566 Mapping of job names to job types
1567 (e.g., payload, final, service).
1568 """
1569 # Later code depends upon insertion order
1570 counts: Counter[str] = Counter() # counts of payload jobs per label
1571 job_name_to_label: dict[str, str] = {}
1572 job_name_to_type: dict[str, WmsNodeType] = {}
1573 for filename in Path(dir_name).glob("*.dag"):
1574 single_counts, single_job_name_to_label, single_job_name_to_type = count_jobs_in_single_dag(filename)
1575 counts += single_counts
1576 _update_dicts(job_name_to_label, single_job_name_to_label)
1577 _update_dicts(job_name_to_type, single_job_name_to_type)
1579 for filename in Path(dir_name).glob("subdags/*/*.dag"):
1580 single_counts, single_job_name_to_label, single_job_name_to_type = count_jobs_in_single_dag(filename)
1581 counts += single_counts
1582 _update_dicts(job_name_to_label, single_job_name_to_label)
1583 _update_dicts(job_name_to_type, single_job_name_to_type)
1585 summary = ";".join([f"{name}:{counts[name]}" for name in counts])
1586 _LOG.debug("summarize_dag: %s %s %s", summary, job_name_to_label, job_name_to_type)
1587 return summary, job_name_to_label, job_name_to_type
1590def pegasus_name_to_label(name):
1591 """Convert pegasus job name to a label for the report.
1593 Parameters
1594 ----------
1595 name : `str`
1596 Name of job.
1598 Returns
1599 -------
1600 label : `str`
1601 Label for job.
1602 """
1603 label = "UNK"
1604 if name.startswith("create_dir") or name.startswith("stage_in") or name.startswith("stage_out"):
1605 label = "pegasus"
1606 else:
1607 m = re.match(r"pipetask_(\d+_)?([^_]+)", name)
1608 if m:
1609 label = m.group(2)
1610 if label == "init":
1611 label = "pipetaskInit"
1613 return label
1616def read_single_dag_status(filename: str | os.PathLike) -> dict[str, Any]:
1617 """Read the node status file for DAG summary information.
1619 Parameters
1620 ----------
1621 filename : `str` or `Path.pathlib`
1622 Node status filename.
1624 Returns
1625 -------
1626 dag_ad : `dict` [`str`, `~typing.Any`]
1627 DAG summary information.
1628 """
1629 dag_ad: dict[str, Any] = {}
1631 # While this is probably more up to date than dag classad, only read from
1632 # file if need to.
1633 try:
1634 node_stat_file = Path(filename)
1635 _LOG.debug("Reading Node Status File %s", node_stat_file)
1636 with open(node_stat_file) as infh:
1637 dag_ad = dict(classad.parseNext(infh)) # pylint: disable=E1101
1639 if not dag_ad:
1640 # Pegasus check here
1641 metrics_file = node_stat_file.with_suffix(".dag.metrics")
1642 if metrics_file.exists():
1643 with open(metrics_file) as infh:
1644 metrics = json.load(infh)
1645 dag_ad["NodesTotal"] = metrics.get("jobs", 0)
1646 dag_ad["NodesFailed"] = metrics.get("jobs_failed", 0)
1647 dag_ad["NodesDone"] = metrics.get("jobs_succeeded", 0)
1648 metrics_file = node_stat_file.with_suffix(".metrics")
1649 with open(metrics_file) as infh:
1650 metrics = json.load(infh)
1651 dag_ad["NodesTotal"] = metrics["wf_metrics"]["total_jobs"]
1652 except (OSError, PermissionError):
1653 pass
1655 _LOG.debug("read_dag_status: %s", dag_ad)
1656 return dag_ad
1659def read_dag_status(wms_path: str | os.PathLike) -> dict[str, Any]:
1660 """Read the node status file for DAG summary information.
1662 Parameters
1663 ----------
1664 wms_path : `str` or `os.PathLike`
1665 Path that includes node status file for a run.
1667 Returns
1668 -------
1669 dag_ad : `dict` [`str`, `~typing.Any`]
1670 DAG summary information, counts summed across any subdags.
1671 """
1672 dag_ads: dict[str, Any] = {}
1673 path = Path(wms_path)
1674 try:
1675 node_stat_file = next(path.glob("*.node_status"))
1676 except StopIteration as exc:
1677 raise FileNotFoundError(f"DAGMan node status not found in {wms_path}") from exc
1679 dag_ads = read_single_dag_status(node_stat_file)
1681 for node_stat_file in path.glob("subdags/*/*.node_status"):
1682 dag_ad = read_single_dag_status(node_stat_file)
1683 dag_ads["JobProcsHeld"] += dag_ad.get("JobProcsHeld", 0)
1684 dag_ads["NodesPost"] += dag_ad.get("NodesPost", 0)
1685 dag_ads["JobProcsIdle"] += dag_ad.get("JobProcsIdle", 0)
1686 dag_ads["NodesTotal"] += dag_ad.get("NodesTotal", 0)
1687 dag_ads["NodesFailed"] += dag_ad.get("NodesFailed", 0)
1688 dag_ads["NodesDone"] += dag_ad.get("NodesDone", 0)
1689 dag_ads["NodesQueued"] += dag_ad.get("NodesQueued", 0)
1690 dag_ads["NodesPre"] += dag_ad.get("NodesReady", 0)
1691 dag_ads["NodesFutile"] += dag_ad.get("NodesFutile", 0)
1692 dag_ads["NodesUnready"] += dag_ad.get("NodesUnready", 0)
1694 return dag_ads
1697def read_single_node_status(filename: str | os.PathLike, init_fake_id: int) -> dict[str, Any]:
1698 """Read entire node status file.
1700 Parameters
1701 ----------
1702 filename : `str` or `pathlib.Path`
1703 Node status filename.
1704 init_fake_id : `int`
1705 Initial fake id value.
1707 Returns
1708 -------
1709 jobs : `dict` [`str`, `~typing.Any`]
1710 DAG summary information compiled from the node status file combined
1711 with the information found in the node event log.
1713 Currently, if the same job attribute is found in both files, its value
1714 from the event log takes precedence over the value from the node status
1715 file.
1716 """
1717 filename = Path(filename)
1719 # Get jobid info from other places to fill in gaps in info from node_status
1720 _, job_name_to_label, job_name_to_type = count_jobs_in_single_dag(filename.with_suffix(".dag"))
1721 loginfo: dict[str, dict[str, Any]] = {}
1722 try:
1723 wms_workflow_id, loginfo = read_single_dag_log(filename.with_suffix(".dag.dagman.log"))
1724 loginfo = read_single_dag_nodes_log(filename.with_suffix(".dag.nodes.log"))
1725 except (OSError, PermissionError):
1726 pass
1728 job_name_to_id: dict[str, str] = {}
1729 _LOG.debug("loginfo = %s", loginfo)
1730 log_job_name_to_id: dict[str, str] = {}
1731 for job_id, job_info in loginfo.items():
1732 if "LogNotes" in job_info:
1733 m = re.match(r"DAG Node: (\S+)", job_info["LogNotes"])
1734 if m:
1735 job_name = m.group(1)
1736 log_job_name_to_id[job_name] = job_id
1737 job_info["DAGNodeName"] = job_name
1738 job_info["wms_node_type"] = job_name_to_type[job_name]
1739 job_info["bps_job_label"] = job_name_to_label[job_name]
1741 jobs = {}
1742 fake_id = init_fake_id # For nodes that do not yet have a job id, give fake one
1743 try:
1744 with open(filename) as fh:
1745 for ad in classad.parseAds(fh):
1746 match ad["Type"]:
1747 case "DagStatus":
1748 # Skip DAG summary.
1749 pass
1750 case "NodeStatus":
1751 job_name = ad["Node"]
1752 if job_name in job_name_to_label:
1753 job_label = job_name_to_label[job_name]
1754 elif "_" in job_name:
1755 job_label = job_name.split("_")[1]
1756 else:
1757 job_label = job_name
1759 job = dict(ad)
1760 if job_name in log_job_name_to_id:
1761 job_id = str(log_job_name_to_id[job_name])
1762 _update_dicts(job, loginfo[job_id])
1763 else:
1764 job_id = str(fake_id)
1765 job = dict(ad)
1766 fake_id -= 1
1767 jobs[job_id] = job
1768 job_name_to_id[job_name] = job_id
1770 # Make job info as if came from condor_q.
1771 job["ClusterId"] = int(float(job_id))
1772 job["DAGManJobID"] = wms_workflow_id
1773 job["DAGNodeName"] = job_name
1774 job["bps_job_label"] = job_label
1775 job["wms_node_type"] = job_name_to_type[job_name]
1777 case "StatusEnd":
1778 # Skip node status file "epilog".
1779 pass
1780 case _:
1781 _LOG.debug(
1782 "Ignoring unknown classad type '%s' in the node status file '%s'",
1783 ad["Type"],
1784 filename,
1785 )
1786 except (OSError, PermissionError):
1787 pass
1789 # Check for missing jobs (e.g., submission failure or not submitted yet)
1790 # Use dag info to create job placeholders
1791 for name in set(job_name_to_label) - set(job_name_to_id):
1792 if name in log_job_name_to_id: # job was in nodes.log, but not node_status
1793 job_id = str(log_job_name_to_id[name])
1794 job = dict(loginfo[job_id])
1795 else:
1796 job_id = str(fake_id)
1797 fake_id -= 1
1798 job = {}
1799 job["NodeStatus"] = NodeStatus.NOT_READY
1801 job["ClusterId"] = int(float(job_id))
1802 job["ProcId"] = 0
1803 job["DAGManJobID"] = wms_workflow_id
1804 job["DAGNodeName"] = name
1805 job["bps_job_label"] = job_name_to_label[name]
1806 job["wms_node_type"] = job_name_to_type[name]
1807 jobs[f"{job['ClusterId']}.{job['ProcId']}"] = job
1809 for job_info in jobs.values():
1810 job_info["from_dag_job"] = f"wms_{filename.stem}"
1812 return jobs
1815def read_node_status(wms_path: str | os.PathLike) -> dict[str, dict[str, Any]]:
1816 """Read entire node status file.
1818 Parameters
1819 ----------
1820 wms_path : `str` or `os.PathLike`
1821 Path that includes node status file for a run.
1823 Returns
1824 -------
1825 jobs : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1826 DAG summary information compiled from the node status file combined
1827 with the information found in the node event log.
1829 Currently, if the same job attribute is found in both files, its value
1830 from the event log takes precedence over the value from the node status
1831 file.
1832 """
1833 jobs: dict[str, dict[str, Any]] = {}
1834 init_fake_id = -1
1836 # subdags may not have run so wouldn't have node_status file
1837 # use dag files and let read_single_node_status handle missing
1838 # node_status file.
1839 for dag_filename in Path(wms_path).glob("*.dag"):
1840 filename = dag_filename.with_suffix(".node_status")
1841 info = read_single_node_status(filename, init_fake_id)
1842 init_fake_id -= len(info)
1843 _update_dicts(jobs, info)
1845 for dag_filename in Path(wms_path).glob("subdags/*/*.dag"):
1846 filename = dag_filename.with_suffix(".node_status")
1847 info = read_single_node_status(filename, init_fake_id)
1848 init_fake_id -= len(info)
1849 _update_dicts(jobs, info)
1851 # Propagate pruned from subdags to jobs
1852 name_to_id: dict[str, str] = {}
1853 missing_status: dict[str, list[str]] = {}
1854 for id_, job in jobs.items():
1855 if job["DAGNodeName"].startswith("wms_"):
1856 name_to_id[job["DAGNodeName"]] = id_
1857 if "NodeStatus" not in job or job["NodeStatus"] == NodeStatus.NOT_READY:
1858 missing_status.setdefault(job["from_dag_job"], []).append(id_)
1860 for name, dag_id in name_to_id.items():
1861 dag_status = jobs[dag_id].get("NodeStatus", NodeStatus.NOT_READY)
1862 if dag_status in {NodeStatus.NOT_READY, NodeStatus.FUTILE}:
1863 for id_ in missing_status.get(name, []):
1864 jobs[id_]["NodeStatus"] = dag_status
1866 return jobs
1869def read_single_dag_log(log_filename: str | os.PathLike) -> tuple[str, dict[str, dict[str, Any]]]:
1870 """Read job information from the DAGMan log file.
1872 Parameters
1873 ----------
1874 log_filename : `str` or `os.PathLike`
1875 DAGMan log filename.
1877 Returns
1878 -------
1879 wms_workflow_id : `str`
1880 HTCondor job id (i.e., <ClusterId>.<ProcId>) of the DAGMan job.
1881 dag_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1882 HTCondor job information read from the log file mapped to HTCondor
1883 job id.
1885 Raises
1886 ------
1887 FileNotFoundError
1888 If cannot find DAGMan log in given wms_path.
1889 """
1890 wms_workflow_id = "0"
1891 dag_info: dict[str, dict[str, Any]] = {}
1893 filename = Path(log_filename)
1894 if filename.exists():
1895 _LOG.debug("dag node log filename: %s", filename)
1897 info: dict[str, Any] = {}
1898 job_event_log = htcondor.JobEventLog(str(filename))
1899 for event in job_event_log.events(stop_after=0):
1900 id_ = f"{event['Cluster']}.{event['Proc']}"
1901 if id_ not in info:
1902 info[id_] = {}
1903 wms_workflow_id = id_ # taking last job id in case of restarts
1904 _update_dicts(info[id_], event)
1905 info[id_][f"{event.type.name.lower()}_time"] = event["EventTime"]
1907 # only save latest DAG job
1908 dag_info = {wms_workflow_id: info[wms_workflow_id]}
1910 return wms_workflow_id, dag_info
1913def read_dag_log(wms_path: str | os.PathLike) -> tuple[str, dict[str, Any]]:
1914 """Read job information from the DAGMan log file.
1916 Parameters
1917 ----------
1918 wms_path : `str` or `os.PathLike`
1919 Path containing the DAGMan log file.
1921 Returns
1922 -------
1923 wms_workflow_id : `str`
1924 HTCondor job id (i.e., <ClusterId>.<ProcId>) of the DAGMan job.
1925 dag_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1926 HTCondor job information read from the log file mapped to HTCondor
1927 job id.
1929 Raises
1930 ------
1931 FileNotFoundError
1932 If cannot find DAGMan log in given wms_path.
1933 """
1934 wms_workflow_id = MISSING_ID
1935 dag_info: dict[str, dict[str, Any]] = {}
1937 path = Path(wms_path)
1938 if path.exists():
1939 try:
1940 filename = next(path.glob("*.dag.dagman.log"))
1941 except StopIteration as exc:
1942 raise FileNotFoundError(f"DAGMan log not found in {wms_path}") from exc
1943 _LOG.debug("dag node log filename: %s", filename)
1944 wms_workflow_id, dag_info = read_single_dag_log(filename)
1946 return wms_workflow_id, dag_info
1949def read_single_dag_nodes_log(filename: str | os.PathLike) -> dict[str, dict[str, Any]]:
1950 """Read job information from the DAGMan nodes log file.
1952 Parameters
1953 ----------
1954 filename : `str` or `os.PathLike`
1955 Path containing the DAGMan nodes log file.
1957 Returns
1958 -------
1959 info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1960 HTCondor job information read from the log file mapped to HTCondor
1961 job id.
1963 Raises
1964 ------
1965 FileNotFoundError
1966 If cannot find DAGMan node log in given wms_path.
1967 """
1968 _LOG.debug("dag node log filename: %s", filename)
1969 filename = Path(filename)
1971 info: dict[str, dict[str, Any]] = {}
1972 if not filename.exists():
1973 raise FileNotFoundError(f"{filename} does not exist")
1975 try:
1976 job_event_log = htcondor.JobEventLog(str(filename))
1977 except htcondor.HTCondorIOError as ex:
1978 _LOG.error("Problem reading nodes log file (%s): %s", filename, ex)
1979 import traceback
1981 traceback.print_stack()
1982 raise
1983 for event in job_event_log.events(stop_after=0):
1984 _LOG.debug("log event type = %s, keys = %s", event["EventTypeNumber"], event.keys())
1986 try:
1987 id_ = f"{event['Cluster']}.{event['Proc']}"
1988 except KeyError:
1989 _LOG.warn(
1990 "Log event missing ids (DAGNodeName=%s, EventTime=%s, EventTypeNumber=%s)",
1991 event.get("DAGNodeName", "UNK"),
1992 event.get("EventTime", "UNK"),
1993 event.get("EventTypeNumber", "UNK"),
1994 )
1995 else:
1996 if id_ not in info:
1997 info[id_] = {}
1998 # Workaround: Please check to see if still problem in
1999 # future HTCondor versions. Sometimes get a
2000 # JobAbortedEvent for a subdag job after it already
2001 # terminated normally. Seems to happen when using job
2002 # plus subdags.
2003 if event["EventTypeNumber"] == 9 and info[id_].get("EventTypeNumber", -1) == 5:
2004 _LOG.debug("Skipping spurious JobAbortedEvent: %s", dict(event))
2005 else:
2006 _update_dicts(info[id_], event)
2007 info[id_][f"{event.type.name.lower()}_time"] = event["EventTime"]
2009 return info
2012def read_dag_nodes_log(wms_path: str | os.PathLike) -> dict[str, dict[str, Any]]:
2013 """Read job information from the DAGMan nodes log file.
2015 Parameters
2016 ----------
2017 wms_path : `str` or `os.PathLike`
2018 Path containing the DAGMan nodes log file.
2020 Returns
2021 -------
2022 info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
2023 HTCondor job information read from the log file mapped to HTCondor
2024 job id.
2026 Raises
2027 ------
2028 FileNotFoundError
2029 If cannot find DAGMan node log in given wms_path.
2030 """
2031 info: dict[str, dict[str, Any]] = {}
2032 for filename in Path(wms_path).glob("*.dag.nodes.log"):
2033 _LOG.debug("dag node log filename: %s", filename)
2034 _update_dicts(info, read_single_dag_nodes_log(filename))
2036 # If submitted, the main nodes log file should exist
2037 if not info:
2038 raise FileNotFoundError(f"DAGMan node log not found in {wms_path}")
2040 # Subdags will not have dag nodes log files if they haven't
2041 # started running yet (so missing is not an error).
2042 for filename in Path(wms_path).glob("subdags/*/*.dag.nodes.log"):
2043 _LOG.debug("dag node log filename: %s", filename)
2044 _update_dicts(info, read_single_dag_nodes_log(filename))
2046 return info
2049def read_dag_info(wms_path: str | os.PathLike) -> dict[str, dict[str, Any]]:
2050 """Read custom DAGMan job information from the file.
2052 Parameters
2053 ----------
2054 wms_path : `str` or `os.PathLike`
2055 Path containing the file with the DAGMan job info.
2057 Returns
2058 -------
2059 dag_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
2060 HTCondor job information.
2062 Raises
2063 ------
2064 FileNotFoundError
2065 If cannot find DAGMan job info file in the given location.
2066 """
2067 dag_info: dict[str, dict[str, Any]] = {}
2068 try:
2069 filename = next(Path(wms_path).glob("*.info.json"))
2070 except StopIteration as exc:
2071 raise FileNotFoundError(f"File with DAGMan job information not found in {wms_path}") from exc
2072 _LOG.debug("DAGMan job information filename: %s", filename)
2073 try:
2074 with open(filename) as fh:
2075 dag_info = json.load(fh)
2076 except (OSError, PermissionError) as exc:
2077 _LOG.debug("Retrieving DAGMan job information failed: %s", exc)
2078 return dag_info
2081def write_dag_info(filename, dag_info):
2082 """Write custom job information about DAGMan job.
2084 Parameters
2085 ----------
2086 filename : `str`
2087 Name of the file where the information will be stored.
2088 dag_info : `dict` [`str` `dict` [`str`, `~typing.Any`]]
2089 Information about the DAGMan job.
2090 """
2091 schedd_name = next(iter(dag_info))
2092 dag_id = next(iter(dag_info[schedd_name]))
2093 dag_ad = dag_info[schedd_name][dag_id]
2094 ad = {"ClusterId": dag_ad["ClusterId"], "GlobalJobId": dag_ad["GlobalJobId"]}
2095 ad.update({key: val for key, val in dag_ad.items() if key.startswith("bps")})
2096 try:
2097 with open(filename, "w") as fh:
2098 info = {schedd_name: {dag_id: ad}}
2099 json.dump(info, fh)
2100 except (KeyError, OSError, PermissionError) as exc:
2101 _LOG.debug("Persisting DAGMan job information failed: %s", exc)
2104def htc_tweak_log_info(wms_path: str | Path, job: dict[str, Any]) -> None:
2105 """Massage the given job info has same structure as if came from condor_q.
2107 Parameters
2108 ----------
2109 wms_path : `str` | `os.PathLike`
2110 Path containing an HTCondor event log file.
2111 job : `dict` [ `str`, `~typing.Any` ]
2112 A mapping between HTCondor job id and job information read from
2113 the log.
2114 """
2115 _LOG.debug("htc_tweak_log_info: %s %s", wms_path, job)
2117 # Use the presence of 'MyType' key as a proxy to determine if the job ad
2118 # contains the info extracted from the event log. Exit early if it doesn't
2119 # (e.g. it is a job ad for a pruned job).
2120 if "MyType" not in job:
2121 return
2123 try:
2124 job["ClusterId"] = job["Cluster"]
2125 job["ProcId"] = job["Proc"]
2126 except KeyError as e:
2127 _LOG.error("Missing key %s in job: %s", str(e), job)
2128 raise
2129 job["Iwd"] = str(wms_path)
2130 job["Owner"] = Path(wms_path).owner()
2132 match job["MyType"]:
2133 case "ExecuteEvent":
2134 job["JobStatus"] = htcondor.JobStatus.RUNNING
2135 case "JobTerminatedEvent" | "PostScriptTerminatedEvent":
2136 job["JobStatus"] = htcondor.JobStatus.COMPLETED
2137 case "SubmitEvent":
2138 job["JobStatus"] = htcondor.JobStatus.IDLE
2139 case "JobAbortedEvent":
2140 job["JobStatus"] = htcondor.JobStatus.REMOVED
2141 case "JobHeldEvent":
2142 job["JobStatus"] = htcondor.JobStatus.HELD
2143 case "JobReleaseEvent":
2144 # Shows up as the last event if a DAG job was held and released,
2145 # so assume the job is running. If a regular job is released,
2146 # there will be other events, so JobReleaseEvent won't be the last.
2147 job["JobStatus"] = htcondor.JobStatus.RUNNING
2148 case _:
2149 _LOG.debug("Unknown log event type: %s", job["MyType"])
2150 job["JobStatus"] = None
2152 if job["JobStatus"] in {
2153 htcondor.JobStatus.COMPLETED,
2154 htcondor.JobStatus.HELD,
2155 htcondor.JobStatus.REMOVED,
2156 }:
2157 new_job = HTC_JOB_AD_HANDLERS.handle(job)
2158 if new_job is not None:
2159 job = new_job
2160 else:
2161 _LOG.error("Could not determine exit status for job '%s.%s'", job["ClusterId"], job["ProcId"])
2163 if "LogNotes" in job:
2164 m = re.match(r"DAG Node: (\S+)", job["LogNotes"])
2165 if m:
2166 job["DAGNodeName"] = m.group(1)
2169def htc_check_dagman_output(wms_path: str | os.PathLike) -> str:
2170 """Check the DAGMan output for error messages.
2172 Parameters
2173 ----------
2174 wms_path : `str` or `os.PathLike`
2175 Directory containing the DAGman output file.
2177 Returns
2178 -------
2179 message : `str`
2180 Message containing error messages from the DAGMan output. Empty
2181 string if no messages.
2183 Raises
2184 ------
2185 FileNotFoundError
2186 If cannot find DAGMan standard output file in given wms_path.
2187 """
2188 try:
2189 filename = next(Path(wms_path).glob("*.dag.dagman.out"))
2190 except StopIteration as exc:
2191 raise FileNotFoundError(f"DAGMan standard output file not found in {wms_path}") from exc
2192 _LOG.debug("dag output filename: %s", filename)
2194 p = re.compile(r"^(\d\d/\d\d/\d\d \d\d:\d\d:\d\d) (Job submit try \d+/\d+ failed|Warning:.*$|ERROR:.*$)")
2196 message = ""
2197 try:
2198 with open(filename) as fh:
2199 last_submit_failed = "" # Since submit retries multiple times only report last one
2200 for line in fh:
2201 m = p.match(line)
2202 if m:
2203 if m.group(2).startswith("Job submit try"):
2204 last_submit_failed = m.group(1)
2205 elif m.group(2).startswith("ERROR: submit attempt failed"):
2206 pass # Should be handled by Job submit try
2207 elif m.group(2).startswith("Warning"):
2208 if ".dag.nodes.log is in /tmp" in m.group(2):
2209 last_warning = "Cannot submit from /tmp."
2210 else:
2211 last_warning = m.group(2)
2212 elif m.group(2) == "ERROR: Warning is fatal error because of DAGMAN_USE_STRICT setting":
2213 message += "ERROR: "
2214 message += last_warning
2215 message += "\n"
2216 elif m.group(2) in [
2217 "ERROR: the following job(s) failed:",
2218 "ERROR: the following Node(s) failed:",
2219 ]:
2220 pass
2221 else:
2222 message += m.group(2)
2223 message += "\n"
2225 if last_submit_failed:
2226 message += f"Warn: Job submission issues (last: {last_submit_failed})"
2227 except (OSError, PermissionError):
2228 message = f"Warn: Could not read dagman output file from {wms_path}."
2229 _LOG.debug("dag output file message: %s", message)
2230 return message
2233def _read_rescue_headers(infh: TextIO) -> tuple[list[str], list[str]]:
2234 """Read header lines from a rescue file.
2236 Parameters
2237 ----------
2238 infh : `TextIO`
2239 The rescue file from which to read the header lines.
2241 Returns
2242 -------
2243 header_lines : `list` [`str`]
2244 Header lines read from the rescue file.
2245 failed_subdags : `list` [`str`]
2246 Names of failed subdag jobs.
2247 """
2248 header_lines: list[str] = []
2249 failed = False
2250 failed_subdags: list[str] = []
2252 for line in infh:
2253 line = line.strip()
2254 if line.startswith("#"):
2255 if line.startswith("# Nodes that failed:"):
2256 failed = True
2257 header_lines.append(line)
2258 elif failed:
2259 orig_failed_nodes = line[1:].strip().split(",")
2260 new_failed_nodes = []
2261 for node in orig_failed_nodes:
2262 if node.startswith("wms_check_status"):
2263 group_node = node[17:]
2264 failed_subdags.append(group_node)
2265 new_failed_nodes.append(group_node)
2266 else:
2267 new_failed_nodes.append(node)
2268 header_lines.append(f"# {','.join(new_failed_nodes)}")
2269 if orig_failed_nodes[-1] == "<ENDLIST>":
2270 failed = False
2271 else:
2272 header_lines.append(line)
2273 elif line.strip() == "": # end of headers
2274 break
2275 return header_lines, failed_subdags
2278def _write_rescue_headers(header_lines: list[str], failed_subdags: list[str], outfh: TextIO) -> None:
2279 """Write the header lines to the new rescue file.
2281 Parameters
2282 ----------
2283 header_lines : `list` [`str`]
2284 Header lines to write to the new rescue file.
2285 failed_subdags : `list` [`str`]
2286 Job names of the failed subdags.
2287 outfh : `TextIO`
2288 New rescue file.
2289 """
2290 done_str = "# Nodes premarked DONE"
2291 pattern = f"^{done_str}:\\s+(\\d+)"
2292 for header_line in header_lines:
2293 m = re.match(pattern, header_line)
2294 if m:
2295 print(f"{done_str}: {int(m.group(1)) - len(failed_subdags)}", file=outfh)
2296 else:
2297 print(header_line, file=outfh)
2299 print("", file=outfh)
2302def _copy_done_lines(failed_subdags: list[str], infh: TextIO, outfh: TextIO) -> None:
2303 """Copy the DONE lines from the original rescue file skipping
2304 the failed group jobs.
2306 Parameters
2307 ----------
2308 failed_subdags : `list` [`str`]
2309 List of job names for the failed subdags
2310 infh : `TextIO`
2311 Original rescue file to copy from.
2312 outfh : `TextIO`
2313 New rescue file to copy to.
2314 """
2315 for line in infh:
2316 line = line.strip()
2317 try:
2318 _, node_name = line.split()
2319 except ValueError:
2320 _LOG.error(f"Unexpected line in rescue file = '{line}'")
2321 raise
2322 if node_name not in failed_subdags:
2323 print(line, file=outfh)
2326def _update_rescue_file(rescue_file: Path) -> None:
2327 """Update the subdag failures in the main rescue file
2328 and backup the failed subdag dirs.
2330 Parameters
2331 ----------
2332 rescue_file : `pathlib.Path`
2333 The main rescue file that needs to be updated.
2334 """
2335 # To reduce memory requirements, not reading entire file into memory.
2336 rescue_tmp = rescue_file.with_suffix(rescue_file.suffix + ".tmp")
2337 with open(rescue_file) as infh:
2338 header_lines, failed_subdags = _read_rescue_headers(infh)
2339 with open(rescue_tmp, "w") as outfh:
2340 _write_rescue_headers(header_lines, failed_subdags, outfh)
2341 _copy_done_lines(failed_subdags, infh, outfh)
2342 rescue_file.unlink()
2343 rescue_tmp.rename(rescue_file)
2344 for failed_subdag in failed_subdags:
2345 htc_backup_files(
2346 rescue_file.parent / "subdags" / failed_subdag, subdir=f"backups/subdags/{failed_subdag}"
2347 )
2350def _update_dicts(dict1, dict2):
2351 """Update dict1 with info in dict2.
2353 (Basically an update for nested dictionaries.)
2355 Parameters
2356 ----------
2357 dict1 : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
2358 HTCondor job information to be updated.
2359 dict2 : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
2360 Additional HTCondor job information.
2361 """
2362 for key, value in dict2.items():
2363 if key in dict1 and isinstance(dict1[key], dict) and isinstance(value, dict):
2364 _update_dicts(dict1[key], value)
2365 else:
2366 dict1[key] = value
2369def _locate_schedds(locate_all=False):
2370 """Find out Scheduler daemons in an HTCondor pool.
2372 Parameters
2373 ----------
2374 locate_all : `bool`, optional
2375 If True, all available schedulers in the HTCondor pool will be located.
2376 False by default which means that the search will be limited to looking
2377 for the Scheduler running on a local host.
2379 Returns
2380 -------
2381 schedds : `dict` [`str`, `htcondor.Schedd`]
2382 A mapping between Scheduler names and Python objects allowing for
2383 interacting with them.
2384 """
2385 coll = htcondor.Collector()
2387 schedd_ads = []
2388 if locate_all:
2389 schedd_ads.extend(coll.locateAll(htcondor.DaemonTypes.Schedd))
2390 else:
2391 schedd_ads.append(coll.locate(htcondor.DaemonTypes.Schedd))
2392 return {ad["Name"]: htcondor.Schedd(ad) for ad in schedd_ads}