Coverage for python/lsst/ctrl/bps/htcondor/common_utils.py: 81%

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

27 

28"""Utility functions used by multiple functions in ctrl_bps_htcondor.""" 

29 

30import logging 

31from enum import IntEnum, auto 

32from pathlib import Path 

33 

34from classad2 import ClassAd 

35from htcondor2 import Collector, DaemonTypes, JobStatus, Schedd 

36 

37from lsst.ctrl.bps import ( 

38 WmsStates, 

39) 

40 

41from .lssthtc import ( 

42 NodeStatus, 

43 condor_history, 

44 condor_q, 

45 read_dag_info, 

46) 

47 

48_LOG = logging.getLogger(__name__) 

49 

50 

51class WmsIdType(IntEnum): 

52 """Type of valid WMS ids.""" 

53 

54 UNKNOWN = auto() 

55 """The type of id cannot be determined. 

56 """ 

57 

58 LOCAL = auto() 

59 """The id is HTCondor job's ClusterId (with optional '.ProcId'). 

60 """ 

61 

62 GLOBAL = auto() 

63 """Id is a HTCondor's global job id. 

64 """ 

65 

66 PATH = auto() 

67 """Id is a submission path. 

68 """ 

69 

70 

71def _htc_status_to_wms_state(job): 

72 """Convert HTCondor job status to generic wms state. 

73 

74 Parameters 

75 ---------- 

76 job : `dict` [`str`, `~typing.Any`] 

77 HTCondor job information. 

78 

79 Returns 

80 ------- 

81 wms_state : `WmsStates` 

82 The equivalent WmsState to given job's status. 

83 """ 

84 wms_state = WmsStates.MISFIT 

85 if "JobStatus" in job: 

86 wms_state = _htc_job_status_to_wms_state(job) 

87 

88 if wms_state == WmsStates.MISFIT and "NodeStatus" in job: 

89 wms_state = _htc_node_status_to_wms_state(job) 

90 return wms_state 

91 

92 

93def _htc_job_status_to_wms_state(job): 

94 """Convert HTCondor job status to generic wms state. 

95 

96 Parameters 

97 ---------- 

98 job : `dict` [`str`, `~typing.Any`] 

99 HTCondor job information. 

100 

101 Returns 

102 ------- 

103 wms_state : `lsst.ctrl.bps.WmsStates` 

104 The equivalent WmsState to given job's status. 

105 """ 

106 _LOG.debug( 

107 "htc_job_status_to_wms_state: %s=%s, %s", job["ClusterId"], job["JobStatus"], type(job["JobStatus"]) 

108 ) 

109 wms_state = WmsStates.MISFIT 

110 if "JobStatus" in job and job["JobStatus"]: 

111 job_status = int(job["JobStatus"]) 

112 

113 _LOG.debug("htc_job_status_to_wms_state: job_status = %s", job_status) 

114 if job_status == JobStatus.IDLE: 

115 wms_state = WmsStates.PENDING 

116 elif job_status == JobStatus.RUNNING: 

117 wms_state = WmsStates.RUNNING 

118 elif job_status == JobStatus.REMOVED: 

119 if (job.get("ExitBySignal", False) and job.get("ExitSignal", 0)) or job.get("ExitCode", 0): 119 ↛ 120line 119 didn't jump to line 120 because the condition on line 119 was never true

120 wms_state = WmsStates.FAILED 

121 else: 

122 wms_state = WmsStates.DELETED 

123 elif job_status == JobStatus.COMPLETED: 

124 if ( 

125 (job.get("ExitBySignal", False) and job.get("ExitSignal", 0)) 

126 or job.get("ExitCode", 0) 

127 or job.get("DAG_Status", 0) 

128 ): 

129 wms_state = WmsStates.FAILED 

130 else: 

131 wms_state = WmsStates.SUCCEEDED 

132 elif job_status == JobStatus.HELD: 132 ↛ 135line 132 didn't jump to line 135 because the condition on line 132 was always true

133 wms_state = WmsStates.HELD 

134 

135 return wms_state 

136 

137 

138def _htc_node_status_to_wms_state(job): 

139 """Convert HTCondor node status to generic wms state. 

140 

141 Parameters 

142 ---------- 

143 job : `dict` [`str`, `~typing.Any`] 

144 HTCondor job information. 

145 

146 Returns 

147 ------- 

148 wms_state : `lsst.ctrl.bps.WmsStates` 

149 The equivalent WmsState to given node's status. 

150 """ 

151 wms_state = WmsStates.MISFIT 

152 match job["NodeStatus"]: 

153 case NodeStatus.NOT_READY: 

154 wms_state = WmsStates.UNREADY 

155 case NodeStatus.READY: 

156 wms_state = WmsStates.READY 

157 case NodeStatus.PRERUN: 

158 wms_state = WmsStates.MISFIT 

159 case NodeStatus.SUBMITTED: 

160 if job["JobProcsHeld"]: 

161 wms_state = WmsStates.HELD 

162 elif job["StatusDetails"] == "not_idle": 

163 wms_state = WmsStates.RUNNING 

164 elif job["JobProcsQueued"]: 164 ↛ 180line 164 didn't jump to line 180 because the condition on line 164 was always true

165 wms_state = WmsStates.PENDING 

166 case NodeStatus.POSTRUN: 

167 wms_state = WmsStates.MISFIT 

168 case NodeStatus.DONE: 

169 wms_state = WmsStates.SUCCEEDED 

170 case NodeStatus.ERROR: 

171 # Use job exit status instead of post script exit status. 

172 if "DAGMAN error 0" in job["StatusDetails"]: 

173 wms_state = WmsStates.SUCCEEDED 

174 elif "ULOG_JOB_ABORTED" in job["StatusDetails"]: 

175 wms_state = WmsStates.DELETED 

176 else: 

177 wms_state = WmsStates.FAILED 

178 case NodeStatus.FUTILE: 178 ↛ 180line 178 didn't jump to line 180 because the pattern on line 178 always matched

179 wms_state = WmsStates.PRUNED 

180 return wms_state 

181 

182 

183def _wms_id_type(wms_id): 

184 """Determine the type of the WMS id. 

185 

186 Parameters 

187 ---------- 

188 wms_id : `str` 

189 WMS id identifying a job. 

190 

191 Returns 

192 ------- 

193 id_type : `lsst.ctrl.bps.htcondor.WmsIdType` 

194 Type of WMS id. 

195 """ 

196 try: 

197 int(float(wms_id)) 

198 except ValueError: 

199 wms_path = Path(wms_id) 

200 if wms_path.is_dir(): 

201 id_type = WmsIdType.PATH 

202 else: 

203 id_type = WmsIdType.GLOBAL 

204 except TypeError: 

205 id_type = WmsIdType.UNKNOWN 

206 else: 

207 id_type = WmsIdType.LOCAL 

208 return id_type 

209 

210 

211def _wms_id_to_cluster(wms_id) -> tuple[ClassAd | None, int | None, WmsIdType]: 

212 """Convert WMS id to cluster id. 

213 

214 Parameters 

215 ---------- 

216 wms_id : `int` or `float` or `str` 

217 HTCondor job id or path. 

218 

219 Returns 

220 ------- 

221 schedd_ad : `classad2.ClassAd` 

222 ClassAd describing the scheduler managing the job with the given id. 

223 cluster_id : `int` 

224 HTCondor cluster id. 

225 id_type : `lsst.ctrl.bps.wms.htcondor.WmsIdType` 

226 The type of the provided id. 

227 """ 

228 coll = Collector() 

229 

230 schedd_ad = None 

231 cluster_id = None 

232 id_type = _wms_id_type(wms_id) 

233 _LOG.debug("id_type = %s", id_type.name) 

234 if id_type == WmsIdType.LOCAL: 234 ↛ 235line 234 didn't jump to line 235 because the condition on line 234 was never true

235 schedd_ad = coll.locate(DaemonTypes.Schedd) 

236 cluster_id = int(float(wms_id)) 

237 elif id_type == WmsIdType.GLOBAL: 237 ↛ 238line 237 didn't jump to line 238 because the condition on line 237 was never true

238 constraint = f'GlobalJobId == "{wms_id}"' 

239 schedd_ads = {ad["Name"]: ad for ad in coll.locateAll(DaemonTypes.Schedd)} 

240 schedds = {name: Schedd(ad) for name, ad in schedd_ads.items()} 

241 job_info = condor_q(constraint=constraint, schedds=schedds) 

242 if job_info: 

243 schedd_name, job_rec = job_info.popitem() 

244 job_id, _ = job_rec.popitem() 

245 schedd_ad = schedd_ads[schedd_name] 

246 cluster_id = int(float(job_id)) 

247 elif id_type == WmsIdType.PATH: 247 ↛ 258line 247 didn't jump to line 258 because the condition on line 247 was always true

248 try: 

249 _, job_info = read_dag_info(wms_id) 

250 except (FileNotFoundError, PermissionError, OSError): 

251 pass 

252 else: 

253 schedd_name, job_rec = job_info.popitem() 

254 job_id, _ = job_rec.popitem() 

255 schedd_ad = coll.locate(DaemonTypes.Schedd, schedd_name) 

256 cluster_id = int(float(job_id)) 

257 else: 

258 pass 

259 return schedd_ad, cluster_id, id_type 

260 

261 

262def _wms_id_to_dir(wms_id): 

263 """Convert WMS id to a submit directory candidate. 

264 

265 The function does not check if the directory exists or if it is a valid 

266 BPS submit directory. 

267 

268 Parameters 

269 ---------- 

270 wms_id : `int` or `float` or `str` 

271 HTCondor job id or path. 

272 

273 Returns 

274 ------- 

275 wms_path : `pathlib.Path` or None 

276 Submit directory candidate for the run with the given job id. If no 

277 directory can be associated with the provided WMS id, it will be set 

278 to None. 

279 id_type : `lsst.ctrl.bps.wms.htcondor.IdType` 

280 The type of the provided id. 

281 

282 Raises 

283 ------ 

284 TypeError 

285 Raised if provided WMS id has invalid type. 

286 """ 

287 coll = Collector() 

288 schedd_ads = [] 

289 

290 constraint = None 

291 wms_path = None 

292 id_type = _wms_id_type(wms_id) 

293 match id_type: 

294 case WmsIdType.LOCAL: 294 ↛ 295line 294 didn't jump to line 295 because the pattern on line 294 never matched

295 constraint = f"ClusterId == {int(float(wms_id))}" 

296 schedd_ads.append(coll.locate(DaemonTypes.Schedd)) 

297 case WmsIdType.GLOBAL: 297 ↛ 298line 297 didn't jump to line 298 because the pattern on line 297 never matched

298 constraint = f'GlobalJobId == "{wms_id}"' 

299 schedd_ads.extend(coll.locateAll(DaemonTypes.Schedd)) 

300 case WmsIdType.PATH: 

301 wms_path = Path(wms_id).resolve() 

302 case WmsIdType.UNKNOWN: 302 ↛ 304line 302 didn't jump to line 304 because the pattern on line 302 always matched

303 raise TypeError(f"Invalid job id type: {wms_id}") 

304 if constraint is not None: 304 ↛ 305line 304 didn't jump to line 305 because the condition on line 304 was never true

305 schedds = {ad["name"]: Schedd(ad) for ad in schedd_ads} 

306 job_info = condor_history(constraint=constraint, schedds=schedds, projection=["Iwd"]) 

307 if job_info: 

308 _, job_rec = job_info.popitem() 

309 _, job_ad = job_rec.popitem() 

310 wms_path = Path(job_ad["Iwd"]) 

311 return wms_path, id_type