Merge pull request #516 from minrk/slurm

Slurm test coverage

Author: minrk

.github/workflows/test.yml

@@ -19,15 +19,17 @@ jobs:
       matrix:
         include:
           - python: "3.9"
-            ssh: ssh
+            cluster_type: ssh
+          - python: "3.8"
+            cluster_type: mpi
+          - python: "3.7"
+            cluster_type: slurm
           - python: "3.6"
             tornado: "5.1.1"
           - python: "3.7"
             controller_ip: "*"
           - python: "3.8"
             runs_on: windows-2019
-          - python: "3.8"
-            mpi: mpi
           - python: "3.9"
             runs_on: macos-10.15
 
@@ -47,15 +49,27 @@ jobs:
           echo "IPP_CONTROLLER_IP=${{ matrix.controller_ip }}" >> $GITHUB_ENV
 
       - name: Set up docker-compose for ssh launcher
-        if: ${{ matrix.ssh }}
+        if: ${{ matrix.cluster_type == 'ssh' }}
         run: |
           export DOCKER_BUILDKIT=1
           export COMPOSE_DOCKER_CLI_BUILD=1
           cd ci/ssh
           docker-compose up -d --build
 
+      - name: Set up slurm
+        if: ${{ matrix.cluster_type == 'slurm' }}
+        run: |
+          sudo rm -rf /var/lib/apt/lists
+          sudo apt-get update && sudo apt-get -f -y install && sudo apt-get -y install slurm-wlm
+          sudo cp ci/slurm/slurm.conf /etc/slurm-llnl/
+          sudo mkdir /var/spool/slurmctl
+          sudo mkdir /var/spool/slurmd
+          sudo service munge start
+          sudo service slurmd start
+          sudo service slurmctld start
+
       - name: Install Python (conda) ${{ matrix.python }}
-        if: matrix.mpi
+        if: ${{ matrix.cluster_type == 'mpi' }}
         run: |
           export MAMBA_ROOT_PREFIX=$HOME/conda
           test -d $MAMBA_ROOT_PREFIX || mkdir $MAMBA_ROOT_PREFIX
@@ -66,7 +80,7 @@ jobs:
           echo "PATH=$MAMBA_ROOT_PREFIX/bin:$PATH" >> $GITHUB_ENV
 
       - name: Install Python ${{ matrix.python }}
-        if: ${{ ! matrix.mpi }}
+        if: ${{ matrix.cluster_type != 'mpi' }}
         uses: actions/setup-python@v2
         with:
           python-version: ${{ matrix.python }}
@@ -85,23 +99,29 @@ jobs:
       - name: Show environment
         run: pip freeze
 
-      - name: Run mpi tests
-        if: ${{ matrix.mpi }}
+      - name: Run ${{ matrix.cluster_type }} tests
+        if: ${{ matrix.cluster_type }}
         run: |
-          pytest -vx --color=yes --cov=ipyparallel ipyparallel/tests/test_cluster.py
-
-      - name: Run ssh tests
-        if: ${{ matrix.ssh }}
-        run: |
-          pytest -ra -v --maxfail=2 --color=yes --cov=ipyparallel ipyparallel/tests/test_ssh.py
+          pytest -ra -v --maxfail=2 --color=yes --cov=ipyparallel ipyparallel/tests/test_${{ matrix.cluster_type }}.py
 
       - name: Run tests
-        if: ${{ ! matrix.mpi && ! matrix.ssh }}
+        if: ${{ ! matrix.cluster_type }}
         # FIXME: --color=yes explicitly set because:
         #        https://github.com/actions/runner/issues/241
         run: |
-          pytest -v --maxfail=3 --color=yes --cov=ipyparallel ipyparallel/tests
+          pytest -ra -v --maxfail=3 --color=yes --cov=ipyparallel ipyparallel/tests
 
       - name: Submit codecov report
         run: |
           codecov
+
+      - name: Report on slurm
+        if: ${{ matrix.cluster_type == 'slurm' && failure() }}
+        run: |
+          set -x
+          slurmd -C
+          ls -l
+          squeue
+          sinfo
+          scontrol show node=localhost
+          sudo cat /var/log/slurm-llnl/*

ci/slurm/slurm.conf

@@ -0,0 +1,53 @@
+# slurm.conf file generated by configurator easy.html.
+# Put this file on all nodes of your cluster.
+# See the slurm.conf man page for more information.
+#
+SlurmctldHost=localhost
+#
+#MailProg=/bin/mail
+MpiDefault=none
+#MpiParams=ports=#-#
+ProctrackType=proctrack/linuxproc
+ReturnToService=1
+SlurmctldPidFile=/var/run/slurmctld.pid
+#SlurmctldPort=6817
+SlurmdPidFile=/var/run/slurmd.pid
+#SlurmdPort=6818
+SlurmdSpoolDir=/var/spool/slurmd
+SlurmUser=root
+#SlurmdUser=root
+StateSaveLocation=/var/spool/slurmctl
+SwitchType=switch/none
+#TaskPlugin=task/affinity
+#CoreSpecPlugin=core_spec/none
+
+#
+# TIMERS
+#KillWait=30
+#MinJobAge=300
+#SlurmctldTimeout=120
+#SlurmdTimeout=300
+#
+#
+# SCHEDULING
+SchedulerType=sched/backfill
+SelectType=select/cons_tres
+SelectTypeParameters=CR_Core
+#
+#
+# LOGGING AND ACCOUNTING
+AccountingStorageType=accounting_storage/none
+ClusterName=cluster
+#JobAcctGatherFrequency=30
+JobAcctGatherType=jobacct_gather/none
+SlurmctldDebug=debug5
+SlurmctldLogFile=/var/log/slurm-llnl/slurmctld.log
+SlurmdDebug=debug5
+SlurmdLogFile=/var/log/slurm-llnl/slurmd.log
+#
+#
+# COMPUTE NODES
+# Note: CPUs apparently cannot be oversubscribed
+# this can only run where at least 2 CPUs are available
+NodeName=localhost NodeHostName=localhost NodeAddr=127.0.0.1 CPUs=2 State=UNKNOWN
+PartitionName=part Nodes=localhost Default=YES MaxTime=INFINITE State=UP OverSubscribe=YES

ipyparallel/cluster/launcher.py

@@ -1586,6 +1586,12 @@ def _cluster_id_default(self):
         help="The name of the command line program used to delete jobs.",
     )
 
+    signal_command = List(
+        [''],
+        config=True,
+        help="The name of the command line program used to send signals to jobs.",
+    )
+
     job_id = Unicode().tag(to_dict=True)
 
     job_id_regexp = CRegExp(
@@ -1616,7 +1622,7 @@ def _cluster_id_default(self):
     def _queue_changed(self, change):
         self._update_context(change)
 
-    n = Integer(1)
+    n = Integer(1).tag(to_dict=True)
 
     @observe('n')
     def _n_changed(self, change):
@@ -1643,7 +1649,7 @@ def _n_changed(self, change):
         This lets you parameterize additional options,
         such as wall_time with a custom template.
         """,
-    )
+    ).tag(to_dict=True)
 
     @default("context")
     def _context_default(self):
@@ -1714,7 +1720,7 @@ def write_batch_script(self, n=1):
         # from user config
         ns.update(self.namespace)
         script_as_string = self.formatter.format(self.batch_template, **ns)
-        self.log.debug('Writing batch script: %s', self.batch_file)
+        self.log.debug(f'Writing batch script: {self.batch_file}\n{script_as_string}')
         with open(self.batch_file, 'w') as f:
             f.write(script_as_string)
         os.chmod(self.batch_file, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)
@@ -1739,35 +1745,44 @@ def start(self, n=1):
         # Here we save profile_dir in the context so they
         # can be used in the batch script template as {profile_dir}
         self.write_batch_script(n)
+
         output = check_output(self.args, env=os.environ)
         output = output.decode(DEFAULT_ENCODING, 'replace')
+        self.log.debug(f"Submitted {shlex_join(self.args)}. Output: {output}")
 
         job_id = self.parse_job_id(output)
         self.notify_start(job_id)
         return job_id
 
     def stop(self):
         try:
-            p = Popen(
+            output = check_output(
                 self.delete_command + [self.job_id],
-                env=os.environ,
-                stdout=PIPE,
-                stderr=PIPE,
-            )
-            out, err = p.communicate()
-            output = out + err
-        except:
+                stdin=None,
+            ).decode(DEFAULT_ENCODING, 'replace')
+        except Exception:
             self.log.exception(
                 "Problem stopping cluster with command: %s"
                 % (self.delete_command + [self.job_id])
             )
             output = ""
-        output = output.decode(DEFAULT_ENCODING, 'replace')
+
         self.notify_stop(
             dict(job_id=self.job_id, output=output)
         )  # Pass the output of the kill cmd
         return output
 
+    def signal(self, sig):
+        cmd = self.signal_command + [str(sig), self.job_id]
+        try:
+            output = check_output(
+                cmd,
+                stdin=None,
+            ).decode(DEFAULT_ENCODING, 'replace')
+        except Exception:
+            self.log.exception("Problem sending signal with: {shlex_join(cmd)}")
+            output = ""
+
 
 class BatchControllerLauncher(BatchSystemLauncher, ControllerLauncher):
     @default("program")
@@ -1813,6 +1828,9 @@ class PBSLauncher(BatchSystemLauncher):
 
     submit_command = List(['qsub'], config=True, help="The PBS submit command ['qsub']")
     delete_command = List(['qdel'], config=True, help="The PBS delete command ['qdel']")
+    signal_command = List(
+        ['qsig', '-s'], config=True, help="The PBS signal command ['qsig']"
+    )
     job_id_regexp = CRegExp(
         r'\d+',
         config=True,
@@ -1868,6 +1886,11 @@ class SlurmLauncher(BatchSystemLauncher):
     delete_command = List(
         ['scancel'], config=True, help="The slurm delete command ['scancel']"
     )
+    signal_command = List(
+        ['scancel', '-s'],
+        config=True,
+        help="The slurm signal command ['scancel', '-s']",
+    )
     job_id_regexp = CRegExp(
         r'\d+',
         config=True,
@@ -2023,9 +2046,12 @@ class SGEEngineSetLauncher(SGELauncher, BatchEngineSetLauncher):
 class LSFLauncher(BatchSystemLauncher):
     """A BatchSystemLauncher subclass for LSF."""
 
-    submit_command = List(['bsub'], config=True, help="The PBS submit command ['bsub']")
+    submit_command = List(['bsub'], config=True, help="The LSF submit command ['bsub']")
     delete_command = List(
-        ['bkill'], config=True, help="The PBS delete command ['bkill']"
+        ['bkill'], config=True, help="The LSF delete command ['bkill']"
+    )
+    signal_command = List(
+        ['bkill', '-s'], config=True, help="The LSF signal command ['bkill', '-s']"
     )
     job_id_regexp = CRegExp(
         r'\d+',

ipyparallel/tests/test_cluster.py

@@ -1,6 +1,5 @@
 import asyncio
 import os
-import shutil
 import signal
 import sys
 import time
@@ -12,10 +11,6 @@
 from ipyparallel import cluster
 from ipyparallel.cluster.launcher import find_launcher_class
 
-_engine_launcher_classes = ["Local"]
-if shutil.which("mpiexec"):
-    _engine_launcher_classes.append("MPI")
-
 _timeout = 30
 
 
@@ -58,9 +53,9 @@ async def test_ipython_log(ipython):
     assert c.log.handlers[0].stream is sys.stdout
 
 
-@pytest.fixture(params=_engine_launcher_classes)
-def engine_launcher_class(request):
-    return request.param
+@pytest.fixture
+def engine_launcher_class():
+    return 'Local'
 
 
 async def test_start_stop_controller(Cluster):
@@ -88,7 +83,7 @@ async def test_start_stop_engines(Cluster, engine_launcher_class):
     cluster = Cluster(engine_launcher_class=engine_launcher_class)
     await cluster.start_controller()
 
-    n = 3
+    n = 2
     engine_set_id = await cluster.start_engines(n)
     assert engine_set_id in cluster._engine_sets
     engine_set = cluster._engine_sets[engine_set_id]
@@ -127,10 +122,10 @@ async def test_start_stop_cluster(Cluster, engine_launcher_class):
 async def test_signal_engines(request, Cluster, engine_launcher_class):
     cluster = Cluster(engine_launcher_class=engine_launcher_class)
     await cluster.start_controller()
-    engine_set_id = await cluster.start_engines(n=3)
+    engine_set_id = await cluster.start_engines(n=2)
     rc = await cluster.connect_client()
     request.addfinalizer(rc.close)
-    rc.wait_for_engines(3)
+    rc.wait_for_engines(2)
     # seems to be a problem if we start too soon...
     await asyncio.sleep(1)
     # ensure responsive
@@ -157,7 +152,7 @@ async def test_signal_engines(request, Cluster, engine_launcher_class):
 
 
 async def test_restart_engines(Cluster, engine_launcher_class):
-    n = 3
+    n = 2
     async with Cluster(engine_launcher_class=engine_launcher_class, n=n) as rc:
         cluster = rc.cluster
         engine_set_id = next(iter(cluster._engine_sets))

ipyparallel/tests/test_mpi.py

@@ -0,0 +1,17 @@
+import shutil
+
+import pytest
+
+from .test_cluster import test_restart_engines  # noqa: F401
+from .test_cluster import test_signal_engines  # noqa: F401
+from .test_cluster import test_start_stop_cluster  # noqa: F401
+from .test_cluster import test_to_from_dict  # noqa: F401
+
+# import tests that use engine_launcher_class fixture
+
+# override engine_launcher_class
+@pytest.fixture
+def engine_launcher_class():
+    if shutil.which("mpiexec") is None:
+        pytest.skip("Requires mpiexec")
+    return 'MPI'

ipyparallel/tests/test_slurm.py

@@ -0,0 +1,17 @@
+import shutil
+
+import pytest
+
+from .test_cluster import test_restart_engines  # noqa: F401
+from .test_cluster import test_signal_engines  # noqa: F401
+from .test_cluster import test_start_stop_cluster  # noqa: F401
+from .test_cluster import test_to_from_dict  # noqa: F401
+
+# import tests that use engine_launcher_class fixture
+
+# override engine_launcher_class
+@pytest.fixture
+def engine_launcher_class():
+    if shutil.which("sbatch") is None:
+        pytest.skip("Requires slurm")
+    return 'Slurm'