Coverage for python/lsst/ctrl/bps/htcondor/htcondor_service.py: 34%
238 statements
« prev ^ index » next coverage.py v7.14.2, created at 2026-06-21 08:50 +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_file = htc_backup_files(wms_path, subdir="backups")
261 if (wms_path / "subdags").exists():
262 _update_rescue_file(rescue_file)
264 # For workflow portability, internal paths are all relative. Hence
265 # the DAG needs to be resubmitted to HTCondor from inside the submit
266 # directory.
267 _LOG.info("Adding workflow to the job queue")
268 run_id, run_name, message = None, None, ""
269 with chdir(wms_path):
270 try:
271 dag_path = next(Path.cwd().glob("*.dag.condor.sub"))
272 except StopIteration:
273 message = f"DAGMan submit description file not found in '{wms_path}'"
274 else:
275 sub = htc_create_submit_from_file(dag_path.name)
276 schedd_dag_info = htc_submit_dag(sub)
278 # Save select information about the DAGMan job to a file. Use
279 # the run name (available in the ClassAd) as the filename.
280 if schedd_dag_info:
281 dag_info = next(iter(schedd_dag_info.values()))
282 dag_ad = next(iter(dag_info.values()))
283 write_dag_info(f"{dag_ad['bps_run']}.info.json", schedd_dag_info)
284 run_id = f"{dag_ad['ClusterId']}.{dag_ad['ProcId']}"
285 run_name = dag_ad["bps_run"]
286 else:
287 message = "DAGMan job information unavailable"
289 return run_id, run_name, message
291 def list_submitted_jobs(self, wms_id=None, user=None, require_bps=True, pass_thru=None, is_global=False):
292 """Query WMS for list of submitted WMS workflows/jobs.
294 This should be a quick lookup function to create list of jobs for
295 other functions.
297 Parameters
298 ----------
299 wms_id : `int` or `str`, optional
300 Id or path that can be used by WMS service to look up job.
301 user : `str`, optional
302 User whose submitted jobs should be listed.
303 require_bps : `bool`, optional
304 Whether to require jobs returned in list to be bps-submitted jobs.
305 pass_thru : `str`, optional
306 Information to pass through to WMS.
307 is_global : `bool`, optional
308 If set, all job queues (and their histories) will be queried for
309 job information. Defaults to False which means that only the local
310 job queue will be queried.
312 Returns
313 -------
314 job_ids : `list` [`~typing.Any`]
315 Only job ids to be used by cancel and other functions. Typically
316 this means top-level jobs (i.e., not children jobs).
317 """
318 _LOG.debug(
319 "list_submitted_jobs params: wms_id=%s, user=%s, require_bps=%s, pass_thru=%s, is_global=%s",
320 wms_id,
321 user,
322 require_bps,
323 pass_thru,
324 is_global,
325 )
327 # Determine which Schedds will be queried for job information.
328 coll = htcondor.Collector()
330 schedd_ads = []
331 if is_global:
332 schedd_ads.extend(coll.locateAll(htcondor.DaemonTypes.Schedd))
333 else:
334 schedd_ads.append(coll.locate(htcondor.DaemonTypes.Schedd))
336 # Construct appropriate constraint expression using provided arguments.
337 constraint = "False"
338 if wms_id is None:
339 if user is not None:
340 constraint = f'(Owner == "{user}")'
341 else:
342 schedd_ad, cluster_id, id_type = _wms_id_to_cluster(wms_id)
343 if cluster_id is not None:
344 constraint = f"(DAGManJobId == {cluster_id} || ClusterId == {cluster_id})"
346 # If provided id is either a submission path or a global id,
347 # make sure the right Schedd will be queried regardless of
348 # 'is_global' value.
349 if id_type in {WmsIdType.GLOBAL, WmsIdType.PATH}:
350 schedd_ads = [schedd_ad]
351 if require_bps:
352 constraint += ' && (bps_isjob == "True")'
353 if pass_thru:
354 if "-forcex" in pass_thru:
355 pass_thru_2 = pass_thru.replace("-forcex", "")
356 if pass_thru_2 and not pass_thru_2.isspace():
357 constraint += f" && ({pass_thru_2})"
358 else:
359 constraint += f" && ({pass_thru})"
361 # Create a list of scheduler daemons which need to be queried.
362 schedds = {ad["Name"]: htcondor.Schedd(ad) for ad in schedd_ads}
364 _LOG.debug("constraint = %s, schedds = %s", constraint, ", ".join(schedds))
365 results = condor_q(constraint=constraint, schedds=schedds)
367 # Prune child jobs where DAG job is in queue (i.e., aren't orphans).
368 job_ids = []
369 for job_info in results.values():
370 for job_id, job_ad in job_info.items():
371 _LOG.debug("job_id=%s DAGManJobId=%s", job_id, job_ad.get("DAGManJobId", "None"))
372 if "DAGManJobId" not in job_ad:
373 job_ids.append(job_ad.get("GlobalJobId", job_id))
374 else:
375 _LOG.debug("Looking for %s", f"{job_ad['DAGManJobId']}.0")
376 _LOG.debug("\tin jobs.keys() = %s", job_info.keys())
377 if f"{job_ad['DAGManJobId']}.0" not in job_info: # orphaned job
378 job_ids.append(job_ad.get("GlobalJobId", job_id))
380 _LOG.debug("job_ids = %s", job_ids)
381 return job_ids
383 def get_status(
384 self,
385 wms_workflow_id: str,
386 hist: float = 1,
387 is_global: bool = False,
388 ) -> tuple[WmsStates, str]:
389 """Return status of run based upon given constraints.
391 Parameters
392 ----------
393 wms_workflow_id : `str`
394 Limit to specific run based on id (queue id or path).
395 hist : `float`, optional
396 Limit history search to this many days. Defaults to 1.
397 is_global : `bool`, optional
398 If set, all job queues (and their histories) will be queried for
399 job information. Defaults to False which means that only the local
400 job queue will be queried.
402 Returns
403 -------
404 state : `lsst.ctrl.bps.WmsStates`
405 Status of single run from given information.
406 message : `str`
407 Extra message for status command to print. This could be pointers
408 to documentation or to WMS specific commands.
409 """
410 _LOG.debug("get_status: id=%s, hist=%s, is_global=%s", wms_workflow_id, hist, is_global)
412 id_type = _wms_id_type(wms_workflow_id)
413 _LOG.debug("id_type = %s", id_type.name)
415 if id_type == WmsIdType.LOCAL:
416 schedulers = _locate_schedds(locate_all=is_global)
417 _LOG.debug("schedulers = %s", schedulers)
418 state, message = _get_status_from_id(wms_workflow_id, hist, schedds=schedulers)
419 elif id_type == WmsIdType.GLOBAL:
420 schedulers = _locate_schedds(locate_all=True)
421 _LOG.debug("schedulers = %s", schedulers)
422 state, message = _get_status_from_id(wms_workflow_id, hist, schedds=schedulers)
423 elif id_type == WmsIdType.PATH:
424 state, message = _get_status_from_path(wms_workflow_id)
425 else:
426 state, message = WmsStates.UNKNOWN, "Invalid job id"
427 _LOG.debug("state: %s, %s", state, message)
429 return state, message
431 def report(
432 self,
433 wms_workflow_id=None,
434 user=None,
435 hist=0,
436 pass_thru=None,
437 is_global=False,
438 return_exit_codes=False,
439 ):
440 """Return run information based upon given constraints.
442 Parameters
443 ----------
444 wms_workflow_id : `str`, optional
445 Limit to specific run based on id.
446 user : `str`, optional
447 Limit results to runs for this user.
448 hist : `float`, optional
449 Limit history search to this many days. Defaults to 0.
450 pass_thru : `str`, optional
451 Constraints to pass through to HTCondor.
452 is_global : `bool`, optional
453 If set, all job queues (and their histories) will be queried for
454 job information. Defaults to False which means that only the local
455 job queue will be queried.
456 return_exit_codes : `bool`, optional
457 If set, return exit codes related to jobs with a
458 non-success status. Defaults to False, which means that only
459 the summary state is returned.
461 Only applicable in the context of a WMS with associated
462 handlers to return exit codes from jobs.
464 Returns
465 -------
466 runs : `list` [`lsst.ctrl.bps.WmsRunReport`]
467 Information about runs from given job information.
468 message : `str`
469 Extra message for report command to print. This could be pointers
470 to documentation or to WMS specific commands.
471 """
472 if wms_workflow_id:
473 id_type = _wms_id_type(wms_workflow_id)
474 if id_type == WmsIdType.LOCAL:
475 schedulers = _locate_schedds(locate_all=is_global)
476 run_reports, message = _report_from_id(wms_workflow_id, hist, schedds=schedulers)
477 elif id_type == WmsIdType.GLOBAL:
478 schedulers = _locate_schedds(locate_all=True)
479 run_reports, message = _report_from_id(wms_workflow_id, hist, schedds=schedulers)
480 elif id_type == WmsIdType.PATH:
481 run_reports, message = _report_from_path(wms_workflow_id)
482 else:
483 run_reports, message = {}, "Invalid job id"
484 else:
485 schedulers = _locate_schedds(locate_all=is_global)
486 run_reports, message = _summary_report(user, hist, pass_thru, schedds=schedulers)
487 _LOG.debug("report: %s, %s", run_reports, message)
489 return list(run_reports.values()), message
491 def cancel(self, wms_id, pass_thru=None):
492 """Cancel submitted workflows/jobs.
494 Parameters
495 ----------
496 wms_id : `str`
497 Id or path of job that should be canceled.
498 pass_thru : `str`, optional
499 Information to pass through to WMS.
501 Returns
502 -------
503 deleted : `bool`
504 Whether successful deletion or not. Currently, if any doubt or any
505 individual jobs not deleted, return False.
506 message : `str`
507 Any message from WMS (e.g., error details).
508 """
509 _LOG.debug("Canceling wms_id = %s", wms_id)
511 schedd_ad, cluster_id, _ = _wms_id_to_cluster(wms_id)
513 if cluster_id is None:
514 deleted = False
515 message = "invalid id"
516 else:
517 _LOG.debug(
518 "Canceling job managed by schedd_name = %s with cluster_id = %s",
519 cluster_id,
520 schedd_ad["Name"],
521 )
522 schedd = htcondor.Schedd(schedd_ad)
524 constraint = f"ClusterId == {cluster_id}"
525 if pass_thru is not None and "-forcex" in pass_thru:
526 pass_thru_2 = pass_thru.replace("-forcex", "")
527 if pass_thru_2 and not pass_thru_2.isspace():
528 constraint += f"&& ({pass_thru_2})"
529 _LOG.debug("JobAction.RemoveX constraint = %s", constraint)
530 results = schedd.act(htcondor.JobAction.RemoveX, constraint)
531 else:
532 if pass_thru:
533 constraint += f"&& ({pass_thru})"
534 _LOG.debug("JobAction.Remove constraint = %s", constraint)
535 results = schedd.act(htcondor.JobAction.Remove, constraint)
536 _LOG.debug("Remove results: %s", results)
538 if results["TotalSuccess"] > 0 and results["TotalError"] == 0:
539 deleted = True
540 message = ""
541 else:
542 deleted = False
543 if results["TotalSuccess"] == 0 and results["TotalError"] == 0:
544 message = "no such bps job in batch queue"
545 else:
546 message = f"unknown problems deleting: {results}"
548 _LOG.debug("deleted: %s; message = %s", deleted, message)
549 return deleted, message
551 def ping(self, pass_thru):
552 """Check whether WMS services are up, reachable, and can authenticate
553 if authentication is required.
555 The services to be checked are those needed for submit, report, cancel,
556 restart, but ping cannot guarantee whether jobs would actually run
557 successfully.
559 Parameters
560 ----------
561 pass_thru : `str`, optional
562 Information to pass through to WMS.
564 Returns
565 -------
566 status : `int`
567 0 for success, non-zero for failure.
568 message : `str`
569 Any message from WMS (e.g., error details).
570 """
571 coll = htcondor.Collector()
572 secman = htcondor.SecMan()
573 status = 0
574 message = ""
575 _LOG.info("Not verifying that compute resources exist.")
576 try:
577 for daemon_type in [htcondor.DaemonTypes.Schedd, htcondor.DaemonTypes.Collector]:
578 _ = secman.ping(coll.locate(daemon_type))
579 except htcondor.HTCondorLocateError:
580 status = 1
581 message = f"Could not locate {daemon_type} service."
582 except htcondor.HTCondorIOError:
583 status = 1
584 message = f"Permission problem with {daemon_type} service."
585 return status, message