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 

34import htcondor 

35 

36from lsst.ctrl.bps import ( 

37 WmsStates, 

38) 

39 

40from .lssthtc import ( 

41 NodeStatus, 

42 condor_history, 

43 condor_q, 

44 read_dag_info, 

45) 

46 

47_LOG = logging.getLogger(__name__) 

48 

49 

50class WmsIdType(IntEnum): 

51 """Type of valid WMS ids.""" 

52 

53 UNKNOWN = auto() 

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

55 """ 

56 

57 LOCAL = auto() 

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

59 """ 

60 

61 GLOBAL = auto() 

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

63 """ 

64 

65 PATH = auto() 

66 """Id is a submission path. 

67 """ 

68 

69 

70def _htc_status_to_wms_state(job): 

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

72 

73 Parameters 

74 ---------- 

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

76 HTCondor job information. 

77 

78 Returns 

79 ------- 

80 wms_state : `WmsStates` 

81 The equivalent WmsState to given job's status. 

82 """ 

83 wms_state = WmsStates.MISFIT 

84 if "JobStatus" in job: 

85 wms_state = _htc_job_status_to_wms_state(job) 

86 

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

88 wms_state = _htc_node_status_to_wms_state(job) 

89 return wms_state 

90 

91 

92def _htc_job_status_to_wms_state(job): 

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

94 

95 Parameters 

96 ---------- 

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

98 HTCondor job information. 

99 

100 Returns 

101 ------- 

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

103 The equivalent WmsState to given job's status. 

104 """ 

105 _LOG.debug( 

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

107 ) 

108 wms_state = WmsStates.MISFIT 

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

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

111 

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

113 if job_status == htcondor.JobStatus.IDLE: 

114 wms_state = WmsStates.PENDING 

115 elif job_status == htcondor.JobStatus.RUNNING: 

116 wms_state = WmsStates.RUNNING 

117 elif job_status == htcondor.JobStatus.REMOVED: 

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

119 wms_state = WmsStates.FAILED 

120 else: 

121 wms_state = WmsStates.DELETED 

122 elif job_status == htcondor.JobStatus.COMPLETED: 

123 if ( 

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

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

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

127 ): 

128 wms_state = WmsStates.FAILED 

129 else: 

130 wms_state = WmsStates.SUCCEEDED 

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

132 wms_state = WmsStates.HELD 

133 

134 return wms_state 

135 

136 

137def _htc_node_status_to_wms_state(job): 

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

139 

140 Parameters 

141 ---------- 

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

143 HTCondor job information. 

144 

145 Returns 

146 ------- 

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

148 The equivalent WmsState to given node's status. 

149 """ 

150 wms_state = WmsStates.MISFIT 

151 match job["NodeStatus"]: 

152 case NodeStatus.NOT_READY: 

153 wms_state = WmsStates.UNREADY 

154 case NodeStatus.READY: 

155 wms_state = WmsStates.READY 

156 case NodeStatus.PRERUN: 

157 wms_state = WmsStates.MISFIT 

158 case NodeStatus.SUBMITTED: 

159 if job["JobProcsHeld"]: 

160 wms_state = WmsStates.HELD 

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

162 wms_state = WmsStates.RUNNING 

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

164 wms_state = WmsStates.PENDING 

165 case NodeStatus.POSTRUN: 

166 wms_state = WmsStates.MISFIT 

167 case NodeStatus.DONE: 

168 wms_state = WmsStates.SUCCEEDED 

169 case NodeStatus.ERROR: 

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

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

172 wms_state = WmsStates.SUCCEEDED 

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

174 wms_state = WmsStates.DELETED 

175 else: 

176 wms_state = WmsStates.FAILED 

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

178 wms_state = WmsStates.PRUNED 

179 return wms_state 

180 

181 

182def _wms_id_type(wms_id): 

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

184 

185 Parameters 

186 ---------- 

187 wms_id : `str` 

188 WMS id identifying a job. 

189 

190 Returns 

191 ------- 

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

193 Type of WMS id. 

194 """ 

195 try: 

196 int(float(wms_id)) 

197 except ValueError: 

198 wms_path = Path(wms_id) 

199 if wms_path.is_dir(): 

200 id_type = WmsIdType.PATH 

201 else: 

202 id_type = WmsIdType.GLOBAL 

203 except TypeError: 

204 id_type = WmsIdType.UNKNOWN 

205 else: 

206 id_type = WmsIdType.LOCAL 

207 return id_type 

208 

209 

210def _wms_id_to_cluster(wms_id): 

211 """Convert WMS id to cluster id. 

212 

213 Parameters 

214 ---------- 

215 wms_id : `int` or `float` or `str` 

216 HTCondor job id or path. 

217 

218 Returns 

219 ------- 

220 schedd_ad : `classad.ClassAd` 

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

222 cluster_id : `int` 

223 HTCondor cluster id. 

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

225 The type of the provided id. 

226 """ 

227 coll = htcondor.Collector() 

228 

229 schedd_ad = None 

230 cluster_id = None 

231 id_type = _wms_id_type(wms_id) 

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

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

234 schedd_ad = coll.locate(htcondor.DaemonTypes.Schedd) 

235 cluster_id = int(float(wms_id)) 

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

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

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

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

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

241 if job_info: 

242 schedd_name, job_rec = job_info.popitem() 

243 job_id, _ = job_rec.popitem() 

244 schedd_ad = schedd_ads[schedd_name] 

245 cluster_id = int(float(job_id)) 

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

247 try: 

248 _, job_info = read_dag_info(wms_id) 

249 except (FileNotFoundError, PermissionError, OSError): 

250 pass 

251 else: 

252 schedd_name, job_rec = job_info.popitem() 

253 job_id, _ = job_rec.popitem() 

254 schedd_ad = coll.locate(htcondor.DaemonTypes.Schedd, schedd_name) 

255 cluster_id = int(float(job_id)) 

256 else: 

257 pass 

258 return schedd_ad, cluster_id, id_type 

259 

260 

261def _wms_id_to_dir(wms_id): 

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

263 

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

265 BPS submit directory. 

266 

267 Parameters 

268 ---------- 

269 wms_id : `int` or `float` or `str` 

270 HTCondor job id or path. 

271 

272 Returns 

273 ------- 

274 wms_path : `pathlib.Path` or None 

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

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

277 to None. 

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

279 The type of the provided id. 

280 

281 Raises 

282 ------ 

283 TypeError 

284 Raised if provided WMS id has invalid type. 

285 """ 

286 coll = htcondor.Collector() 

287 schedd_ads = [] 

288 

289 constraint = None 

290 wms_path = None 

291 id_type = _wms_id_type(wms_id) 

292 match id_type: 

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

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

295 schedd_ads.append(coll.locate(htcondor.DaemonTypes.Schedd)) 

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

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

298 schedd_ads.extend(coll.locateAll(htcondor.DaemonTypes.Schedd)) 

299 case WmsIdType.PATH: 

300 wms_path = Path(wms_id).resolve() 

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

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

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

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

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

306 if job_info: 

307 _, job_rec = job_info.popitem() 

308 _, job_ad = job_rec.popitem() 

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

310 return wms_path, id_type