DM-56633: Add skip_history option to condor_search

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.

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})"

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
 
 

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.
 

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."""

tests/test_lssthtc.py

@@ -36,15 +36,15 @@
 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
 from lsst.ctrl.bps.htcondor import dagman_configurator, htcondor_config, lssthtc
 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()