Merge pull request #179 from lsst/tickets/DM-41672

DM-41672: Create WMS-id softlink to submit run directory.

doc/changes/DM-41672.feature.rst

@@ -0,0 +1 @@
+``bps submit`` and ``bps restart`` can make a softlink to the submit run directory.  The softlink is named after the WMS job id.  ``idLinkPath`` determines the parent directory for the link (defaults to ``${PWD}/bps_links``).  ``makeIdLink`` is a boolean used to turn on/off the creation of the link (defaults to ``False``).

doc/lsst.ctrl.bps/quickstart.rst

@@ -176,6 +176,17 @@ The following subset of ``pipetask`` options are also usable on ``bps submit`` c
 - ``-p, --pipeline FILE``
 - ``-g, --qgraph FILENAME``
 
+**Other Options**
+
+The following miscellaneous options are also usable on ``bps submit`` command lines.
+
+- ``--make-id-link``
+  Turns on the creation of an id-based softlink to the submit run directory.
+  Defaults to False.  Replaces variable ``makeIdLink``.
+- ``--id-link-path``
+  The parent directory for the id-based softlink.  Defaults to ``${PWD}/bps_links``.
+  Replaces variable ``idLinkPath``.
+
 .. _bps-report:
 
 Checking status
@@ -630,6 +641,14 @@ Supported settings
     graphs/workflows.  The plots will be practically illegible for graphs which
     number of nodes exceeds order of tens.
 
+**makeIdLink**
+    A boolean flag.  If set to true, ``bps submit`` will create a WMS-id softlink
+    to the submit run directory.  Defaults to ``False``.  (see :ref:`bps-softlink`)
+
+**idLinkPath**
+    The parent directory for the WMS-id link.  Defaults to ``${PWD}/bps_links``.
+    (see :ref:`bps-softlink`)
+
 .. _pickle: https://docs.python.org/3/library/pickle.html
 .. _ref: https://stackoverflow.com/a/38971446
 
@@ -1245,6 +1264,35 @@ Relevant Config Entries:
        # requestCpus: N              # Overrides for jobs in this cluster
        # requestMemory: NNNN         # MB, Overrides for jobs in this cluster
 
+.. _bps-softlink:
+
+WMS-id softlink
+---------------
+
+``bps submit`` makes a directory in which to put submit-time files.  The
+location is based on the output run collection which normally ends in a
+timestamp.  WMS workflow ids are normally shorter and easier to remember.
+BPS can make a softlink, named with the WMS workflow id, to the submit
+run directory.  The following are the two values that control this:
+
+- ``idLinkPath`` determines the parent directory for the link (defaults to ``${PWD}/bps_links``).
+- ``makeIdLink`` is a boolean used to turn on/off the creation of the link (defaults to ``False``).
+
+Both of these values can also be set on the ``bps submit`` command-line.
+
+``bps restart`` will use the settings defined in the original submission.
+If the WMS restart returns a new id, bps will create a new softlink if
+``makeIdLink`` is true.
+
+BPS cannot make an id softlink if the WMS does not return an id (e.g.,
+Parsl).
+
+.. warning::
+
+    BPS does **NOT** manage the softlinks once created.  It is the user's
+    responsibility to remove them once no longer needed.  The removal
+    should be done regularly to avoid too many in single directory.
+
 .. _bps-troubleshooting:
 
 Troubleshooting

python/lsst/ctrl/bps/bps_utils.py

@@ -37,6 +37,7 @@
     "parse_count_summary",
     "_dump_pkg_info",
     "_dump_env_info",
+    "_make_id_link",
 ]
 
 import contextlib
@@ -256,3 +257,48 @@ def _dump_env_info(filename):
         file = file.with_suffix(f"{file.suffix}.yaml")
     with open(file, "w", encoding="utf-8") as fh:
         yaml.dump(dict(os.environ), fh)
+
+
+def _make_id_link(config, run_id):
+    """Make id softlink to the submit run directory if makeIdLink
+    is true.
+
+    Parameters
+    ----------
+    config : `lsst.ctrl.bps.BpsConfig`
+        BPS configuration.
+    run_id : `str`
+        WMS run ID.
+    """
+    _, make_id_link = config.search("makeIdLink")
+    if make_id_link:
+        if run_id is None:
+            _LOG.info("Run ID is None.  Skipping making id link.")
+        else:
+            found, submit_path = config.search("submitPath")
+            # pathlib.Path.symlink_to() does not care if target exists
+            # so we check it ourselves.
+            if found and Path(submit_path).exists():
+                _, id_link_path = config.search("idLinkPath")
+                _LOG.debug("submit_path=%s, id_link_path=%s", submit_path, id_link_path)
+                id_link_path = Path(id_link_path)
+                id_link_path = id_link_path / f"{run_id}"
+                _LOG.debug("submit_path=%s, id_link_path=%s", submit_path, id_link_path)
+                if (
+                    id_link_path.exists()
+                    and id_link_path.is_symlink()
+                    and str(id_link_path.readlink()) == submit_path
+                ):
+                    _LOG.debug("Correct softlink already exists (%s)", id_link_path)
+                else:
+                    _LOG.debug("Softlink doesn't already exist (%s)", id_link_path)
+                    try:
+                        id_link_path.parent.mkdir(parents=True, exist_ok=True)
+                        id_link_path.symlink_to(submit_path)
+                        _LOG.info("Made id softlink: %s", id_link_path)
+                    except (OSError, FileExistsError, PermissionError) as exc:
+                        _LOG.warning("Could not make id softlink: %s", exc)
+            else:
+                _LOG.warning("Could not make id softlink: submitPath does not exist (%s)", submit_path)
+    else:
+        _LOG.debug("Not asked to make id link (makeIdLink=%s)", make_id_link)

python/lsst/ctrl/bps/cli/opt/option_groups.py

@@ -45,6 +45,8 @@
     extra_qgraph_option,
     extra_run_quantum_option,
     extra_update_qgraph_option,
+    id_link_path_option,
+    make_id_link_option,
 )
 
 
@@ -70,4 +72,6 @@ def __init__(self):
             extra_update_qgraph_option(),
             extra_init_option(),
             extra_run_quantum_option(),
+            make_id_link_option(),
+            id_link_path_option(),
         ]

python/lsst/ctrl/bps/cli/opt/options.py

@@ -34,6 +34,8 @@
     "extra_run_quantum_option",
     "wms_service_option",
     "compute_site_option",
+    "make_id_link_option",
+    "id_link_path_option",
 ]
 
 from lsst.daf.butler.cli.utils import MWOptionDecorator
@@ -64,3 +66,17 @@
     "compute_site",
     help="The compute site used to run the workflow.",
 )
+
+make_id_link_option = MWOptionDecorator(
+    "--make-id-link",
+    "make_id_link",
+    default=False,
+    is_flag=True,
+    help="Make a link to the submit directory using the workflow id.",
+)
+
+id_link_path_option = MWOptionDecorator(
+    "--id-link-path",
+    "id_link_path",
+    help="Location in which to make id soft link to the submit directory." "default ('${PWD}/bps_links')",
+)

python/lsst/ctrl/bps/drivers.py

@@ -60,7 +60,7 @@
 from lsst.utils.usage import get_peak_mem_usage
 
 from . import BPS_DEFAULTS, BPS_SEARCH_ORDER, DEFAULT_MEM_FMT, DEFAULT_MEM_UNIT, BpsConfig
-from .bps_utils import _dump_env_info, _dump_pkg_info
+from .bps_utils import _dump_env_info, _dump_pkg_info, _make_id_link
 from .cancel import cancel
 from .ping import ping
 from .pre_transform import acquire_quantum_graph, cluster_quanta
@@ -470,6 +470,8 @@ def submit_driver(config_file, **kwargs):
             *tuple(f"{val.to(DEFAULT_MEM_UNIT):{DEFAULT_MEM_FMT}}" for val in get_peak_mem_usage()),
         )
 
+    _make_id_link(wms_workflow_config, wms_workflow.run_id)
+
     print(f"Run Id: {wms_workflow.run_id}")
     print(f"Run Name: {wms_workflow.name}")
 
@@ -494,6 +496,9 @@ def restart_driver(wms_service, run_id):
         if path.exists():
             _dump_env_info(f"{run_id}/{run_name}.env.info.yaml")
             _dump_pkg_info(f"{run_id}/{run_name}.pkg.info.yaml")
+            config = BpsConfig(f"{run_id}/{run_name}_config.yaml")
+            _make_id_link(config, new_run_id)
+
         print(f"Run Id: {new_run_id}")
         print(f"Run Name: {run_name}")
     else:

python/lsst/ctrl/bps/etc/bps_defaults.yaml

@@ -137,3 +137,8 @@ finalJob:
 # own defaults.  Currently, needs to be override in user's config
 # if a different limit needs to be used.
 memoryLimit: 491520
+
+# Default values for making id soft link to submit directory
+makeIdLink: False
+idLinkPath: "${PWD}/bps_links"
+

tests/test_bps_utils.py

@@ -24,38 +24,176 @@
 #
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
-import os
+import logging
 import shutil
 import tempfile
 import unittest
+from pathlib import Path
 
-from lsst.ctrl.bps.bps_utils import chdir
-
-TESTDIR = os.path.abspath(os.path.dirname(__file__))
+from lsst.ctrl.bps import BpsConfig
+from lsst.ctrl.bps.bps_utils import _make_id_link, chdir
 
 
 class TestChdir(unittest.TestCase):
     """Test directory changing."""
 
     def setUp(self):
-        self.cwd = os.getcwd()
-        self.tmpdir = tempfile.mkdtemp(dir=TESTDIR)
+        self.tmpdir = Path(tempfile.mkdtemp())
 
     def tearDown(self):
         shutil.rmtree(self.tmpdir, ignore_errors=True)
 
     def testSuccessfulChdir(self):
-        self.assertNotEqual(os.getcwd(), self.tmpdir)
+        cwd = Path.cwd()
+        self.assertFalse(self.tmpdir.samefile(cwd))
         with chdir(self.tmpdir):
-            self.assertEqual(os.getcwd(), self.tmpdir)
-        self.assertNotEqual(os.getcwd(), self.tmpdir)
+            self.assertTrue(self.tmpdir.samefile(Path.cwd()))
+        self.assertTrue(cwd.samefile(Path.cwd()))
 
     def testFailingChdir(self):
-        dir_not_there = os.path.join(self.tmpdir, "notthere")
+        dir_not_there = self.tmpdir / "notthere"
         with self.assertRaises(FileNotFoundError):
             with chdir(dir_not_there):
                 pass  # should not get here
 
 
+class TestMakeIdLink(unittest.TestCase):
+    """Test _make_id_link function."""
+
+    def setUp(self):
+        self.tmpdir = Path(tempfile.mkdtemp())
+
+    def tearDown(self):
+        shutil.rmtree(self.tmpdir, ignore_errors=True)
+
+    def testMakeIdLinkFalse(self):
+        """Test skipping making link."""
+        config = BpsConfig({"makeIdLink": False})
+        submit_path = self.tmpdir / "test_submit/testrun/1"
+        submit_path.mkdir(parents=True)
+        config["submitPath"] = str(submit_path)
+
+        dir_of_links = self.tmpdir / "bps_links"
+        with self.assertLogs("lsst.ctrl.bps.bps_utils", level=logging.DEBUG) as cm:
+            _make_id_link(config, "100.0")
+            self.assertRegex(cm.records[-1].getMessage(), "Not asked to make id link")
+        self.assertFalse(dir_of_links.exists())
+
+    def testNoRunID(self):
+        """Test no link made if no run id."""
+        config = BpsConfig({"makeIdLink": True})
+        submit_path = self.tmpdir / "test_submit/testrun/2"
+        submit_path.mkdir(parents=True)
+        config["submitPath"] = str(submit_path)
+
+        dir_of_links = self.tmpdir / "bps_links"
+        with self.assertLogs("lsst.ctrl.bps.bps_utils", level=logging.DEBUG) as cm:
+            _make_id_link(config, None)
+            self.assertRegex(cm.records[-1].getMessage(), "Run ID is None.  Skipping making id link.")
+        self.assertFalse(dir_of_links.exists())
+
+    def testSuccessfulLink(self):
+        """Test successfully made id link."""
+        config = BpsConfig({"makeIdLink": True})
+
+        submit_path = self.tmpdir / "test_submit/testrun/3"
+        submit_path.mkdir(parents=True)
+        config["submitPath"] = str(submit_path)
+
+        # Make sure can make multiple dirs
+        dir_of_links = self.tmpdir / "test_bps/links"
+        config["idLinkPath"] = str(dir_of_links)
+
+        with self.assertLogs("lsst.ctrl.bps.bps_utils", level=logging.INFO) as cm:
+            _make_id_link(config, "100.0")
+            self.assertRegex(cm.records[-1].getMessage(), "Made id softlink:")
+
+        link_path = dir_of_links / "100.0"
+        self.assertTrue(link_path.is_symlink())
+        self.assertEqual(link_path.readlink(), submit_path)
+
+    def testSubmitDoesNotExist(self):
+        """Test checking that submit directory exists."""
+        config = BpsConfig({"makeIdLink": True})
+
+        submit_path = self.tmpdir / "test_submit/testrun/4"
+        submit_path.mkdir(parents=True)
+        config["submitPath"] = str(submit_path / "notthere")
+
+        dir_of_links = self.tmpdir / "test_bps_links"
+        config["idLinkPath"] = str(dir_of_links)
+
+        with self.assertLogs("lsst.ctrl.bps.bps_utils", level=logging.WARNING) as cm:
+            _make_id_link(config, "100.0")
+            self.assertRegex(
+                cm.records[-1].getMessage(), "Could not make id softlink: submitPath does not exist"
+            )
+        self.assertFalse(dir_of_links.exists())
+
+    def testLinkAlreadyExists(self):
+        """Test skipping if link already correctly exists
+        for example if a restart gives same id.
+        """
+        config = BpsConfig({"makeIdLink": True})
+
+        submit_path = self.tmpdir / "test_submit/testrun/5"
+        submit_path.mkdir(parents=True)
+        config["submitPath"] = str(submit_path)
+
+        dir_of_links = self.tmpdir / "test_bps_links"
+        config["idLinkPath"] = str(dir_of_links)
+
+        # Make the softlink
+        dir_of_links.mkdir(parents=True)
+        link_path = dir_of_links / "100.0"
+        link_path.symlink_to(submit_path)
+
+        with self.assertLogs("lsst.ctrl.bps.bps_utils", level=logging.DEBUG) as cm:
+            _make_id_link(config, "100.0")
+            self.assertRegex(cm.records[-1].getMessage(), "Correct softlink already exists")
+        self.assertTrue(link_path.is_symlink())
+        self.assertEqual(link_path.readlink(), submit_path)
+
+    def testFileExistsError(self):
+        """Test catching of FileExistsError."""
+        config = BpsConfig({"makeIdLink": True})
+
+        submit_path = self.tmpdir / "test_submit/testrun/6"
+        submit_path.mkdir(parents=True)
+        config["submitPath"] = str(submit_path)
+
+        dir_of_links = self.tmpdir / "test_bps_links"
+        config["idLinkPath"] = str(dir_of_links)
+
+        # Make a directory with the link path so
+        # that get FileExistsError
+        link_path = dir_of_links / "100.0"
+        link_path.mkdir(parents=True)
+
+        with self.assertLogs("lsst.ctrl.bps.bps_utils", level=logging.WARNING) as cm:
+            _make_id_link(config, "100.0")
+            self.assertRegex(cm.records[-1].getMessage(), "Could not make id softlink:.*File exists")
+        self.assertFalse(link_path.is_symlink())
+
+    def testPermissionError(self):
+        """Test catching of PermissionError."""
+        config = BpsConfig({"makeIdLink": True})
+
+        submit_path = self.tmpdir / "test_submit/testrun/7"
+        submit_path.mkdir(parents=True)
+        config["submitPath"] = str(submit_path)
+
+        dir_of_links = self.tmpdir / "test_bps_links"
+        # Create dir without write permissions to cause the error.
+        dir_of_links.mkdir(mode=0o555, parents=True)
+        config["idLinkPath"] = str(dir_of_links)
+
+        with self.assertLogs("lsst.ctrl.bps.bps_utils", level=logging.WARNING) as cm:
+            _make_id_link(config, "100.0")
+            self.assertRegex(cm.records[-1].getMessage(), "Could not make id softlink:.*Permission denied")
+        link_path = dir_of_links / "100.0"
+        self.assertFalse(link_path.is_symlink())
+
+
 if __name__ == "__main__":
     unittest.main()