Coverage for python/lsst/ctrl/bps/htcondor/handlers.py: 99%
122 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-13 09:09 +0000
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"""Definitions of handlers of HTCondor job ClassAds."""
30__all__ = [
31 "HTC_JOB_AD_HANDLERS",
32 "Chain",
33 "Handler",
34 "JobAbortedByPeriodicRemoveHandler",
35 "JobAbortedByUserHandler",
36 "JobCompletedHandler",
37 "JobHeldByOtherHandler",
38 "JobHeldBySignalHandler",
39 "JobHeldByUserHandler",
40]
43import abc
44import logging
45import re
46from collections.abc import Sequence
47from typing import Any, overload
49_LOG = logging.getLogger(__name__)
52class Handler(abc.ABC):
53 """Abstract base class defining Handler interface."""
55 @abc.abstractmethod
56 def handle(self, ad: dict[str, Any]) -> dict[str, Any] | None:
57 """Handle a ClassAd.
59 Parameters
60 ----------
61 ad : `dict[`str`, Any]`
62 The dictionary representing ClassAd that need to be processed.
64 Returns
65 -------
66 ad : `dict[`str`, Any]` | None
67 The dictionary representing ClassAd after processing and ``None``
68 if the handler was not able to process the ad.
70 Notes
71 -----
72 To optimize the memory usage, the implementation of this method may
73 modify the ClassAd in place. In such a case, the ClassAd returned by
74 the method will be the same object that was passed to it as
75 the argument, but including any modifications that were made.
76 """
79class Chain(Sequence):
80 """Class defining chaining of handlers.
82 Parameters
83 ----------
84 handlers : `Sequence` [`Handler`]
85 List of handlers that will be used to initialize the chain.
86 """
88 def __init__(self, handlers: Sequence[Handler] | None = None) -> None:
89 self._handlers: list[Handler] = []
90 if handlers is not None:
91 for handler in handlers:
92 self.append(handler)
94 @overload
95 def __getitem__(self, index: int) -> Handler: ... 95 ↛ exitline 95 didn't return from function '__getitem__' because
96 @overload
97 def __getitem__(self, index: slice) -> Sequence[Handler]: ... 97 ↛ exitline 97 didn't return from function '__getitem__' because
98 def __getitem__(self, index):
99 return self._handlers[index]
101 def __len__(self) -> int:
102 return len(self._handlers)
104 def append(self, handler: Handler) -> None:
105 """Append a handler to the chain.
107 Parameters
108 ----------
109 handler : `Handler`
110 The handler that needs to be added to the chain.
112 Raises
113 ------
114 TypeError
115 Raised if the passed object in not a ``Handler``.
116 """
117 if not isinstance(handler, Handler):
118 raise TypeError(f"append() argument must be a 'Handler', not a '{type(handler)}'")
119 self._handlers.append(handler)
121 def handle(self, ad: dict[str, Any]) -> dict[str, Any] | None:
122 """Handle a ClassAd.
124 Parameters
125 ----------
126 ad : `dict[`str`, Any]`
127 The dictionary representing a ClassAd that need to be handled.
129 Returns
130 -------
131 ad : `dict[`str`, Any]`
132 A modified ClassAd if any handler in the chain was able to
133 process the ad, None otherwise.
134 """
135 new_ad = None
136 for handler in self:
137 try:
138 new_ad = handler.handle(ad)
139 except Exception as e:
140 _LOG.warning(
141 "Handler '%s' raised an exception while processing the ad: %s. "
142 "Proceeding to the next handler (if any).",
143 type(handler).__name__,
144 repr(e),
145 )
146 else:
147 if new_ad is not None:
148 break
149 return new_ad
152class JobCompletedHandler(Handler):
153 """Handler of ClassAds for completed jobs."""
155 def handle(self, ad: dict[str, Any]) -> dict[str, Any] | None:
156 if not ad["MyType"].endswith("TerminatedEvent"):
157 _LOG.debug(
158 "Handler '%s': refusing to process the ad for the job '%s.%s': job not completed",
159 self.__class__.__name__,
160 ad["ClusterId"],
161 ad["ProcId"],
162 )
163 return None
165 ad["ExitBySignal"] = not ad["TerminatedNormally"]
166 if ad["ExitBySignal"]:
167 ad["ExitSignal"] = ad["TerminatedBySignal"]
168 else:
169 ad["ExitCode"] = ad["ReturnValue"]
170 return ad
173class JobHeldByOtherHandler(Handler):
174 """Handler of ClassAds for jobs put on hold."""
176 def handle(self, ad: dict[str, Any]) -> dict[str, Any] | None:
177 if not ad["MyType"].endswith("HeldEvent"):
178 _LOG.debug(
179 "Handler '%s': refusing to process the ad for the job '%s.%s': job not held",
180 self.__class__.__name__,
181 ad["ClusterId"],
182 ad["ProcId"],
183 )
184 return None
185 if ad["HoldReasonCode"] not in {1, 3}:
186 ad["ExitBySignal"] = False
187 ad["ExitCode"] = ad["HoldReasonCode"]
188 else:
189 _LOG.debug(
190 "Handler '%s': refusing to process the ad for the job '%s.%s': "
191 "invalid hold reason code: HoldReasonCode = %s",
192 self.__class__.__name__,
193 ad["ClusterId"],
194 ad["ProcId"],
195 ad["HoldReasonCode"],
196 )
197 return None
198 return ad
201class JobHeldBySignalHandler(Handler):
202 """Handler of ClassAds for jobs put on hold by signals."""
204 def handle(self, ad: dict[str, Any]) -> dict[str, Any] | None:
205 if not ad["MyType"].endswith("HeldEvent"):
206 _LOG.debug(
207 "Handler '%s': refusing to process the ad for the job '%s.%s': job not held",
208 self.__class__.__name__,
209 ad["ClusterId"],
210 ad["ProcId"],
211 )
212 return None
213 if ad["HoldReasonCode"] == 3:
214 match = re.search(r"signal (\d+)", ad["HoldReason"])
215 if match is not None:
216 ad["ExitBySignal"] = True
217 ad["ExitSignal"] = match.group(1)
218 else:
219 _LOG.debug(
220 "Handler '%s': refusing to process the ad for the job '%s.%s': "
221 "signal not found: HoldReason = %s",
222 self.__class__.__name__,
223 ad["ClusterId"],
224 ad["ProcId"],
225 ad["HoldReason"],
226 )
227 return None
228 else:
229 _LOG.debug(
230 "Handler '%s': refusing to process the ad for the job '%s.%s': "
231 "job not held by a signal: HoldReasonCode = %s, HoldReason = %s",
232 self.__class__.__name__,
233 ad["ClusterId"],
234 ad["ProcId"],
235 ad["HoldReasonCode"],
236 ad["HoldReason"],
237 )
238 return None
239 return ad
242class JobHeldByUserHandler(Handler):
243 """Handler of ClassAds for jobs put on hold by the user."""
245 def handle(self, ad: dict[str, Any]) -> dict[str, Any] | None:
246 if not ad["MyType"].endswith("HeldEvent"):
247 _LOG.debug(
248 "Handler '%s': refusing to process the ad for the job '%s.%s': job not held",
249 self.__class__.__name__,
250 ad["ClusterId"],
251 ad["ProcId"],
252 )
253 return None
254 if ad["HoldReasonCode"] == 1:
255 ad["ExitBySignal"] = False
256 ad["ExitCode"] = 0
257 else:
258 _LOG.debug(
259 "Handler '%s': refusing to process the ad for the job '%s.%s': "
260 "job not held by the user: HoldReasonCode = %s, HoldReason = %s",
261 self.__class__.__name__,
262 ad["ClusterId"],
263 ad["ProcId"],
264 ad["HoldReasonCode"],
265 ad["HoldReason"],
266 )
267 return None
268 return ad
271class JobAbortedByPeriodicRemoveHandler(Handler):
272 """Handler of ClassAds for jobs deleted by periodic remove policy."""
274 def handle(self, ad: dict[str, Any]) -> dict[str, Any] | None:
275 if not ad["MyType"].endswith("AbortedEvent"):
276 _LOG.debug(
277 "Handler '%s': refusing to process the ad for the job '%s.%s': job not removed",
278 self.__class__.__name__,
279 ad["ClusterId"],
280 ad["ProcId"],
281 )
282 return None
283 if "Reason" in ad:
284 if "PeriodicRemove" in ad["Reason"]:
285 ad["ExitBySignal"] = True
287 ad["ExitSignal"] = -1
288 if "HoldReason" in ad:
289 match = re.search(r"signal (\d+)", ad["HoldReason"])
290 if match is not None:
291 ad["ExitSignal"] = int(match.group(1))
293 else:
294 _LOG.debug(
295 "Handler '%s': refusing to process the ad for the job '%s.%s': "
296 "job was not removed by the periodic removal policy: Reason = %s",
297 self.__class__.__name__,
298 ad["ClusterId"],
299 ad["ProcId"],
300 ad["Reason"],
301 )
302 return None
303 else:
304 _LOG.debug(
305 "Handler '%s': refusing to process the ad for the job '%s.%s': "
306 "unable to determine the reason for the removal.",
307 self.__class__.__name__,
308 ad["ClusterId"],
309 ad["ProcId"],
310 )
311 return None
312 return ad
315class JobAbortedByUserHandler(Handler):
316 """Handler of ClassAds for jobs deleted by the user."""
318 def handle(self, ad: dict[str, Any]) -> dict[str, Any] | None:
319 if not ad["MyType"].endswith("AbortedEvent"):
320 _LOG.debug(
321 "Handler '%s': refusing to process the ad for the job '%s.%s': job not removed",
322 self.__class__.__name__,
323 ad["ClusterId"],
324 ad["ProcId"],
325 )
326 return None
327 if "Reason" in ad:
328 patterns = (
329 "Python-initiated action", # DAGMan job removed by the user
330 "DAG Removed", # payload job removed by the user
331 "OtherJobRemoveRequirements", # a subdag job removed by the user
332 )
333 for patt in patterns:
334 if patt in ad["Reason"]:
335 ad["ExitBySignal"] = False
336 ad["ExitCode"] = 0
337 break
338 else:
339 _LOG.debug(
340 "Handler '%s': refusing to process the ad for the job '%s.%s': "
341 "job not removed by the user: Reason = %s",
342 self.__class__.__name__,
343 ad["ClusterId"],
344 ad["ProcId"],
345 ad["Reason"],
346 )
347 return None
348 else:
349 _LOG.debug(
350 "Handler '%s': refusing to process the ad for the job '%s.%s': "
351 "unable to determine the reason for the removal.",
352 self.__class__.__name__,
353 ad["ClusterId"],
354 ad["ProcId"],
355 )
356 return None
357 return ad
360_handlers = [
361 JobAbortedByPeriodicRemoveHandler(),
362 JobAbortedByUserHandler(),
363 JobHeldByUserHandler(),
364 JobHeldBySignalHandler(),
365 JobHeldByOtherHandler(),
366 JobCompletedHandler(),
367]
368HTC_JOB_AD_HANDLERS = Chain(handlers=_handlers)