Add slurm runner

ci/run_tests.sh

@@ -36,3 +36,5 @@ echo "###################################"
 ci/test/offline
 echo "###################################"
 ci/test/engine-conf
+echo "###################################"
+ci/test/slurm

ci/test/engine-conf

@@ -17,58 +17,58 @@ action "run" {
 }
 EOF
 
-# config file called settings.py in the project root.
-cat <<EOF > settings.py
-ENGINE = {
-    "name": "dont use this name",
-    "image": "abc/xyz",
-    "hostname": "xYz.local"
-}
+# config file called settings.yml in the project root.
+cat <<EOF > settings.yml
+engine:
+  name: docker
+  options:
+    image: abc/xyz
+    hostname: xYz.local
 EOF
 
-popper run --wfile main.workflow --conf settings.py > output
+popper run --wfile main.workflow --conf settings.yml > output
 grep -Fxq "xYz.local" output
 
 popper run --wfile main.workflow > output
 grep -vFxq "xYz.local" output
 
 # config file with different name in the project root.
-cat <<EOF > myconf.py
-ENGINE = {
-    "name": "dont use this name",
-    "image": "abc/xyz",
-    "hostname": "xYz.local"
-}
+cat <<EOF > myconf.yml
+engine:
+  name: docker
+  options:
+    image: abc/xyz
+    hostname: xYz.local
 EOF
 
-popper run --wfile main.workflow --conf myconf.py > output
+popper run --wfile main.workflow --conf myconf.yml > output
 grep -Fxq "xYz.local" output
 
 popper run --wfile main.workflow > output
 grep -vFxq "xYz.local" output
 
 # config file in different directory than project root.
 mkdir -p /tmp/myengineconf/
-cat <<EOF > /tmp/myengineconf/mysettings.py
-ENGINE = {
-    "name": "dont use this name",
-    "image": "abc/xyz",
-    "hostname": "xYz.local"
-}
+cat <<EOF > /tmp/myengineconf/mysettings.yml
+engine:
+  name: docker
+  options:
+    image: abc/xyz
+    hostname: xYz.local
 EOF
 
-popper run --wfile main.workflow --conf /tmp/myengineconf/mysettings.py > output
+popper run --wfile main.workflow --conf /tmp/myengineconf/mysettings.yml > output
 grep -Fxq "xYz.local" output
 
 popper run --wfile main.workflow > output
 grep -vFxq "xYz.local" output
 
 cat <<EOF > settings
-ENGINE = {
-    "name": "dont use this name",
-    "image": "abc/xyz",
-    "hostname": "xYz.local"
-}
+engine:
+  name: docker
+  options:
+    image: abc/xyz
+    hostname: xYz.local
 EOF
 
 popper run --wfile main.workflow --conf settings && exit 1

ci/test/slurm

@@ -0,0 +1,44 @@
+#!/bin/bash
+set -ex
+# shellcheck source=./ci/test/common
+source ./ci/test/common
+init_test_repo
+cd "$test_repo_path"
+
+cat <<EOF > wf.yml
+steps:
+- id: one
+  uses: docker://alpine
+  args: echo step one
+
+- id: two
+  uses: sh
+  runs: echo step two
+EOF
+
+cat <<EOF > settings.yml
+engine:
+  name: docker
+
+resource_manager:
+  name: slurm
+EOF
+
+cat <<EOF > test.sh
+#!/bin/bash
+set -ex
+
+export LANG=en_US.utf8
+
+popper run -f wf.yml -c settings.yml > output
+grep -q "step one" output
+grep -q "step two" output
+echo "TEST SLURM PASSED"
+EOF
+
+chmod +x test.sh
+
+# run the tests in the cluster
+docker pull popperized/docker-slurm-cluster:latest
+docker run -h ernie --workdir /src -v /var/run/docker.sock:/var/run/docker.sock -v $PWD:/src popperized/docker-slurm-cluster:latest /src/test.sh
+

cli/popper/parser.py

@@ -623,7 +623,7 @@ def complete_graph(self):
                         if not self.visited.get(tuple(next_set), None):
                             step['next'] = next_set
                             for nsa in next_set:
-                                self.steps[nsa]['needs'] = id
+                                self.steps[nsa]['needs'] = [id]
                             self.visited[tuple(curr_set)] = True
 
         # Finally, generate the root.

cli/popper/runner.py

@@ -36,24 +36,29 @@ def __init__(self, config_file=None, workspace_dir=os.getcwd(),
             # read options from config file
             config_from_file = pu.load_config_file(config_file)
 
-            if hasattr(config_from_file, 'ENGINE'):
-                self.config.engine_options = config_from_file.ENGINE
-            if hasattr(config_from_file, 'RESOURCE_MANAGER'):
-                self.config.resman_options = config_from_file.RESOURCE_MANAGER
-
-        if not self.config.resman:
-            self.config.resman = 'host'
+            # lets consider for now, name property will be present.
+            if config_from_file.get('engine', None):
+                self.config.engine_name = config_from_file['engine']['name']
+                self.config.engine_options = config_from_file['engine'].get(
+                    'options', None)
+            if config_from_file.get('resource_manager', None):
+                self.config.resman_name = config_from_file['resource_manager']['name']
+                self.config.resman_options = config_from_file['resource_manager'].get(
+                    'options', None)
+
+        if not self.config.resman_name:
+            self.config.resman_name = 'host'
 
         # dynamically load resource manager
-        resman_mod_name = f'popper.runner_{self.config.resman}'
+        resman_mod_name = f'popper.runner_{self.config.resman_name}'
         resman_spec = importlib.util.find_spec(resman_mod_name)
         if not resman_spec:
-            raise ValueError(f'Invalid resource manager: {self.config.resman}')
+            raise ValueError(
+                f'Invalid resource manager: {self.config.resman_name}')
         self.resman_mod = importlib.import_module(resman_mod_name)
 
         log.debug(f'WorkflowRunner config:\n{pu.prettystr(self.config)}')
 
-
     def __enter__(self):
         return self
 
@@ -202,6 +207,7 @@ def step_runner(self, engine_name, step):
 
 class StepRunner(object):
     """Base class for step runners, assumed to be singletons."""
+
     def __init__(self, config):
         self.config = config
 

cli/popper/runner_host.py

@@ -10,7 +10,6 @@
 from popper.runner import StepRunner as StepRunner
 
 
-
 class HostRunner(StepRunner):
     """Run an step on the Host Machine."""
 

cli/popper/runner_slurm.py

@@ -0,0 +1,127 @@
+import os
+from subprocess import PIPE, Popen, STDOUT, SubprocessError
+
+import docker
+
+from popper import utils as pu
+from popper import scm
+from popper.cli import log as log
+from popper.runner import StepRunner as StepRunner
+from popper.runner_host import DockerRunner as HostDockerRunner
+from popper.runner_host import HostRunner
+
+
+class SlurmRunner(StepRunner):
+
+    spawned_processes = set()
+
+    def __init__(self, config):
+        super(SlurmRunner, self).__init__(config)
+
+    def __exit__(self, exc_type, exc, traceback):
+        SlurmRunner.spawned_processes = set()
+
+    def exec_srun_cmd(self, cmd):
+        try:
+            cmd.insert(0, 'srun')
+            with Popen(cmd, stdout=PIPE, stderr=STDOUT,
+                       universal_newlines=True, preexec_fn=os.setsid,
+                       cwd=self.config.workspace_dir) as p:
+                SlurmRunner.spawned_processes.add(p)
+
+                log.debug('Reading process output')
+
+                for line in iter(p.stdout.readline, ''):
+                    line_decoded = pu.decode(line)
+                    log.step_info(line_decoded[:-1])
+
+                p.wait()
+                ecode = p.poll()
+                SlurmRunner.spawned_processes.remove(p)
+
+            log.debug(f'Code returned by process: {ecode}')
+
+        except SubprocessError as ex:
+            ecode = ex.returncode
+            log.step_info(f"Command '{cmd[0]}' failed with: {ex}")
+        except Exception as ex:
+            ecode = 1
+            log.step_info(f"Command raised non-SubprocessError error: {ex}")
+
+        return ecode
+
+    def stop_srun_cmd(self):
+        for p in SlurmRunner.spawned_processes:
+            log.info(f'Stopping proces {p.pid}')
+            p.kill()
+
+
+class DockerRunner(SlurmRunner, HostDockerRunner):
+    """Runs steps in docker."""
+
+    def __init__(self, config):
+        super(DockerRunner, self).__init__(config)
+
+    def run(self, step):
+        """Execute the given step in docker."""
+        cid = pu.sanitized_name(step['name'], self.config.wid)
+
+        build, img, dockerfile = HostDockerRunner.get_build_info(
+            step, self.config.workspace_dir, self.config.workspace_sha)
+
+        container = HostDockerRunner.find_container(cid)
+
+        if container and not self.config.reuse and not self.config.dry_run:
+            container.remove(force=True)
+
+        # build or pull
+        if build:
+            HostDockerRunner.docker_build(step, img, dockerfile,
+                                          self.config.dry_run)
+        elif not self.config.skip_pull and not step.get('skip_pull', False):
+            HostDockerRunner.docker_pull(step, img, self.config.dry_run)
+
+        msg = f'{img} {step.get("runs", "")} {step.get("args", "")}'
+        log.info(f'[{step["name"]}] docker create {msg}')
+
+        if not self.config.dry_run:
+            engine_config = {
+                "image": img,
+                "command": step.get('args', None),
+                "name": cid,
+                "volumes": [
+                    f'{self.config.workspace_dir}:/workspace',
+                    '/var/run/docker.sock:/var/run/docker.sock'
+                ],
+                "working_dir": '/workspace',
+                "environment": SlurmRunner.prepare_environment(step),
+                "entrypoint": step.get('runs', None),
+                "detach": True
+            }
+
+            if self.config.engine_options:
+                HostDockerRunner.update_engine_config(
+                    engine_config, self.config.engine_options)
+            log.debug(f'Engine configuration: {pu.prettystr(engine_config)}\n')
+
+            container = HostDockerRunner.d.containers.create(**engine_config)
+
+        log.info(f'[{step["name"]}] srun docker start')
+
+        if self.config.dry_run:
+            return 0
+
+        HostDockerRunner.spawned_containers.append(container)
+        ecode = self.start_container(cid)
+        return ecode
+
+    def start_container(self, cid):
+        docker_cmd = f"docker start --attach {cid}"
+        ecode = self.exec_srun_cmd(docker_cmd.split(" "))
+        return ecode
+
+    def stop_running_tasks(self):
+        for c in HostDockerRunner.spawned_containers:
+            log.info(f'Stopping container {c.name}')
+            c.stop()
+        self.stop_srun_cmd()

cli/popper/utils.py

@@ -147,22 +147,6 @@ def write_file(path, content=''):
     f.close()
 
 
-def module_from_file(module_name, file_path):
-    """Import a file as a module.
-
-    Args:
-      module_name(str): The name of the module.
-      file_path(str): The path to the file to be imported.
-
-    Returns:
-      module
-    """
-    spec = importlib.util.spec_from_file_location(module_name, file_path)
-    module = importlib.util.module_from_spec(spec)
-    spec.loader.exec_module(module)
-    return module
-
-
 def load_config_file(config_file):
     """Validate and parse the engine configuration file.
 
@@ -175,13 +159,13 @@ def load_config_file(config_file):
     if not os.path.exists(config_file):
         log.fail(f'File {config_file} was not found.')
 
-    if not config_file.endswith('.py'):
-        log.fail('Configuration file must be a python source file.')
+    if not config_file.endswith('.yml'):
+        log.fail('Configuration file must be a YAML file.')
 
-    module_name = os.path.basename(config_file)[:-3]
-    module = module_from_file(module_name, config_file)
+    with open(config_file, 'r') as cf:
+        data = yaml.load(cf, Loader=yaml.Loader)
 
-    return module
+    return data
 
 
 def assert_executable_exists(command):

cli/setup.py

@@ -15,11 +15,12 @@
     include_package_data=True,
     install_requires=[
         'dotmap',
+        'testfixtures',
         'python-vagrant',
         'GitPython',
         'spython',
         'click',
-        'pyhcl',
+        'pyhcl==0.4.0',
         'pyyaml',
         'docker'
     ],