Coverage for python/lsst/ctrl/bps/htcondor/report_utils.py: 63%
295 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-31 01:53 -0700
1# This file is part of ctrl_bps_htcondor.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (https://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# This software is dual licensed under the GNU General Public License and also
10# under a 3-clause BSD license. Recipients may choose which of these licenses
11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt,
12# respectively. If you choose the GPL option then the following text applies
13# (but note that there is still no warranty even if you opt for BSD instead):
14#
15# This program is free software: you can redistribute it and/or modify
16# it under the terms of the GNU General Public License as published by
17# the Free Software Foundation, either version 3 of the License, or
18# (at your option) any later version.
19#
20# This program is distributed in the hope that it will be useful,
21# but WITHOUT ANY WARRANTY; without even the implied warranty of
22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
23# GNU General Public License for more details.
24#
25# You should have received a copy of the GNU General Public License
26# along with this program. If not, see <https://www.gnu.org/licenses/>.
28"""Utility functions used for reporting."""
30import logging
31import os
32import re
33from pathlib import Path
34from typing import Any
36from htcondor2 import JobStatus, Schedd
38from lsst.ctrl.bps import (
39 WmsJobReport,
40 WmsRunReport,
41 WmsSpecificInfo,
42 WmsStates,
43)
45from .common_utils import _htc_status_to_wms_state
46from .lssthtc import (
47 MISSING_ID,
48 WmsNodeType,
49 condor_search,
50 htc_check_dagman_output,
51 htc_tweak_log_info,
52 pegasus_name_to_label,
53 read_dag_info,
54 read_dag_log,
55 read_dag_status,
56 read_node_status,
57 summarize_dag,
58)
60_LOG = logging.getLogger(__name__)
63def _get_status_from_id(
64 wms_workflow_id: str, hist: float, schedds: dict[str, Schedd]
65) -> tuple[WmsStates, str]:
66 """Gather run information using workflow id.
68 Parameters
69 ----------
70 wms_workflow_id : `str`
71 Limit to specific run based on id.
72 hist : `float`
73 Limit history search to this many days.
74 schedds : `dict` [ `str`, `htcondor.Schedd` ]
75 HTCondor schedulers which to query for job information. If empty
76 dictionary, all queries will be run against the local scheduler only.
78 Returns
79 -------
80 state : `lsst.ctrl.bps.WmsStates`
81 Status for the corresponding run.
82 message : `str`
83 Message with extra error information.
84 """
85 _LOG.debug("_get_status_from_id: id=%s, hist=%s, schedds=%s", wms_workflow_id, hist, schedds)
87 message = ""
89 # Collect information about the job by querying HTCondor schedd and
90 # HTCondor history.
91 schedd_dag_info = _get_info_from_schedd(wms_workflow_id, hist, schedds, skip_history=True)
92 if len(schedd_dag_info) == 1:
93 schedd_name = next(iter(schedd_dag_info))
94 dag_id = next(iter(schedd_dag_info[schedd_name]))
95 dag_ad = schedd_dag_info[schedd_name][dag_id]
96 state = _htc_status_to_wms_state(dag_ad)
97 else:
98 state = WmsStates.UNKNOWN
99 message = f"DAGMan job {wms_workflow_id} not found in queue or history. Check id or try path."
100 return state, message
103def _get_status_from_path(wms_path: str | os.PathLike) -> tuple[WmsStates, str]:
104 """Gather run status from a given run directory.
106 Parameters
107 ----------
108 wms_path : `str` | `os.PathLike`
109 The directory containing the submit side files (e.g., HTCondor files).
111 Returns
112 -------
113 state : `lsst.ctrl.bps.WmsStates`
114 Status for the run.
115 message : `str`
116 Message to be printed.
117 """
118 wms_path = Path(wms_path).resolve()
119 message = ""
120 dag_ad: dict[str, Any] = {}
121 try:
122 wms_workflow_id, dag_ad = read_dag_log(wms_path)
123 except FileNotFoundError:
124 wms_workflow_id = MISSING_ID
125 message = f"DAGMan log not found in {wms_path}. Check path."
127 if wms_workflow_id == MISSING_ID:
128 state = WmsStates.UNKNOWN
129 else:
130 htc_tweak_log_info(wms_path, dag_ad[wms_workflow_id])
131 state = _htc_status_to_wms_state(dag_ad[wms_workflow_id])
133 return state, message
136def _report_from_path(wms_path):
137 """Gather run information from a given run directory.
139 Parameters
140 ----------
141 wms_path : `str`
142 The directory containing the submit side files (e.g., HTCondor files).
144 Returns
145 -------
146 run_reports : `dict` [`str`, `lsst.ctrl.bps.WmsRunReport`]
147 Run information for the detailed report. The key is the HTCondor id
148 and the value is a collection of report information for that run.
149 message : `str`
150 Message to be printed with the summary report.
151 """
152 wms_workflow_id, jobs, message = _get_info_from_path(wms_path)
153 if wms_workflow_id == MISSING_ID:
154 run_reports = {}
155 else:
156 run_reports = _create_detailed_report_from_jobs(wms_workflow_id, jobs)
157 return run_reports, message
160def _report_from_id(wms_workflow_id, hist, schedds=None):
161 """Gather run information using workflow id.
163 Parameters
164 ----------
165 wms_workflow_id : `str`
166 Limit to specific run based on id.
167 hist : `float`
168 Limit history search to this many days.
169 schedds : `dict` [ `str`, `htcondor.Schedd` ], optional
170 HTCondor schedulers which to query for job information. If None
171 (default), all queries will be run against the local scheduler only.
173 Returns
174 -------
175 run_reports : `dict` [`str`, `lsst.ctrl.bps.WmsRunReport`]
176 Run information for the detailed report. The key is the HTCondor id
177 and the value is a collection of report information for that run.
178 message : `str`
179 Message to be printed with the summary report.
180 """
181 messages = []
183 # Collect information about the job by querying HTCondor schedd and
184 # HTCondor history.
185 schedd_dag_info = _get_info_from_schedd(wms_workflow_id, hist, schedds)
186 if len(schedd_dag_info) == 1:
187 # Extract the DAG info without altering the results of the query.
188 schedd_name = next(iter(schedd_dag_info))
189 dag_id = next(iter(schedd_dag_info[schedd_name]))
190 dag_ad = schedd_dag_info[schedd_name][dag_id]
192 # If the provided workflow id does not correspond to the one extracted
193 # from the DAGMan log file in the submit directory, rerun the query
194 # with the id found in the file.
195 #
196 # This is to cover the situation in which the user provided the old job
197 # id of a restarted run.
198 try:
199 path_dag_id, _ = read_dag_log(dag_ad["Iwd"])
200 except FileNotFoundError as exc:
201 # At the moment missing DAGMan log is pretty much a fatal error.
202 # So empty the DAG info to finish early (see the if statement
203 # below).
204 schedd_dag_info.clear()
205 messages.append(f"Cannot create the report for '{dag_id}': {exc}")
206 else:
207 if path_dag_id != dag_id:
208 schedd_dag_info = _get_info_from_schedd(path_dag_id, hist, schedds)
209 messages.append(
210 f"WARNING: Found newer workflow executions in same submit directory as id '{dag_id}'. "
211 "This normally occurs when a run is restarted. The report shown is for the most "
212 f"recent status with run id '{path_dag_id}'"
213 )
215 if len(schedd_dag_info) == 0:
216 run_reports = {}
217 elif len(schedd_dag_info) == 1:
218 _, dag_info = schedd_dag_info.popitem()
219 dag_id, dag_ad = dag_info.popitem()
221 # Create a mapping between jobs and their classads. The keys will
222 # be of format 'ClusterId.ProcId'.
223 job_info = {dag_id: dag_ad}
225 # Find jobs (nodes) belonging to that DAGMan job.
226 job_constraint = f"DAGManJobId == {int(float(dag_id))}"
227 schedd_job_info = condor_search(constraint=job_constraint, hist=hist, schedds=schedds)
228 if schedd_job_info:
229 _, node_info = schedd_job_info.popitem()
230 job_info.update(node_info)
232 # Collect additional pieces of information about jobs using HTCondor
233 # files in the submission directory.
234 _, path_jobs, message = _get_info_from_path(dag_ad["Iwd"])
235 _update_jobs(job_info, path_jobs)
236 if message:
237 messages.append(message)
238 run_reports = _create_detailed_report_from_jobs(dag_id, job_info)
239 else:
240 ids = [ad["GlobalJobId"] for dag_info in schedd_dag_info.values() for ad in dag_info.values()]
241 message = (
242 f"More than one job matches id '{wms_workflow_id}', "
243 f"their global ids are: {', '.join(ids)}. Rerun with one of the global ids"
244 )
245 messages.append(message)
246 run_reports = {}
248 message = "\n".join(messages)
249 return run_reports, message
252def _get_info_from_schedd(
253 wms_workflow_id: str, hist: float, schedds: dict[str, Schedd], skip_history: bool = False
254) -> dict[str, dict[str, dict[str, Any]]]:
255 """Gather run information from HTCondor.
257 Parameters
258 ----------
259 wms_workflow_id : `str`
260 Limit to specific run based on id.
261 hist : `float`
262 Limit history search to this many days.
263 schedds : `dict` [ `str`, `htcondor.Schedd` ]
264 HTCondor schedulers which to query for job information. If empty
265 dictionary, all queries will be run against the local scheduler only.
266 skip_history : `bool`, optional
267 If True, skip searching history if searching the queue returns any
268 information. Defaults to False.
270 Returns
271 -------
272 schedd_dag_info : `dict` [`str`, `dict` [`str`, `dict` [`str` Any]]]
273 Information about jobs satisfying the search criteria where for each
274 Scheduler, local HTCondor job ids are mapped to their respective
275 classads.
276 """
277 _LOG.debug(
278 "_get_info_from_schedd: id=%s, hist=%s, schedds=%s, skip_history=%s",
279 wms_workflow_id,
280 hist,
281 schedds,
282 skip_history,
283 )
285 dag_constraint = 'regexp("dagman$", Cmd)'
286 try:
287 cluster_id = int(float(wms_workflow_id))
288 except ValueError:
289 dag_constraint += f' && GlobalJobId == "{wms_workflow_id}"'
290 else:
291 dag_constraint += f" && ClusterId == {cluster_id}"
293 # With the current implementation of the condor_* functions the query
294 # will always return only one match per Scheduler.
295 #
296 # Even in the highly unlikely situation where HTCondor history (which
297 # condor_search queries too) is long enough to have jobs from before
298 # the cluster ids were rolled over (and as a result there is more then
299 # one job with the same cluster id) they will not show up in
300 # the results.
301 schedd_dag_info = condor_search(
302 constraint=dag_constraint, hist=hist, schedds=schedds, skip_history=skip_history
303 )
304 return schedd_dag_info
307def _get_info_from_path(wms_path: str | os.PathLike) -> tuple[str, dict[str, dict[str, Any]], str]:
308 """Gather run information from a given run directory.
310 Parameters
311 ----------
312 wms_path : `str` or `os.PathLike`
313 Directory containing HTCondor files.
315 Returns
316 -------
317 wms_workflow_id : `str`
318 The run id which is a DAGman job id.
319 jobs : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
320 Information about jobs read from files in the given directory.
321 The key is the HTCondor id and the value is a dictionary of HTCondor
322 keys and values.
323 message : `str`
324 Message to be printed with the summary report.
325 """
326 # Ensure path is absolute, in particular for folks helping
327 # debug failures that need to dig around submit files.
328 wms_path = Path(wms_path).resolve()
330 messages = []
331 try:
332 wms_workflow_id, jobs = read_dag_log(wms_path)
333 _LOG.debug("_get_info_from_path: from dag log %s = %s", wms_workflow_id, jobs)
334 _update_jobs(jobs, read_node_status(wms_path))
335 _LOG.debug("_get_info_from_path: after node status %s = %s", wms_workflow_id, jobs)
337 # Add more info for DAGman job
338 job = jobs[wms_workflow_id]
339 job.update(read_dag_status(wms_path))
341 job["total_jobs"], job["state_counts"] = _get_state_counts_from_jobs(wms_workflow_id, jobs)
342 if "bps_run" not in job: 342 ↛ 345line 342 didn't jump to line 345 because the condition on line 342 was always true
343 _add_run_info(wms_path, job)
345 message = htc_check_dagman_output(wms_path)
346 if message: 346 ↛ 347line 346 didn't jump to line 347 because the condition on line 346 was never true
347 messages.append(message)
348 _LOG.debug(
349 "_get_info: id = %s, total_jobs = %s", wms_workflow_id, jobs[wms_workflow_id]["total_jobs"]
350 )
352 # Add extra pieces of information which cannot be found in HTCondor
353 # generated files like 'GlobalJobId'.
354 #
355 # Do not treat absence of this file as a serious error. Neither runs
356 # submitted with earlier versions of the plugin nor the runs submitted
357 # with Pegasus plugin will have it at the moment. However, once enough
358 # time passes and Pegasus plugin will have its own report() method
359 # (instead of sneakily using HTCondor's one), the lack of that file
360 # should be treated as seriously as lack of any other file.
361 try:
362 _, job_info = read_dag_info(wms_path)
363 except FileNotFoundError as exc:
364 message = f"Warn: Some information may not be available: {exc}"
365 messages.append(message)
366 else:
367 schedd_name = next(iter(job_info))
368 job_ad = next(iter(job_info[schedd_name].values()))
369 job.update(job_ad)
370 except FileNotFoundError as err:
371 message = f"Could not find HTCondor files in '{wms_path}' ({err})"
372 _LOG.debug(message)
373 messages.append(message)
374 try:
375 message = htc_check_dagman_output(wms_path)
376 except FileNotFoundError as err:
377 message = (
378 f"Could not find DAGMan standard output file in '{wms_path}'.\n"
379 "Check that path is a valid HTCondor submission directory.\n"
380 "Check that the condor_dagman executable path is accessible from the AP machine."
381 )
382 if message:
383 messages.append(message)
384 wms_workflow_id = MISSING_ID
385 jobs = {}
387 # Add more condor_q-like info.
388 for job in jobs.values():
389 htc_tweak_log_info(wms_path, job)
391 message = "\n".join([msg for msg in messages if msg])
392 _LOG.debug("wms_workflow_id = %s, jobs = %s", wms_workflow_id, jobs.keys())
393 _LOG.debug("message = %s", message)
394 return wms_workflow_id, jobs, message
397def _create_detailed_report_from_jobs(
398 wms_workflow_id: str, jobs: dict[str, dict[str, Any]]
399) -> dict[str, WmsRunReport]:
400 """Gather run information to be used in generating summary reports.
402 Parameters
403 ----------
404 wms_workflow_id : `str`
405 The run id to create the report for.
406 jobs : `dict` [`str`, `dict` [`str`, Any]]
407 Mapping HTCondor job id to job information.
409 Returns
410 -------
411 run_reports : `dict` [`str`, `lsst.ctrl.bps.WmsRunReport`]
412 Run information for the detailed report. The key is the given HTCondor
413 id and the value is a collection of report information for that run.
414 """
415 _LOG.debug("_create_detailed_report: id = %s, job = %s", wms_workflow_id, jobs[wms_workflow_id])
417 dag_ad = jobs[wms_workflow_id]
419 report = WmsRunReport(
420 wms_id=f"{dag_ad['ClusterId']}.{dag_ad['ProcId']}",
421 global_wms_id=dag_ad.get("GlobalJobId", "MISS"),
422 path=dag_ad["Iwd"],
423 label=dag_ad.get("bps_job_label", "MISS"),
424 run=dag_ad.get("bps_run", "MISS"),
425 site=dag_ad.get("bps_runsite", ""),
426 project=dag_ad.get("bps_project", "MISS"),
427 campaign=dag_ad.get("bps_campaign", "MISS"),
428 payload=dag_ad.get("bps_payload", "MISS"),
429 operator=_get_owner(dag_ad),
430 run_summary=_get_run_summary(dag_ad),
431 state=_htc_status_to_wms_state(dag_ad),
432 total_number_jobs=0,
433 jobs=[],
434 job_state_counts=dict.fromkeys(WmsStates, 0),
435 exit_code_summary={},
436 )
438 payload_jobs = {} # keep track for later processing
439 specific_info = WmsSpecificInfo()
440 for job_id, job_ad in jobs.items():
441 if job_ad.get("wms_node_type", WmsNodeType.UNKNOWN) in [WmsNodeType.PAYLOAD, WmsNodeType.FINAL]:
442 try:
443 name = job_ad.get("DAGNodeName", job_id)
444 wms_state = _htc_status_to_wms_state(job_ad)
445 job_report = WmsJobReport(
446 wms_id=job_id,
447 name=name,
448 label=job_ad.get("bps_job_label", pegasus_name_to_label(name)),
449 state=wms_state,
450 )
451 if job_report.label == "init": 451 ↛ 452line 451 didn't jump to line 452 because the condition on line 451 was never true
452 job_report.label = "pipetaskInit"
453 report.job_state_counts[wms_state] += 1
454 report.jobs.append(job_report)
455 payload_jobs[job_id] = job_ad
456 except KeyError as ex:
457 _LOG.error("Job missing key '%s': %s", str(ex), job_ad)
458 raise
459 elif is_service_job(job_ad):
460 _LOG.debug(
461 "Found service job: id='%s', name='%s', label='%s', NodeStatus='%s', JobStatus='%s'",
462 job_id,
463 job_ad["DAGNodeName"],
464 job_ad.get("bps_job_label", "MISS"),
465 job_ad.get("NodeStatus", "MISS"),
466 job_ad.get("JobStatus", "MISS"),
467 )
468 _add_service_job_specific_info(job_ad, specific_info)
470 report.total_number_jobs = len(payload_jobs)
471 report.exit_code_summary = _get_exit_code_summary(payload_jobs)
472 if specific_info:
473 report.specific_info = specific_info
475 # Workflow will exit with non-zero DAG_STATUS if problem with
476 # any of the wms jobs. So change FAILED to SUCCEEDED if all
477 # payload jobs SUCCEEDED.
478 if report.total_number_jobs == report.job_state_counts[WmsStates.SUCCEEDED]:
479 report.state = WmsStates.SUCCEEDED
481 run_reports = {report.wms_id: report}
482 _LOG.debug("_create_detailed_report: run_reports = %s", run_reports)
483 return run_reports
486def _add_service_job_specific_info(job_ad: dict[str, Any], specific_info: WmsSpecificInfo) -> None:
487 """Generate report information for service job.
489 Parameters
490 ----------
491 job_ad : `dict` [`str`, `~typing.Any`]
492 Provisioning job information.
493 specific_info : `lsst.ctrl.bps.WmsSpecificInfo`
494 Where to add message.
495 """
496 status_details = ""
497 job_status = _htc_status_to_wms_state(job_ad)
499 # Service jobs in queue are deleted when DAG is done.
500 # To get accurate status, need to check other info.
501 if (
502 job_status == WmsStates.DELETED
503 and "Reason" in job_ad
504 and (
505 "Removed by DAGMan" in job_ad["Reason"]
506 or "removed because <OtherJobRemoveRequirements = DAGManJobId =?=" in job_ad["Reason"]
507 or "DAG is exiting and writing rescue file." in job_ad["Reason"]
508 )
509 ):
510 if "HoldReason" in job_ad:
511 # HoldReason exists even if released, so check.
512 if "job_released_time" in job_ad and job_ad["job_held_time"] < job_ad["job_released_time"]:
513 # If released, assume running until deleted.
514 job_status = WmsStates.SUCCEEDED
515 status_details = ""
516 else:
517 # If job held when deleted by DAGMan, still want to
518 # report hold reason
519 status_details = f"(Job was held for the following reason: {job_ad['HoldReason']})"
521 else:
522 job_status = WmsStates.SUCCEEDED
523 elif job_status == WmsStates.SUCCEEDED:
524 status_details = "(Note: Finished before workflow.)"
525 elif job_status == WmsStates.HELD:
526 status_details = f"({job_ad['HoldReason']})"
528 template = "Status of {job_name}: {status} {status_details}"
529 context = {
530 "job_name": job_ad["DAGNodeName"],
531 "status": job_status.name,
532 "status_details": status_details,
533 }
534 specific_info.add_message(template=template, context=context)
537def _summary_report(user, hist, pass_thru, schedds=None):
538 """Gather run information to be used in generating summary reports.
540 Parameters
541 ----------
542 user : `str`
543 Run lookup restricted to given user.
544 hist : `float`
545 How many previous days to search for run information.
546 pass_thru : `str`
547 Advanced users can define the HTCondor constraint to be used
548 when searching queue and history.
550 Returns
551 -------
552 run_reports : `dict` [`str`, `lsst.ctrl.bps.WmsRunReport`]
553 Run information for the summary report. The keys are HTCondor ids and
554 the values are collections of report information for each run.
555 message : `str`
556 Message to be printed with the summary report.
557 """
558 # only doing summary report so only look for dagman jobs
559 if pass_thru:
560 constraint = pass_thru
561 else:
562 # Notes:
563 # * bps_isjob == 'True' isn't getting set for DAG jobs that are
564 # manually restarted.
565 # * Any job with DAGManJobID isn't a DAG job
566 constraint = 'bps_isjob == "True" && JobUniverse == 7 && DAGManJobID =?= Undefined'
567 if user:
568 constraint += f' && (Owner == "{user}" || bps_operator == "{user}")'
570 job_info = condor_search(constraint=constraint, hist=hist, schedds=schedds)
572 # Have list of DAGMan jobs, need to get run_report info.
573 run_reports = {}
574 msg = ""
575 for jobs in job_info.values():
576 for job_id, job in jobs.items():
577 total_jobs, state_counts = _get_state_counts_from_dag_job(job)
578 # If didn't get from queue information (e.g., Kerberos bug),
579 # try reading from file.
580 if total_jobs == 0:
581 try:
582 job.update(read_dag_status(job["Iwd"]))
583 total_jobs, state_counts = _get_state_counts_from_dag_job(job)
584 except (StopIteration, FileNotFoundError):
585 pass # don't kill report can't find htcondor files
587 if "bps_run" not in job:
588 _add_run_info(job["Iwd"], job)
589 report = WmsRunReport(
590 wms_id=job_id,
591 global_wms_id=job["GlobalJobId"],
592 path=job["Iwd"],
593 label=job.get("bps_job_label", "MISS"),
594 run=job.get("bps_run", "MISS"),
595 project=job.get("bps_project", "MISS"),
596 campaign=job.get("bps_campaign", "MISS"),
597 payload=job.get("bps_payload", "MISS"),
598 site=job.get("bps_runsite", ""),
599 operator=_get_owner(job),
600 run_summary=_get_run_summary(job),
601 state=_htc_status_to_wms_state(job),
602 jobs=[],
603 total_number_jobs=total_jobs,
604 job_state_counts=state_counts,
605 )
606 run_reports[report.global_wms_id] = report
608 return run_reports, msg
611def _add_run_info(wms_path, job):
612 """Find BPS run information elsewhere for runs without bps attributes.
614 Parameters
615 ----------
616 wms_path : `str`
617 Path to submit files for the run.
618 job : `dict` [`str`, `~typing.Any`]
619 HTCondor dag job information.
621 Raises
622 ------
623 StopIteration
624 If cannot find file it is looking for. Permission errors are
625 caught and job's run is marked with error.
626 """
627 path = Path(wms_path) / "jobs"
628 try:
629 subfile = next(path.glob("**/*.sub"))
630 except (StopIteration, PermissionError):
631 job["bps_run"] = "Unavailable"
632 else:
633 _LOG.debug("_add_run_info: subfile = %s", subfile)
634 try:
635 with open(subfile, encoding="utf-8") as fh:
636 for line in fh:
637 if line.startswith("+bps_"):
638 m = re.match(r"\+(bps_[^\s]+)\s*=\s*(.+)$", line)
639 if m:
640 _LOG.debug("Matching line: %s", line)
641 job[m.group(1)] = m.group(2).replace('"', "")
642 else:
643 _LOG.debug("Could not parse attribute: %s", line)
644 except PermissionError:
645 job["bps_run"] = "PermissionError"
646 _LOG.debug("After adding job = %s", job)
649def _get_owner(job):
650 """Get the owner of a dag job.
652 Parameters
653 ----------
654 job : `dict` [`str`, `~typing.Any`]
655 HTCondor dag job information.
657 Returns
658 -------
659 owner : `str`
660 Owner of the dag job.
661 """
662 owner = job.get("bps_operator", None)
663 if not owner: 663 ↛ 664line 663 didn't jump to line 664 because the condition on line 663 was never true
664 owner = job.get("Owner", None)
665 if not owner:
666 _LOG.warning("Could not get Owner from htcondor job: %s", job)
667 owner = "MISS"
668 return owner
671def _get_run_summary(job):
672 """Get the run summary for a job.
674 Parameters
675 ----------
676 job : `dict` [`str`, `~typing.Any`]
677 HTCondor dag job information.
679 Returns
680 -------
681 summary : `str`
682 Number of jobs per PipelineTask label in approximate pipeline order.
683 Format: <label>:<count>[;<label>:<count>]+
684 """
685 summary = job.get("bps_job_summary", job.get("bps_run_summary", None))
686 if not summary:
687 summary, _, _ = summarize_dag(job["Iwd"])
688 if not summary:
689 _LOG.warning("Could not get run summary for htcondor job: %s", job)
690 _LOG.debug("_get_run_summary: summary=%s", summary)
692 # Workaround sometimes using init vs pipetaskInit
693 summary = summary.replace("init:", "pipetaskInit:")
695 if "pegasus_version" in job and "pegasus" not in summary: 695 ↛ 696line 695 didn't jump to line 696 because the condition on line 695 was never true
696 summary += ";pegasus:0"
698 return summary
701def _get_exit_code_summary(jobs):
702 """Get the exit code summary for a run.
704 Parameters
705 ----------
706 jobs : `dict` [`str`, `dict` [`str`, Any]]
707 Mapping HTCondor job id to job information.
709 Returns
710 -------
711 summary : `dict` [`str`, `list` [`int`]]
712 Jobs' exit codes per job label.
713 """
714 summary = {}
715 for job_id, job_ad in jobs.items():
716 job_label = job_ad["bps_job_label"]
717 summary.setdefault(job_label, [])
718 try:
719 exit_code = 0
720 job_status = job_ad["JobStatus"]
721 match job_status:
722 case JobStatus.COMPLETED | JobStatus.HELD | JobStatus.REMOVED:
723 exit_code = job_ad["ExitSignal"] if job_ad["ExitBySignal"] else job_ad["ExitCode"]
724 case JobStatus.IDLE | JobStatus.RUNNING | JobStatus.TRANSFERRING_OUTPUT | JobStatus.SUSPENDED:
725 pass
726 case _:
727 _LOG.debug("Unknown 'JobStatus' value ('%d') in classad for job '%s'", job_status, job_id)
728 if exit_code != 0:
729 summary[job_label].append(exit_code)
730 except KeyError as ex:
731 _LOG.debug("Attribute '%s' not found in the classad for job '%s'", ex, job_id)
732 return summary
735def _get_state_counts_from_jobs(
736 wms_workflow_id: str, jobs: dict[str, dict[str, Any]]
737) -> tuple[int, dict[WmsStates, int]]:
738 """Count number of jobs per WMS state.
740 The workflow job and the service jobs are excluded from the count.
742 Parameters
743 ----------
744 wms_workflow_id : `str`
745 HTCondor job id.
746 jobs : `dict [`dict` [`str`, `~typing.Any`]]
747 HTCondor dag job information.
749 Returns
750 -------
751 total_count : `int`
752 Total number of dag nodes.
753 state_counts : `dict` [`lsst.ctrl.bps.WmsStates`, `int`]
754 Keys are the different WMS states and values are counts of jobs
755 that are in that WMS state.
756 """
757 state_counts = dict.fromkeys(WmsStates, 0)
758 for job_id, job_ad in jobs.items():
759 if job_id != wms_workflow_id and job_ad.get("wms_node_type", WmsNodeType.UNKNOWN) in [
760 WmsNodeType.PAYLOAD,
761 WmsNodeType.FINAL,
762 ]:
763 state_counts[_htc_status_to_wms_state(job_ad)] += 1
764 total_count = sum(state_counts.values())
766 return total_count, state_counts
769def _get_state_counts_from_dag_job(job):
770 """Count number of jobs per WMS state.
772 Parameters
773 ----------
774 job : `dict` [`str`, `~typing.Any`]
775 HTCondor dag job information.
777 Returns
778 -------
779 total_count : `int`
780 Total number of dag nodes.
781 state_counts : `dict` [`lsst.ctrl.bps.WmsStates`, `int`]
782 Keys are the different WMS states and values are counts of jobs
783 that are in that WMS state.
784 """
785 _LOG.debug("_get_state_counts_from_dag_job: job = %s %s", type(job), len(job))
786 state_counts = dict.fromkeys(WmsStates, 0)
787 if "DAG_NodesReady" in job: 787 ↛ 799line 787 didn't jump to line 799 because the condition on line 787 was always true
788 state_counts = {
789 WmsStates.UNREADY: job.get("DAG_NodesUnready", 0),
790 WmsStates.READY: job.get("DAG_NodesReady", 0),
791 WmsStates.HELD: job.get("DAG_JobsHeld", 0),
792 WmsStates.SUCCEEDED: job.get("DAG_NodesDone", 0),
793 WmsStates.FAILED: job.get("DAG_NodesFailed", 0),
794 WmsStates.PRUNED: job.get("DAG_NodesFutile", 0),
795 WmsStates.MISFIT: job.get("DAG_NodesPre", 0) + job.get("DAG_NodesPost", 0),
796 }
797 total_jobs = job.get("DAG_NodesTotal")
798 _LOG.debug("_get_state_counts_from_dag_job: from DAG_* keys, total_jobs = %s", total_jobs)
799 elif "NodesFailed" in job:
800 state_counts = {
801 WmsStates.UNREADY: job.get("NodesUnready", 0),
802 WmsStates.READY: job.get("NodesReady", 0),
803 WmsStates.HELD: job.get("JobProcsHeld", 0),
804 WmsStates.SUCCEEDED: job.get("NodesDone", 0),
805 WmsStates.FAILED: job.get("NodesFailed", 0),
806 WmsStates.PRUNED: job.get("NodesFutile", 0),
807 WmsStates.MISFIT: job.get("NodesPre", 0) + job.get("NodesPost", 0),
808 }
809 try:
810 total_jobs = job.get("NodesTotal")
811 except KeyError as ex:
812 _LOG.error("Job missing %s. job = %s", str(ex), job)
813 raise
814 _LOG.debug("_get_state_counts_from_dag_job: from NODES* keys, total_jobs = %s", total_jobs)
815 else:
816 # With Kerberos job auth and Kerberos bug, if warning would be printed
817 # for every DAG.
818 _LOG.debug("Can't get job state counts %s", job["Iwd"])
819 total_jobs = 0
821 _LOG.debug("total_jobs = %s, state_counts: %s", total_jobs, state_counts)
822 return total_jobs, state_counts
825def _update_jobs(jobs1, jobs2):
826 """Update jobs1 with info in jobs2.
828 (Basically an update for nested dictionaries.)
830 Parameters
831 ----------
832 jobs1 : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
833 HTCondor job information to be updated.
834 jobs2 : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
835 Additional HTCondor job information.
836 """
837 for job_id, job_ad in jobs2.items():
838 if job_id in jobs1: 838 ↛ 839line 838 didn't jump to line 839 because the condition on line 838 was never true
839 jobs1[job_id].update(job_ad)
840 else:
841 jobs1[job_id] = job_ad
844def is_service_job(job_ad: dict[str, Any]) -> bool:
845 """Determine if a job is a service one.
847 Parameters
848 ----------
849 job_ad : `dict` [`str`, Any]
850 Information about an HTCondor job.
852 Returns
853 -------
854 is_service_job : `bool`
855 True if the job is a service one, false otherwise.
857 Notes
858 -----
859 At the moment, HTCondor does not provide a native way to distinguish
860 between payload and service jobs in the workflow. This code depends
861 on read_node_status adding wms_node_type.
862 """
863 return job_ad.get("wms_node_type", WmsNodeType.UNKNOWN) == WmsNodeType.SERVICE