Merge pull request #110 from lsst/tickets/DM-35144

DM-35144: Add ping subcommand.

.github/workflows/build.yaml

@@ -24,13 +24,6 @@ jobs:
         run: |
           conda install -y -q pip wheel
 
-      - name: Install IDDS for tests
-        shell: bash -l {0}
-        env:
-          IDDS_CONFIG: ${{ github.workspace }}/tests/idds_test.cfg
-        run: |
-          conda install -y -q idds-common idds-client panda-client argcomplete
-
       - name: Install dependencies
         shell: bash -l {0}
         run: |
@@ -57,8 +50,6 @@ jobs:
 
       - name: Run tests
         shell: bash -l {0}
-        env:
-          IDDS_CONFIG: ${{ github.workspace }}/tests/idds_test.cfg
         run: |
           pytest -r a -v -n 3 --open-files --cov=lsst.ctrl.bps --cov=tests --cov-report=xml --cov-report=term --cov-branch
       - name: Upload coverage to codecov

doc/changes/DM-35144.feature.rst

@@ -0,0 +1 @@
+Add ping subcommand to test whether the workflow services are available.

doc/lsst.ctrl.bps/quickstart.rst

@@ -17,13 +17,35 @@ how to find it.
 The location of the plugin can be specified as listed below (in the increasing
 order of priority):
 
-#. by setting ``WMS_SERVICE_CLASS`` environment variable,
+#. by setting ``BPS_WMS_SERVICE_CLASS`` environment variable,
 #. in the config file *via* ``wmsServiceClass`` setting,
 #. using command-line option ``--wms-service-class``.
 
 If plugin location is not specified explicitly using one of the methods above,
 a default value, ``lsst.ctrl.bps.htcondor.HTCondorService``, will be used.
 
+.. _bps-ping:
+
+Checking status of WMS services
+-------------------------------
+
+Run `bps ping` to check the status of the WMS services.  This subcommand
+requires specifying the WMS plugin (see :ref:`bps-wmsclass`).  If the plugin
+provides such functionality, it will check whether the WMS services
+necessary for workflow management (submission, reporting, canceling,
+etc) are usable.  If the WMS services require authentication, that will
+also be tested.
+
+If services are ready for use, then `bps ping` will log an INFO success
+message and exit with 0.  If not, it will log ERROR messages and exit
+with a non-0 exit code.  If the WMS plugin did not implement the ping
+functionality, a NotImplementedError will be thrown.
+
+.. note::
+
+   `bps ping` does *not* test whether compute resources are available or
+   that jobs will run.
+
 .. .. _bps-authenticating:
 
 .. Authenticating

python/lsst/ctrl/bps/cli/cmd/__init__.py

@@ -19,6 +19,6 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
-__all__ = ["acquire", "cluster", "transform", "prepare", "submit", "restart", "report", "cancel"]
+__all__ = ["acquire", "cluster", "transform", "prepare", "submit", "restart", "report", "cancel", "ping"]
 
-from .commands import acquire, cancel, cluster, prepare, report, restart, submit, transform
+from .commands import acquire, cancel, cluster, ping, prepare, report, restart, submit, transform

python/lsst/ctrl/bps/cli/cmd/commands.py

@@ -27,6 +27,7 @@
     acquire_qgraph_driver,
     cancel_driver,
     cluster_qgraph_driver,
+    ping_driver,
     prepare_driver,
     report_driver,
     restart_driver,
@@ -130,3 +131,13 @@ def report(*args, **kwargs):
 def cancel(*args, **kwargs):
     """Cancel submitted workflow(s)."""
     cancel_driver(*args, **kwargs)
+
+
+@click.command(cls=BpsCommand)
+@opt.wms_service_option()
+@click.option("--pass-thru", "pass_thru", default=str(), help="Pass the given string to the WMS service.")
+def ping(*args, **kwargs):
+    """Ping workflow services."""
+    # Note: Using return statement doesn't actually return the value
+    # to the shell.  Using click function instead.
+    click.get_current_context().exit(ping_driver(*args, **kwargs))

python/lsst/ctrl/bps/drivers.py

@@ -35,6 +35,7 @@
     "report_driver",
     "restart_driver",
     "cancel_driver",
+    "ping_driver",
 ]
 
 
@@ -55,6 +56,7 @@
 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 .cancel import cancel
+from .ping import ping
 from .pre_transform import acquire_quantum_graph, cluster_quanta
 from .prepare import prepare
 from .report import report
@@ -496,3 +498,43 @@ def cancel_driver(wms_service, run_id, user, require_bps, pass_thru, is_global=F
         default_config = BpsConfig(BPS_DEFAULTS)
         wms_service = os.environ.get("BPS_WMS_SERVICE_CLASS", default_config["wmsServiceClass"])
     cancel(wms_service, run_id, user, require_bps, pass_thru, is_global=is_global)
+
+
+def ping_driver(wms_service=None, pass_thru=None):
+    """Checks whether WMS services are up, reachable, and any authentication,
+    if needed, succeeds.
+
+    The services to be checked are those needed for submit, report, cancel,
+    restart, but ping cannot guarantee whether jobs would actually run
+    successfully.
+
+    Parameters
+    ----------
+    wms_service : `str`, optional
+        Name of the Workload Management System service class.
+    pass_thru : `str`, optional
+        Information to pass through to WMS.
+
+    Returns
+    -------
+    success : `int`
+        Whether services are up and usable (0) or not (non-zero).
+    """
+    if wms_service is None:
+        default_config = BpsConfig(BPS_DEFAULTS)
+        wms_service = os.environ.get("BPS_WMS_SERVICE_CLASS", default_config["wmsServiceClass"])
+    status, message = ping(wms_service, pass_thru)
+
+    if message:
+        if not status:
+            _LOG.info(message)
+        else:
+            _LOG.error(message)
+
+    # Log overall status message
+    if not status:
+        _LOG.info("Ping successful.")
+    else:
+        _LOG.error("Ping failed (%d).", status)
+
+    return status

python/lsst/ctrl/bps/ping.py

@@ -0,0 +1,59 @@
+# This file is part of ctrl_bps.
+#
+# Developed for the LSST Data Management System.
+# This product includes software developed by the LSST Project
+# (https://www.lsst.org).
+# See the COPYRIGHT file at the top-level directory of this distribution
+# for details of code ownership.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+"""Supporting functions for pinging WMS service.
+"""
+
+__all__ = ["ping"]
+
+import logging
+
+from lsst.utils import doImport
+
+_LOG = logging.getLogger(__name__)
+
+
+def ping(wms_service, pass_thru=None):
+    """Checks whether WMS services are up, reachable, and any authentication,
+    if needed, succeeds.
+
+    The services to be checked are those needed for submit, report, cancel,
+    restart, but ping cannot guarantee whether jobs would actually run
+    successfully.
+
+    Parameters
+    ----------
+    wms_service : `str`
+        Name of the class.
+    pass_thru : `str`
+        A string to pass directly to the WMS service class.
+
+    Returns
+    -------
+    status : `int`
+        Services are available (0) or problems (not 0)
+    message : `str`
+        Any message from WMS (e.g., error details).
+    """
+    wms_service_class = doImport(wms_service)
+    wms_service = wms_service_class({})
+
+    return wms_service.ping(pass_thru)

python/lsst/ctrl/bps/wms_service.py

@@ -260,6 +260,28 @@ def run_submission_checks(self):
         """
         raise NotImplementedError
 
+    def ping(self, pass_thru):
+        """Checks whether WMS services are up, reachable, and can authenticate
+        if authentication is required.
+
+        The services to be checked are those needed for submit, report, cancel,
+        restart, but ping cannot guarantee whether jobs would actually run
+        successfully.
+
+        Parameters
+        ----------
+        pass_thru : `str`, optional
+            Information to pass through to WMS.
+
+        Returns
+        -------
+        status : `int`
+            0 for success, non-zero for failure
+        message : `str`
+            Any message from WMS (e.g., error details).
+        """
+        raise NotImplementedError
+
 
 class BaseWmsWorkflow(metaclass=ABCMeta):
     """Interface for single workflow specific to a WMS.

tests/test_cli_commands.py

@@ -0,0 +1,78 @@
+# This file is part of ctrl_bps.
+#
+# Developed for the LSST Data Management System.
+# This product includes software developed by the LSST Project
+# (https://www.lsst.org).
+# See the COPYRIGHT file at the top-level directory of this distribution
+# for details of code ownership.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# 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 unittest
+
+from lsst.ctrl.bps.cli import bps
+from lsst.daf.butler.cli.utils import LogCliRunner
+
+
+class TestCommandPing(unittest.TestCase):
+    """Test executing the ping subcommand."""
+
+    def setUp(self):
+        self.runner = LogCliRunner()
+
+    def testPingNoArgs(self):
+        with unittest.mock.patch("lsst.ctrl.bps.cli.cmd.commands.ping_driver") as mock_driver:
+            mock_driver.return_value = 0
+            result = self.runner.invoke(bps.cli, ["ping"])
+            self.assertEqual(result.exit_code, 0)
+            mock_driver.assert_called_with(wms_service=None, pass_thru="")
+
+    def testPingClass(self):
+        with unittest.mock.patch("lsst.ctrl.bps.cli.cmd.commands.ping_driver") as mock_driver:
+            mock_driver.return_value = 0
+            result = self.runner.invoke(
+                bps.cli, ["ping", "--wms-service-class", "wms_test_utils.WmsServiceSuccess"]
+            )
+            self.assertEqual(result.exit_code, 0)
+            mock_driver.assert_called_with(wms_service="wms_test_utils.WmsServiceSuccess", pass_thru="")
+
+    def testPingFailure(self):
+        with unittest.mock.patch("lsst.ctrl.bps.cli.cmd.commands.ping_driver") as mock_driver:
+            mock_driver.return_value = 64  # avoid 1 and 2 as returned when cli problems
+            result = self.runner.invoke(
+                bps.cli, ["ping", "--wms-service-class", "wms_test_utils.WmsServiceFailure"]
+            )
+            self.assertEqual(result.exit_code, 64)
+            mock_driver.assert_called_with(wms_service="wms_test_utils.WmsServiceFailure", pass_thru="")
+
+    def testPingPassthru(self):
+        with unittest.mock.patch("lsst.ctrl.bps.cli.cmd.commands.ping_driver") as mock_driver:
+            mock_driver.return_value = 0
+            result = self.runner.invoke(
+                bps.cli,
+                [
+                    "ping",
+                    "--wms-service-class",
+                    "wms_test_utils.WmsServicePassThru",
+                    "--pass-thru",
+                    "EXTRA_VALUES",
+                ],
+            )
+            self.assertEqual(result.exit_code, 0)
+            mock_driver.assert_called_with(
+                wms_service="wms_test_utils.WmsServicePassThru", pass_thru="EXTRA_VALUES"
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_drivers.py

@@ -18,12 +18,15 @@
 #
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
+"""Unit tests for drivers.py."""
+import logging
 import os
 import shutil
 import tempfile
 import unittest
 
-from lsst.ctrl.bps.drivers import _init_submission_driver
+from lsst.ctrl.bps import BpsConfig
+from lsst.ctrl.bps.drivers import _init_submission_driver, ping_driver
 
 TESTDIR = os.path.abspath(os.path.dirname(__file__))
 
@@ -49,5 +52,40 @@ def testMissingSubmitPath(self):
             _init_submission_driver({"payload": {"outputRun": "bad"}})
 
 
+class TestPingDriver(unittest.TestCase):
+    def testWmsServiceSuccess(self):
+        retval = ping_driver("wms_test_utils.WmsServiceSuccess")
+        self.assertEqual(retval, 0)
+
+    def testWmsServiceFailure(self):
+        with self.assertLogs(level=logging.ERROR) as cm:
+            retval = ping_driver("wms_test_utils.WmsServiceFailure")
+            self.assertNotEqual(retval, 0)
+            self.assertEqual(cm.records[0].getMessage(), "Couldn't contact service X")
+
+    def testWmsServiceEnvVar(self):
+        with unittest.mock.patch.dict(
+            os.environ, {"BPS_WMS_SERVICE_CLASS": "wms_test_utils.WmsServiceSuccess"}
+        ):
+            retval = ping_driver()
+            self.assertEqual(retval, 0)
+
+    @unittest.mock.patch.dict(os.environ, {})
+    def testWmsServiceNone(self):
+        # Override default wms to be the test one
+        with unittest.mock.patch.object(BpsConfig, "__getitem__") as mock_function:
+            mock_function.return_value = "wms_test_utils.WmsServiceDefault"
+            with self.assertLogs(level=logging.INFO) as cm:
+                retval = ping_driver()
+                self.assertEqual(retval, 0)
+                self.assertEqual(cm.records[0].getMessage(), "DEFAULT None")
+
+    def testWmsServicePassThru(self):
+        with self.assertLogs(level=logging.INFO) as cm:
+            retval = ping_driver("wms_test_utils.WmsServicePassThru", "EXTRA_VALUES")
+            self.assertEqual(retval, 0)
+            self.assertRegex(cm.output[0], "INFO.+EXTRA_VALUES")
+
+
 if __name__ == "__main__":
     unittest.main()