brings back ingularity runner (#806)

Adds singularity runner in the new refactored design. This commit also
implements a singularity runner for SLURM. Tests for both host and SLURM
are added in this commit.

.travis.yml

@@ -7,7 +7,7 @@ python:
 - "3.8"
 env:
 - ENGINE=docker
-#- ENGINE=singularity
+- ENGINE=singularity
 services: docker
 before_install:
 - ci/scripts/install_scripts.sh

ci/scripts/install_scripts.sh

@@ -2,18 +2,9 @@
 set -ex
 
 if [ "$ENGINE" == "singularity" ]; then
+    sudo add-apt-repository 'deb http://ftp.de.debian.org/debian bullseye main'
     sudo apt-get update
-    sudo apt-get install -y build-essential libssl-dev uuid-dev libgpgme11-dev libseccomp-dev pkg-config squashfs-tools
-    mkdir -p ${GOPATH}/src/github.com/sylabs
-    cd ${GOPATH}/src/github.com/sylabs
-    git clone https://github.com/sylabs/singularity.git
-    cd singularity
-    git checkout v3.2.0
-    cd ${GOPATH}/src/github.com/sylabs/singularity
-    ./mconfig
-    cd ./builddir
-    make
-    sudo make install
+    wget http://ftp.us.debian.org/debian/pool/main/s/singularity-container/singularity-container_3.5.2+ds1-1_amd64.deb
+    sudo apt-get -f --allow-unauthenticated -y install ./singularity-container_3.5.2+ds1-1_amd64.deb
     singularity version
-    cd $TRAVIS_BUILD_DIR
 fi

ci/test/engine-conf

@@ -20,9 +20,8 @@ EOF
 # config file called settings.yml in the project root.
 cat <<EOF > settings.yml
 engine:
-  name: docker
+  name: $ENGINE
   options:
-    image: abc/xyz
     hostname: xYz.local
 EOF
 

ci/test/offline

@@ -86,7 +86,7 @@ if [ "$ENGINE" == "singularity" ]; then
   rm -rf "$HOME/.cache/popper/singularity/$ID"
   mkdir "$HOME/.cache/popper/singularity/$ID"
   singularity pull --name debian_"$ID".sif docker://debian:stretch-slim
-  mv debian_"$ID".sif "$HOME/.cache/popper/singularity/$ID/popper_docker___debian_stretch-slim_$ID.sif"
+  mv debian_"$ID".sif "$HOME/.cache/popper/singularity/$ID/popper_sample_action_one_$ID.sif"
 else
   docker pull debian:stretch-slim
 fi

cli/popper/parser.py

@@ -102,6 +102,10 @@ def normalize(self):
     @staticmethod
     def new(file, step=None, skipped_steps=[], substitutions=[],
             allow_loose=False, include_step_dependencies=False):
+
+        if not os.path.exists(file):
+            log.fail(f"File {file} was not found.")
+
         if file.endswith('.workflow'):
             wf = HCLWorkflow(file)
         elif file.endswith('.yml') or file.endswith('.yaml'):

cli/popper/runner.py

@@ -123,7 +123,11 @@ def _clone_repos(self, wf):
 
         for _, a in wf.steps.items():
             uses = a['uses']
-            if 'docker://' in uses or './' in uses or uses == 'sh':
+            if ('docker://' in uses
+                or 'shub://' in uses
+                or 'library://' in uses
+                or './' in uses
+                    or uses == 'sh'):
                 continue
 
             url, service, user, repo, step_dir, version = scm.parse(

cli/popper/runner_host.py

@@ -1,14 +1,19 @@
 import os
 import signal
+import threading
 
 import docker
 
-from subprocess import Popen, STDOUT, PIPE, SubprocessError
+from subprocess import Popen, STDOUT, PIPE, SubprocessError, CalledProcessError
+
+import spython
+from spython.main.parse.parsers import DockerParser
+from spython.main.parse.writers import SingularityWriter
 
 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 import StepRunner as StepRunner, WorkflowRunner
 
 
 class HostRunner(StepRunner):
@@ -115,7 +120,6 @@ def __init__(self, init_docker_client=True, **kw):
     def __exit__(self, exc_type, exc_value, exc_traceback):
         if self._d:
             self._d.close()
-        self._spawned_containers = set()
 
     def run(self, step):
         """Execute the given step in docker."""
@@ -158,7 +162,7 @@ def _get_build_info(self, step):
         """
         build = True
         img = None
-        build_source = None
+        build_ctx_path = None
 
         if 'docker://' in step['uses']:
             img = step['uses'].replace('docker://', '')
@@ -170,25 +174,24 @@ def _get_build_info(self, step):
         elif './' in step['uses']:
             img = f'{pu.sanitized_name(step["name"], "step")}'
             tag = f'{self._config.workspace_sha}'
-            build_source = os.path.join(self._config.workspace_dir,
-                                        step['uses'])
+            build_ctx_path = os.path.join(self._config.workspace_dir,
+                                          step['uses'])
         else:
             _, _, user, repo, _, version = scm.parse(step['uses'])
             img = f'{user}/{repo}'.lower()
             tag = version
-            build_source = os.path.join(step['repo_dir'], step['step_dir'])
+            build_ctx_path = os.path.join(step['repo_dir'], step['step_dir'])
 
-        return (build, img, tag, build_source)
+        return (build, img, tag, build_ctx_path)
 
     def _create_container(self, cid, step):
-        build, img, tag, dockerfile = self._get_build_info(step)
+        build, img, tag, build_ctx_path = self._get_build_info(step)
 
         if build:
             log.info(
-                f'[{step["name"]}] docker build {img}:{tag} '
-                f'{os.path.dirname(dockerfile)}')
+                f'[{step["name"]}] docker build {img}:{tag} {build_ctx_path}')
             if not self._config.dry_run:
-                self._d.images.build(path=dockerfile, tag=f'{img}:{tag}',
+                self._d.images.build(path=build_ctx_path, tag=f'{img}:{tag}',
                                      rm=True, pull=True)
         elif not self._config.skip_pull and not step.get('skip_pull', False):
             log.info(f'[{step["name"]}] docker pull {img}:{tag}')
@@ -260,3 +263,196 @@ def _update_with_engine_config(self, container_args):
         for k, v in update_with.items():
             if k not in container_args.keys():
                 container_args[k] = update_with[k]
+
+
+class SingularityRunner(StepRunner):
+    """Runs steps in singularity on the local machine."""
+    lock = threading.Lock()
+
+    def __init__(self, init_spython_client=True, **kw):
+        super(SingularityRunner, self).__init__(**kw)
+
+        self._spawned_containers = set()
+        self._s = None
+
+        if self._config.reuse:
+            log.fail('Reuse not supported for SingularityRunner.')
+
+        if not init_spython_client:
+            return
+
+        self._s = spython.main.Client
+        self._s.quiet = True
+
+    def run(self, step):
+        self._setup_singularity_cache()
+        cid = pu.sanitized_name(step['name'], self._config.wid) + '.sif'
+        self._container = os.path.join(self._singularity_cache, cid)
+
+        exists = os.path.exists(self._container)
+        if exists and not self._config.dry_run and not self._config.skip_pull:
+            os.remove(self._container)
+
+        self._create_container(step, cid)
+        ecode = self._singularity_start(step, cid)
+        return ecode
+
+    @staticmethod
+    def _convert(dockerfile, singularityfile):
+        parser = DockerParser(dockerfile)
+        for p in parser.recipe.files:
+            p[0] = p[0].strip('\"')
+            p[1] = p[1].strip('\"')
+            if os.path.isdir(p[0]):
+                p[0] += '/.'
+
+        writer = SingularityWriter(parser.recipe)
+        recipe = writer.convert()
+        with open(singularityfile, 'w') as sf:
+            sf.write(recipe)
+        return singularityfile
+
+    @staticmethod
+    def _get_recipe_file(build_ctx_path, cid):
+        dockerfile = os.path.join(build_ctx_path, 'Dockerfile')
+        singularityfile = os.path.join(
+            build_ctx_path, 'Singularity.{}'.format(cid[:-4]))
+
+        if os.path.isfile(dockerfile):
+            return SingularityRunner._convert(dockerfile, singularityfile)
+        else:
+            log.fail('No Dockerfile was found.')
+
+    def _build_from_recipe(self, build_ctx_path, build_dest, cid):
+        SingularityRunner.lock.acquire()
+        pwd = os.getcwd()
+        os.chdir(build_ctx_path)
+        recipefile = SingularityRunner._get_recipe_file(build_ctx_path, cid)
+        self._s.build(
+            recipe=recipefile,
+            image=cid,
+            build_folder=build_dest,
+            force=True)
+        os.chdir(pwd)
+        SingularityRunner.lock.release()
+
+    def _get_build_info(self, step):
+        build = True
+        img = None
+        build_ctx_path = None
+
+        if ('docker://' in step['uses']
+            or 'shub://' in step['uses']
+                or 'library://' in step['uses']):
+            img = step['uses']
+            build = False
+
+        elif './' in step['uses']:
+            img = f'{pu.sanitized_name(step["name"], "step")}'
+            build_ctx_path = os.path.join(self._config.workspace_dir,
+                                          step['uses'])
+        else:
+            _, _, user, repo, _, version = scm.parse(step['uses'])
+            img = f'{user}/{repo}'.lower()
+            build_ctx_path = os.path.join(step['repo_dir'], step['step_dir'])
+
+        return (build, img, build_ctx_path)
+
+    def _setup_singularity_cache(self):
+        self._singularity_cache = os.path.join(
+            WorkflowRunner._setup_base_cache(),
+            'singularity',
+            self._config.wid)
+        if not os.path.exists(self._singularity_cache):
+            os.makedirs(self._singularity_cache, exist_ok=True)
+
+    def _update_with_engine_config(self, container_args):
+        update_with = self._config.engine_opts
+        if not update_with:
+            return
+
+        container_args["bind"] = [*container_args["bind"],
+                                  *update_with.get('bind', list())]
+
+        for k, v in update_with.items():
+            if k not in container_args.keys():
+                container_args[k] = update_with[k]
+
+    def _get_container_options(self):
+        container_args = {
+            'userns': True,
+            'pwd': '/workspace',
+            'bind': [f'{self._config.workspace_dir}:/workspace']
+        }
+
+        self._update_with_engine_config(container_args)
+
+        options = []
+        for k, v in container_args.items():
+            if isinstance(v, list):
+                for item in v:
+                    options.append(pu.key_value_to_flag(k, item))
+            else:
+                options.append(pu.key_value_to_flag(k, v))
+
+        options = ' '.join(options).split(' ')
+        log.debug(f'container options: {options}\n')
+
+        return options
+
+    def _create_container(self, step, cid):
+        build, image, build_ctx_path = self._get_build_info(step)
+
+        if build:
+            log.info(
+                f'[{step["name"]}] singularity build {cid} {build_ctx_path}')
+            if not self._config.dry_run:
+                self._build_from_recipe(
+                    build_ctx_path, self._singularity_cache, cid)
+        elif not self._config.skip_pull and not step.get('skip_pull', False):
+            log.info(f'[{step["name"]}] singularity pull {cid} {image}')
+            if not self._config.dry_run:
+                self._s.pull(
+                    image=image,
+                    name=cid,
+                    pull_folder=self._singularity_cache)
+
+    def _singularity_start(self, step, cid):
+        env = StepRunner.prepare_environment(step)
+
+        # set the environment variables
+        for k, v in env.items():
+            os.environ[k] = v
+
+        args = step.get('args', None)
+        runs = step.get('runs', None)
+        ecode = None
+
+        if runs:
+            info = f'[{step["name"]}] singularity exec {cid} {runs}'
+            commands = runs
+            start_fn = self._s.execute
+        else:
+            info = f'[{step["name"]}] singularity run {cid} {args}'
+            commands = args
+            start_fn = self._s.run
+
+        log.info(info)
+
+        if self._config.dry_run:
+            return 0
+
+        options = self._get_container_options()
+        output = start_fn(self._container, commands,
+                          stream=True, options=options)
+        try:
+            for line in output:
+                log.step_info(line.strip('\n'))
+            ecode = 0
+        except CalledProcessError as ex:
+            ecode = ex.returncode
+
+        return ecode
+
+    def stop_running_tasks(self):
+        pass