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