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