diff --git a/doc/changes/DM-56633.misc.rst b/doc/changes/DM-56633.misc.rst new file mode 100644 index 0000000..6134899 --- /dev/null +++ b/doc/changes/DM-56633.misc.rst @@ -0,0 +1 @@ +Added the capability to ``condor_search`` to skip checking history if found information in the queue. The skipping is enabled by a new optional parameter. diff --git a/python/lsst/ctrl/bps/htcondor/lssthtc.py b/python/lsst/ctrl/bps/htcondor/lssthtc.py index 31279aa..08f82da 100644 --- a/python/lsst/ctrl/bps/htcondor/lssthtc.py +++ b/python/lsst/ctrl/bps/htcondor/lssthtc.py @@ -69,6 +69,7 @@ ] +import contextlib import itertools import json import logging @@ -634,10 +635,13 @@ def htc_submit_dag(sub: Submit): schedd_ad = coll.locate(DaemonTypes.Schedd) schedd = Schedd(schedd_ad) - # If Schedd.submit() fails, the method will raise an exception. Usually, - # that implies issues with the HTCondor pool which BPS can't address. - # Hence, no effort is made to handle the exception. - submit_result = schedd.submit(sub, itemdata=None) + # Suppress stdout during job submission seen with v24 where the + # dagman submit description is dumped to stdout. + with contextlib.redirect_stdout(open(os.devnull, "w")): # pragma: no cover + # If Schedd.submit() fails, the method will raise an exception. + # Usually, that implies issues with the HTCondor pool which BPS + # can't address. Hence, no effort is made to handle the exception. + submit_result = schedd.submit(sub, itemdata=None) # Sadly, the ClassAd from Schedd.submit() (see above) does not have # 'GlobalJobId' so we need to run a regular query to get it anyway. @@ -1338,7 +1342,12 @@ def condor_query(constraint=None, schedds=None, query_func=htc_query_present, ** return {key: val for key, val in job_info.items() if val} -def condor_search(constraint=None, hist=None, schedds=None): +def condor_search( + constraint: str | None = None, + hist: float | None = None, + schedds: dict[str, dict[str, Schedd]] | None = None, + skip_history: bool = False, +) -> dict[str, dict[str, dict[str, Any]]]: """Search for running and finished jobs satisfying given criteria. Parameters @@ -1350,6 +1359,9 @@ def condor_search(constraint=None, hist=None, schedds=None): schedds : `dict` [`str`, `htcondor.Schedd`], optional The list of the HTCondor schedulers which to query for job information. If None (default), only the local scheduler will be queried. + skip_history : `bool`, optional + If True, skip searching history if searching the queue returns any + information. Defaults to False. Returns ------- @@ -1367,7 +1379,7 @@ def condor_search(constraint=None, hist=None, schedds=None): schedds = {schedd_ad["Name"]: Schedd(location=schedd_ad)} job_info = condor_q(constraint=constraint, schedds=schedds) - if hist is not None: + if hist is not None and (not skip_history or not job_info): _LOG.debug("Searching history going back %s days", hist) epoch = (datetime.now() - timedelta(days=hist)).timestamp() constraint += f" && (CompletionDate >= {epoch} || JobFinishedHookDone >= {epoch})" diff --git a/python/lsst/ctrl/bps/htcondor/report_utils.py b/python/lsst/ctrl/bps/htcondor/report_utils.py index 4166b23..f9a7499 100644 --- a/python/lsst/ctrl/bps/htcondor/report_utils.py +++ b/python/lsst/ctrl/bps/htcondor/report_utils.py @@ -88,7 +88,7 @@ def _get_status_from_id( # Collect information about the job by querying HTCondor schedd and # HTCondor history. - schedd_dag_info = _get_info_from_schedd(wms_workflow_id, hist, schedds) + schedd_dag_info = _get_info_from_schedd(wms_workflow_id, hist, schedds, skip_history=True) if len(schedd_dag_info) == 1: schedd_name = next(iter(schedd_dag_info)) dag_id = next(iter(schedd_dag_info[schedd_name])) @@ -250,7 +250,7 @@ def _report_from_id(wms_workflow_id, hist, schedds=None): def _get_info_from_schedd( - wms_workflow_id: str, hist: float, schedds: dict[str, Schedd] + wms_workflow_id: str, hist: float, schedds: dict[str, Schedd], skip_history: bool = False ) -> dict[str, dict[str, dict[str, Any]]]: """Gather run information from HTCondor. @@ -263,6 +263,9 @@ def _get_info_from_schedd( schedds : `dict` [ `str`, `htcondor.Schedd` ] HTCondor schedulers which to query for job information. If empty dictionary, all queries will be run against the local scheduler only. + skip_history : `bool`, optional + If True, skip searching history if searching the queue returns any + information. Defaults to False. Returns ------- @@ -271,7 +274,13 @@ def _get_info_from_schedd( Scheduler, local HTCondor job ids are mapped to their respective classads. """ - _LOG.debug("_get_info_from_schedd: id=%s, hist=%s, schedds=%s", wms_workflow_id, hist, schedds) + _LOG.debug( + "_get_info_from_schedd: id=%s, hist=%s, schedds=%s, skip_history=%s", + wms_workflow_id, + hist, + schedds, + skip_history, + ) dag_constraint = 'regexp("dagman$", Cmd)' try: @@ -289,7 +298,9 @@ def _get_info_from_schedd( # the cluster ids were rolled over (and as a result there is more then # one job with the same cluster id) they will not show up in # the results. - schedd_dag_info = condor_search(constraint=dag_constraint, hist=hist, schedds=schedds) + schedd_dag_info = condor_search( + constraint=dag_constraint, hist=hist, schedds=schedds, skip_history=skip_history + ) return schedd_dag_info diff --git a/tests/dag_test_utils.py b/tests/dag_test_utils.py index 9acabf9..a85b919 100644 --- a/tests/dag_test_utils.py +++ b/tests/dag_test_utils.py @@ -28,13 +28,76 @@ """Utility functions for ctrl_bps_htcondor tests.""" import logging +import re from pathlib import Path +from htcondor2 import Schedd, classad + from lsst.ctrl.bps.htcondor.lssthtc import HTC_VALID_JOB_KEYS, HTCDag, HTCJob, RestrictedDict _LOG = logging.getLogger(__name__) +LOCATE_SUCCESS = classad.ClassAd("""[ + CondorPlatform = "$CondorPlatform: X86_64-CentOS_7.9 $"; + MyType = "Scheduler"; + Machine = "testmachine"; + Name = "testmachine"; + CondorVersion = "$CondorVersion: 23.0.3 2024-04-04 $"; + MyAddress = "<127.0.0.1:9618?addrs=127.0.0.1-9618+snip>" + ] +""") + +PING_SUCCESS = classad.ClassAd("""[ + AuthCommand = 60011; + AuthMethods = "FS_REMOTE"; + Command = 60040; + AuthorizationSucceeded = true; + ValidCommands = "60002,60003,60011,60014,60045,60046,60047,60048,60049,60050,60052,523"; + TriedAuthentication = true; + RemoteVersion = "$CondorVersion: 10.9.0 2023-09-28 BuildID: 678228 PackageID: 10.9.0-1 $"; + MyRemoteUserName = "testuser@testmachine"; + Authentication = "YES"; + ] +""") + + +class ScheddMatch: + """Matches any Schedd object whose address equals the expected + address string. Useful for assert_called_once_with. + + Parameters + ---------- + schedd_classad : `str` + Expected value to be compared against in __eq__. + """ + + def __init__(self, schedd_classad: classad.ClassAd): + self.addr_str = schedd_classad["MyAddress"] + + def __eq__(self, other_schedd: Schedd): + return other_schedd._addr == self.addr_str + + +class ConstraintMatch: + """Matches any constraint string against the expected pattern + (regular expressions allowed). Useful for assert_called_once_with. + + Parameters + ---------- + expected : `str` + Expected value to be compared against in __eq__. + """ + + def __init__(self, expected: str): + self.expected_pattern = expected + + def __eq__(self, other: object): + if not isinstance(other, str): # pragma: no cover + return NotImplemented + return re.match(self.expected_pattern, other) is not None + + def make_lazy_dag(name: str, final: bool = True) -> tuple[HTCDag, list[str]]: """Manually make a simple HTCDag with a lazy subdag for unit testing. diff --git a/tests/test_htcondor_service.py b/tests/test_htcondor_service.py index d884798..dcf6047 100644 --- a/tests/test_htcondor_service.py +++ b/tests/test_htcondor_service.py @@ -44,32 +44,11 @@ from lsst.ctrl.bps.tests.gw_test_utils import make_3_label_workflow from lsst.daf.butler import Config +from .dag_test_utils import LOCATE_SUCCESS, PING_SUCCESS + logger = logging.getLogger("lsst.ctrl.bps.htcondor") TESTDIR = os.path.abspath(os.path.dirname(__file__)) -LOCATE_SUCCESS = """[ - CondorPlatform = "$CondorPlatform: X86_64-CentOS_7.9 $"; - MyType = "Scheduler"; - Machine = "testmachine"; - Name = "testmachine"; - CondorVersion = "$CondorVersion: 23.0.3 2024-04-04 $"; - MyAddress = "<127.0.0.1:9618?addrs=127.0.0.1-9618+snip>" - ] -""" - -PING_SUCCESS = """[ - AuthCommand = 60011; - AuthMethods = "FS_REMOTE"; - Command = 60040; - AuthorizationSucceeded = true; - ValidCommands = "60002,60003,60011,60014,60045,60046,60047,60048,60049,60050,60052,523"; - TriedAuthentication = true; - RemoteVersion = "$CondorVersion: 10.9.0 2023-09-28 BuildID: 678228 PackageID: 10.9.0-1 $"; - MyRemoteUserName = "testuser@testmachine"; - Authentication = "YES"; - ] -""" - class HTCondorServiceTestCase(unittest.TestCase): """Test selected methods of the HTCondor WMS service class.""" diff --git a/tests/test_lssthtc.py b/tests/test_lssthtc.py index 81da359..a654bac 100644 --- a/tests/test_lssthtc.py +++ b/tests/test_lssthtc.py @@ -36,7 +36,7 @@ from shutil import copy2, copytree, ignore_patterns, rmtree, which from unittest.mock import patch -from htcondor2 import JobStatus +from htcondor2 import Collector, JobStatus, Schedd from lsst.ctrl.bps import BpsConfig from lsst.ctrl.bps.bps_utils import chdir @@ -44,7 +44,7 @@ from lsst.daf.butler import Config from lsst.utils.tests import temporaryDirectory -from .dag_test_utils import make_lazy_dag +from .dag_test_utils import LOCATE_SUCCESS, ConstraintMatch, ScheddMatch, make_lazy_dag logger = logging.getLogger("lsst.ctrl.bps.htcondor") TESTDIR = os.path.abspath(os.path.dirname(__file__)) @@ -1823,5 +1823,124 @@ def testWrite(self): self.assertEqual(read_data, self.data) +class CondorSearchTestCase(unittest.TestCase): + """Test condor_search.""" + + def setUp(self): + self.constraint = "ClusterId == 29169 || DAGManJobId == 29169" + hist_pattern = r"(CompletionDate >= \d+ || JobFinishedHookDone >= \d+)" + self.constraint_pattern = f"{self.constraint} && {hist_pattern}" + self.queue_results = { + "testmachine": { + "29169.0": { + "ClusterId": 29169, + "Cmd": "/path/to/condor_dagman", + "GlobalJobId": "testmachine#29169.0#1782598587", + "JobStatus": JobStatus.RUNNING, + }, + "29172.0": { + "ClusterId": 29172, + "Cmd": "/path/to/pipetask", + "GlobalJobId": "testmachine#29172.0#1782598587", + "JobStatus": JobStatus.IDLE, + "DAGManJobId": 29169, + "DAGNodeName": "isr_96908_41", + }, + } + } + self.history_results = { + "testmachine": { + "29171.0": { + "ClusterId": 29171, + "Cmd": "/path/to/pipetask", + "GlobalJobId": "testmachine#29171.0#1782598587", + "JobStatus": JobStatus.COMPLETED, + "DAGManJobId": 29169, + "DAGNodeName": "pipetaskInit", + } + } + } + + @patch.object(Collector, "locate") + @patch("lsst.ctrl.bps.htcondor.lssthtc.condor_q") + @patch("lsst.ctrl.bps.htcondor.lssthtc.condor_history") + def testNoParams(self, mock_history, mock_q, mock_locate): + mock_locate.return_value = LOCATE_SUCCESS + mock_q.return_value = {} + mock_history.return_value = {} + + results = lssthtc.condor_search() + mock_locate.assert_called_once() + mock_q.assert_called_once_with(constraint="", schedds={"testmachine": ScheddMatch(LOCATE_SUCCESS)}) + mock_history.assert_not_called() + self.assertEqual(results, {}) + + @patch.object(Collector, "locate") + @patch("lsst.ctrl.bps.htcondor.lssthtc.condor_q") + @patch("lsst.ctrl.bps.htcondor.lssthtc.condor_history") + def testSearchParams(self, mock_history, mock_q, mock_locate): + self.maxDiff = None + mock_locate.return_value = LOCATE_SUCCESS + mock_q.return_value = self.queue_results + mock_history.return_value = self.history_results + + results = lssthtc.condor_search( + constraint=self.constraint, hist=3, schedds={"testmachine": Schedd(LOCATE_SUCCESS)} + ) + mock_locate.assert_not_called() + mock_q.assert_called_once_with( + constraint=self.constraint, schedds={"testmachine": ScheddMatch(LOCATE_SUCCESS)} + ) + mock_history.assert_called_once_with( + ConstraintMatch(self.constraint_pattern), schedds={"testmachine": ScheddMatch(LOCATE_SUCCESS)} + ) + all_results = {"testmachine": self.queue_results["testmachine"] | self.history_results["testmachine"]} + self.assertEqual(results, all_results) + + @patch.object(Collector, "locate") + @patch("lsst.ctrl.bps.htcondor.lssthtc.condor_q") + @patch("lsst.ctrl.bps.htcondor.lssthtc.condor_history") + def testSkipHistory(self, mock_history, mock_q, mock_locate): + mock_locate.return_value = LOCATE_SUCCESS + mock_q.return_value = self.queue_results + mock_history.return_value = self.history_results + + results = lssthtc.condor_search( + constraint=self.constraint, + hist=3, + skip_history=True, + schedds={"testmachine": Schedd(LOCATE_SUCCESS)}, + ) + mock_locate.assert_not_called() + mock_q.assert_called_once_with( + constraint=self.constraint, schedds={"testmachine": ScheddMatch(LOCATE_SUCCESS)} + ) + mock_history.assert_not_called() + self.assertEqual(results, self.queue_results) + + @patch.object(Collector, "locate") + @patch("lsst.ctrl.bps.htcondor.lssthtc.condor_q") + @patch("lsst.ctrl.bps.htcondor.lssthtc.condor_history") + def testSkipHistoryNoQueue(self, mock_history, mock_q, mock_locate): + mock_locate.return_value = LOCATE_SUCCESS + mock_q.return_value = {} + mock_history.return_value = self.history_results + + results = lssthtc.condor_search( + constraint=self.constraint, + hist=3, + skip_history=True, + schedds={"testmachine": Schedd(LOCATE_SUCCESS)}, + ) + mock_locate.assert_not_called() + mock_q.assert_called_once_with( + constraint=self.constraint, schedds={"testmachine": ScheddMatch(LOCATE_SUCCESS)} + ) + mock_history.assert_called_once_with( + ConstraintMatch(self.constraint_pattern), schedds={"testmachine": ScheddMatch(LOCATE_SUCCESS)} + ) + self.assertEqual(results, self.history_results) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_report_utils.py b/tests/test_report_utils.py index 3240471..b65c4a9 100644 --- a/tests/test_report_utils.py +++ b/tests/test_report_utils.py @@ -34,7 +34,7 @@ from shutil import copy2, copytree from unittest.mock import patch -from htcondor2 import JobStatus +from htcondor2 import JobStatus, Schedd from lsst.ctrl.bps import ( WmsSpecificInfo, @@ -43,20 +43,12 @@ from lsst.ctrl.bps.htcondor import lssthtc, report_utils from lsst.utils.tests import temporaryDirectory +from .dag_test_utils import LOCATE_SUCCESS + logger = logging.getLogger("lsst.ctrl.bps.htcondor") TESTDIR = os.path.abspath(os.path.dirname(__file__)) -LOCATE_SUCCESS = """[ - CondorPlatform = "$CondorPlatform: X86_64-CentOS_7.9 $"; - MyType = "Scheduler"; - Machine = "testmachine"; - Name = "testmachine"; - CondorVersion = "$CondorVersion: 23.0.3 2024-04-04 $"; - MyAddress = "<127.0.0.1:9618?addrs=127.0.0.1-9618+snip>" - ] -""" - class GetExitCodeSummaryTestCase(unittest.TestCase): """Test the function responsible for creating exit code summary.""" @@ -806,7 +798,7 @@ def testNotFound(self, mock_get): state, message = report_utils._get_status_from_id("100", 0, {}) - mock_get.assert_called_once_with("100", 0, {}) + mock_get.assert_called_once_with("100", 0, {}, skip_history=True) self.assertEqual(state, WmsStates.UNKNOWN) self.assertEqual(message, "DAGMan job 100 not found in queue or history. Check id or try path.") @@ -821,7 +813,7 @@ def testFound(self, mock_get, mock_conversion): state, message = report_utils._get_status_from_id(fake_id, 0, {}) - mock_get.assert_called_once_with(fake_id, 0, {}) + mock_get.assert_called_once_with(fake_id, 0, {}, skip_history=True) mock_conversion.assert_called_once_with(dag_ads[fake_id]) self.assertEqual(state, WmsStates.RUNNING) @@ -862,5 +854,28 @@ def testFailure(self): self.assertEqual(message, "") +class GetInfoFromScheddTestCase(unittest.TestCase): + """Test _get_info_from_schedd function.""" + + @patch("lsst.ctrl.bps.htcondor.report_utils.condor_search") + def testNumericID(self, mock_search): + schedds = {"testmachine": Schedd(location=LOCATE_SUCCESS)} + _ = report_utils._get_info_from_schedd("26159", 2.5, schedds, True) + constraint = 'regexp("dagman$", Cmd) && ClusterId == 26159' + mock_search.assert_called_once_with( + constraint=constraint, hist=2.5, schedds=schedds, skip_history=True + ) + + @patch("lsst.ctrl.bps.htcondor.report_utils.condor_search") + def testGlobalID(self, mock_search): + schedds = {"testmachine": Schedd(location=LOCATE_SUCCESS)} + global_id = "testmachine#26159.0#1791323977" + _ = report_utils._get_info_from_schedd(global_id, 2.5, schedds, True) + constraint = f'regexp("dagman$", Cmd) && GlobalJobId == "{global_id}"' + mock_search.assert_called_once_with( + constraint=constraint, hist=2.5, schedds=schedds, skip_history=True + ) + + if __name__ == "__main__": unittest.main()