diff --git a/doc/changes/DM-55194.other.rst b/doc/changes/DM-55194.other.rst new file mode 100644 index 00000000..1ed3ca5e --- /dev/null +++ b/doc/changes/DM-55194.other.rst @@ -0,0 +1 @@ +Added dagman.conf value to DAGMAN_MAX_JOBS_IDLE workaround. diff --git a/python/lsst/ctrl/bps/htcondor/htcondor_service.py b/python/lsst/ctrl/bps/htcondor/htcondor_service.py index 6e8b1ea4..4ef025ad 100644 --- a/python/lsst/ctrl/bps/htcondor/htcondor_service.py +++ b/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: diff --git a/python/lsst/ctrl/bps/htcondor/lssthtc.py b/python/lsst/ctrl/bps/htcondor/lssthtc.py index 36cedc6f..14612149 100644 --- a/python/lsst/ctrl/bps/htcondor/lssthtc.py +++ b/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) diff --git a/tests/test_htcondor_service.py b/tests/test_htcondor_service.py index 49c0342e..31df2de4 100644 --- a/tests/test_htcondor_service.py +++ b/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) diff --git a/tests/test_lssthtc.py b/tests/test_lssthtc.py index 7edc2cd8..8318cfe0 100644 --- a/tests/test_lssthtc.py +++ b/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