Coverage for python/lsst/ctrl/bps/htcondor/htcondor_service.py: 59%
261 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-09 10:31 +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 importlib.metadata import version
36from pathlib import Path
38from htcondor2 import Collector, DaemonTypes, JobAction, Schedd
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_compat import HTCondorException, ping
51from .htcondor_config import HTC_DEFAULTS_URI
52from .htcondor_workflow import HTCondorWorkflow
53from .lssthtc import (
54 _locate_schedds,
55 _update_rescue_file,
56 condor_q,
57 htc_backup_files,
58 htc_create_submit_from_dag,
59 htc_create_submit_from_file,
60 htc_submit_dag,
61 read_dag_info,
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.info("Using HTCondor Python bindings version %s", version("htcondor"))
107 _LOG.debug("out_prefix = '%s'", out_prefix)
108 with time_this(log=_LOG, level=logging.INFO, prefix=None, msg="Completed HTCondor workflow creation"):
109 _, enable_provisioning = config.search("provisionResources")
111 # If bps is doing provisioning, force a unique nodeset
112 # to reduce complications if user also manually does
113 # provisioning.
114 if enable_provisioning:
115 config[".bps_defined.nodeset"] = config[".bps_defined.timestamp"]
117 workflow = HTCondorWorkflow.from_generic_workflow(
118 config,
119 generic_workflow,
120 out_prefix,
121 f"{self.__class__.__module__}.{self.__class__.__name__}",
122 )
124 if enable_provisioning:
125 provisioner = Provisioner(config)
126 provisioner.configure()
127 provisioner.prepare("provisioningJob.bash", prefix=out_prefix)
128 provisioner.provision(workflow.dag)
130 try:
131 configurator = DagmanConfigurator(config)
132 except KeyError:
133 _LOG.debug(
134 "No DAGMan-specific settings were found in BPS config; "
135 "skipping writing DAG-specific configuration file."
136 )
137 else:
138 configurator.prepare("dagman.conf", prefix=out_prefix)
139 configurator.configure(workflow.dag)
141 with time_this(
142 log=_LOG, level=logging.INFO, prefix=None, msg="Completed writing out HTCondor workflow"
143 ):
144 workflow.write(out_prefix)
145 return workflow
147 def submit(self, workflow, **kwargs):
148 """Submit a single HTCondor workflow.
150 Parameters
151 ----------
152 workflow : `lsst.ctrl.bps.htcondor.HTCondorWorkflow`
153 A single HTCondor workflow to submit. run_id is updated after
154 successful submission to WMS.
155 **kwargs : `~typing.Any`
156 Keyword arguments for the options.
157 """
158 dag = workflow.dag
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 wms_config_path = None
166 if "bps_wms_config_path" in dag.graph["attr"]:
167 wms_config_path = dag.graph["attr"]["bps_wms_config_path"]
168 sub = htc_create_submit_from_dag(
169 dag.graph["dag_filename"], dag.graph["submit_options"], wms_config_path
170 )
171 except Exception:
172 _LOG.error(
173 "Problems creating HTCondor submit object from filename: %s", dag.graph["dag_filename"]
174 )
175 raise
177 _LOG.info("Submitting from directory: %s", os.getcwd())
178 schedd_dag_info = htc_submit_dag(sub)
179 if schedd_dag_info:
180 _, dag_info = next(iter(schedd_dag_info.items()))
181 dag_id, dag_ad = next(iter(dag_info.items()))
183 write_dag_info(f"{dag_ad['bps_run']}.info.json", schedd_dag_info)
185 dag.run_id = f"{dag_ad['ClusterId']}.{dag_ad['ProcId']}"
186 workflow.run_id = dag.run_id
187 else:
188 raise RuntimeError("Submission failed: unable to retrieve DAGMan job information")
190 def restart(self, wms_workflow_id):
191 """Restart a failed DAGMan workflow.
193 Parameters
194 ----------
195 wms_workflow_id : `str`
196 The directory with HTCondor files.
198 Returns
199 -------
200 run_id : `str`
201 HTCondor id of the restarted DAGMan job. If restart failed, it will
202 be set to None.
203 run_name : `str`
204 Name of the restarted workflow. If restart failed, it will be set
205 to None.
206 message : `str`
207 A message describing any issues encountered during the restart.
208 If there were no issues, an empty string is returned.
209 """
210 wms_path, id_type = _wms_id_to_dir(wms_workflow_id)
211 if wms_path is None:
212 return (
213 None,
214 None,
215 (
216 f"workflow with run id '{wms_workflow_id}' not found. "
217 "Hint: use run's submit directory as the id instead"
218 ),
219 )
221 if id_type in {WmsIdType.GLOBAL, WmsIdType.LOCAL}:
222 if not wms_path.is_dir(): 222 ↛ 225line 222 didn't jump to line 225 because the condition on line 222 was always true
223 return None, None, f"submit directory '{wms_path}' for run id '{wms_workflow_id}' not found."
225 _LOG.info("Restarting workflow from directory '%s'", wms_path)
226 rescue_dags = list(wms_path.glob("*.dag.rescue*"))
227 if not rescue_dags:
228 return None, None, f"HTCondor rescue DAG(s) not found in '{wms_path}'"
230 _LOG.info("Verifying that the workflow is not already in the job queue")
231 schedd_dag_info = condor_q(constraint=f'regexp("dagman$", Cmd) && Iwd == "{wms_path}"')
232 if schedd_dag_info:
233 _, dag_info = schedd_dag_info.popitem()
234 _, dag_ad = dag_info.popitem()
235 id_ = dag_ad["GlobalJobId"]
236 return None, None, f"Workflow already in the job queue (global job id: '{id_}')"
238 _LOG.info("Checking execution status of the workflow")
239 warn = False
240 dag_ad = read_dag_status(str(wms_path))
241 if dag_ad: 241 ↛ 250line 241 didn't jump to line 250 because the condition on line 241 was always true
242 nodes_total = dag_ad.get("NodesTotal", 0)
243 if nodes_total != 0: 243 ↛ 248line 243 didn't jump to line 248 because the condition on line 243 was always true
244 nodes_done = dag_ad.get("NodesDone", 0)
245 if nodes_total == nodes_done:
246 return None, None, "All jobs in the workflow finished successfully"
247 else:
248 warn = True
249 else:
250 warn = True
251 if warn: 251 ↛ 252line 251 didn't jump to line 252 because the condition on line 251 was never true
252 _LOG.warning(
253 "Cannot determine the execution status of the workflow, continuing with restart regardless"
254 )
256 # In the case of lazy DAGs, workflow summaries can change at
257 # runtime. So read the workflow's info.json file before moving
258 # it to backup dir and use to update the summaries later before
259 # writing the new info.json file.
260 dag_info_filename, old_dag_schedd_info = read_dag_info(wms_path)
261 old_dag_info = next(iter(old_dag_schedd_info.values()))
262 old_dag_ad = next(iter(old_dag_info.values()))
264 _LOG.info("Backing up select HTCondor files from previous run attempt")
265 rescue_files = sorted(wms_path.glob("*.rescue[0-9][0-9][0-9]"))
266 last_rescue_file = Path(rescue_files[-1]) if rescue_files else None
267 has_subdags = (wms_path / "subdags").exists()
268 failed_subdags = None
269 if last_rescue_file and has_subdags: 269 ↛ 270line 269 didn't jump to line 270 because the condition on line 269 was never true
270 failed_subdags = set(_update_rescue_file(last_rescue_file))
271 htc_backup_files(wms_path, subdir="backups", failed_subdags=failed_subdags)
273 # For workflow portability, internal paths are all relative. Hence
274 # the DAG needs to be resubmitted to HTCondor from inside the submit
275 # directory.
276 _LOG.info("Adding workflow to the job queue")
277 run_id, run_name, message = None, None, ""
278 with chdir(wms_path):
279 try:
280 dag_path = next(Path.cwd().glob("*.dag.condor.sub"))
281 except StopIteration:
282 message = f"DAGMan submit description file not found in '{wms_path}'"
283 else:
284 sub = htc_create_submit_from_file(dag_path.name)
285 schedd_dag_info = htc_submit_dag(sub)
287 # Save select information about the DAGMan job to a file. Use
288 # the run name (available in the ClassAd) as the filename.
289 if schedd_dag_info:
290 dag_info = next(iter(schedd_dag_info.values()))
291 dag_ad = next(iter(dag_info.values()))
293 # Just in case lazy DAGs, update the summaries.
294 dag_ad["bps_job_summary"] = old_dag_ad["bps_job_summary"]
295 dag_ad["bps_run_quanta"] = old_dag_ad["bps_run_quanta"]
297 write_dag_info(dag_info_filename, schedd_dag_info)
298 run_id = f"{dag_ad['ClusterId']}.{dag_ad['ProcId']}"
299 run_name = dag_ad["bps_run"]
300 else:
301 message = "DAGMan job information unavailable"
303 return run_id, run_name, message
305 def list_submitted_jobs(self, wms_id=None, user=None, require_bps=True, pass_thru=None, is_global=False):
306 """Query WMS for list of submitted WMS workflows/jobs.
308 This should be a quick lookup function to create list of jobs for
309 other functions.
311 Parameters
312 ----------
313 wms_id : `int` or `str`, optional
314 Id or path that can be used by WMS service to look up job.
315 user : `str`, optional
316 User whose submitted jobs should be listed.
317 require_bps : `bool`, optional
318 Whether to require jobs returned in list to be bps-submitted jobs.
319 pass_thru : `str`, optional
320 Information to pass through to WMS.
321 is_global : `bool`, optional
322 If set, all job queues (and their histories) will be queried for
323 job information. Defaults to False which means that only the local
324 job queue will be queried.
326 Returns
327 -------
328 job_ids : `list` [`~typing.Any`]
329 Only job ids to be used by cancel and other functions. Typically
330 this means top-level jobs (i.e., not children jobs).
331 """
332 _LOG.debug(
333 "list_submitted_jobs params: wms_id=%s, user=%s, require_bps=%s, pass_thru=%s, is_global=%s",
334 wms_id,
335 user,
336 require_bps,
337 pass_thru,
338 is_global,
339 )
341 # Determine which Schedds will be queried for job information.
342 coll = Collector()
344 schedd_ads = []
345 if is_global:
346 schedd_ads.extend(coll.locateAll(DaemonTypes.Schedd))
347 else:
348 schedd_ads.append(coll.locate(DaemonTypes.Schedd))
350 # Construct appropriate constraint expression using provided arguments.
351 constraint = "False"
352 if wms_id is None:
353 if user is not None:
354 constraint = f'(Owner == "{user}")'
355 else:
356 schedd_ad, cluster_id, id_type = _wms_id_to_cluster(wms_id)
357 if cluster_id is not None:
358 constraint = f"(DAGManJobId == {cluster_id} || ClusterId == {cluster_id})"
360 # If provided id is either a submission path or a global id,
361 # make sure the right Schedd will be queried regardless of
362 # 'is_global' value.
363 if id_type in {WmsIdType.GLOBAL, WmsIdType.PATH}:
364 schedd_ads = [schedd_ad]
365 if require_bps:
366 constraint += ' && (bps_isjob == "True")'
367 if pass_thru:
368 if "-forcex" in pass_thru:
369 pass_thru_2 = pass_thru.replace("-forcex", "")
370 if pass_thru_2 and not pass_thru_2.isspace():
371 constraint += f" && ({pass_thru_2})"
372 else:
373 constraint += f" && ({pass_thru})"
375 # Create a list of scheduler daemons which need to be queried.
376 schedds = {ad["Name"]: Schedd(ad) for ad in schedd_ads}
378 _LOG.debug("constraint = %s, schedds = %s", constraint, ", ".join(schedds))
379 results = condor_q(constraint=constraint, schedds=schedds)
381 # Prune child jobs where DAG job is in queue (i.e., aren't orphans).
382 job_ids = []
383 for job_info in results.values():
384 for job_id, job_ad in job_info.items():
385 _LOG.debug("job_id=%s DAGManJobId=%s", job_id, job_ad.get("DAGManJobId", "None"))
386 if "DAGManJobId" not in job_ad:
387 job_ids.append(job_ad.get("GlobalJobId", job_id))
388 else:
389 _LOG.debug("Looking for %s", f"{job_ad['DAGManJobId']}.0")
390 _LOG.debug("\tin jobs.keys() = %s", job_info.keys())
391 if f"{job_ad['DAGManJobId']}.0" not in job_info: # orphaned job
392 job_ids.append(job_ad.get("GlobalJobId", job_id))
394 _LOG.debug("job_ids = %s", job_ids)
395 return job_ids
397 def get_status(
398 self,
399 wms_workflow_id: str,
400 hist: float = 1,
401 is_global: bool = False,
402 ) -> tuple[WmsStates, str]:
403 """Return status of run based upon given constraints.
405 Parameters
406 ----------
407 wms_workflow_id : `str`
408 Limit to specific run based on id (queue id or path).
409 hist : `float`, optional
410 Limit history search to this many days. Defaults to 1.
411 is_global : `bool`, optional
412 If set, all job queues (and their histories) will be queried for
413 job information. Defaults to False which means that only the local
414 job queue will be queried.
416 Returns
417 -------
418 state : `lsst.ctrl.bps.WmsStates`
419 Status of single run from given information.
420 message : `str`
421 Extra message for status command to print. This could be pointers
422 to documentation or to WMS specific commands.
423 """
424 _LOG.debug("get_status: id=%s, hist=%s, is_global=%s", wms_workflow_id, hist, is_global)
426 id_type = _wms_id_type(wms_workflow_id)
427 _LOG.debug("id_type = %s", id_type.name)
429 if id_type == WmsIdType.LOCAL:
430 schedulers = _locate_schedds(locate_all=is_global)
431 _LOG.debug("schedulers = %s", schedulers)
432 state, message = _get_status_from_id(wms_workflow_id, hist, schedds=schedulers)
433 elif id_type == WmsIdType.GLOBAL:
434 schedulers = _locate_schedds(locate_all=True)
435 _LOG.debug("schedulers = %s", schedulers)
436 state, message = _get_status_from_id(wms_workflow_id, hist, schedds=schedulers)
437 elif id_type == WmsIdType.PATH:
438 state, message = _get_status_from_path(wms_workflow_id)
439 else:
440 state, message = WmsStates.UNKNOWN, "Invalid job id"
441 _LOG.debug("state: %s, %s", state, message)
443 return state, message
445 def report(
446 self,
447 wms_workflow_id=None,
448 user=None,
449 hist=0,
450 pass_thru=None,
451 is_global=False,
452 return_exit_codes=False,
453 ):
454 """Return run information based upon given constraints.
456 Parameters
457 ----------
458 wms_workflow_id : `str`, optional
459 Limit to specific run based on id.
460 user : `str`, optional
461 Limit results to runs for this user.
462 hist : `float`, optional
463 Limit history search to this many days. Defaults to 0.
464 pass_thru : `str`, optional
465 Constraints to pass through to HTCondor.
466 is_global : `bool`, optional
467 If set, all job queues (and their histories) will be queried for
468 job information. Defaults to False which means that only the local
469 job queue will be queried.
470 return_exit_codes : `bool`, optional
471 If set, return exit codes related to jobs with a
472 non-success status. Defaults to False, which means that only
473 the summary state is returned.
475 Only applicable in the context of a WMS with associated
476 handlers to return exit codes from jobs.
478 Returns
479 -------
480 runs : `list` [`lsst.ctrl.bps.WmsRunReport`]
481 Information about runs from given job information.
482 message : `str`
483 Extra message for report command to print. This could be pointers
484 to documentation or to WMS specific commands.
485 """
486 if wms_workflow_id:
487 id_type = _wms_id_type(wms_workflow_id)
488 if id_type == WmsIdType.LOCAL:
489 schedulers = _locate_schedds(locate_all=is_global)
490 run_reports, message = _report_from_id(wms_workflow_id, hist, schedds=schedulers)
491 elif id_type == WmsIdType.GLOBAL:
492 schedulers = _locate_schedds(locate_all=True)
493 run_reports, message = _report_from_id(wms_workflow_id, hist, schedds=schedulers)
494 elif id_type == WmsIdType.PATH:
495 run_reports, message = _report_from_path(wms_workflow_id)
496 else:
497 run_reports, message = {}, "Invalid job id"
498 else:
499 schedulers = _locate_schedds(locate_all=is_global)
500 run_reports, message = _summary_report(user, hist, pass_thru, schedds=schedulers)
501 _LOG.debug("report: %s, %s", run_reports, message)
503 return list(run_reports.values()), message
505 def cancel(self, wms_id, pass_thru=None):
506 """Cancel submitted workflows/jobs.
508 Parameters
509 ----------
510 wms_id : `str`
511 Id or path of job that should be canceled.
512 pass_thru : `str`, optional
513 Information to pass through to WMS.
515 Returns
516 -------
517 deleted : `bool`
518 Whether successful deletion or not. Currently, if any doubt or any
519 individual jobs not deleted, return False.
520 message : `str`
521 Any message from WMS (e.g., error details).
522 """
523 _LOG.debug("Canceling wms_id = %s", wms_id)
525 schedd_ad, cluster_id, _ = _wms_id_to_cluster(wms_id)
527 if cluster_id is None:
528 deleted = False
529 message = "invalid id"
530 else:
531 _LOG.debug(
532 "Canceling job managed by schedd_name = %s with cluster_id = %s",
533 cluster_id,
534 schedd_ad["Name"],
535 )
536 schedd = Schedd(schedd_ad)
538 constraint = f"ClusterId == {cluster_id}"
539 if pass_thru is not None and "-forcex" in pass_thru:
540 pass_thru_2 = pass_thru.replace("-forcex", "")
541 if pass_thru_2 and not pass_thru_2.isspace():
542 constraint += f"&& ({pass_thru_2})"
543 _LOG.debug("JobAction.RemoveX constraint = %s", constraint)
544 results = schedd.act(JobAction.RemoveX, constraint)
545 else:
546 if pass_thru:
547 constraint += f"&& ({pass_thru})"
548 _LOG.debug("JobAction.Remove constraint = %s", constraint)
549 results = schedd.act(JobAction.Remove, constraint)
550 _LOG.debug("Remove results: %s", results)
552 if results["TotalSuccess"] > 0 and results["TotalError"] == 0:
553 deleted = True
554 message = ""
555 else:
556 deleted = False
557 if results["TotalSuccess"] == 0 and results["TotalError"] == 0:
558 message = "no such bps job in batch queue"
559 else:
560 message = f"unknown problems deleting: {results}"
562 _LOG.debug("deleted: %s; message = %s", deleted, message)
563 return deleted, message
565 def ping(self, pass_thru):
566 """Check whether WMS services are up, reachable, and can authenticate
567 if authentication is required.
569 The services to be checked are those needed for submit, report, cancel,
570 restart, but ping cannot guarantee whether jobs would actually run
571 successfully.
573 Parameters
574 ----------
575 pass_thru : `str`, optional
576 Information to pass through to WMS.
578 Returns
579 -------
580 status : `int`
581 0 for success, non-zero for failure.
582 message : `str`
583 Any message from WMS (e.g., error details).
584 """
585 coll = Collector()
586 status = 0
587 message = ""
588 _LOG.info("Not verifying that compute resources exist.")
589 daemon_type = DaemonTypes.Schedd # To avoid possibly undefined error
590 try:
591 for daemon_type in [DaemonTypes.Schedd, DaemonTypes.Collector]:
592 ping(coll.locate(daemon_type))
593 except HTCondorException as e:
594 status = 1
595 if "unable to locate" in str(e).lower():
596 message = f"Could not locate {daemon_type.name} service."
597 elif "failed to connect" in str(e).lower(): 597 ↛ 599line 597 didn't jump to line 599 because the condition on line 597 was always true
598 message = f"Permission problem with {daemon_type.name} service."
599 return status, message
601 def run_submission_checks(self):
602 """Check to run at start if running WMS specific submission steps.
604 Any exception other than NotImplementedError will halt submission.
605 Submit directory may not yet exist when this is called.
606 """
607 # Some early config sanity checks
608 found, value = self.config.search("bpsMakeCommand")
609 bps_make_command = value if found else True
610 if not bps_make_command:
611 found, value = self.config.search("payloadCommand", opt={"replaceVars": False})
612 if not found:
613 raise KeyError("Missing 'payloadCommand' in config while bpsMakeCommand=True")
615 if "setupEnv" in value:
616 found, value = self.config.search("setupEnv", opt={"replaceVars": False})
617 if not found:
618 raise KeyError("Missing 'setupEnv' in config, but appears in payloadCommand")
620 if "lsstVersion" in value:
621 found, value = self.config.search("lsstVersion", opt={"replaceVars": False})
622 if not found:
623 raise KeyError("Missing 'lsstVersion' in config, but appears in setupEnv in config")