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