Coverage for python/lsst/ctrl/bps/report.py: 14%

64 statements  

« prev     ^ index     » next       coverage.py v7.11.0, created at 2025-11-20 01:56 -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/>. 

27 

28"""Supporting functions for reporting on runs submitted to a WMS. 

29 

30Note: Expectations are that future reporting effort will revolve around LSST 

31oriented database tables. 

32""" 

33 

34__all__ = ["BPS_POSTPROCESSORS", "display_report", "retrieve_report"] 

35 

36import logging 

37import sys 

38from collections.abc import Callable, Sequence 

39from typing import TextIO 

40 

41from lsst.utils import doImportType 

42 

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 

51 

52BPS_POSTPROCESSORS = (compile_job_summary, compile_code_summary) 

53"""Postprocessors for massaging run reports 

54(`tuple` [`Callable` [[`WmsRunReport`], None]). 

55""" 

56 

57_LOG = logging.getLogger(__name__) 

58 

59 

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. 

69 

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). 

84 

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. 

91 

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 ) 

110 

111 if is_detailed: 

112 fields = [(" ", "S")] + [(state.name, "i") for state in WmsStates] + [("EXPECTED", "i")] 

113 run_report = DetailedRunReport(fields) 

114 

115 for run in runs: 

116 run_brief.add(run, use_global_id=is_global) 

117 

118 run_report.add(run, use_global_id=is_global) 

119 if run_report.message: 

120 messages.append(run_report.message) 

121 

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) 

130 

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() 

146 

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) 

154 

155 if messages: 

156 uniques = list(dict.fromkeys(messages)) 

157 print("\n".join(uniques), file=file) 

158 print("\n", file=file) 

159 

160 

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. 

173 

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. 

190 

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. 

197 

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: 

203 

204 - ``report``: run report (`lsst.ctrl.bps.WmsRunReport`) 

205 

206 If None (default), each run report returned by the plugin (if any) 

207 will be returned as is. 

208 

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. 

216 

217 Raises 

218 ------ 

219 TypeError 

220 Raised if the WMS service class is not a subclass of BaseWmsService. 

221 """ 

222 messages: list[str] = [] 

223 

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({}) 

230 

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) 

241 

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 ) 

252 

253 return reports, messages