Coverage for python/lsst/ctrl/bps/htcondor/lssthtc.py: 81%
937 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-31 09:02 +0000
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_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 contextlib
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 networkx
88from classad2 import parseAds, parseNext
89from htcondor2 import (
90 Collector,
91 DaemonTypes,
92 JobEventLog,
93 JobEventType,
94 JobStatus,
95 Schedd,
96 Submit,
97 param,
98)
100from .handlers import HTC_JOB_AD_HANDLERS
101from .htcondor_compat import HTCondorException, compat_submit_ad
103_LOG = logging.getLogger(__name__)
105MISSING_ID = "-99999"
108class DagStatus(IntEnum):
109 """HTCondor DAGMan's statuses for a DAG."""
111 OK = 0
112 ERROR = 1 # an error condition different than those listed here
113 FAILED = 2 # one or more nodes in the DAG have failed
114 ABORTED = 3 # the DAG has been aborted by an ABORT-DAG-ON specification
115 REMOVED = 4 # the DAG has been removed by condor_rm
116 CYCLE = 5 # a cycle was found in the DAG
117 SUSPENDED = 6 # the DAG has been suspended (see section 2.10.8)
120class NodeStatus(IntEnum):
121 """HTCondor's statuses for DAGman nodes."""
123 # (STATUS_NOT_READY): At least one parent has not yet finished or the node
124 # is a FINAL node.
125 NOT_READY = 0
127 # (STATUS_READY): All parents have finished, but the node is not yet
128 # running.
129 READY = 1
131 # (STATUS_PRERUN): The node’s PRE script is running.
132 PRERUN = 2
134 # (STATUS_SUBMITTED): The node’s HTCondor job(s) are in the queue.
135 # StatusDetails = "not_idle" -> running.
136 # JobProcsHeld = 1-> hold.
137 # JobProcsQueued = 1 -> idle.
138 SUBMITTED = 3
140 # (STATUS_POSTRUN): The node’s POST script is running.
141 POSTRUN = 4
143 # (STATUS_DONE): The node has completed successfully.
144 DONE = 5
146 # (STATUS_ERROR): The node has failed. StatusDetails has info (e.g.,
147 # ULOG_JOB_ABORTED for deleted job).
148 ERROR = 6
150 # (STATUS_FUTILE): The node will never run because ancestor node failed.
151 FUTILE = 7
154class WmsNodeType(IntEnum):
155 """HTCondor plugin node types to help with payload reporting."""
157 UNKNOWN = auto()
158 """Dummy value when missing."""
160 PAYLOAD = auto()
161 """Payload job."""
163 FINAL = auto()
164 """Final job."""
166 SERVICE = auto()
167 """Service job."""
169 NOOP = auto()
170 """NOOP job used for ordering jobs."""
172 SUBDAG = auto()
173 """SUBDAG job used for ordering jobs."""
175 SUBDAG_CHECK = auto()
176 """Job used to correctly prune jobs after a subdag."""
179HTC_QUOTE_KEYS = {"environment", "arguments"}
180HTC_VALID_JOB_KEYS = {
181 "universe",
182 "executable",
183 "arguments",
184 "environment",
185 "log",
186 "error",
187 "output",
188 "should_transfer_files",
189 "when_to_transfer_output",
190 "getenv",
191 "notification",
192 "notify_user",
193 "concurrency_limit",
194 "transfer_executable",
195 "transfer_input_files",
196 "transfer_output_files",
197 "transfer_output_remaps",
198 "request_cpus",
199 "request_memory",
200 "request_disk",
201 "priority",
202 "category",
203 "requirements",
204 "on_exit_hold",
205 "on_exit_hold_reason",
206 "on_exit_hold_subcode",
207 "max_retries",
208 "retry_until",
209 "periodic_release",
210 "periodic_remove",
211 "accounting_group",
212 "accounting_group_user",
213 "kill_sig",
214 "want_graceful_removal",
215 "job_max_vacate_time",
216}
217HTC_VALID_JOB_DAG_KEYS = {
218 "dir",
219 "noop",
220 "done",
221 "vars",
222 "pre",
223 "post",
224 "retry",
225 "retry_unless_exit",
226 "abort_dag_on",
227 "abort_exit",
228 "priority",
229}
232class RestrictedDict(MutableMapping):
233 """A dictionary that only allows certain keys.
235 Parameters
236 ----------
237 valid_keys : `~collections.abc.Container`
238 Strings that are valid keys.
239 init_data : `dict` or `RestrictedDict`, optional
240 Initial data.
242 Raises
243 ------
244 KeyError
245 If invalid key(s) in init_data.
246 """
248 def __init__(self, valid_keys, init_data=()):
249 self.valid_keys = valid_keys
250 self.data = {}
251 self.update(init_data)
253 def __getitem__(self, key):
254 """Return value for given key if exists.
256 Parameters
257 ----------
258 key : `str`
259 Identifier for value to return.
261 Returns
262 -------
263 value : `~typing.Any`
264 Value associated with given key.
266 Raises
267 ------
268 KeyError
269 If key doesn't exist.
270 """
271 return self.data[key]
273 def __delitem__(self, key):
274 """Delete value for given key if exists.
276 Parameters
277 ----------
278 key : `str`
279 Identifier for value to delete.
281 Raises
282 ------
283 KeyError
284 If key doesn't exist.
285 """
286 del self.data[key]
288 def __setitem__(self, key, value):
289 """Store key,value in internal dict only if key is valid.
291 Parameters
292 ----------
293 key : `str`
294 Identifier to associate with given value.
295 value : `~typing.Any`
296 Value to store.
298 Raises
299 ------
300 KeyError
301 If key is invalid.
302 """
303 if key not in self.valid_keys: 303 ↛ 304line 303 didn't jump to line 304 because the condition on line 303 was never true
304 raise KeyError(f"Invalid key {key}")
305 self.data[key] = value
307 def __iter__(self):
308 return self.data.__iter__()
310 def __len__(self):
311 return len(self.data)
313 def __str__(self):
314 return str(self.data)
317def htc_backup_files(
318 wms_path: str | os.PathLike,
319 subdir: str | os.PathLike | None = None,
320 limit: int = 100,
321 failed_subdags: set[str] | None = None,
322) -> None:
323 """Backup select HTCondor files in the submit directory.
325 Files will be saved in separate subdirectories which will be created in
326 the submit directory where the files are located. These subdirectories
327 will be consecutive, zero-padded integers. Their values will correspond to
328 the number of HTCondor rescue DAGs in the submit directory.
330 Hence, with the default settings, copies after the initial failed run will
331 be placed in '001' subdirectory, '002' after the first restart, and so on
332 until the limit of backups is reached. If there's no rescue DAG yet, files
333 will be copied to '000' subdirectory.
335 This is not a generic function for making backups. It is intended to be
336 used once, just before a restart, to make snapshots of files which will be
337 overwritten by HTCondor after during the next run.
339 Parameters
340 ----------
341 wms_path : `str` or `os.PathLike`
342 Path to the submit directory either absolute or relative.
343 subdir : `str` or `os.PathLike`, optional
344 A path, relative to the submit directory, where all subdirectories with
345 backup files will be kept. Defaults to None which means that the backup
346 subdirectories will be placed directly in the submit directory.
347 limit : `int`, optional
348 Maximal number of backups. If the number of backups reaches the limit,
349 the last backup files will be overwritten. The default value is 100
350 to match the default value of HTCondor's DAGMAN_MAX_RESCUE_NUM in
351 version 8.8+.
352 failed_subdags : `set` [`str`], optional
353 Names of subdag jobs that failed. Only files of these subdags will be
354 backed up. If None (the default), files of all subdags with rescue
355 files are backed up.
357 Raises
358 ------
359 FileNotFoundError
360 If the submit directory or the file that needs to be backed up does not
361 exist.
362 OSError
363 If the submit directory cannot be accessed or backing up a file failed
364 either due to permission or filesystem related issues.
365 """
366 width = len(str(limit))
368 path = Path(wms_path).resolve()
369 if not path.is_dir():
370 raise FileNotFoundError(f"Directory {path} not found")
372 # Initialize the backup counter.
373 # If using control DAG, don't want to include nested DAGs.
374 rescue_dags = list(path.glob("*_ctrl.dag.rescue[0-9][0-9][0-9]"))
375 if not rescue_dags: 375 ↛ 377line 375 didn't jump to line 377 because the condition on line 375 was always true
376 rescue_dags = list(path.glob("*.rescue[0-9][0-9][0-9]"))
377 counter = min(len(rescue_dags), limit)
379 # Create the backup directory and move select files there.
380 dest = path
381 if subdir:
382 # PurePath.is_relative_to() is not available before Python 3.9. Hence
383 # we need to check is 'subdir' is in the submit directory in some other
384 # way if it is an absolute path.
385 subdir = Path(subdir)
386 if subdir.is_absolute():
387 subdir = subdir.resolve() # Since resolve was run on path, must run it here
388 if dest not in subdir.parents:
389 _LOG.warning(
390 "Invalid backup location: '%s' not in the submit directory, will use '%s' instead.",
391 subdir,
392 wms_path,
393 )
394 else:
395 dest /= subdir
396 else:
397 dest /= subdir
398 dest /= f"{counter:0{width}}"
399 _LOG.debug("dest = %s", dest)
400 try:
401 dest.mkdir(parents=True, exist_ok=False if counter < limit else True)
402 except FileExistsError:
403 _LOG.warning("Refusing to do backups: target directory '%s' already exists", dest)
404 else:
405 htc_backup_files_single_path(path, dest)
407 # Back up selected files for failed subdags as well.
408 #
409 # Do NOT back up files of subdags that succeeded! These files need to stay
410 # in their respective directories as they will not be recreated by HTCondor
411 # after the run is restarted. As HTCondorService.report() uses information
412 # in these files to determine job statuses, their absence may lead to
413 # reporting incorrect job status counts.
414 for subdag_dir in {file.parent for file in path.glob("subdags/*/*.rescue*")}:
415 if failed_subdags is not None and subdag_dir.name not in failed_subdags: 415 ↛ 416line 415 didn't jump to line 416 because the condition on line 415 was never true
416 continue
417 subdag_dest = dest / subdag_dir.relative_to(path)
418 subdag_dest.mkdir(parents=True, exist_ok=False)
419 htc_backup_files_single_path(subdag_dir, subdag_dest)
422def htc_backup_files_single_path(src: str | os.PathLike, dest: str | os.PathLike) -> None:
423 """Move particular htc files to a different directory for later debugging.
425 Parameters
426 ----------
427 src : `str` or `os.PathLike`
428 Directory from which to back up particular files.
429 dest : `str` or `os.PathLike`
430 Directory to which particular files are moved.
432 Raises
433 ------
434 RuntimeError
435 If given dest directory matches given src directory.
436 OSError
437 If problems moving file.
438 FileNotFoundError
439 Item matching pattern in src directory isn't a file.
440 """
441 src = Path(src)
442 dest = Path(dest)
443 if dest.samefile(src):
444 raise RuntimeError(f"Destination directory is same as the source directory ({src})")
446 for patt in [
447 "*.info.*",
448 "*.dag.metrics",
449 "*.dag.nodes.log",
450 "*.node_status",
451 "wms_*.dag.post.out",
452 "wms_*.status.txt",
453 ]:
454 for source in src.glob(patt):
455 if source.is_file(): 455 ↛ 462line 455 didn't jump to line 462 because the condition on line 455 was always true
456 target = dest / source.relative_to(src)
457 try:
458 source.rename(target)
459 except OSError as exc:
460 raise type(exc)(f"Backing up '{source}' failed: {exc.strerror}") from None
461 else:
462 raise FileNotFoundError(f"Backing up '{source}' failed: not a file")
465def htc_escape(value):
466 """Escape characters in given value based upon HTCondor syntax.
468 Parameters
469 ----------
470 value : `~typing.Any`
471 Value that needs to have characters escaped if string.
473 Returns
474 -------
475 new_value : `~typing.Any`
476 Given value with characters escaped appropriate for HTCondor if string.
477 """
478 if isinstance(value, str):
479 newval = value.replace('"', '""').replace("'", "''").replace(""", '"')
480 else:
481 newval = value
483 return newval
486def htc_write_attribs(stream, attrs):
487 """Write job attributes in HTCondor format to writeable stream.
489 Parameters
490 ----------
491 stream : `~typing.TextIO`
492 Output text stream (typically an open file).
493 attrs : `dict`
494 HTCondor job attributes (dictionary of attribute key, value).
495 """
496 for key, value in attrs.items():
497 # Make sure strings are syntactically correct for HTCondor.
498 if isinstance(value, str): 498 ↛ 501line 498 didn't jump to line 501 because the condition on line 498 was always true
499 pval = f'"{htc_escape(value)}"'
500 else:
501 pval = value
503 print(f"+{key} = {pval}", file=stream)
506def htc_write_condor_file(
507 filename: str | os.PathLike, job_name: str, job: RestrictedDict, job_attrs: dict[str, Any]
508) -> None:
509 """Write an HTCondor submit file.
511 Parameters
512 ----------
513 filename : `str` or `os.PathLike`
514 Filename for the HTCondor submit file.
515 job_name : `str`
516 Job name to use in submit file.
517 job : `RestrictedDict`
518 Submit script information.
519 job_attrs : `dict`
520 Job attributes.
521 """
522 os.makedirs(os.path.dirname(filename), exist_ok=True)
523 with open(filename, "w") as fh:
524 for key, value in job.items():
525 if value is not None: 525 ↛ 524line 525 didn't jump to line 524 because the condition on line 525 was always true
526 if key in HTC_QUOTE_KEYS: # Assumes internal quotes are already escaped correctly
527 print(f'{key}="{value}"', file=fh)
528 else:
529 print(f"{key}={value}", file=fh)
530 for key in ["output", "error", "log"]:
531 if key not in job:
532 filename = f"{job_name}.$(Cluster).{'out' if key != 'log' else key}"
533 print(f"{key}={filename}", file=fh)
535 if job_attrs is not None:
536 htc_write_attribs(fh, job_attrs)
537 print("queue", file=fh)
540def htc_tune_schedd_args(**kwargs):
541 """Ensure that arguments for Schedd are version appropriate.
543 This is the fallback function if no version specific alteration are
544 necessary. Effectively, a no-op.
546 Parameters
547 ----------
548 **kwargs
549 Any keyword arguments that Schedd.history(), Schedd.query(), and
550 Schedd.xquery() accepts.
552 Returns
553 -------
554 kwargs : `dict` [`str`, `~typing.Any`]
555 Keywords arguments that were passed to the function.
557 Notes
558 -----
559 Left as a placeholder for any future changes to the keyword arguments
560 for ``Schedd.(history|query|xquery)``.
561 """
562 return kwargs
565def htc_query_history(schedds, **kwargs):
566 """Fetch history records from the condor_schedd daemon.
568 Parameters
569 ----------
570 schedds : `htcondor.Schedd`
571 HTCondor schedulers which to query for job information.
572 **kwargs
573 Any keyword arguments that Schedd.history() accepts.
575 Yields
576 ------
577 schedd_name : `str`
578 Name of the HTCondor scheduler managing the job queue.
579 job_ad : `dict` [`str`, `~typing.Any`]
580 A dictionary representing HTCondor ClassAd describing a job. It maps
581 job attributes names to values of the ClassAd expressions they
582 represent.
583 """
584 # If not set, provide defaults for positional arguments.
585 kwargs.setdefault("constraint", None)
586 kwargs.setdefault("projection", [])
587 kwargs = htc_tune_schedd_args(**kwargs)
588 for schedd_name, schedd in schedds.items():
589 for job_ad in schedd.history(**kwargs):
590 yield schedd_name, dict(job_ad)
593def htc_query_present(schedds, **kwargs):
594 """Query the condor_schedd daemon for job ads.
596 Parameters
597 ----------
598 schedds : `htcondor.Schedd`
599 HTCondor schedulers which to query for job information.
600 **kwargs
601 Any keyword arguments that Schedd.xquery() accepts.
603 Yields
604 ------
605 schedd_name : `str`
606 Name of the HTCondor scheduler managing the job queue.
607 job_ad : `dict` [`str`, `~typing.Any`]
608 A dictionary representing HTCondor ClassAd describing a job. It maps
609 job attributes names to values of the ClassAd expressions they
610 represent.
611 """
612 kwargs = htc_tune_schedd_args(**kwargs)
613 for schedd_name, schedd in schedds.items():
614 for job_ad in schedd.query(**kwargs):
615 yield schedd_name, dict(job_ad)
618def htc_submit_dag(sub: Submit):
619 """Submit job for execution.
621 Parameters
622 ----------
623 sub : `htcondor.Submit`
624 An object representing a job submit description.
626 Returns
627 -------
628 schedd_job_info : `dict` [`str`, `dict` [`str`, \
629 `dict` [`str`, `~typing.Any`]]]
630 Information about jobs satisfying the search criteria where for each
631 Scheduler, local HTCondor job ids are mapped to their respective
632 classads.
633 """
634 coll = Collector()
635 schedd_ad = coll.locate(DaemonTypes.Schedd)
636 schedd = Schedd(schedd_ad)
638 # Suppress stdout during job submission seen with v24 where the
639 # dagman submit description is dumped to stdout.
640 with contextlib.redirect_stdout(open(os.devnull, "w")): # pragma: no cover
641 # If Schedd.submit() fails, the method will raise an exception.
642 # Usually, that implies issues with the HTCondor pool which BPS
643 # can't address. Hence, no effort is made to handle the exception.
644 submit_result = schedd.submit(sub, itemdata=None)
646 # Sadly, the ClassAd from Schedd.submit() (see above) does not have
647 # 'GlobalJobId' so we need to run a regular query to get it anyway.
648 schedd_name = schedd_ad["Name"]
649 schedd_dag_info = condor_q(
650 constraint=f"ClusterId == {submit_result.cluster()}", schedds={schedd_name: schedd}
651 )
652 return schedd_dag_info
655def htc_create_submit_from_dag(
656 dag_filename: str, submit_options: dict[str, Any], dagman_conf_filename: str | os.PathLike | None = None
657) -> Submit:
658 """Create a DAGMan job submit description.
660 Parameters
661 ----------
662 dag_filename : `str`
663 Name of file containing HTCondor DAG commands.
664 submit_options : `dict` [`str`, `~typing.Any`], optional
665 Contains extra options for command line (Value of None means flag).
666 dagman_conf_filename : `str` or `os.PathLike`, optional
667 Location of DAGMan configuration file, if Any. Defaults to no
668 file (None).
670 Returns
671 -------
672 sub : `htcondor.Submit`
673 An object representing a job submit description.
675 Notes
676 -----
677 Use with HTCondor versions which support htcondor.Submit.from_dag(),
678 i.e., 8.9.3 or newer.
679 """
680 # Config and environment variables do not seem to override -MaxIdle
681 # on the .dag.condor.sub's command line (broken in some 24.0.x versions).
682 # Explicitly forward them as a submit_option if either exists.
683 # Note: auto generated subdag submit files are still the -MaxIdle=1000
684 # in the broken versions.
685 if "MaxIdle" not in submit_options:
686 max_jobs_idle: int | None = None
687 config_var_name = "DAGMAN_MAX_JOBS_IDLE"
689 if dagman_conf_filename:
690 _LOG.debug("Checking DAGMan config file = %s", dagman_conf_filename)
691 with open(dagman_conf_filename) as fh:
692 for line in fh:
693 _LOG.debug("DAGMan config file line = %s", line)
694 parts = line.split("=")
695 _LOG.debug("DAGMan config file line parts = %s", parts)
696 if len(parts) == 2 and parts[0].strip() == config_var_name:
697 max_jobs_idle = int(parts[1].strip())
698 _LOG.debug("Found %s = %s", config_var_name, max_jobs_idle)
699 break
700 if max_jobs_idle is None:
701 if f"_CONDOR_{config_var_name}" in os.environ:
702 max_jobs_idle = int(os.environ[f"_CONDOR_{config_var_name}"])
703 elif config_var_name in param:
704 max_jobs_idle = param[config_var_name]
706 if max_jobs_idle:
707 submit_options["MaxIdle"] = max_jobs_idle
708 else:
709 _LOG.debug("MaxIdle already in submit_options: %s", submit_options)
711 _LOG.debug("Using submit_options = %s", submit_options)
712 submit_ad = Submit.from_dag(dag_filename, submit_options)
713 submit_ad = compat_submit_ad(submit_ad)
714 return submit_ad
717def htc_create_submit_from_cmd(dag_filename, submit_options=None) -> Submit:
718 """Create a DAGMan job submit description.
720 Create a DAGMan job submit description by calling ``condor_submit_dag``
721 on given DAG description file.
723 Parameters
724 ----------
725 dag_filename : `str`
726 Name of file containing HTCondor DAG commands.
727 submit_options : `dict` [`str`, `~typing.Any`], optional
728 Contains extra options for command line (Value of None means flag).
730 Returns
731 -------
732 sub : `htcondor.Submit`
733 An object representing a job submit description.
735 Notes
736 -----
737 Use with HTCondor versions which do not support htcondor.Submit.from_dag(),
738 i.e., older than 8.9.3.
739 """
740 # Run command line condor_submit_dag command.
741 cmd = "condor_submit_dag -f -no_submit -notification never -autorescue 1 -UseDagDir -no_recurse "
743 if submit_options is not None:
744 for opt, val in submit_options.items():
745 cmd += f" -{opt} {val or ''}"
746 cmd += f"{dag_filename}"
748 process = subprocess.Popen(
749 cmd.split(), shell=False, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, encoding="utf-8"
750 )
751 process.wait()
753 if process.returncode != 0:
754 print(f"Exit code: {process.returncode}")
755 print(process.communicate()[0])
756 raise RuntimeError("Problems running condor_submit_dag")
758 return htc_create_submit_from_file(f"{dag_filename}.condor.sub")
761def htc_create_submit_from_file(submit_file):
762 """Parse a submission file.
764 Parameters
765 ----------
766 submit_file : `str`
767 Name of the HTCondor submit file.
769 Returns
770 -------
771 sub : `htcondor.Submit`
772 An object representing a job submit description.
773 """
774 descriptors = {}
775 with open(submit_file) as fh:
776 for line in fh:
777 line = line.strip()
778 if not line.startswith("#") and not line == "queue":
779 (key, val) = re.split(r"\s*=\s*", line, maxsplit=1)
780 descriptors[key] = val
782 # Avoid UserWarning: the line 'copy_to_spool = False' was
783 # unused by Submit object. Is it a typo?
784 try:
785 del descriptors["copy_to_spool"]
786 except KeyError:
787 pass
789 submit_ad = Submit(descriptors)
790 submit_ad = compat_submit_ad(submit_ad)
791 return submit_ad
794def _htc_write_job_commands(stream, name, commands, node_type="JOB"):
795 """Output the DAGMan job lines for single job in DAG.
797 Parameters
798 ----------
799 stream : `~typing.TextIO`
800 Writeable text stream (typically an opened file).
801 name : `str`
802 Job name.
803 commands : `RestrictedDict`
804 DAG commands for a job.
805 node_type : `str`, optional
806 Type of DAGMan node (JOB, FINAL, SERVICE). Defaults to "JOB".
807 """
808 # Note: optional pieces of commands include a space at the beginning.
809 # also making sure values aren't empty strings as placeholders.
810 if "pre" in commands and commands["pre"]:
811 defer = ""
812 if "defer" in commands["pre"] and commands["pre"]["defer"]:
813 defer = f" DEFER {commands['pre']['defer']['status']} {commands['pre']['defer']['time']}"
815 debug = ""
816 if "debug" in commands["pre"] and commands["pre"]["debug"]:
817 debug = f" DEBUG {commands['pre']['debug']['filename']} {commands['pre']['debug']['type']}"
819 arguments = ""
820 if "arguments" in commands["pre"] and commands["pre"]["arguments"]:
821 arguments = f" {commands['pre']['arguments']}"
823 executable = commands["pre"]["executable"]
824 print(f"SCRIPT{defer}{debug} PRE {name} {executable}{arguments}", file=stream)
826 if "post" in commands and commands["post"]:
827 defer = ""
828 if "defer" in commands["post"] and commands["post"]["defer"]:
829 defer = f" DEFER {commands['post']['defer']['status']} {commands['post']['defer']['time']}"
831 debug = ""
832 if "debug" in commands["post"] and commands["post"]["debug"]:
833 debug = f" DEBUG {commands['post']['debug']['filename']} {commands['post']['debug']['type']}"
835 arguments = ""
836 if "arguments" in commands["post"] and commands["post"]["arguments"]:
837 arguments = f" {commands['post']['arguments']}"
839 executable = commands["post"]["executable"]
840 print(f"SCRIPT{defer}{debug} POST {name} {executable}{arguments}", file=stream)
842 if "vars" in commands and commands["vars"]:
843 for key, value in commands["vars"].items():
844 print(f'VARS {name} {key}="{htc_escape(value)}"', file=stream)
846 if "pre_skip" in commands and commands["pre_skip"]:
847 print(f"PRE_SKIP {name} {commands['pre_skip']}", file=stream)
849 # FINAL node cannot have a DAGMan retry, abort-dag-on, priority, category
850 if node_type != "FINAL":
851 if "retry" in commands and commands["retry"]:
852 print(f"RETRY {name} {commands['retry']}", end="", file=stream)
853 if "retry_unless_exit" in commands and commands["retry_unless_exit"]:
854 print(f" UNLESS-EXIT {commands['retry_unless_exit']}", end="", file=stream)
855 print("", file=stream) # Since previous prints don't include new line
857 if "abort_dag_on" in commands and commands["abort_dag_on"]:
858 print(
859 f"ABORT-DAG-ON {name} {commands['abort_dag_on']['node_exit']}"
860 f" RETURN {commands['abort_dag_on']['abort_exit']}",
861 file=stream,
862 )
864 if "priority" in commands and commands["priority"]:
865 print(
866 f"PRIORITY {name} {commands['priority']}",
867 file=stream,
868 )
871class HTCJob:
872 """HTCondor job for use in building DAG.
874 Parameters
875 ----------
876 name : `str`
877 Name of the job.
878 label : `str`
879 Label that can used for grouping or lookup.
880 initcmds : `RestrictedDict`
881 Initial job commands for submit file.
882 initdagcmds : `RestrictedDict`
883 Initial commands for job inside DAG.
884 initattrs : `dict`
885 Initial dictionary of job attributes.
886 """
888 def __init__(self, name, label=None, initcmds=(), initdagcmds=(), initattrs=None):
889 self.name = name
890 self.label = label
891 self.cmds = RestrictedDict(HTC_VALID_JOB_KEYS, initcmds)
892 self.dagcmds = RestrictedDict(HTC_VALID_JOB_DAG_KEYS, initdagcmds)
893 self.attrs = initattrs
894 self.subfile = None
895 self.subdir = None
896 self.subdag = None
898 def __str__(self):
899 return self.name
901 def add_job_cmds(self, new_commands):
902 """Add commands to Job (overwrite existing).
904 Parameters
905 ----------
906 new_commands : `dict`
907 Submit file commands to be added to Job.
908 """
909 self.cmds.update(new_commands)
911 def add_dag_cmds(self, new_commands):
912 """Add DAG commands to Job (overwrite existing).
914 Parameters
915 ----------
916 new_commands : `dict`
917 DAG file commands to be added to Job.
918 """
919 self.dagcmds.update(new_commands)
921 def add_job_attrs(self, new_attrs):
922 """Add attributes to Job (overwrite existing).
924 Parameters
925 ----------
926 new_attrs : `dict`
927 Attributes to be added to Job.
928 """
929 if self.attrs is None:
930 self.attrs = {}
931 if new_attrs:
932 self.attrs.update(new_attrs)
934 def write_submit_file(self, submit_path: str | os.PathLike) -> None:
935 """Write job description to submit file.
937 Parameters
938 ----------
939 submit_path : `str` or `os.PathLike`
940 Prefix path for the submit file.
941 """
942 if not self.subfile:
943 self.subfile = f"{self.name}.sub"
945 subfile = self.subfile
946 if self.subdir:
947 subfile = Path(self.subdir) / subfile
949 subfile = Path(os.path.expandvars(subfile))
950 if not subfile.is_absolute():
951 subfile = Path(submit_path) / subfile
952 if not subfile.exists():
953 _LOG.debug("Writing subfile: %s", subfile)
954 htc_write_condor_file(subfile, self.name, self.cmds, self.attrs)
955 else:
956 _LOG.debug("Using existing subfile: %s", subfile)
958 def write_dag_commands(self, stream, dag_rel_path, command_name="JOB"):
959 """Write DAG commands for single job to output stream.
961 Parameters
962 ----------
963 stream : `~typing.TextIO`
964 Output Stream.
965 dag_rel_path : `str`
966 Relative path of dag to submit directory.
967 command_name : `str`
968 Name of the DAG command (e.g., JOB, FINAL).
969 """
970 subfile = os.path.expandvars(self.subfile)
972 # JOB NodeName SubmitDescription [DIR directory] [NOOP] [DONE]
973 job_line = f'{command_name} {self.name} "{subfile}"'
974 if "dir" in self.dagcmds:
975 dir_val = self.dagcmds["dir"]
976 if dag_rel_path:
977 dir_val = os.path.join(dag_rel_path, dir_val)
978 job_line += f' DIR "{dir_val}"'
979 if self.dagcmds.get("noop", False):
980 job_line += " NOOP"
982 print(job_line, file=stream)
983 if self.dagcmds:
984 _htc_write_job_commands(stream, self.name, self.dagcmds, command_name)
986 def dump(self, fh):
987 """Dump job information to output stream.
989 Parameters
990 ----------
991 fh : `~typing.TextIO`
992 Output stream.
993 """
994 printer = pprint.PrettyPrinter(indent=4, stream=fh)
995 printer.pprint(self.name)
996 printer.pprint(self.cmds)
997 printer.pprint(self.attrs)
1000class HTCDag(networkx.DiGraph):
1001 """HTCondor DAG.
1003 Parameters
1004 ----------
1005 data : `~typing.Any`
1006 Initial graph data of any format that is supported
1007 by the to_network_graph() function.
1008 name : `str`
1009 Name for DAG.
1010 """
1012 def __init__(self, data=None, name=""):
1013 super().__init__(data=data, name=name)
1015 self.graph["attr"] = {}
1016 self.graph["run_id"] = None
1017 self.graph["submit_path"] = None
1018 self.graph["final_job"] = None
1019 self.graph["service_job"] = None
1020 self.graph["submit_options"] = {}
1022 def __str__(self):
1023 """Represent basic DAG info as string.
1025 Returns
1026 -------
1027 info : `str`
1028 String containing basic DAG info.
1029 """
1030 return f"{self.graph['name']} {len(self)}"
1032 def add_attribs(self, attribs=None):
1033 """Add attributes to the DAG.
1035 Parameters
1036 ----------
1037 attribs : `dict`
1038 DAG attributes.
1039 """
1040 if attribs is not None: 1040 ↛ exitline 1040 didn't return from function 'add_attribs' because the condition on line 1040 was always true
1041 self.graph["attr"].update(attribs)
1043 def add_job(self, job, parent_names=None, child_names=None):
1044 """Add an HTCJob to the HTCDag.
1046 Parameters
1047 ----------
1048 job : `HTCJob`
1049 HTCJob to add to the HTCDag.
1050 parent_names : `~collections.abc.Iterable` [`str`], optional
1051 Names of parent jobs.
1052 child_names : `~collections.abc.Iterable` [`str`], optional
1053 Names of child jobs.
1054 """
1055 assert isinstance(job, HTCJob)
1056 _LOG.debug("Adding job %s to dag", job.name)
1058 # Add dag level attributes to each job
1059 job.add_job_attrs(self.graph["attr"])
1061 self.add_node(job.name, data=job)
1063 if parent_names is not None: 1063 ↛ 1064line 1063 didn't jump to line 1064 because the condition on line 1063 was never true
1064 self.add_job_relationships(parent_names, [job.name])
1066 if child_names is not None: 1066 ↛ 1067line 1066 didn't jump to line 1067 because the condition on line 1066 was never true
1067 self.add_job_relationships(child_names, [job.name])
1069 def add_job_relationships(self, parents, children):
1070 """Add DAG edge between parents and children jobs.
1072 Parameters
1073 ----------
1074 parents : `list` [`str`]
1075 Contains parent job name(s).
1076 children : `list` [`str`]
1077 Contains children job name(s).
1078 """
1079 self.add_edges_from(itertools.product(parents, children))
1081 def add_final_job(self, job):
1082 """Add an HTCJob for the FINAL job in HTCDag.
1084 Parameters
1085 ----------
1086 job : `HTCJob`
1087 HTCJob to add to the HTCDag as a FINAL job.
1088 """
1089 # Add dag level attributes to each job
1090 job.add_job_attrs(self.graph["attr"])
1092 self.graph["final_job"] = job
1094 def add_service_job(self, job):
1095 """Add an HTCJob for the SERVICE job in HTCDag.
1097 Parameters
1098 ----------
1099 job : `HTCJob`
1100 HTCJob to add to the HTCDag as a SERVICE job.
1101 """
1102 # Add dag level attributes to each job
1103 job.add_job_attrs(self.graph["attr"])
1105 self.graph["service_job"] = job
1107 def del_job(self, job_name):
1108 """Delete the job from the DAG.
1110 Parameters
1111 ----------
1112 job_name : `str`
1113 Name of job in DAG to delete.
1114 """
1115 # Reconnect edges around node to delete
1116 parents = self.predecessors(job_name)
1117 children = self.successors(job_name)
1118 self.add_edges_from(itertools.product(parents, children))
1120 # Delete job node (which deletes its edges).
1121 self.remove_node(job_name)
1123 def write(self, submit_path, job_subdir="", dag_subdir="", dag_rel_path=""):
1124 """Write DAG to a file.
1126 Parameters
1127 ----------
1128 submit_path : `str`
1129 Prefix path for all outputs.
1130 job_subdir : `str`, optional
1131 Template for job subdir (submit_path + job_subdir).
1132 dag_subdir : `str`, optional
1133 DAG subdir (submit_path + dag_subdir).
1134 dag_rel_path : `str`, optional
1135 Prefix to job_subdir for jobs inside subdag.
1136 """
1137 self.graph["submit_path"] = submit_path
1138 self.graph["dag_filename"] = os.path.join(dag_subdir, f"{self.graph['name']}.dag")
1139 full_filename = os.path.join(submit_path, self.graph["dag_filename"])
1140 os.makedirs(os.path.dirname(full_filename), exist_ok=True)
1142 try:
1143 dagman_config_path = Path(self.graph["attr"]["bps_wms_config_path"])
1144 except KeyError:
1145 dagman_config_path = None
1146 with open(full_filename, "w") as fh:
1147 if dagman_config_path is not None:
1148 fh.write(f"CONFIG {dag_rel_path / dagman_config_path}\n")
1150 for name, nodeval in self.nodes().items():
1151 try:
1152 job = nodeval["data"]
1153 except KeyError:
1154 _LOG.error("Job %s doesn't have data (keys: %s).", name, nodeval.keys())
1155 raise
1156 if job.subdag: 1156 ↛ 1157line 1156 didn't jump to line 1157 because the condition on line 1156 was never true
1157 if job.subfile:
1158 this_dag_rel_path = ""
1159 else:
1160 this_dag_rel_path = "../.."
1161 dag_subdir = f"subdags/{job.name}"
1162 if "dir" in job.dagcmds:
1163 subdir = job.dagcmds["dir"]
1164 else:
1165 subdir = job_subdir
1166 if dagman_config_path is not None:
1167 job.subdag.add_attribs({"bps_wms_config_path": str(dagman_config_path)})
1168 job.subdag.write(submit_path, subdir, dag_subdir, this_dag_rel_path)
1169 fh.write(f"SUBDAG EXTERNAL {job.name} {Path(job.subdag.graph['dag_filename']).name}")
1170 if dag_subdir:
1171 fh.write(f" DIR {dag_subdir}")
1172 fh.write("\n")
1173 if job.dagcmds:
1174 _htc_write_job_commands(fh, job.name, job.dagcmds)
1175 else:
1176 job.write_submit_file(submit_path)
1177 job.write_dag_commands(fh, dag_rel_path)
1179 for edge in self.edges():
1180 print(f"PARENT {edge[0]} CHILD {edge[1]}", file=fh)
1182 if self.graph.get("write_dot", False):
1183 print(f"DOT {self.name}.dot", file=fh)
1185 print(f"NODE_STATUS_FILE {self.name}.node_status", file=fh)
1187 # Add bps attributes to dag submission
1188 for key, value in self.graph["attr"].items():
1189 print(f'SET_JOB_ATTR {key}= "{htc_escape(value)}"', file=fh)
1191 # Add special nodes if any.
1192 special_jobs = {
1193 "FINAL": self.graph["final_job"],
1194 "SERVICE": self.graph["service_job"],
1195 }
1196 for dagcmd, job in special_jobs.items():
1197 if job is not None:
1198 job.write_submit_file(submit_path)
1199 job.write_dag_commands(fh, dag_rel_path, dagcmd)
1201 def dump(self, fh):
1202 """Dump DAG info to output stream.
1204 Parameters
1205 ----------
1206 fh : `typing.IO`
1207 Where to dump DAG info as text.
1208 """
1209 for key, value in self.graph:
1210 print(f"{key}={value}", file=fh)
1211 for name, data in self.nodes().items():
1212 print(f"{name}:", file=fh)
1213 data.dump(fh)
1214 for edge in self.edges():
1215 print(f"PARENT {edge[0]} CHILD {edge[1]}", file=fh)
1216 if self.graph["final_job"]:
1217 print(f"FINAL {self.graph['final_job'].name}:", file=fh)
1218 self.graph["final_job"].dump(fh)
1220 def write_dot(self, filename):
1221 """Write a dot version of the DAG.
1223 Parameters
1224 ----------
1225 filename : `str`
1226 Name of the dot file.
1227 """
1228 pos = networkx.nx_agraph.graphviz_layout(self)
1229 networkx.draw(self, pos=pos)
1230 networkx.drawing.nx_pydot.write_dot(self, filename)
1233def condor_q(constraint=None, schedds=None, **kwargs):
1234 """Get information about the jobs in the HTCondor job queue(s).
1236 Parameters
1237 ----------
1238 constraint : `str`, optional
1239 Constraints to be passed to job query.
1240 schedds : `dict` [`str`, `htcondor.Schedd`], optional
1241 HTCondor schedulers which to query for job information. If None
1242 (default), the query will be run against local scheduler only.
1243 **kwargs : `~typing.Any`
1244 Additional keyword arguments that need to be passed to the internal
1245 query method.
1247 Returns
1248 -------
1249 job_info : `dict` [`str`, `dict` [`str`, `dict` [`str`, `~typing.Any`]]]
1250 Information about jobs satisfying the search criteria where for each
1251 Scheduler, local HTCondor job ids are mapped to their respective
1252 classads.
1253 """
1254 return condor_query(constraint, schedds, htc_query_present, **kwargs)
1257def condor_history(constraint=None, schedds=None, **kwargs):
1258 """Get information about the jobs from HTCondor history records.
1260 Parameters
1261 ----------
1262 constraint : `str`, optional
1263 Constraints to be passed to job query.
1264 schedds : `dict` [`str`, `htcondor.Schedd`], optional
1265 HTCondor schedulers which to query for job information. If None
1266 (default), the query will be run against the history file of
1267 the local scheduler only.
1268 **kwargs : `~typing.Any`
1269 Additional keyword arguments that need to be passed to the internal
1270 query method.
1272 Returns
1273 -------
1274 job_info : `dict` [`str`, `dict` [`str`, `dict` [`str`, `~typing.Any`]]]
1275 Information about jobs satisfying the search criteria where for each
1276 Scheduler, local HTCondor job ids are mapped to their respective
1277 classads.
1278 """
1279 return condor_query(constraint, schedds, htc_query_history, **kwargs)
1282def condor_query(constraint=None, schedds=None, query_func=htc_query_present, **kwargs):
1283 """Get information about HTCondor jobs.
1285 Parameters
1286 ----------
1287 constraint : `str`, optional
1288 Constraints to be passed to job query.
1289 schedds : `dict` [`str`, `htcondor.Schedd`], optional
1290 HTCondor schedulers which to query for job information. If None
1291 (default), the query will be run against the history file of
1292 the local scheduler only.
1293 query_func : `~collections.abc.Callable`
1294 An query function which takes following arguments:
1296 - ``schedds``: Schedulers to query (`list` [`htcondor.Schedd`]).
1297 - ``**kwargs``: Keyword arguments that will be passed to the query
1298 function.
1299 **kwargs : `~typing.Any`
1300 Additional keyword arguments that need to be passed to the query
1301 method.
1303 Returns
1304 -------
1305 job_info : `dict` [`str`, `dict` [`str`, `dict` [`str`, `~typing.Any`]]]
1306 Information about jobs satisfying the search criteria where for each
1307 Scheduler, local HTCondor job ids are mapped to their respective
1308 classads.
1309 """
1310 if not schedds:
1311 coll = Collector()
1312 schedd_ad = coll.locate(DaemonTypes.Schedd)
1313 schedds = {schedd_ad["Name"]: Schedd(schedd_ad)}
1315 # Make sure that 'ClusterId' and 'ProcId' attributes are always included
1316 # in the job classad. They are needed to construct the job id.
1317 added_attrs = set()
1318 if "projection" in kwargs and kwargs["projection"]:
1319 requested_attrs = set(kwargs["projection"])
1320 required_attrs = {"ClusterId", "ProcId"}
1321 added_attrs = required_attrs - requested_attrs
1322 for attr in added_attrs:
1323 kwargs["projection"].append(attr)
1325 unwanted_attrs = {"Env", "Environment"} | added_attrs
1326 job_info = defaultdict(dict)
1327 for schedd_name, job_ad in query_func(schedds, constraint=constraint, **kwargs):
1328 id_ = f"{job_ad['ClusterId']}.{job_ad['ProcId']}"
1329 for attr in set(job_ad) & unwanted_attrs:
1330 del job_ad[attr]
1331 job_info[schedd_name][id_] = job_ad
1332 _LOG.debug("query returned %d jobs", sum(len(val) for val in job_info.values()))
1334 # Restore the list of the requested attributes to its original value
1335 # if needed.
1336 if added_attrs:
1337 for attr in added_attrs:
1338 kwargs["projection"].remove(attr)
1340 # When returning the results filter out entries for schedulers with no jobs
1341 # matching the search criteria.
1342 return {key: val for key, val in job_info.items() if val}
1345def condor_search(
1346 constraint: str | None = None,
1347 hist: float | None = None,
1348 schedds: dict[str, dict[str, Schedd]] | None = None,
1349 skip_history: bool = False,
1350) -> dict[str, dict[str, dict[str, Any]]]:
1351 """Search for running and finished jobs satisfying given criteria.
1353 Parameters
1354 ----------
1355 constraint : `str`, optional
1356 Constraints to be passed to job query.
1357 hist : `float`
1358 Limit history search to this many days.
1359 schedds : `dict` [`str`, `htcondor.Schedd`], optional
1360 The list of the HTCondor schedulers which to query for job information.
1361 If None (default), only the local scheduler will be queried.
1362 skip_history : `bool`, optional
1363 If True, skip searching history if searching the queue returns any
1364 information. Defaults to False.
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 constraint is None:
1374 constraint = ""
1376 if not schedds:
1377 coll = Collector()
1378 schedd_ad = coll.locate(DaemonTypes.Schedd)
1379 schedds = {schedd_ad["Name"]: Schedd(location=schedd_ad)}
1381 job_info = condor_q(constraint=constraint, schedds=schedds)
1382 if hist is not None and (not skip_history or not job_info):
1383 _LOG.debug("Searching history going back %s days", hist)
1384 epoch = (datetime.now() - timedelta(days=hist)).timestamp()
1385 constraint += f" && (CompletionDate >= {epoch} || JobFinishedHookDone >= {epoch})"
1386 hist_info = condor_history(constraint, schedds=schedds)
1387 update_job_info(job_info, hist_info)
1388 return job_info
1391def condor_status(constraint=None, coll=None):
1392 """Get information about HTCondor pool.
1394 Parameters
1395 ----------
1396 constraint : `str`, optional
1397 Constraints to be passed to the query.
1398 coll : `htcondor.Collector`, optional
1399 Object representing HTCondor collector daemon.
1401 Returns
1402 -------
1403 pool_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1404 Mapping between HTCondor slot names and slot information (classAds).
1405 """
1406 if coll is None:
1407 coll = Collector()
1408 try:
1409 pool_ads = coll.query(constraint=constraint)
1410 except HTCondorException as ex:
1411 raise RuntimeError(f"Problem querying the Collector. (Constraint='{constraint}')") from ex
1413 pool_info = {}
1414 for slot in pool_ads:
1415 pool_info[slot["name"]] = dict(slot)
1416 _LOG.debug("condor_status returned %d ads", len(pool_info))
1417 return pool_info
1420def update_job_info(job_info, other_info):
1421 """Update results of a job query with results from another query.
1423 Parameters
1424 ----------
1425 job_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1426 Results of the job query that needs to be updated.
1427 other_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1428 Results of the other job query.
1430 Returns
1431 -------
1432 job_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1433 The updated results.
1434 """
1435 for schedd_name, others in other_info.items():
1436 try:
1437 jobs = job_info[schedd_name]
1438 except KeyError:
1439 job_info[schedd_name] = others
1440 else:
1441 for id_, ad in others.items():
1442 jobs.setdefault(id_, {}).update(ad)
1443 return job_info
1446def count_jobs_in_single_dag(
1447 filename: str | os.PathLike,
1448) -> tuple[Counter[str], dict[str, str], dict[str, WmsNodeType]]:
1449 """Build bps_run_summary string from dag file.
1451 Parameters
1452 ----------
1453 filename : `str`
1454 Path that includes dag file for a run.
1456 Returns
1457 -------
1458 counts : `Counter` [`str`]
1459 Semi-colon separated list of job labels and counts.
1460 (Same format as saved in dag classad).
1461 job_name_to_label : `dict` [`str`, `str`]
1462 Mapping of job names to job labels.
1463 job_name_to_type : `dict` [`str`, `lsst.ctrl.bps.htcondor.WmsNodeType`]
1464 Mapping of job names to job types
1465 (e.g., payload, final, service).
1466 """
1467 # Later code depends upon insertion order
1468 counts: Counter = Counter() # counts of payload jobs per label
1469 job_name_to_label: dict[str, str] = {}
1470 job_name_to_type: dict[str, WmsNodeType] = {}
1471 with open(filename) as fh:
1472 for line in fh:
1473 # Skip any line that contains commands irrelevant to job counting.
1474 if not line.startswith(
1475 (
1476 "JOB",
1477 "FINAL",
1478 "SERVICE",
1479 "SUBDAG EXTERNAL",
1480 )
1481 ):
1482 continue
1484 m = re.match(
1485 r"(?P<command>JOB|FINAL|SERVICE|SUBDAG EXTERNAL)\s+"
1486 r'(?P<jobname>(?P<wms>wms_)?\S+)\s+"?(?P<subfile>\S+)"?\s*'
1487 r'(DIR "?(?P<dir>[^\s"]+)"?)?\s*(?P<noop>NOOP)?',
1488 line,
1489 )
1490 if m: 1490 ↛ 1543line 1490 didn't jump to line 1543 because the condition on line 1490 was always true
1491 job_name = m.group("jobname")
1492 job_type = WmsNodeType.UNKNOWN
1493 name_parts = job_name.split("_")
1495 label = ""
1496 if m.group("dir"):
1497 dir_match = re.search(r"jobs/([^\s/]+)", m.group("dir"))
1498 if dir_match:
1499 label = dir_match.group(1)
1500 else:
1501 _LOG.debug("Parse DAG: unparsed dir = %s", line)
1502 elif m.group("subfile"): 1502 ↛ 1509line 1502 didn't jump to line 1509 because the condition on line 1502 was always true
1503 subfile_match = re.search(r"jobs/([^\s/]+)", m.group("subfile"))
1504 if subfile_match:
1505 label = m.group("subfile").split("/")[1]
1506 else:
1507 label = pegasus_name_to_label(job_name)
1509 match m.group("command"):
1510 case "JOB":
1511 if m.group("noop"):
1512 job_type = WmsNodeType.NOOP
1513 # wms_noop_label
1514 label = name_parts[2]
1515 elif m.group("wms"):
1516 if name_parts[1] == "check": 1516 ↛ 1521line 1516 didn't jump to line 1521 because the condition on line 1516 was always true
1517 job_type = WmsNodeType.SUBDAG_CHECK
1518 # wms_check_status_wms_group_label
1519 label = name_parts[5]
1520 else:
1521 _LOG.warning(
1522 "Unexpected skipping of dag line due to unknown wms job: %s", line
1523 )
1524 else:
1525 job_type = WmsNodeType.PAYLOAD
1526 if label == "init": 1526 ↛ 1527line 1526 didn't jump to line 1527 because the condition on line 1526 was never true
1527 label = "pipetaskInit"
1528 counts[label] += 1
1529 case "FINAL":
1530 job_type = WmsNodeType.FINAL
1531 counts[label] += 1 # final counts a payload job.
1532 case "SERVICE":
1533 job_type = WmsNodeType.SERVICE
1534 case "SUBDAG EXTERNAL": 1534 ↛ 1538line 1534 didn't jump to line 1538 because the pattern on line 1534 always matched
1535 job_type = WmsNodeType.SUBDAG
1536 label = name_parts[2]
1538 job_name_to_label[job_name] = label
1539 job_name_to_type[job_name] = job_type
1540 else:
1541 # The line should, but didn't match the pattern above. Probably
1542 # problems with regex.
1543 _LOG.warning("Unexpected skipping of dag line: %s", line)
1545 return counts, job_name_to_label, job_name_to_type
1548def summarize_dag(dir_name: str) -> tuple[str, dict[str, str], dict[str, WmsNodeType]]:
1549 """Build bps_run_summary string from dag file.
1551 Parameters
1552 ----------
1553 dir_name : `str`
1554 Path that includes dag file for a run.
1556 Returns
1557 -------
1558 summary : `str`
1559 Semi-colon separated list of job labels and counts
1560 (Same format as saved in dag classad).
1561 job_name_to_label : `dict` [`str`, `str`]
1562 Mapping of job names to job labels.
1563 job_name_to_type : `dict` [`str`, `lsst.ctrl.bps.htcondor.WmsNodeType`]
1564 Mapping of job names to job types
1565 (e.g., payload, final, service).
1566 """
1567 # Later code depends upon insertion order
1568 counts: Counter[str] = Counter() # counts of payload jobs per label
1569 job_name_to_label: dict[str, str] = {}
1570 job_name_to_type: dict[str, WmsNodeType] = {}
1571 for filename in Path(dir_name).glob("*.dag"):
1572 single_counts, single_job_name_to_label, single_job_name_to_type = count_jobs_in_single_dag(filename)
1573 counts += single_counts
1574 _update_dicts(job_name_to_label, single_job_name_to_label)
1575 _update_dicts(job_name_to_type, single_job_name_to_type)
1577 for filename in Path(dir_name).glob("subdags/*/*.dag"):
1578 single_counts, single_job_name_to_label, single_job_name_to_type = count_jobs_in_single_dag(filename)
1579 counts += single_counts
1580 _update_dicts(job_name_to_label, single_job_name_to_label)
1581 _update_dicts(job_name_to_type, single_job_name_to_type)
1583 summary = ";".join([f"{name}:{counts[name]}" for name in counts])
1584 _LOG.debug("summarize_dag: %s %s %s", summary, job_name_to_label, job_name_to_type)
1585 return summary, job_name_to_label, job_name_to_type
1588def pegasus_name_to_label(name):
1589 """Convert pegasus job name to a label for the report.
1591 Parameters
1592 ----------
1593 name : `str`
1594 Name of job.
1596 Returns
1597 -------
1598 label : `str`
1599 Label for job.
1600 """
1601 label = "UNK"
1602 if name.startswith("create_dir") or name.startswith("stage_in") or name.startswith("stage_out"): 1602 ↛ 1603line 1602 didn't jump to line 1603 because the condition on line 1602 was never true
1603 label = "pegasus"
1604 else:
1605 m = re.match(r"pipetask_(\d+_)?([^_]+)", name)
1606 if m: 1606 ↛ 1607line 1606 didn't jump to line 1607 because the condition on line 1606 was never true
1607 label = m.group(2)
1608 if label == "init":
1609 label = "pipetaskInit"
1611 return label
1614def read_single_dag_status(filename: str | os.PathLike) -> dict[str, Any]:
1615 """Read the node status file for DAG summary information.
1617 Parameters
1618 ----------
1619 filename : `str` or `Path.pathlib`
1620 Node status filename.
1622 Returns
1623 -------
1624 dag_ad : `dict` [`str`, `~typing.Any`]
1625 DAG summary information.
1626 """
1627 dag_ad: dict[str, Any] = {}
1629 # While this is probably more up to date than dag classad, only read from
1630 # file if need to.
1631 try:
1632 node_stat_file = Path(filename)
1633 _LOG.debug("Reading Node Status File %s", node_stat_file)
1634 with open(node_stat_file) as infh:
1635 dag_ad = dict(parseNext(infh)) # pylint: disable=E1101
1637 if not dag_ad: 1637 ↛ 1639line 1637 didn't jump to line 1639 because the condition on line 1637 was never true
1638 # Pegasus check here
1639 metrics_file = node_stat_file.with_suffix(".dag.metrics")
1640 if metrics_file.exists():
1641 with open(metrics_file) as infh:
1642 metrics = json.load(infh)
1643 dag_ad["NodesTotal"] = metrics.get("jobs", 0)
1644 dag_ad["NodesFailed"] = metrics.get("jobs_failed", 0)
1645 dag_ad["NodesDone"] = metrics.get("jobs_succeeded", 0)
1646 metrics_file = node_stat_file.with_suffix(".metrics")
1647 with open(metrics_file) as infh:
1648 metrics = json.load(infh)
1649 dag_ad["NodesTotal"] = metrics["wf_metrics"]["total_jobs"]
1650 except (OSError, PermissionError):
1651 pass
1653 _LOG.debug("read_dag_status: %s", dag_ad)
1654 return dag_ad
1657def read_dag_status(wms_path: str | os.PathLike) -> dict[str, Any]:
1658 """Read the node status file for DAG summary information.
1660 Parameters
1661 ----------
1662 wms_path : `str` or `os.PathLike`
1663 Path that includes node status file for a run.
1665 Returns
1666 -------
1667 dag_ad : `dict` [`str`, `~typing.Any`]
1668 DAG summary information, counts summed across any subdags.
1669 """
1670 dag_ads: dict[str, Any] = {}
1671 path = Path(wms_path)
1672 try:
1673 node_stat_file = next(path.glob("*.node_status"))
1674 except StopIteration as exc:
1675 raise FileNotFoundError(f"DAGMan node status not found in {wms_path}") from exc
1677 dag_ads = read_single_dag_status(node_stat_file)
1679 for node_stat_file in path.glob("subdags/*/*.node_status"):
1680 dag_ad = read_single_dag_status(node_stat_file)
1681 dag_ads["JobProcsHeld"] += dag_ad.get("JobProcsHeld", 0)
1682 dag_ads["NodesPost"] += dag_ad.get("NodesPost", 0)
1683 dag_ads["JobProcsIdle"] += dag_ad.get("JobProcsIdle", 0)
1684 dag_ads["NodesTotal"] += dag_ad.get("NodesTotal", 0)
1685 dag_ads["NodesFailed"] += dag_ad.get("NodesFailed", 0)
1686 dag_ads["NodesDone"] += dag_ad.get("NodesDone", 0)
1687 dag_ads["NodesQueued"] += dag_ad.get("NodesQueued", 0)
1688 dag_ads["NodesPre"] += dag_ad.get("NodesReady", 0)
1689 dag_ads["NodesFutile"] += dag_ad.get("NodesFutile", 0)
1690 dag_ads["NodesUnready"] += dag_ad.get("NodesUnready", 0)
1692 return dag_ads
1695def read_single_node_status(filename: str | os.PathLike, init_fake_id: int) -> dict[str, Any]:
1696 """Read entire node status file.
1698 Parameters
1699 ----------
1700 filename : `str` or `pathlib.Path`
1701 Node status filename.
1702 init_fake_id : `int`
1703 Initial fake id value.
1705 Returns
1706 -------
1707 jobs : `dict` [`str`, `~typing.Any`]
1708 DAG summary information compiled from the node status file combined
1709 with the information found in the node event log.
1711 Currently, if the same job attribute is found in both files, its value
1712 from the event log takes precedence over the value from the node status
1713 file.
1714 """
1715 filename = Path(filename)
1717 # Get jobid info from other places to fill in gaps in info from node_status
1718 _, job_name_to_label, job_name_to_type = count_jobs_in_single_dag(filename.with_suffix(".dag"))
1719 loginfo: dict[str, dict[str, Any]] = {}
1720 wms_workflow_id = MISSING_ID
1721 try:
1722 wms_workflow_id, _ = read_single_dag_log(filename.with_suffix(".dag.dagman.log"))
1723 loginfo = read_single_dag_nodes_log(filename.with_suffix(".dag.nodes.log"))
1724 except (HTCondorException, OSError, PermissionError):
1725 pass
1727 job_name_to_id: dict[str, str] = {}
1728 _LOG.debug("loginfo = %s", loginfo)
1729 log_job_name_to_id: dict[str, str] = {}
1730 for job_id, job_info in loginfo.items():
1731 if "LogNotes" in job_info: 1731 ↛ 1730line 1731 didn't jump to line 1730 because the condition on line 1731 was always true
1732 m = re.match(r"DAG Node: (\S+)", job_info["LogNotes"])
1733 if m: 1733 ↛ 1730line 1733 didn't jump to line 1730 because the condition on line 1733 was always true
1734 job_name = m.group(1)
1735 log_job_name_to_id[job_name] = job_id
1736 job_info["DAGNodeName"] = job_name
1737 job_info["wms_node_type"] = job_name_to_type[job_name]
1738 job_info["bps_job_label"] = job_name_to_label[job_name]
1740 jobs = {}
1741 fake_id = init_fake_id # For nodes that do not yet have a job id, give fake one
1742 try:
1743 with open(filename) as fh:
1744 for ad in parseAds(fh):
1745 match ad["Type"]:
1746 case "DagStatus":
1747 # Skip DAG summary.
1748 pass
1749 case "NodeStatus":
1750 job_name = ad["Node"]
1751 if job_name in job_name_to_label: 1751 ↛ 1753line 1751 didn't jump to line 1753 because the condition on line 1751 was always true
1752 job_label = job_name_to_label[job_name]
1753 elif "_" in job_name:
1754 job_label = job_name.split("_")[1]
1755 else:
1756 job_label = job_name
1758 job = dict(ad)
1759 if job_name in log_job_name_to_id:
1760 job_id = str(log_job_name_to_id[job_name])
1761 _update_dicts(job, loginfo[job_id])
1762 else:
1763 job_id = str(fake_id)
1764 job = dict(ad)
1765 fake_id -= 1
1766 jobs[job_id] = job
1767 job_name_to_id[job_name] = job_id
1769 # Make job info as if came from condor_q.
1770 job["ClusterId"] = int(float(job_id))
1771 job["DAGManJobID"] = wms_workflow_id
1772 job["DAGNodeName"] = job_name
1773 job["bps_job_label"] = job_label
1774 job["wms_node_type"] = job_name_to_type[job_name]
1776 case "StatusEnd": 1776 ↛ 1779line 1776 didn't jump to line 1779 because the pattern on line 1776 always matched
1777 # Skip node status file "epilog".
1778 pass
1779 case _:
1780 _LOG.debug(
1781 "Ignoring unknown classad type '%s' in the node status file '%s'",
1782 ad["Type"],
1783 filename,
1784 )
1785 except (OSError, PermissionError):
1786 pass
1788 # Check for missing jobs (e.g., submission failure or not submitted yet)
1789 # Use dag info to create job placeholders
1790 for name in set(job_name_to_label) - set(job_name_to_id):
1791 if name in log_job_name_to_id: # job was in nodes.log, but not node_status
1792 job_id = str(log_job_name_to_id[name])
1793 job = dict(loginfo[job_id])
1794 else:
1795 job_id = str(fake_id)
1796 fake_id -= 1
1797 job = {}
1798 job["NodeStatus"] = NodeStatus.NOT_READY
1800 job["ClusterId"] = int(float(job_id))
1801 job["ProcId"] = 0
1802 job["DAGManJobID"] = wms_workflow_id
1803 job["DAGNodeName"] = name
1804 job["bps_job_label"] = job_name_to_label[name]
1805 job["wms_node_type"] = job_name_to_type[name]
1806 jobs[f"{job['ClusterId']}.{job['ProcId']}"] = job
1808 for job_info in jobs.values():
1809 job_info["from_dag_job"] = f"wms_{filename.stem}"
1811 return jobs
1814def read_node_status(wms_path: str | os.PathLike) -> dict[str, dict[str, Any]]:
1815 """Read entire node status file.
1817 Parameters
1818 ----------
1819 wms_path : `str` or `os.PathLike`
1820 Path that includes node status file for a run.
1822 Returns
1823 -------
1824 jobs : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1825 DAG summary information compiled from the node status file combined
1826 with the information found in the node event log.
1828 Currently, if the same job attribute is found in both files, its value
1829 from the event log takes precedence over the value from the node status
1830 file.
1831 """
1832 jobs: dict[str, dict[str, Any]] = {}
1833 init_fake_id = -1
1835 # subdags may not have run so wouldn't have node_status file
1836 # use dag files and let read_single_node_status handle missing
1837 # node_status file.
1838 for dag_filename in Path(wms_path).glob("*.dag"):
1839 filename = dag_filename.with_suffix(".node_status")
1840 info = read_single_node_status(filename, init_fake_id)
1841 init_fake_id -= len(info)
1842 _update_dicts(jobs, info)
1844 for dag_filename in Path(wms_path).glob("subdags/*/*.dag"):
1845 filename = dag_filename.with_suffix(".node_status")
1846 info = read_single_node_status(filename, init_fake_id)
1847 init_fake_id -= len(info)
1848 _update_dicts(jobs, info)
1850 # Propagate pruned from subdags to jobs
1851 name_to_id: dict[str, str] = {}
1852 missing_status: dict[str, list[str]] = {}
1853 for id_, job in jobs.items():
1854 if job["DAGNodeName"].startswith("wms_"):
1855 name_to_id[job["DAGNodeName"]] = id_
1856 if "NodeStatus" not in job or job["NodeStatus"] == NodeStatus.NOT_READY:
1857 missing_status.setdefault(job["from_dag_job"], []).append(id_)
1859 for name, dag_id in name_to_id.items():
1860 dag_status = jobs[dag_id].get("NodeStatus", NodeStatus.NOT_READY)
1861 if dag_status in {NodeStatus.NOT_READY, NodeStatus.FUTILE}:
1862 for id_ in missing_status.get(name, []):
1863 jobs[id_]["NodeStatus"] = dag_status
1865 return jobs
1868def read_single_dag_log(log_filename: str | os.PathLike) -> tuple[str, dict[str, dict[str, Any]]]:
1869 """Read job information from the DAGMan log file.
1871 Parameters
1872 ----------
1873 log_filename : `str` or `os.PathLike`
1874 DAGMan log filename.
1876 Returns
1877 -------
1878 wms_workflow_id : `str`
1879 HTCondor job id (i.e., <ClusterId>.<ProcId>) of the DAGMan job.
1880 dag_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1881 HTCondor job information read from the log file mapped to HTCondor
1882 job id.
1884 Raises
1885 ------
1886 FileNotFoundError
1887 If cannot find DAGMan log in given wms_path.
1888 """
1889 wms_workflow_id = MISSING_ID
1890 dag_info: dict[str, dict[str, Any]] = {}
1892 filename = Path(log_filename)
1893 if filename.exists():
1894 _LOG.debug("dag node log filename: %s", filename)
1896 info: dict[str, Any] = {}
1897 job_event_log = JobEventLog(str(filename))
1898 for event in job_event_log.events(stop_after=0):
1899 id_ = f"{event['Cluster']}.{event['Proc']}"
1900 if id_ not in info:
1901 info[id_] = {}
1902 wms_workflow_id = id_ # taking last job id in case of restarts
1903 _update_dicts(info[id_], event)
1904 info[id_][f"{event.type.name.lower()}_time"] = event["EventTime"]
1906 # only save latest DAG job
1907 dag_info = {wms_workflow_id: info[wms_workflow_id]}
1909 return wms_workflow_id, dag_info
1912def read_dag_log(wms_path: str | os.PathLike) -> tuple[str, dict[str, Any]]:
1913 """Read job information from the DAGMan log file.
1915 Parameters
1916 ----------
1917 wms_path : `str` or `os.PathLike`
1918 Path containing the DAGMan log file.
1920 Returns
1921 -------
1922 wms_workflow_id : `str`
1923 HTCondor job id (i.e., <ClusterId>.<ProcId>) of the DAGMan job.
1924 dag_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1925 HTCondor job information read from the log file mapped to HTCondor
1926 job id.
1928 Raises
1929 ------
1930 FileNotFoundError
1931 If cannot find DAGMan log in given wms_path.
1932 """
1933 wms_workflow_id = MISSING_ID
1934 dag_info: dict[str, dict[str, Any]] = {}
1936 path = Path(wms_path)
1937 if path.exists(): 1937 ↛ 1949line 1937 didn't jump to line 1949 because the condition on line 1937 was always true
1938 # Can be more than one dag file in directory. Assume one
1939 # with lowest ID is main DAG
1940 ids = []
1941 for filename in path.glob("*.dag.dagman.log"):
1942 _LOG.debug("dag log filename: %s", filename)
1943 single_id, single_dag_info = read_single_dag_log(filename)
1944 _update_dicts(dag_info, single_dag_info)
1945 ids.append(single_id)
1946 if ids:
1947 wms_workflow_id = min(ids)
1949 if wms_workflow_id == MISSING_ID:
1950 raise FileNotFoundError(f"DAGMan log not found in {wms_path}")
1952 return wms_workflow_id, dag_info
1955def read_single_dag_nodes_log(filename: str | os.PathLike) -> dict[str, dict[str, Any]]:
1956 """Read job information from the DAGMan nodes log file.
1958 Parameters
1959 ----------
1960 filename : `str` or `os.PathLike`
1961 Path containing the DAGMan nodes log file.
1963 Returns
1964 -------
1965 info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
1966 HTCondor job information read from the log file mapped to HTCondor
1967 job id.
1969 Raises
1970 ------
1971 FileNotFoundError
1972 If cannot find DAGMan node log in given wms_path.
1973 """
1974 _LOG.debug("dag node log filename: %s", filename)
1975 filename = Path(filename)
1977 info: dict[str, dict[str, Any]] = {}
1978 if not filename.exists():
1979 raise FileNotFoundError(f"{filename} does not exist")
1981 try:
1982 job_event_log = JobEventLog(str(filename))
1983 except HTCondorException as ex:
1984 _LOG.error("Problem reading nodes log file (%s): %s", filename, ex)
1985 import traceback
1987 traceback.print_stack()
1988 raise
1989 for event in job_event_log.events(stop_after=0):
1990 _LOG.debug("log event type = %s, keys = %s", event["EventTypeNumber"], event.keys())
1992 try:
1993 id_ = f"{event['Cluster']}.{event['Proc']}"
1994 except KeyError:
1995 _LOG.warning(
1996 "Log event missing ids (DAGNodeName=%s, EventTime=%s, EventTypeNumber=%s)",
1997 event.get("DAGNodeName", "UNK"),
1998 event.get("EventTime", "UNK"),
1999 event.get("EventTypeNumber", "UNK"),
2000 )
2001 else:
2002 if id_ not in info:
2003 info[id_] = {}
2005 # Workaround: Please check to see if still problem in
2006 # future HTCondor versions. Sometimes get a
2007 # JobAbortedEvent for a subdag job after it already
2008 # terminated normally. Seems to happen when using job
2009 # plus subdags.
2010 if ( 2010 ↛ 2014line 2010 didn't jump to line 2014 because the condition on line 2010 was never true
2011 event["EventTypeNumber"] == JobEventType.JOB_ABORTED
2012 and info[id_].get("EventTypeNumber", -1) == JobEventType.JOB_TERMINATED
2013 ):
2014 _LOG.debug("Skipping spurious JobAbortedEvent: %s", dict(event))
2015 elif event["EventTypeNumber"] == 16 and event["DAGNodeName"] == "finalJob":
2016 # FINAL job's post script exit code is special and indicates
2017 # status of DAG instead of just the FINAL job. Save the
2018 # information separately.
2019 info[id_]["post"] = dict(event)
2020 else:
2021 _update_dicts(info[id_], event)
2022 info[id_][f"{event.type.name.lower()}_time"] = event["EventTime"]
2024 return info
2027def read_dag_nodes_log(wms_path: str | os.PathLike) -> dict[str, dict[str, Any]]:
2028 """Read job information from the DAGMan nodes log file.
2030 Parameters
2031 ----------
2032 wms_path : `str` or `os.PathLike`
2033 Path containing the DAGMan nodes log file.
2035 Returns
2036 -------
2037 info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
2038 HTCondor job information read from the log file mapped to HTCondor
2039 job id.
2041 Raises
2042 ------
2043 FileNotFoundError
2044 If cannot find DAGMan node log in given wms_path.
2045 """
2046 info: dict[str, dict[str, Any]] = {}
2047 for filename in Path(wms_path).glob("*.dag.nodes.log"):
2048 _LOG.debug("dag node log filename: %s", filename)
2049 _update_dicts(info, read_single_dag_nodes_log(filename))
2051 # If submitted, the main nodes log file should exist
2052 if not info:
2053 raise FileNotFoundError(f"DAGMan node log not found in {wms_path}")
2055 # Subdags will not have dag nodes log files if they haven't
2056 # started running yet (so missing is not an error).
2057 for filename in Path(wms_path).glob("subdags/*/*.dag.nodes.log"):
2058 _LOG.debug("dag node log filename: %s", filename)
2059 _update_dicts(info, read_single_dag_nodes_log(filename))
2061 return info
2064def read_dag_info(wms_path: str | os.PathLike) -> tuple[Path, dict[str, dict[str, Any]]]:
2065 """Read custom DAGMan job information from the file.
2067 Parameters
2068 ----------
2069 wms_path : `str` or `os.PathLike`
2070 Path containing the file with the DAGMan job info.
2072 Returns
2073 -------
2074 filename : `pathlib.Path`
2075 Name of file containing the dag information.
2076 dag_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
2077 HTCondor job information.
2079 Raises
2080 ------
2081 FileNotFoundError
2082 If cannot find DAGMan job info file in the given location.
2083 """
2084 dag_info: dict[str, dict[str, Any]] = {}
2085 try:
2086 filename = next(Path(wms_path).glob("*.info.json"))
2087 except StopIteration as exc:
2088 raise FileNotFoundError(f"File with DAGMan job information not found in {wms_path}") from exc
2089 _LOG.debug("DAGMan job information filename: %s", filename)
2090 try:
2091 with open(filename) as fh:
2092 dag_info = json.load(fh)
2093 except (OSError, PermissionError) as exc:
2094 _LOG.debug("Retrieving DAGMan job information failed: %s", exc)
2095 return filename, dag_info
2098def write_dag_info(filename: str, schedd_dag_info: dict[str, dict[str, Any]]):
2099 """Write custom job information about DAGMan job.
2101 Parameters
2102 ----------
2103 filename : `str`
2104 Name of the file where the information will be stored. If not given,
2105 creates a filename using bps_run value.
2106 schedd_dag_info : `dict` [`str` `dict` [`str`, `~typing.Any`]]
2107 Information about the DAGMan job.
2108 """
2109 _LOG.debug("schedd_dag_info = %s", schedd_dag_info)
2110 schedd_name, dag_info = next(iter(schedd_dag_info.items()))
2111 dag_id, dag_ad = next(iter(dag_info.items()))
2112 ad = {"ClusterId": dag_ad["ClusterId"], "GlobalJobId": dag_ad["GlobalJobId"]}
2113 ad.update({key: val for key, val in dag_ad.items() if key.startswith("bps")})
2114 try:
2115 with open(filename, "w") as fh:
2116 info = {schedd_name: {dag_id: ad}}
2117 json.dump(info, fh)
2118 except (KeyError, OSError, PermissionError) as exc:
2119 _LOG.debug("Persisting DAGMan job information failed: %s", exc)
2121 return filename
2124def htc_tweak_log_info(wms_path: str | Path, job: dict[str, Any]) -> None:
2125 """Massage the given job info has same structure as if came from condor_q.
2127 Parameters
2128 ----------
2129 wms_path : `str` | `os.PathLike`
2130 Path containing an HTCondor event log file.
2131 job : `dict` [ `str`, `~typing.Any` ]
2132 A mapping between HTCondor job id and job information read from
2133 the log.
2135 Notes
2136 -----
2137 FIXME: this function works by side effect, replacing the ``job`` reference
2138 with a new object. It should instead return the new object (or ``None``)
2139 and resolve the object references in the caller.
2140 """
2141 _LOG.debug("htc_tweak_log_info: %s %s", wms_path, job)
2143 # Use the presence of 'MyType' key as a proxy to determine if the job ad
2144 # contains the info extracted from the event log. Exit early if it doesn't
2145 # (e.g. it is a job ad for a pruned job).
2146 if "MyType" not in job:
2147 return
2149 try:
2150 job["ClusterId"] = job["Cluster"]
2151 job["ProcId"] = job["Proc"]
2152 except KeyError as e:
2153 _LOG.error("Missing key %s in job: %s", str(e), job)
2154 raise
2155 job["Iwd"] = str(wms_path)
2156 job["Owner"] = Path(wms_path).owner()
2158 if "LogNotes" in job:
2159 m = re.match(r"DAG Node: (\S+)", job["LogNotes"])
2160 if m: 2160 ↛ 2163line 2160 didn't jump to line 2163 because the condition on line 2160 was always true
2161 job["DAGNodeName"] = m.group(1)
2163 match job["MyType"]:
2164 case "ExecuteEvent":
2165 job["JobStatus"] = JobStatus.RUNNING
2166 case "JobTerminatedEvent" | "PostScriptTerminatedEvent":
2167 job["JobStatus"] = JobStatus.COMPLETED
2168 case "SubmitEvent":
2169 job["JobStatus"] = JobStatus.IDLE
2170 case "JobAbortedEvent":
2171 job["JobStatus"] = JobStatus.REMOVED
2172 case "JobHeldEvent":
2173 job["JobStatus"] = JobStatus.HELD
2174 case "JobReleaseEvent":
2175 # If the job managing the execution of the root DAG is held and
2176 # released this will be the last event showing up in its
2177 # job event log even if the job is still running. If this is
2178 # the last event for a job corresponding to the workflow node
2179 # (either a normal payload job or the job managing the execution
2180 # of an inner DAG), its final status will be determined later
2181 # using node status log (see _htc_status_to_wms_state()).
2182 job["JobStatus"] = JobStatus.RUNNING if "DAGNodeName" not in job else None
2183 case _:
2184 _LOG.debug("Unknown log event type: %s", job["MyType"])
2185 job["JobStatus"] = None
2187 # Use available information to add either "ExitCode" or "ExitSignal"
2188 # attribute that captures respectively job's exit status (if it finished
2189 # on its own accord) or its exit signal (if it was terminated by
2190 # a signal). Also, include a flag "ExitBySignal" to make distinguishing
2191 # between these two cases easy later on.
2192 if job["JobStatus"] in {
2193 JobStatus.COMPLETED,
2194 JobStatus.HELD,
2195 JobStatus.REMOVED,
2196 }:
2197 new_job = HTC_JOB_AD_HANDLERS.handle(job)
2198 if new_job is not None:
2199 job = new_job
2200 else:
2201 _LOG.error("Could not determine exit status for job '%s.%s'", job["ClusterId"], job["ProcId"])
2204def htc_check_dagman_output(wms_path: str | os.PathLike) -> str:
2205 """Check the DAGMan output for error messages.
2207 Parameters
2208 ----------
2209 wms_path : `str` or `os.PathLike`
2210 Directory containing the DAGman output file.
2212 Returns
2213 -------
2214 message : `str`
2215 Message containing error messages from the DAGMan output. Empty
2216 string if no messages.
2218 Raises
2219 ------
2220 FileNotFoundError
2221 If cannot find DAGMan standard output file in given wms_path.
2222 """
2223 try:
2224 filename = next(Path(wms_path).glob("*.dag.dagman.out"))
2225 except StopIteration as exc:
2226 raise FileNotFoundError(f"DAGMan standard output file not found in {wms_path}") from exc
2227 _LOG.debug("dag output filename: %s", filename)
2229 p = re.compile(r"^(\d\d/\d\d/\d\d \d\d:\d\d:\d\d) (Job submit try \d+/\d+ failed|Warning:.*$|ERROR:.*$)")
2231 message = ""
2232 # Shouldn't normally see this warning. Initializing it just in case.
2233 last_warning = (
2234 "Missing warning: Usually means the .dagman.out file has been truncated "
2235 "or temporarily had a full quota."
2236 )
2237 try:
2238 with open(filename) as fh:
2239 last_submit_failed = "" # Since submit retries multiple times only report last one
2240 for line in fh:
2241 m = p.match(line)
2242 if m:
2243 if m.group(2).startswith("Job submit try"):
2244 last_submit_failed = m.group(1)
2245 elif m.group(2).startswith("ERROR: submit attempt failed"):
2246 pass # Should be handled by Job submit try
2247 elif m.group(2).startswith("Warning"):
2248 if ".dag.nodes.log is in /tmp" in m.group(2):
2249 last_warning = "Cannot submit from /tmp."
2250 else:
2251 last_warning = m.group(2)
2252 elif m.group(2) == "ERROR: Warning is fatal error because of DAGMAN_USE_STRICT setting":
2253 message += "ERROR: "
2254 message += last_warning
2255 message += "\n"
2256 elif m.group(2) in [ 2256 ↛ 2262line 2256 didn't jump to line 2262 because the condition on line 2256 was always true
2257 "ERROR: the following job(s) failed:",
2258 "ERROR: the following Node(s) failed:",
2259 ]:
2260 pass
2261 else:
2262 message += m.group(2)
2263 message += "\n"
2265 if last_submit_failed:
2266 message += f"Warn: Job submission issues (last: {last_submit_failed})"
2267 except (OSError, PermissionError):
2268 message = f"Warn: Could not read dagman output file from {wms_path}."
2269 _LOG.debug("dag output file message: %s", message)
2270 return message
2273def _read_rescue_headers(infh: TextIO) -> list[str]:
2274 """Read header lines from a rescue file.
2276 Parameters
2277 ----------
2278 infh : `TextIO`
2279 The rescue file from which to read the header lines.
2281 Returns
2282 -------
2283 header_lines : `list` [`str`]
2284 Header lines read from the rescue file.
2285 """
2286 header_lines = []
2287 for line in infh:
2288 line = line.strip()
2289 if not line.startswith("#"):
2290 break
2291 header_lines.append(line)
2292 return header_lines
2295def _update_rescue_headers(header_lines: list[str]) -> list[str]:
2296 """Update rescue header lines in place.
2298 Replaces ``wms_check_status`` node name prefixes with the corresponding
2299 group node names and adjusts the count of nodes premarked DONE.
2301 Parameters
2302 ----------
2303 header_lines : `list` [`str`]
2304 Header lines to update in place.
2306 Returns
2307 -------
2308 failed_subdags : `list` [`str`]
2309 Names of failed subdag jobs.
2310 """
2311 failed_subdags = []
2313 # Relies on HTCondor writing all failed node names on a single line
2314 # (see src/condor_dagman/dag.cpp in https://github.com/htcondor/htcondor).
2315 for i, line in enumerate(header_lines):
2316 if line.startswith("# Nodes that failed:"):
2317 nodes = header_lines[i + 1][1:].strip().split(",")
2318 new_nodes = []
2319 for node in nodes:
2320 if node.startswith("wms_check_status"):
2321 node = node[17:]
2322 failed_subdags.append(node)
2323 new_nodes.append(node)
2324 header_lines[i + 1] = f"# {','.join(new_nodes)}"
2325 break
2327 for i, line in enumerate(header_lines):
2328 m = re.match(r"^(# Nodes premarked DONE:)\s+(\d+)", line)
2329 if m:
2330 header_lines[i] = f"{m.group(1)} {int(m.group(2)) - len(failed_subdags)}"
2331 break
2333 return failed_subdags
2336def _write_rescue_headers(header_lines: list[str], outfh: TextIO) -> None:
2337 """Write the header lines to the new rescue file.
2339 Parameters
2340 ----------
2341 header_lines : `list` [`str`]
2342 Header lines to write to the new rescue file.
2343 outfh : `TextIO`
2344 New rescue file.
2345 """
2346 for line in header_lines:
2347 print(line, file=outfh)
2348 print("", file=outfh)
2351def _copy_done_lines(failed_subdags: list[str], infh: TextIO, outfh: TextIO) -> None:
2352 """Copy the DONE lines from the original rescue file skipping
2353 the failed group jobs.
2355 Parameters
2356 ----------
2357 failed_subdags : `list` [`str`]
2358 List of job names for the failed subdags
2359 infh : `TextIO`
2360 Original rescue file to copy from.
2361 outfh : `TextIO`
2362 New rescue file to copy to.
2363 """
2364 for line in infh:
2365 line = line.strip()
2366 try:
2367 _, node_name = line.split()
2368 except ValueError:
2369 _LOG.error(f"Unexpected line in rescue file = '{line}'")
2370 raise
2371 if node_name not in failed_subdags:
2372 print(line, file=outfh)
2375def _update_rescue_file(rescue_file: Path) -> list[str]:
2376 """Update the subdag failures in the main rescue file.
2378 Parameters
2379 ----------
2380 rescue_file : `pathlib.Path`
2381 The main rescue file that needs to be updated.
2383 Returns
2384 -------
2385 failed_subdags : `list` [`str`]
2386 Names of failed subdag jobs found in the DAGMan rescue file.
2387 """
2388 # To reduce memory requirements, not reading entire file into memory.
2389 rescue_tmp = rescue_file.with_suffix(rescue_file.suffix + ".tmp")
2390 with open(rescue_file) as infh:
2391 header_lines = _read_rescue_headers(infh)
2392 failed_subdags = _update_rescue_headers(header_lines)
2393 with open(rescue_tmp, "w") as outfh:
2394 _write_rescue_headers(header_lines, outfh)
2395 _copy_done_lines(failed_subdags, infh, outfh)
2396 rescue_file.unlink()
2397 rescue_tmp.rename(rescue_file)
2398 return failed_subdags
2401def _update_dicts(dict1, dict2):
2402 """Update dict1 with info in dict2.
2404 (Basically an update for nested dictionaries.)
2406 Parameters
2407 ----------
2408 dict1 : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
2409 HTCondor job information to be updated.
2410 dict2 : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
2411 Additional HTCondor job information.
2412 """
2413 for key, value in dict2.items():
2414 if key in dict1 and isinstance(dict1[key], dict) and isinstance(value, dict): 2414 ↛ 2415line 2414 didn't jump to line 2415 because the condition on line 2414 was never true
2415 _update_dicts(dict1[key], value)
2416 else:
2417 dict1[key] = value
2420def _locate_schedds(locate_all=False):
2421 """Find out Scheduler daemons in an HTCondor pool.
2423 Parameters
2424 ----------
2425 locate_all : `bool`, optional
2426 If True, all available schedulers in the HTCondor pool will be located.
2427 False by default which means that the search will be limited to looking
2428 for the Scheduler running on a local host.
2430 Returns
2431 -------
2432 schedds : `dict` [`str`, `htcondor.Schedd`]
2433 A mapping between Scheduler names and Python objects allowing for
2434 interacting with them.
2435 """
2436 coll = Collector()
2438 schedd_ads = []
2439 if locate_all:
2440 schedd_ads.extend(coll.locateAll(DaemonTypes.Schedd))
2441 else:
2442 schedd_ads.append(coll.locate(DaemonTypes.Schedd))
2443 return {ad["Name"]: Schedd(ad) for ad in schedd_ads}