Coverage for python/lsst/ctrl/bps/htcondor/htcondor_service.py: 34%
242 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-22 17:25 +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"""Interface between generic workflow to HTCondor workflow system."""
30__all__ = ["HTCondorService"]
33import logging
34import os
35from pathlib import Path
37import htcondor
38from packaging import version
40from lsst.ctrl.bps import (
41 BaseWmsService,
42 WmsStates,
43)
44from lsst.ctrl.bps.bps_utils import chdir
45from lsst.daf.butler import Config
46from lsst.utils.timer import time_this
48from .common_utils import WmsIdType, _wms_id_to_cluster, _wms_id_to_dir, _wms_id_type
49from .dagman_configurator import DagmanConfigurator
50from .htcondor_config import HTC_DEFAULTS_URI
51from .htcondor_workflow import HTCondorWorkflow
52from .lssthtc import (
53 _locate_schedds,
54 _update_rescue_file,
55 condor_q,
56 htc_backup_files,
57 htc_create_submit_from_cmd,
58 htc_create_submit_from_dag,
59 htc_create_submit_from_file,
60 htc_submit_dag,
61 htc_version,
62 read_dag_status,
63 write_dag_info,
64)
65from .provisioner import Provisioner
66from .report_utils import (
67 _get_status_from_id,
68 _get_status_from_path,
69 _report_from_id,
70 _report_from_path,
71 _summary_report,
72)
74_LOG = logging.getLogger(__name__)
77class HTCondorService(BaseWmsService):
78 """HTCondor version of WMS service."""
80 @property
81 def defaults(self):
82 return Config(HTC_DEFAULTS_URI)
84 @property
85 def defaults_uri(self):
86 return HTC_DEFAULTS_URI
88 def prepare(self, config, generic_workflow, out_prefix=None):
89 """Convert generic workflow to an HTCondor DAG ready for submission.
91 Parameters
92 ----------
93 config : `lsst.ctrl.bps.BpsConfig`
94 BPS configuration that includes necessary submit/runtime
95 information.
96 generic_workflow : `lsst.ctrl.bps.GenericWorkflow`
97 The generic workflow (e.g., has executable name and arguments).
98 out_prefix : `str`
99 The root directory into which all WMS-specific files are written.
101 Returns
102 -------
103 workflow : `lsst.ctrl.bps.htcondor.HTCondorWorkflow`
104 HTCondor workflow ready to be run.
105 """
106 _LOG.debug("out_prefix = '%s'", out_prefix)
107 with time_this(log=_LOG, level=logging.INFO, prefix=None, msg="Completed HTCondor workflow creation"):
108 _, enable_provisioning = config.search("provisionResources")
110 # If bps is doing provisioning, force a unique nodeset
111 # to reduce complications if user also manually does
112 # provisioning.
113 if enable_provisioning:
114 config[".bps_defined.nodeset"] = config[".bps_defined.timestamp"]
116 workflow = HTCondorWorkflow.from_generic_workflow(
117 config,
118 generic_workflow,
119 out_prefix,
120 f"{self.__class__.__module__}.{self.__class__.__name__}",
121 )
123 if enable_provisioning:
124 provisioner = Provisioner(config)
125 provisioner.configure()
126 provisioner.prepare("provisioningJob.bash", prefix=out_prefix)
127 provisioner.provision(workflow.dag)
129 try:
130 configurator = DagmanConfigurator(config)
131 except KeyError:
132 _LOG.debug(
133 "No DAGMan-specific settings were found in BPS config; "
134 "skipping writing DAG-specific configuration file."
135 )
136 else:
137 configurator.prepare("dagman.conf", prefix=out_prefix)
138 configurator.configure(workflow.dag)
140 with time_this(
141 log=_LOG, level=logging.INFO, prefix=None, msg="Completed writing out HTCondor workflow"
142 ):
143 workflow.write(out_prefix)
144 return workflow
146 def submit(self, workflow, **kwargs):
147 """Submit a single HTCondor workflow.
149 Parameters
150 ----------
151 workflow : `lsst.ctrl.bps.htcondor.HTCondorWorkflow`
152 A single HTCondor workflow to submit. run_id is updated after
153 successful submission to WMS.
154 **kwargs : `~typing.Any`
155 Keyword arguments for the options.
156 """
157 dag = workflow.dag
158 ver = version.parse(htc_version())
160 # For workflow portability, internal paths are all relative. Hence
161 # the DAG needs to be submitted to HTCondor from inside the submit
162 # directory.
163 with chdir(workflow.submit_path):
164 try:
165 if ver >= version.parse("8.9.3"): 165 ↛ 173line 165 didn't jump to line 173 because the condition on line 165 was always true
166 wms_config_path = None
167 if "bps_wms_config_path" in dag.graph["attr"]:
168 wms_config_path = dag.graph["attr"]["bps_wms_config_path"]
169 sub = htc_create_submit_from_dag(
170 dag.graph["dag_filename"], dag.graph["submit_options"], wms_config_path
171 )
172 else:
173 sub = htc_create_submit_from_cmd(dag.graph["dag_filename"], dag.graph["submit_options"])
174 except Exception:
175 _LOG.error(
176 "Problems creating HTCondor submit object from filename: %s", dag.graph["dag_filename"]
177 )
178 raise
180 _LOG.info("Submitting from directory: %s", os.getcwd())
181 schedd_dag_info = htc_submit_dag(sub)
182 if schedd_dag_info:
183 write_dag_info(f"{dag.name}.info.json", schedd_dag_info)
185 _, dag_info = schedd_dag_info.popitem()
186 _, dag_ad = dag_info.popitem()
188 dag.run_id = f"{dag_ad['ClusterId']}.{dag_ad['ProcId']}"
189 workflow.run_id = dag.run_id
190 else:
191 raise RuntimeError("Submission failed: unable to retrieve DAGMan job information")
193 def restart(self, wms_workflow_id):
194 """Restart a failed DAGMan workflow.
196 Parameters
197 ----------
198 wms_workflow_id : `str`
199 The directory with HTCondor files.
201 Returns
202 -------
203 run_id : `str`
204 HTCondor id of the restarted DAGMan job. If restart failed, it will
205 be set to None.
206 run_name : `str`
207 Name of the restarted workflow. If restart failed, it will be set
208 to None.
209 message : `str`
210 A message describing any issues encountered during the restart.
211 If there were no issues, an empty string is returned.
212 """
213 wms_path, id_type = _wms_id_to_dir(wms_workflow_id)
214 if wms_path is None:
215 return (
216 None,
217 None,
218 (
219 f"workflow with run id '{wms_workflow_id}' not found. "
220 "Hint: use run's submit directory as the id instead"
221 ),
222 )
224 if id_type in {WmsIdType.GLOBAL, WmsIdType.LOCAL}:
225 if not wms_path.is_dir():
226 return None, None, f"submit directory '{wms_path}' for run id '{wms_workflow_id}' not found."
228 _LOG.info("Restarting workflow from directory '%s'", wms_path)
229 rescue_dags = list(wms_path.glob("*.dag.rescue*"))
230 if not rescue_dags:
231 return None, None, f"HTCondor rescue DAG(s) not found in '{wms_path}'"
233 _LOG.info("Verifying that the workflow is not already in the job queue")
234 schedd_dag_info = condor_q(constraint=f'regexp("dagman$", Cmd) && Iwd == "{wms_path}"')
235 if schedd_dag_info:
236 _, dag_info = schedd_dag_info.popitem()
237 _, dag_ad = dag_info.popitem()
238 id_ = dag_ad["GlobalJobId"]
239 return None, None, f"Workflow already in the job queue (global job id: '{id_}')"
241 _LOG.info("Checking execution status of the workflow")
242 warn = False
243 dag_ad = read_dag_status(str(wms_path))
244 if dag_ad:
245 nodes_total = dag_ad.get("NodesTotal", 0)
246 if nodes_total != 0:
247 nodes_done = dag_ad.get("NodesDone", 0)
248 if nodes_total == nodes_done:
249 return None, None, "All jobs in the workflow finished successfully"
250 else:
251 warn = True
252 else:
253 warn = True
254 if warn:
255 _LOG.warning(
256 "Cannot determine the execution status of the workflow, continuing with restart regardless"
257 )
259 _LOG.info("Backing up select HTCondor files from previous run attempt")
260 rescue_files = sorted(wms_path.glob("*.rescue[0-9][0-9][0-9]"))
261 last_rescue_file = Path(rescue_files[-1]) if rescue_files else None
262 has_subdags = (wms_path / "subdags").exists()
263 failed_subdags = None
264 if last_rescue_file and has_subdags:
265 failed_subdags = set(_update_rescue_file(last_rescue_file))
266 htc_backup_files(wms_path, subdir="backups", failed_subdags=failed_subdags)
268 # For workflow portability, internal paths are all relative. Hence
269 # the DAG needs to be resubmitted to HTCondor from inside the submit
270 # directory.
271 _LOG.info("Adding workflow to the job queue")
272 run_id, run_name, message = None, None, ""
273 with chdir(wms_path):
274 try:
275 dag_path = next(Path.cwd().glob("*.dag.condor.sub"))
276 except StopIteration:
277 message = f"DAGMan submit description file not found in '{wms_path}'"
278 else:
279 sub = htc_create_submit_from_file(dag_path.name)
280 schedd_dag_info = htc_submit_dag(sub)
282 # Save select information about the DAGMan job to a file. Use
283 # the run name (available in the ClassAd) as the filename.
284 if schedd_dag_info:
285 dag_info = next(iter(schedd_dag_info.values()))
286 dag_ad = next(iter(dag_info.values()))
287 write_dag_info(f"{dag_ad['bps_run']}.info.json", schedd_dag_info)
288 run_id = f"{dag_ad['ClusterId']}.{dag_ad['ProcId']}"
289 run_name = dag_ad["bps_run"]
290 else:
291 message = "DAGMan job information unavailable"
293 return run_id, run_name, message
295 def list_submitted_jobs(self, wms_id=None, user=None, require_bps=True, pass_thru=None, is_global=False):
296 """Query WMS for list of submitted WMS workflows/jobs.
298 This should be a quick lookup function to create list of jobs for
299 other functions.
301 Parameters
302 ----------
303 wms_id : `int` or `str`, optional
304 Id or path that can be used by WMS service to look up job.
305 user : `str`, optional
306 User whose submitted jobs should be listed.
307 require_bps : `bool`, optional
308 Whether to require jobs returned in list to be bps-submitted jobs.
309 pass_thru : `str`, optional
310 Information to pass through to WMS.
311 is_global : `bool`, optional
312 If set, all job queues (and their histories) will be queried for
313 job information. Defaults to False which means that only the local
314 job queue will be queried.
316 Returns
317 -------
318 job_ids : `list` [`~typing.Any`]
319 Only job ids to be used by cancel and other functions. Typically
320 this means top-level jobs (i.e., not children jobs).
321 """
322 _LOG.debug(
323 "list_submitted_jobs params: wms_id=%s, user=%s, require_bps=%s, pass_thru=%s, is_global=%s",
324 wms_id,
325 user,
326 require_bps,
327 pass_thru,
328 is_global,
329 )
331 # Determine which Schedds will be queried for job information.
332 coll = htcondor.Collector()
334 schedd_ads = []
335 if is_global:
336 schedd_ads.extend(coll.locateAll(htcondor.DaemonTypes.Schedd))
337 else:
338 schedd_ads.append(coll.locate(htcondor.DaemonTypes.Schedd))
340 # Construct appropriate constraint expression using provided arguments.
341 constraint = "False"
342 if wms_id is None:
343 if user is not None:
344 constraint = f'(Owner == "{user}")'
345 else:
346 schedd_ad, cluster_id, id_type = _wms_id_to_cluster(wms_id)
347 if cluster_id is not None:
348 constraint = f"(DAGManJobId == {cluster_id} || ClusterId == {cluster_id})"
350 # If provided id is either a submission path or a global id,
351 # make sure the right Schedd will be queried regardless of
352 # 'is_global' value.
353 if id_type in {WmsIdType.GLOBAL, WmsIdType.PATH}:
354 schedd_ads = [schedd_ad]
355 if require_bps:
356 constraint += ' && (bps_isjob == "True")'
357 if pass_thru:
358 if "-forcex" in pass_thru:
359 pass_thru_2 = pass_thru.replace("-forcex", "")
360 if pass_thru_2 and not pass_thru_2.isspace():
361 constraint += f" && ({pass_thru_2})"
362 else:
363 constraint += f" && ({pass_thru})"
365 # Create a list of scheduler daemons which need to be queried.
366 schedds = {ad["Name"]: htcondor.Schedd(ad) for ad in schedd_ads}
368 _LOG.debug("constraint = %s, schedds = %s", constraint, ", ".join(schedds))
369 results = condor_q(constraint=constraint, schedds=schedds)
371 # Prune child jobs where DAG job is in queue (i.e., aren't orphans).
372 job_ids = []
373 for job_info in results.values():
374 for job_id, job_ad in job_info.items():
375 _LOG.debug("job_id=%s DAGManJobId=%s", job_id, job_ad.get("DAGManJobId", "None"))
376 if "DAGManJobId" not in job_ad:
377 job_ids.append(job_ad.get("GlobalJobId", job_id))
378 else:
379 _LOG.debug("Looking for %s", f"{job_ad['DAGManJobId']}.0")
380 _LOG.debug("\tin jobs.keys() = %s", job_info.keys())
381 if f"{job_ad['DAGManJobId']}.0" not in job_info: # orphaned job
382 job_ids.append(job_ad.get("GlobalJobId", job_id))
384 _LOG.debug("job_ids = %s", job_ids)
385 return job_ids
387 def get_status(
388 self,
389 wms_workflow_id: str,
390 hist: float = 1,
391 is_global: bool = False,
392 ) -> tuple[WmsStates, str]:
393 """Return status of run based upon given constraints.
395 Parameters
396 ----------
397 wms_workflow_id : `str`
398 Limit to specific run based on id (queue id or path).
399 hist : `float`, optional
400 Limit history search to this many days. Defaults to 1.
401 is_global : `bool`, optional
402 If set, all job queues (and their histories) will be queried for
403 job information. Defaults to False which means that only the local
404 job queue will be queried.
406 Returns
407 -------
408 state : `lsst.ctrl.bps.WmsStates`
409 Status of single run from given information.
410 message : `str`
411 Extra message for status command to print. This could be pointers
412 to documentation or to WMS specific commands.
413 """
414 _LOG.debug("get_status: id=%s, hist=%s, is_global=%s", wms_workflow_id, hist, is_global)
416 id_type = _wms_id_type(wms_workflow_id)
417 _LOG.debug("id_type = %s", id_type.name)
419 if id_type == WmsIdType.LOCAL:
420 schedulers = _locate_schedds(locate_all=is_global)
421 _LOG.debug("schedulers = %s", schedulers)
422 state, message = _get_status_from_id(wms_workflow_id, hist, schedds=schedulers)
423 elif id_type == WmsIdType.GLOBAL:
424 schedulers = _locate_schedds(locate_all=True)
425 _LOG.debug("schedulers = %s", schedulers)
426 state, message = _get_status_from_id(wms_workflow_id, hist, schedds=schedulers)
427 elif id_type == WmsIdType.PATH:
428 state, message = _get_status_from_path(wms_workflow_id)
429 else:
430 state, message = WmsStates.UNKNOWN, "Invalid job id"
431 _LOG.debug("state: %s, %s", state, message)
433 return state, message
435 def report(
436 self,
437 wms_workflow_id=None,
438 user=None,
439 hist=0,
440 pass_thru=None,
441 is_global=False,
442 return_exit_codes=False,
443 ):
444 """Return run information based upon given constraints.
446 Parameters
447 ----------
448 wms_workflow_id : `str`, optional
449 Limit to specific run based on id.
450 user : `str`, optional
451 Limit results to runs for this user.
452 hist : `float`, optional
453 Limit history search to this many days. Defaults to 0.
454 pass_thru : `str`, optional
455 Constraints to pass through to HTCondor.
456 is_global : `bool`, optional
457 If set, all job queues (and their histories) will be queried for
458 job information. Defaults to False which means that only the local
459 job queue will be queried.
460 return_exit_codes : `bool`, optional
461 If set, return exit codes related to jobs with a
462 non-success status. Defaults to False, which means that only
463 the summary state is returned.
465 Only applicable in the context of a WMS with associated
466 handlers to return exit codes from jobs.
468 Returns
469 -------
470 runs : `list` [`lsst.ctrl.bps.WmsRunReport`]
471 Information about runs from given job information.
472 message : `str`
473 Extra message for report command to print. This could be pointers
474 to documentation or to WMS specific commands.
475 """
476 if wms_workflow_id:
477 id_type = _wms_id_type(wms_workflow_id)
478 if id_type == WmsIdType.LOCAL:
479 schedulers = _locate_schedds(locate_all=is_global)
480 run_reports, message = _report_from_id(wms_workflow_id, hist, schedds=schedulers)
481 elif id_type == WmsIdType.GLOBAL:
482 schedulers = _locate_schedds(locate_all=True)
483 run_reports, message = _report_from_id(wms_workflow_id, hist, schedds=schedulers)
484 elif id_type == WmsIdType.PATH:
485 run_reports, message = _report_from_path(wms_workflow_id)
486 else:
487 run_reports, message = {}, "Invalid job id"
488 else:
489 schedulers = _locate_schedds(locate_all=is_global)
490 run_reports, message = _summary_report(user, hist, pass_thru, schedds=schedulers)
491 _LOG.debug("report: %s, %s", run_reports, message)
493 return list(run_reports.values()), message
495 def cancel(self, wms_id, pass_thru=None):
496 """Cancel submitted workflows/jobs.
498 Parameters
499 ----------
500 wms_id : `str`
501 Id or path of job that should be canceled.
502 pass_thru : `str`, optional
503 Information to pass through to WMS.
505 Returns
506 -------
507 deleted : `bool`
508 Whether successful deletion or not. Currently, if any doubt or any
509 individual jobs not deleted, return False.
510 message : `str`
511 Any message from WMS (e.g., error details).
512 """
513 _LOG.debug("Canceling wms_id = %s", wms_id)
515 schedd_ad, cluster_id, _ = _wms_id_to_cluster(wms_id)
517 if cluster_id is None:
518 deleted = False
519 message = "invalid id"
520 else:
521 _LOG.debug(
522 "Canceling job managed by schedd_name = %s with cluster_id = %s",
523 cluster_id,
524 schedd_ad["Name"],
525 )
526 schedd = htcondor.Schedd(schedd_ad)
528 constraint = f"ClusterId == {cluster_id}"
529 if pass_thru is not None and "-forcex" in pass_thru:
530 pass_thru_2 = pass_thru.replace("-forcex", "")
531 if pass_thru_2 and not pass_thru_2.isspace():
532 constraint += f"&& ({pass_thru_2})"
533 _LOG.debug("JobAction.RemoveX constraint = %s", constraint)
534 results = schedd.act(htcondor.JobAction.RemoveX, constraint)
535 else:
536 if pass_thru:
537 constraint += f"&& ({pass_thru})"
538 _LOG.debug("JobAction.Remove constraint = %s", constraint)
539 results = schedd.act(htcondor.JobAction.Remove, constraint)
540 _LOG.debug("Remove results: %s", results)
542 if results["TotalSuccess"] > 0 and results["TotalError"] == 0:
543 deleted = True
544 message = ""
545 else:
546 deleted = False
547 if results["TotalSuccess"] == 0 and results["TotalError"] == 0:
548 message = "no such bps job in batch queue"
549 else:
550 message = f"unknown problems deleting: {results}"
552 _LOG.debug("deleted: %s; message = %s", deleted, message)
553 return deleted, message
555 def ping(self, pass_thru):
556 """Check whether WMS services are up, reachable, and can authenticate
557 if authentication is required.
559 The services to be checked are those needed for submit, report, cancel,
560 restart, but ping cannot guarantee whether jobs would actually run
561 successfully.
563 Parameters
564 ----------
565 pass_thru : `str`, optional
566 Information to pass through to WMS.
568 Returns
569 -------
570 status : `int`
571 0 for success, non-zero for failure.
572 message : `str`
573 Any message from WMS (e.g., error details).
574 """
575 coll = htcondor.Collector()
576 secman = htcondor.SecMan()
577 status = 0
578 message = ""
579 _LOG.info("Not verifying that compute resources exist.")
580 try:
581 for daemon_type in [htcondor.DaemonTypes.Schedd, htcondor.DaemonTypes.Collector]:
582 _ = secman.ping(coll.locate(daemon_type))
583 except htcondor.HTCondorLocateError:
584 status = 1
585 message = f"Could not locate {daemon_type} service."
586 except htcondor.HTCondorIOError:
587 status = 1
588 message = f"Permission problem with {daemon_type} service."
589 return status, message