Coverage for python/lsst/ctrl/bps/report.py: 14%
64 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-14 01:23 -0800
1# This file is part of ctrl_bps.
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"""Supporting functions for reporting on runs submitted to a WMS.
30Note: Expectations are that future reporting effort will revolve around LSST
31oriented database tables.
32"""
34__all__ = ["BPS_POSTPROCESSORS", "display_report", "retrieve_report"]
36import logging
37import sys
38from collections.abc import Callable, Sequence
39from typing import TextIO
41from lsst.utils import doImportType
43from .bps_reports import (
44 DetailedRunReport,
45 ExitCodesReport,
46 SummaryRunReport,
47 compile_code_summary,
48 compile_job_summary,
49)
50from .wms_service import BaseWmsService, WmsRunReport, WmsStates
52BPS_POSTPROCESSORS = (compile_job_summary, compile_code_summary)
53"""Postprocessors for massaging run reports
54(`tuple` [`Callable` [[`WmsRunReport`], None]).
55"""
57_LOG = logging.getLogger(__name__)
60def display_report(
61 runs: list[WmsRunReport],
62 messages: list[str],
63 is_detailed: bool = False,
64 is_global: bool = False,
65 return_exit_codes: bool = False,
66 file: TextIO = sys.stdout,
67) -> None:
68 """Print out summary of jobs submitted for execution.
70 Parameters
71 ----------
72 runs : `list` [`str`]
73 Runs to include in the summary.
74 messages : `list` [`str`]
75 Errors that happened during report and/or processing. Empty if
76 no issues were encountered.
77 is_detailed : `bool`, optional
78 If set, the function prints out a detailed report including statuses
79 of each task in the workflow grouped by task labels. By default, only
80 a brief summary of each run is displayed.
81 is_global : `bool`, optional
82 If set, a global run id(s) will be used when displaying the report.
83 By default, the report will use local run id(s).
85 Only applicable in the context of a WMS using distributed job queues
86 (e.g., HTCondor).
87 return_exit_codes : `bool`, optional
88 If set, return exit codes related to jobs with a
89 non-success status. Defaults to False, which means that only
90 the summary state is returned.
92 Only applicable in the context of a WMS with associated
93 handlers to return exit codes from jobs.
94 file : TextIO
95 File or file-like object to write the output to.
96 """
97 run_brief = SummaryRunReport(
98 [
99 ("X", "S"),
100 ("STATE", "S"),
101 ("%S", "S"),
102 ("ID", "S"),
103 ("OPERATOR", "S"),
104 ("PROJECT", "S"),
105 ("CAMPAIGN", "S"),
106 ("PAYLOAD", "S"),
107 ("RUN", "S"),
108 ]
109 )
111 if is_detailed:
112 fields = [(" ", "S")] + [(state.name, "i") for state in WmsStates] + [("EXPECTED", "i")]
113 run_report = DetailedRunReport(fields)
115 for run in runs:
116 run_brief.add(run, use_global_id=is_global)
118 run_report.add(run, use_global_id=is_global)
119 if run_report.message:
120 messages.append(run_report.message)
122 print(run_brief, file=file)
123 print("\n", file=file)
124 print(f"Path: {run.path}", file=file)
125 print(f"Global job id: {run.global_wms_id}", file=file)
126 if run.specific_info:
127 print(run.specific_info, file=file)
128 print("\n", file=file)
129 print(run_report, file=file)
131 if return_exit_codes:
132 fields = [
133 (" ", "S"),
134 ("PAYLOAD ERROR COUNT", "i"),
135 ("PAYLOAD ERROR CODES", "S"),
136 ("INFRASTRUCTURE ERROR COUNT", "i"),
137 ("INFRASTRUCTURE ERROR CODES", "S"),
138 ]
139 run_exits_report = ExitCodesReport(fields)
140 run_exits_report.add(run, use_global_id=is_global)
141 if run_exits_report.message:
142 messages.append(run_exits_report.message)
143 print("\n", file=file)
144 print(run_exits_report, file=file)
145 run_exits_report.clear()
147 run_brief.clear()
148 run_report.clear()
149 else:
150 for run in runs:
151 run_brief.add(run, use_global_id=is_global)
152 run_brief.sort("ID")
153 print(run_brief, file=file)
155 if messages:
156 uniques = list(dict.fromkeys(messages))
157 print("\n".join(uniques), file=file)
158 print("\n", file=file)
161def retrieve_report(
162 wms_service_fqn: str,
163 *,
164 run_id: str | None = None,
165 user: str | None = None,
166 hist: float | None = None,
167 pass_thru: str | None = None,
168 is_global: bool = False,
169 return_exit_codes: bool = False,
170 postprocessors: Sequence[Callable[[WmsRunReport], None]] | None = None,
171) -> tuple[list[WmsRunReport], list[str]]:
172 """Retrieve summary of jobs submitted for execution.
174 Parameters
175 ----------
176 wms_service_fqn : `str`
177 Name of the WMS service class.
178 run_id : `str`, optional
179 A run id the report will be restricted to.
180 user : `str`, optional
181 A username the report will be restricted to.
182 hist : `float`, optional
183 Include runs from the given number of past days.
184 pass_thru : `str`, optional
185 A string to pass directly to the WMS service class.
186 is_global : `bool`, optional
187 If set, all available job queues will be queried for job information.
188 Defaults to False which means that only a local job queue will be
189 queried for information.
191 Only applicable in the context of a WMS using distributed job queues
192 (e.g., HTCondor).
193 return_exit_codes : `bool`, optional
194 If set, return exit codes related to jobs with a
195 non-success status. Defaults to False, which means that only
196 the summary state is returned.
198 Only applicable in the context of a WMS with associated
199 handlers to return exit codes from jobs.
200 postprocessors : `collections.abc.Sequence` [callable], optional
201 List of functions for "massaging" reports returned by the plugin. Each
202 function must take one positional argument:
204 - ``report``: run report (`lsst.ctrl.bps.WmsRunReport`)
206 If None (default), each run report returned by the plugin (if any)
207 will be returned as is.
209 Returns
210 -------
211 reports : `list` [`WmsRunReport`]
212 Run reports satisfying the search criteria.
213 messages : `list` [`str`]
214 Errors that happened during report retrieval and/or processing.
215 Empty if no issues were encountered.
217 Raises
218 ------
219 TypeError
220 Raised if the WMS service class is not a subclass of BaseWmsService.
221 """
222 messages: list[str] = []
224 wms_service_class = doImportType(wms_service_fqn)
225 if not issubclass(wms_service_class, BaseWmsService):
226 raise TypeError(
227 f"Invalid WMS service class '{wms_service_fqn}'; must be a subclass of BaseWmsService"
228 )
229 wms_service = wms_service_class({})
231 reports, message = wms_service.report(
232 wms_workflow_id=run_id,
233 user=user,
234 hist=hist,
235 pass_thru=pass_thru,
236 is_global=is_global,
237 return_exit_codes=return_exit_codes,
238 )
239 if message:
240 messages.append(message)
242 if postprocessors:
243 for report in reports:
244 for postprocessor in postprocessors:
245 if warnings := postprocessor(report):
246 for warning in warnings:
247 messages.append(
248 f"WARNING: Report may be incomplete. "
249 f"There was an issue with report postprocessing for '{report.wms_id}': "
250 f"{warning} (origin: {postprocessor.__name__})"
251 )
253 return reports, messages