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