DM-55194: Add dagman.conf value to MaxIdle workaround

doc/changes/DM-55194.other.rst

@@ -0,0 +1 @@
+Added dagman.conf value to DAGMAN_MAX_JOBS_IDLE workaround.

python/lsst/ctrl/bps/htcondor/htcondor_service.py

@@ -163,7 +163,12 @@ def submit(self, workflow, **kwargs):
         with chdir(workflow.submit_path):
             try:
                 if ver >= version.parse("8.9.3"):
-                    sub = htc_create_submit_from_dag(dag.graph["dag_filename"], dag.graph["submit_options"])
+                    wms_config_path = None
+                    if "bps_wms_config_path" in dag.graph["attr"]:
+                        wms_config_path = dag.graph["attr"]["bps_wms_config_path"]
+                    sub = htc_create_submit_from_dag(
+                        dag.graph["dag_filename"], dag.graph["submit_options"], wms_config_path
+                    )
                 else:
                     sub = htc_create_submit_from_cmd(dag.graph["dag_filename"], dag.graph["submit_options"])
             except Exception:

python/lsst/ctrl/bps/htcondor/lssthtc.py

@@ -706,7 +706,9 @@ def htc_submit_dag(sub):
     return schedd_dag_info
 
 
-def htc_create_submit_from_dag(dag_filename: str, submit_options: dict[str, Any]) -> htcondor.Submit:
+def htc_create_submit_from_dag(
+    dag_filename: str, submit_options: dict[str, Any], dagman_conf_filename: str | os.PathLike | None = None
+) -> htcondor.Submit:
     """Create a DAGMan job submit description.
 
     Parameters
@@ -715,6 +717,9 @@ def htc_create_submit_from_dag(dag_filename: str, submit_options: dict[str, Any]
         Name of file containing HTCondor DAG commands.
     submit_options : `dict` [`str`, `~typing.Any`], optional
         Contains extra options for command line (Value of None means flag).
+    dagman_conf_filename : `str` or `os.PathLike`, optional
+        Location of DAGMan configuration file, if Any.  Defaults to no
+        file (None).
 
     Returns
     -------
@@ -734,13 +739,30 @@ def htc_create_submit_from_dag(dag_filename: str, submit_options: dict[str, Any]
     if "MaxIdle" not in submit_options:
         max_jobs_idle: int | None = None
         config_var_name = "DAGMAN_MAX_JOBS_IDLE"
-        if f"_CONDOR_{config_var_name}" in os.environ:
-            max_jobs_idle = int(os.environ[f"_CONDOR_{config_var_name}"])
-        elif config_var_name in htcondor.param:
-            max_jobs_idle = htcondor.param[config_var_name]
+
+        if dagman_conf_filename:
+            _LOG.debug("Checking DAGMan config file = %s", dagman_conf_filename)
+            with open(dagman_conf_filename) as fh:
+                for line in fh:
+                    _LOG.debug("DAGMan config file line = %s", line)
+                    parts = line.split("=")
+                    _LOG.debug("DAGMan config file line parts = %s", parts)
+                    if len(parts) == 2 and parts[0].strip() == config_var_name:
+                        max_jobs_idle = int(parts[1].strip())
+                        _LOG.debug("Found %s = %s", config_var_name, max_jobs_idle)
+                        break
+        if max_jobs_idle is None:
+            if f"_CONDOR_{config_var_name}" in os.environ:
+                max_jobs_idle = int(os.environ[f"_CONDOR_{config_var_name}"])
+            elif config_var_name in htcondor.param:
+                max_jobs_idle = htcondor.param[config_var_name]
+
         if max_jobs_idle:
             submit_options["MaxIdle"] = max_jobs_idle
+    else:
+        _LOG.debug("MaxIdle already in submit_options: %s", submit_options)
 
+    _LOG.debug("Using submit_options = %s", submit_options)
     return htcondor.Submit.from_dag(dag_filename, submit_options)
 
 

tests/test_htcondor_service.py

@@ -28,19 +28,23 @@
 """Unit tests for the HTCondor WMS service class and related functions."""
 
 import logging
+import os
 import tempfile
 import unittest
 from pathlib import Path
 
 import htcondor
 
+import lsst.ctrl.bps.htcondor.lssthtc as lssthtc
 from lsst.ctrl.bps import BpsConfig, WmsStates
 from lsst.ctrl.bps.htcondor import htcondor_service
 from lsst.ctrl.bps.htcondor.htcondor_config import HTC_DEFAULTS_URI
+from lsst.ctrl.bps.htcondor.htcondor_workflow import HTCondorWorkflow
 from lsst.ctrl.bps.tests.gw_test_utils import make_3_label_workflow
 from lsst.daf.butler import Config
 
 logger = logging.getLogger("lsst.ctrl.bps.htcondor")
+TESTDIR = os.path.abspath(os.path.dirname(__file__))
 
 LOCATE_SUCCESS = """[
         CondorPlatform = "$CondorPlatform: X86_64-CentOS_7.9 $";
@@ -224,3 +228,63 @@ def testPrepareProvision(self, mock_write):
             self.assertTrue(prov_script.is_file())
             script_contents = prov_script.read_text()
             self.assertIn(f"--nodeset {timestamp}", script_contents)
+
+    def testSubmitWithConfigPath(self):
+        """Only testing value for wms_config_path being passed
+        correctly to htc_create_submit_from_dag.  Aborting submission
+        after that call to skip rest of submit function.
+        """
+
+        def _fake_htc_create_submit_from_dag(filename, submit_options, wms_config_path):
+            raise RuntimeError("Fake exception from mock")
+
+        dag_filename = "should_not_matter.dag"
+        wms_config_path = "dagman.conf"
+        submit_options = {"DAGMAN_MAX_JOBS_SUBMITTED": 30}
+        attribs = {"bps_wms_config_path": wms_config_path}
+
+        workflow = HTCondorWorkflow("testSuccess")
+        workflow.dag = lssthtc.HTCDag("testSuccess")
+        workflow.dag.graph["dag_filename"] = dag_filename
+        workflow.dag.graph["attr"] = dict(attribs)
+        workflow.dag.graph["submit_options"] = dict(submit_options)
+
+        with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
+            workflow.submit_path = tmpdir
+            with unittest.mock.patch(
+                "lsst.ctrl.bps.htcondor.htcondor_service.htc_create_submit_from_dag"
+            ) as create_mock:
+                create_mock.side_effect = _fake_htc_create_submit_from_dag
+                with self.assertRaisesRegex(RuntimeError, "Fake exception from mock"):
+                    self.service.submit(workflow)
+                create_mock.assert_called_once_with(dag_filename, submit_options, wms_config_path)
+
+    def testSubmitWithoutConfigPath(self):
+        """Only testing that values are being passed correctly to
+        htc_create_submit_from_dag when there isn't a wms config path.
+        Aborting submission after that call to skip rest of submit function.
+        """
+
+        def _fake_htc_create_submit_from_dag(filename, submit_options, wms_config_path):
+            raise RuntimeError("Fake exception from mock")
+
+        dag_filename = "should_not_matter.dag"
+        wms_config_path = None
+        submit_options = {"DAGMAN_MAX_JOBS_SUBMITTED": 30}
+        attribs = {}
+
+        workflow = HTCondorWorkflow("testSuccess")
+        workflow.dag = lssthtc.HTCDag("testSuccess")
+        workflow.dag.graph["dag_filename"] = dag_filename
+        workflow.dag.graph["attr"] = dict(attribs)
+        workflow.dag.graph["submit_options"] = dict(submit_options)
+
+        with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
+            workflow.submit_path = tmpdir
+            with unittest.mock.patch(
+                "lsst.ctrl.bps.htcondor.htcondor_service.htc_create_submit_from_dag"
+            ) as create_mock:
+                create_mock.side_effect = _fake_htc_create_submit_from_dag
+                with self.assertRaisesRegex(RuntimeError, "Fake exception from mock"):
+                    self.service.submit(workflow)
+                create_mock.assert_called_once_with(dag_filename, submit_options, wms_config_path)

tests/test_lssthtc.py

@@ -1234,6 +1234,28 @@ def testMaxIdleEnvVar(self):
             submit = lssthtc.htc_create_submit_from_dag(str(dag_filename), {})
             self.assertIn("-MaxIdle 42", submit["arguments"])
 
+    @unittest.mock.patch.dict(os.environ, {"_CONDOR_DAGMAN_MAX_JOBS_IDLE": "42"})
+    def testMaxIdleInDAGManConfig(self):
+        with temporaryDirectory() as tmp_dir:
+            copy2(f"{TESTDIR}/data/tiny_success/tiny_success.dag", tmp_dir)
+            dag_filename = pathlib.Path(tmp_dir) / "tiny_success.dag"
+            config_filename = pathlib.Path(tmp_dir) / "dagman.conf"
+            with open(config_filename, "w") as fh:
+                print("DAGMAN_MAX_JOBS_IDLE = 300", file=fh)
+            submit = lssthtc.htc_create_submit_from_dag(str(dag_filename), {}, config_filename)
+            self.assertIn("-MaxIdle 300", submit["arguments"])
+
+    @unittest.mock.patch.dict(os.environ, {"_CONDOR_DAGMAN_MAX_JOBS_IDLE": "42"})
+    def testMaxIdleNotInDAGManConfig(self):
+        with temporaryDirectory() as tmp_dir:
+            copy2(f"{TESTDIR}/data/tiny_success/tiny_success.dag", tmp_dir)
+            dag_filename = pathlib.Path(tmp_dir) / "tiny_success.dag"
+            config_filename = pathlib.Path(tmp_dir) / "dagman.conf"
+            with open(config_filename, "w") as fh:
+                print("DAGMAN_MAX_JOBS_SUBMITTED = 300", file=fh)
+            submit = lssthtc.htc_create_submit_from_dag(str(dag_filename), {}, config_filename)
+            self.assertIn("-MaxIdle 42", submit["arguments"])
+
     @unittest.mock.patch.dict(os.environ, {})
     def testMaxIdleGiven(self):
         with temporaryDirectory() as tmp_dir:
@@ -1242,6 +1264,27 @@ def testMaxIdleGiven(self):
             submit = lssthtc.htc_create_submit_from_dag(str(dag_filename), {"MaxIdle": 37})
             self.assertIn("-MaxIdle 37", submit["arguments"])
 
+    @unittest.mock.patch.dict(os.environ, {})
+    def testMaxJobsIdleParam(self):
+        def _fake_params_contains(key):
+            if key == "DAGMAN_MAX_JOBS_IDLE":
+                return True
+            return False  # pragma: no cover
+
+        def _fake_params_get(key):
+            if key == "DAGMAN_MAX_JOBS_IDLE":
+                return 16
+            return "FAKE_VAL"  # pragma: no cover
+
+        with temporaryDirectory() as tmp_dir:
+            copy2(f"{TESTDIR}/data/tiny_success/tiny_success.dag", tmp_dir)
+            dag_filename = pathlib.Path(tmp_dir) / "tiny_success.dag"
+            with unittest.mock.patch("htcondor.param") as mock_param:
+                mock_param.__contains__.side_effect = _fake_params_contains
+                mock_param.__getitem__.side_effect = _fake_params_get
+                submit = lssthtc.htc_create_submit_from_dag(str(dag_filename), {}, None)
+                self.assertIn("-MaxIdle 16", submit["arguments"])
+
     @unittest.mock.patch.dict(os.environ, {})
     def testNoMaxJobsIdle(self):
         """Note: Since the produced arguments differ depending on