Coverage for python / lsst / ctrl / bps / htcondor / htcondor_service.py: 10%
233 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-27 01:28 -0800
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.wms.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 workflow = HTCondorWorkflow.from_generic_workflow(
109 config,
110 generic_workflow,
111 out_prefix,
112 f"{self.__class__.__module__}.{self.__class__.__name__}",
113 )
115 _, enable_provisioning = config.search("provisionResources")
116 if enable_provisioning:
117 provisioner = Provisioner(config)
118 provisioner.configure()
119 provisioner.prepare("provisioningJob.bash", prefix=out_prefix)
120 provisioner.provision(workflow.dag)
122 try:
123 configurator = DagmanConfigurator(config)
124 except KeyError:
125 _LOG.debug(
126 "No DAGMan-specific settings were found in BPS config; "
127 "skipping writing DAG-specific configuration file."
128 )
129 else:
130 configurator.prepare("dagman.conf", prefix=out_prefix)
131 configurator.configure(workflow.dag)
133 with time_this(
134 log=_LOG, level=logging.INFO, prefix=None, msg="Completed writing out HTCondor workflow"
135 ):
136 workflow.write(out_prefix)
137 return workflow
139 def submit(self, workflow, **kwargs):
140 """Submit a single HTCondor workflow.
142 Parameters
143 ----------
144 workflow : `lsst.ctrl.bps.BaseWorkflow`
145 A single HTCondor workflow to submit. run_id is updated after
146 successful submission to WMS.
147 **kwargs : `~typing.Any`
148 Keyword arguments for the options.
149 """
150 dag = workflow.dag
151 ver = version.parse(htc_version())
153 # For workflow portability, internal paths are all relative. Hence
154 # the DAG needs to be submitted to HTCondor from inside the submit
155 # directory.
156 with chdir(workflow.submit_path):
157 try:
158 if ver >= version.parse("8.9.3"):
159 sub = htc_create_submit_from_dag(dag.graph["dag_filename"], dag.graph["submit_options"])
160 else:
161 sub = htc_create_submit_from_cmd(dag.graph["dag_filename"], dag.graph["submit_options"])
162 except Exception:
163 _LOG.error(
164 "Problems creating HTCondor submit object from filename: %s", dag.graph["dag_filename"]
165 )
166 raise
168 _LOG.info("Submitting from directory: %s", os.getcwd())
169 schedd_dag_info = htc_submit_dag(sub)
170 if schedd_dag_info:
171 write_dag_info(f"{dag.name}.info.json", schedd_dag_info)
173 _, dag_info = schedd_dag_info.popitem()
174 _, dag_ad = dag_info.popitem()
176 dag.run_id = f"{dag_ad['ClusterId']}.{dag_ad['ProcId']}"
177 workflow.run_id = dag.run_id
178 else:
179 raise RuntimeError("Submission failed: unable to retrieve DAGMan job information")
181 def restart(self, wms_workflow_id):
182 """Restart a failed DAGMan workflow.
184 Parameters
185 ----------
186 wms_workflow_id : `str`
187 The directory with HTCondor files.
189 Returns
190 -------
191 run_id : `str`
192 HTCondor id of the restarted DAGMan job. If restart failed, it will
193 be set to None.
194 run_name : `str`
195 Name of the restarted workflow. If restart failed, it will be set
196 to None.
197 message : `str`
198 A message describing any issues encountered during the restart.
199 If there were no issues, an empty string is returned.
200 """
201 wms_path, id_type = _wms_id_to_dir(wms_workflow_id)
202 if wms_path is None:
203 return (
204 None,
205 None,
206 (
207 f"workflow with run id '{wms_workflow_id}' not found. "
208 "Hint: use run's submit directory as the id instead"
209 ),
210 )
212 if id_type in {WmsIdType.GLOBAL, WmsIdType.LOCAL}:
213 if not wms_path.is_dir():
214 return None, None, f"submit directory '{wms_path}' for run id '{wms_workflow_id}' not found."
216 _LOG.info("Restarting workflow from directory '%s'", wms_path)
217 rescue_dags = list(wms_path.glob("*.dag.rescue*"))
218 if not rescue_dags:
219 return None, None, f"HTCondor rescue DAG(s) not found in '{wms_path}'"
221 _LOG.info("Verifying that the workflow is not already in the job queue")
222 schedd_dag_info = condor_q(constraint=f'regexp("dagman$", Cmd) && Iwd == "{wms_path}"')
223 if schedd_dag_info:
224 _, dag_info = schedd_dag_info.popitem()
225 _, dag_ad = dag_info.popitem()
226 id_ = dag_ad["GlobalJobId"]
227 return None, None, f"Workflow already in the job queue (global job id: '{id_}')"
229 _LOG.info("Checking execution status of the workflow")
230 warn = False
231 dag_ad = read_dag_status(str(wms_path))
232 if dag_ad:
233 nodes_total = dag_ad.get("NodesTotal", 0)
234 if nodes_total != 0:
235 nodes_done = dag_ad.get("NodesDone", 0)
236 if nodes_total == nodes_done:
237 return None, None, "All jobs in the workflow finished successfully"
238 else:
239 warn = True
240 else:
241 warn = True
242 if warn:
243 _LOG.warning(
244 "Cannot determine the execution status of the workflow, continuing with restart regardless"
245 )
247 _LOG.info("Backing up select HTCondor files from previous run attempt")
248 rescue_file = htc_backup_files(wms_path, subdir="backups")
249 if (wms_path / "subdags").exists():
250 _update_rescue_file(rescue_file)
252 # For workflow portability, internal paths are all relative. Hence
253 # the DAG needs to be resubmitted to HTCondor from inside the submit
254 # directory.
255 _LOG.info("Adding workflow to the job queue")
256 run_id, run_name, message = None, None, ""
257 with chdir(wms_path):
258 try:
259 dag_path = next(Path.cwd().glob("*.dag.condor.sub"))
260 except StopIteration:
261 message = f"DAGMan submit description file not found in '{wms_path}'"
262 else:
263 sub = htc_create_submit_from_file(dag_path.name)
264 schedd_dag_info = htc_submit_dag(sub)
266 # Save select information about the DAGMan job to a file. Use
267 # the run name (available in the ClassAd) as the filename.
268 if schedd_dag_info:
269 dag_info = next(iter(schedd_dag_info.values()))
270 dag_ad = next(iter(dag_info.values()))
271 write_dag_info(f"{dag_ad['bps_run']}.info.json", schedd_dag_info)
272 run_id = f"{dag_ad['ClusterId']}.{dag_ad['ProcId']}"
273 run_name = dag_ad["bps_run"]
274 else:
275 message = "DAGMan job information unavailable"
277 return run_id, run_name, message
279 def list_submitted_jobs(self, wms_id=None, user=None, require_bps=True, pass_thru=None, is_global=False):
280 """Query WMS for list of submitted WMS workflows/jobs.
282 This should be a quick lookup function to create list of jobs for
283 other functions.
285 Parameters
286 ----------
287 wms_id : `int` or `str`, optional
288 Id or path that can be used by WMS service to look up job.
289 user : `str`, optional
290 User whose submitted jobs should be listed.
291 require_bps : `bool`, optional
292 Whether to require jobs returned in list to be bps-submitted jobs.
293 pass_thru : `str`, optional
294 Information to pass through to WMS.
295 is_global : `bool`, optional
296 If set, all job queues (and their histories) will be queried for
297 job information. Defaults to False which means that only the local
298 job queue will be queried.
300 Returns
301 -------
302 job_ids : `list` [`~typing.Any`]
303 Only job ids to be used by cancel and other functions. Typically
304 this means top-level jobs (i.e., not children jobs).
305 """
306 _LOG.debug(
307 "list_submitted_jobs params: wms_id=%s, user=%s, require_bps=%s, pass_thru=%s, is_global=%s",
308 wms_id,
309 user,
310 require_bps,
311 pass_thru,
312 is_global,
313 )
315 # Determine which Schedds will be queried for job information.
316 coll = htcondor.Collector()
318 schedd_ads = []
319 if is_global:
320 schedd_ads.extend(coll.locateAll(htcondor.DaemonTypes.Schedd))
321 else:
322 schedd_ads.append(coll.locate(htcondor.DaemonTypes.Schedd))
324 # Construct appropriate constraint expression using provided arguments.
325 constraint = "False"
326 if wms_id is None:
327 if user is not None:
328 constraint = f'(Owner == "{user}")'
329 else:
330 schedd_ad, cluster_id, id_type = _wms_id_to_cluster(wms_id)
331 if cluster_id is not None:
332 constraint = f"(DAGManJobId == {cluster_id} || ClusterId == {cluster_id})"
334 # If provided id is either a submission path or a global id,
335 # make sure the right Schedd will be queried regardless of
336 # 'is_global' value.
337 if id_type in {WmsIdType.GLOBAL, WmsIdType.PATH}:
338 schedd_ads = [schedd_ad]
339 if require_bps:
340 constraint += ' && (bps_isjob == "True")'
341 if pass_thru:
342 if "-forcex" in pass_thru:
343 pass_thru_2 = pass_thru.replace("-forcex", "")
344 if pass_thru_2 and not pass_thru_2.isspace():
345 constraint += f" && ({pass_thru_2})"
346 else:
347 constraint += f" && ({pass_thru})"
349 # Create a list of scheduler daemons which need to be queried.
350 schedds = {ad["Name"]: htcondor.Schedd(ad) for ad in schedd_ads}
352 _LOG.debug("constraint = %s, schedds = %s", constraint, ", ".join(schedds))
353 results = condor_q(constraint=constraint, schedds=schedds)
355 # Prune child jobs where DAG job is in queue (i.e., aren't orphans).
356 job_ids = []
357 for job_info in results.values():
358 for job_id, job_ad in job_info.items():
359 _LOG.debug("job_id=%s DAGManJobId=%s", job_id, job_ad.get("DAGManJobId", "None"))
360 if "DAGManJobId" not in job_ad:
361 job_ids.append(job_ad.get("GlobalJobId", job_id))
362 else:
363 _LOG.debug("Looking for %s", f"{job_ad['DAGManJobId']}.0")
364 _LOG.debug("\tin jobs.keys() = %s", job_info.keys())
365 if f"{job_ad['DAGManJobId']}.0" not in job_info: # orphaned job
366 job_ids.append(job_ad.get("GlobalJobId", job_id))
368 _LOG.debug("job_ids = %s", job_ids)
369 return job_ids
371 def get_status(
372 self,
373 wms_workflow_id: str,
374 hist: float = 1,
375 is_global: bool = False,
376 ) -> tuple[WmsStates, str]:
377 """Return status of run based upon given constraints.
379 Parameters
380 ----------
381 wms_workflow_id : `str`
382 Limit to specific run based on id (queue id or path).
383 hist : `float`, optional
384 Limit history search to this many days. Defaults to 1.
385 is_global : `bool`, optional
386 If set, all job queues (and their histories) will be queried for
387 job information. Defaults to False which means that only the local
388 job queue will be queried.
390 Returns
391 -------
392 state : `lsst.ctrl.bps.WmsStates`
393 Status of single run from given information.
394 message : `str`
395 Extra message for status command to print. This could be pointers
396 to documentation or to WMS specific commands.
397 """
398 _LOG.debug("get_status: id=%s, hist=%s, is_global=%s", wms_workflow_id, hist, is_global)
400 id_type = _wms_id_type(wms_workflow_id)
401 _LOG.debug("id_type = %s", id_type.name)
403 if id_type == WmsIdType.LOCAL:
404 schedulers = _locate_schedds(locate_all=is_global)
405 _LOG.debug("schedulers = %s", schedulers)
406 state, message = _get_status_from_id(wms_workflow_id, hist, schedds=schedulers)
407 elif id_type == WmsIdType.GLOBAL:
408 schedulers = _locate_schedds(locate_all=True)
409 _LOG.debug("schedulers = %s", schedulers)
410 state, message = _get_status_from_id(wms_workflow_id, hist, schedds=schedulers)
411 elif id_type == WmsIdType.PATH:
412 state, message = _get_status_from_path(wms_workflow_id)
413 else:
414 state, message = WmsStates.UNKNOWN, "Invalid job id"
415 _LOG.debug("state: %s, %s", state, message)
417 return state, message
419 def report(
420 self,
421 wms_workflow_id=None,
422 user=None,
423 hist=0,
424 pass_thru=None,
425 is_global=False,
426 return_exit_codes=False,
427 ):
428 """Return run information based upon given constraints.
430 Parameters
431 ----------
432 wms_workflow_id : `str`, optional
433 Limit to specific run based on id.
434 user : `str`, optional
435 Limit results to runs for this user.
436 hist : `float`, optional
437 Limit history search to this many days. Defaults to 0.
438 pass_thru : `str`, optional
439 Constraints to pass through to HTCondor.
440 is_global : `bool`, optional
441 If set, all job queues (and their histories) will be queried for
442 job information. Defaults to False which means that only the local
443 job queue will be queried.
444 return_exit_codes : `bool`, optional
445 If set, return exit codes related to jobs with a
446 non-success status. Defaults to False, which means that only
447 the summary state is returned.
449 Only applicable in the context of a WMS with associated
450 handlers to return exit codes from jobs.
452 Returns
453 -------
454 runs : `list` [`lsst.ctrl.bps.WmsRunReport`]
455 Information about runs from given job information.
456 message : `str`
457 Extra message for report command to print. This could be pointers
458 to documentation or to WMS specific commands.
459 """
460 if wms_workflow_id:
461 id_type = _wms_id_type(wms_workflow_id)
462 if id_type == WmsIdType.LOCAL:
463 schedulers = _locate_schedds(locate_all=is_global)
464 run_reports, message = _report_from_id(wms_workflow_id, hist, schedds=schedulers)
465 elif id_type == WmsIdType.GLOBAL:
466 schedulers = _locate_schedds(locate_all=True)
467 run_reports, message = _report_from_id(wms_workflow_id, hist, schedds=schedulers)
468 elif id_type == WmsIdType.PATH:
469 run_reports, message = _report_from_path(wms_workflow_id)
470 else:
471 run_reports, message = {}, "Invalid job id"
472 else:
473 schedulers = _locate_schedds(locate_all=is_global)
474 run_reports, message = _summary_report(user, hist, pass_thru, schedds=schedulers)
475 _LOG.debug("report: %s, %s", run_reports, message)
477 return list(run_reports.values()), message
479 def cancel(self, wms_id, pass_thru=None):
480 """Cancel submitted workflows/jobs.
482 Parameters
483 ----------
484 wms_id : `str`
485 Id or path of job that should be canceled.
486 pass_thru : `str`, optional
487 Information to pass through to WMS.
489 Returns
490 -------
491 deleted : `bool`
492 Whether successful deletion or not. Currently, if any doubt or any
493 individual jobs not deleted, return False.
494 message : `str`
495 Any message from WMS (e.g., error details).
496 """
497 _LOG.debug("Canceling wms_id = %s", wms_id)
499 schedd_ad, cluster_id, _ = _wms_id_to_cluster(wms_id)
501 if cluster_id is None:
502 deleted = False
503 message = "invalid id"
504 else:
505 _LOG.debug(
506 "Canceling job managed by schedd_name = %s with cluster_id = %s",
507 cluster_id,
508 schedd_ad["Name"],
509 )
510 schedd = htcondor.Schedd(schedd_ad)
512 constraint = f"ClusterId == {cluster_id}"
513 if pass_thru is not None and "-forcex" in pass_thru:
514 pass_thru_2 = pass_thru.replace("-forcex", "")
515 if pass_thru_2 and not pass_thru_2.isspace():
516 constraint += f"&& ({pass_thru_2})"
517 _LOG.debug("JobAction.RemoveX constraint = %s", constraint)
518 results = schedd.act(htcondor.JobAction.RemoveX, constraint)
519 else:
520 if pass_thru:
521 constraint += f"&& ({pass_thru})"
522 _LOG.debug("JobAction.Remove constraint = %s", constraint)
523 results = schedd.act(htcondor.JobAction.Remove, constraint)
524 _LOG.debug("Remove results: %s", results)
526 if results["TotalSuccess"] > 0 and results["TotalError"] == 0:
527 deleted = True
528 message = ""
529 else:
530 deleted = False
531 if results["TotalSuccess"] == 0 and results["TotalError"] == 0:
532 message = "no such bps job in batch queue"
533 else:
534 message = f"unknown problems deleting: {results}"
536 _LOG.debug("deleted: %s; message = %s", deleted, message)
537 return deleted, message
539 def ping(self, pass_thru):
540 """Check whether WMS services are up, reachable, and can authenticate
541 if authentication is required.
543 The services to be checked are those needed for submit, report, cancel,
544 restart, but ping cannot guarantee whether jobs would actually run
545 successfully.
547 Parameters
548 ----------
549 pass_thru : `str`, optional
550 Information to pass through to WMS.
552 Returns
553 -------
554 status : `int`
555 0 for success, non-zero for failure.
556 message : `str`
557 Any message from WMS (e.g., error details).
558 """
559 coll = htcondor.Collector()
560 secman = htcondor.SecMan()
561 status = 0
562 message = ""
563 _LOG.info("Not verifying that compute resources exist.")
564 try:
565 for daemon_type in [htcondor.DaemonTypes.Schedd, htcondor.DaemonTypes.Collector]:
566 _ = secman.ping(coll.locate(daemon_type))
567 except htcondor.HTCondorLocateError:
568 status = 1
569 message = f"Could not locate {daemon_type} service."
570 except htcondor.HTCondorIOError:
571 status = 1
572 message = f"Permission problem with {daemon_type} service."
573 return status, message