slurm tweaks

.travis.yml

@@ -13,7 +13,7 @@ before_install:
 - ci/scripts/install_scripts.sh
 - pip install coverage
 install:
-- pip install cli/
+- pip install cli/[dev]
 script:
 - coverage run -m unittest discover --start-directory cli/test
 - ci/run_tests.sh

CONTRIBUTING.md

@@ -20,7 +20,7 @@ Popper adheres to our code of conduct, [posted in this repository](CODE_OF_CONDU
 Take a look at the issues in our [list of 
 projects](https://github.com/systemslab/popper/projects) to get started!
 
-### Keywords in Issue Titles
+## Keywords in Issue Titles
 
 The title of each open issue is prefixed with a keyword that denotes the 
 following:
@@ -40,7 +40,7 @@ following:
     Workflow that showcases available features of the CLI tool, as 
     well as the catalog of pre-defined actions.
 
-### Branches
+## Branches
 
 There are two main branches of the codebase:
 
@@ -51,7 +51,7 @@ There are two main branches of the codebase:
   * [`master`](./). This tracks the latest 2.x series, which adopted 
     Github actions workflows as the main supported format.
 
-### Install Popper from Source
+## Install Popper from Source
 
 To install Popper in "development mode", we suggest the following 
 approach:
@@ -68,7 +68,7 @@ git clone git@github.com:systemslab/popper
 cd popper
 
 # install popper from source
-pip install -e cli
+pip install -e cli/[dev]
 ```
 
 the `-e` flag passed to `pip` tells it to install the package from the 
@@ -78,6 +78,21 @@ the above approach you have both (1) popper installed in your machine
 and (2) an environment where you can modify popper and test the 
 results of such modifications.
 
+## Running the unittests locally
+
+To run the unittests on your local machine, we suggest the following
+approach:
+
+```bash
+cd popper/
+
+# activate the virtualenv
+source $HOME/venvs/popper/bin/activate
+
+# run the tests
+ENGINE=docker python -X tracemalloc -m unittest discover -f cli/test/
+```
+
 ## How to contribute changes
 
 Once you've identified one of the issues above that you want to contribute to, you're ready to make a change to the project repository!

README.md

@@ -30,7 +30,8 @@ popper run -f wf.yml
 ```
 
 Keep reading down to find [installation instructions](#installation). 
-For more information on the YAML syntax, see [here][cnwf].
+The full example above can be found [here][minimalpython]. For more 
+information on the YAML syntax, see [here][cnwf].
 
 The high-level goals of this project are to provide:
 
@@ -114,3 +115,4 @@ us](mailto:ivo@cs.ucsc.edu).
 [cn]: https://cloudblogs.microsoft.com/opensource/2018/04/23/5-reasons-you-should-be-doing-container-native-development/
 [compose]: https://docs.docker.com/compose/
 [podman]: https://podman.io
+[minimalpython]: https://github.com/popperized/popper-examples/tree/master/workflows/minimal-python

cli/popper/commands/cmd_run.py

@@ -1,8 +1,10 @@
 import click
 import os
+import traceback
 
 from popper import log as logging
 from popper.cli import log, pass_context
+from popper.config import PopperConfig
 from popper.parser import Workflow
 from popper.runner import WorkflowRunner
 
@@ -118,7 +120,23 @@
 def cli(ctx, step, wfile, debug, dry_run, log_file, quiet, reuse,
         engine, resource_manager, skip, skip_pull, skip_clone,
         substitution, allow_loose, with_dependencies, workspace, conf):
-    """Runs a Popper workflow. Only execute STEP if given."""
+    """Runs a Popper workflow. Only executes STEP if given.
+
+    This command allows specifying the engine and the resource manager
+    in two different ways.
+
+    * Using the `--engine/-e` option and `--resource-manager/-r` option.
+
+    * Through a configuration file specified with the `--conf/-c` option.
+
+    NOTE:
+
+    1. If none of the above are given, popper uses docker as the
+       default engine and host as the default resource manager.
+
+    2. If the engine or resource manager is specified through CLI and
+       config file both, CLI is given preference over config file.
+    """
     # set the logging levels.
     level = 'STEP_INFO'
     if quiet:
@@ -146,14 +164,15 @@ def cli(ctx, step, wfile, debug, dry_run, log_file, quiet, reuse,
                       substitutions=substitution, allow_loose=allow_loose,
                       include_step_dependencies=with_dependencies)
 
-    # instantiate the runner
-    runner = WorkflowRunner(
-        engine,
-        resource_manager,
-        config_file=conf,
-        dry_run=dry_run,
-        reuse=reuse,
-        skip_pull=skip_pull,
-        skip_clone=skip_clone,
-        workspace_dir=workspace)
-    runner.run(wf)
+    config = PopperConfig(engine_name=engine, resman_name=resource_manager,
+                          config_file=conf, reuse=reuse, dry_run=dry_run,
+                          skip_pull=skip_pull, skip_clone=skip_clone,
+                          workspace_dir=workspace)
+
+    runner = WorkflowRunner(config)
+
+    try:
+        runner.run(wf)
+    except Exception as e:
+        log.debug(traceback.format_exc())
+        log.fail(e)

cli/popper/config.py

@@ -1,59 +1,100 @@
 import os
+import yaml
 
-from hashlib import shake_256
+import popper.scm as scm
 
+from hashlib import shake_256
 from popper.cli import log as log
 
-import popper.scm as scm
-import popper.utils as pu
-
 
 class PopperConfig(object):
-    def __init__(self, **kwargs):
+    def __init__(self, engine_name=None, resman_name=None, config_file=None,
+                 workspace_dir=os.getcwd(), reuse=False,
+                 dry_run=False, quiet=False, skip_pull=False,
+                 skip_clone=False):
+
+        self.workspace_dir = os.path.realpath(workspace_dir)
+        self.reuse = reuse
+        self.dry_run = dry_run
+        self.quiet = quiet
+        self.skip_pull = skip_pull
+        self.skip_clone = skip_clone
         self.repo = scm.new_repo()
-        self.workspace_dir = os.path.realpath(kwargs['workspace_dir'])
-        self.wid = shake_256(self.workspace_dir.encode('utf-8')).hexdigest(4)
         self.workspace_sha = scm.get_sha(self.repo)
-        self.config_file = kwargs['config_file']
-        self.dry_run = kwargs['dry_run']
-        self.skip_clone = kwargs['skip_clone']
-        self.skip_pull = kwargs['skip_pull']
-        self.quiet = kwargs['quiet']
-        self.reuse = kwargs['reuse']
-        self.engine_name = kwargs.get('engine', None)
-        self.resman_name = kwargs.get('resource_manager', None)
-        self.engine_options = kwargs['engine_options']
-        self.resman_options = kwargs['resman_options']
-        self.config_from_file = pu.load_config_file(self.config_file)
-
-    def parse(self):
-        self.validate()
-        self.normalize()
-
-    def validate(self):
-        if self.config_from_file.get('engine', None):
-            if not self.config_from_file['engine'].get('name', None):
-                log.fail(
-                    'engine config must have the name property.')
-
-        if self.config_from_file.get('resource_manager', None):
-            if not self.config_from_file['resource_manager'].get('name', None):
-                log.fail(
-                    'resource_manager config must have the name property.')
-
-    def normalize(self):
-        if not self.engine_name:
-            if self.config_from_file.get('engine', None):
-                self.engine_name = self.config_from_file['engine']['name']
-                self.engine_options = self.config_from_file['engine'].get(
-                    'options', dict())
-            else:
-                self.engine_name = 'docker'
-
-        if not self.resman_name:
-            if self.config_from_file.get('resource_manager', None):
-                self.resman_name = self.config_from_file['resource_manager']['name']
-                self.resman_options = self.config_from_file['resource_manager'].get(
-                    'options', dict())
-            else:
-                self.resman_name = 'host'
+
+        wid = shake_256(self.workspace_dir.encode('utf-8')).hexdigest(4)
+        self.wid = wid
+
+        from_file = self._load_config_from_file(config_file, engine_name,
+                                                resman_name)
+
+        self.engine_name = from_file['engine_name']
+        self.resman_name = from_file['resman_name']
+        self.engine_opts = from_file['engine_opts']
+        self.resman_opts = from_file['resman_opts']
+
+    def _load_config_from_file(self, config_file, engine_name, resman_name):
+        from_file = PopperConfig.__load_config_file(config_file)
+        loaded_conf = {}
+
+        eng_section = from_file.get('engine', None)
+        eng_from_file = from_file.get('engine', {}).get('name')
+        if from_file and eng_section and not eng_from_file:
+            log.fail('No engine name given.')
+
+        resman_section = from_file.get('resource_manager', None)
+        resman_from_file = from_file.get('resource_manager', {}).get('name')
+        if from_file and resman_section and not resman_from_file:
+            log.fail('No resource manager name given.')
+
+        # set name in precedence order (or assigne default values)
+        if engine_name:
+            loaded_conf['engine_name'] = engine_name
+        elif eng_from_file:
+            loaded_conf['engine_name'] = eng_from_file
+        else:
+            loaded_conf['engine_name'] = 'docker'
+
+        if resman_name:
+            loaded_conf['resman_name'] = resman_name
+        elif resman_from_file:
+            loaded_conf['resman_name'] = resman_from_file
+        else:
+            loaded_conf['resman_name'] = 'host'
+
+        engine_opts = from_file.get('engine', {}).get('options', {})
+        resman_opts = from_file.get('resource_manager', {}).get('options', {})
+        loaded_conf['engine_opts'] = engine_opts
+        loaded_conf['resman_opts'] = resman_opts
+
+        return loaded_conf
+
+    @staticmethod
+    def __load_config_file(config_file):
+        """Validate and parse the engine configuration file.
+
+        Args:
+          config_file(str): Path to the file to be parsed.
+
+        Returns:
+          dict: Engine configuration.
+        """
+        if isinstance(config_file, dict):
+            return config_file
+
+        if not config_file:
+            return dict()
+
+        if not os.path.exists(config_file):
+            log.fail(f'File {config_file} was not found.')
+
+        if not config_file.endswith('.yml'):
+            log.fail('Configuration file must be a YAML file.')
+
+        with open(config_file, 'r') as cf:
+            data = yaml.load(cf, Loader=yaml.Loader)
+
+        if not data:
+            log.fail('Configuration file is empty.')
+
+        return data

cli/popper/parser.py

@@ -3,13 +3,13 @@
 import re
 import hcl
 import os
+import threading
 import yaml
 
 from copy import deepcopy
 from builtins import str, dict
 from popper.cli import log as log
 
-import popper.scm as scm
 import popper.utils as pu
 
 
@@ -24,6 +24,45 @@
     "next"]
 
 
+class threadsafe_iter_3:
+    """Takes an iterator/generator and makes it thread-safe by serializing call
+    to the `next` method of given iterator/generator."""
+
+    def __init__(self, it):
+        self.it = it
+        self.lock = threading.Lock()
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        with self.lock:
+            return self.it.__next__()
+
+
+def threadsafe_generator(f):
+    """A decorator that takes a generator function and makes it thread-safe.
+
+    Args:
+      f(function): Generator function
+
+    Returns:
+      None
+    """
+    def g(*args, **kwargs):
+        """
+
+        Args:
+          *args(list): List of non-key worded,variable length arguments.
+          **kwargs(dict): List of key-worded,variable length arguments.
+
+        Returns:
+          function: The thread-safe function.
+        """
+        return threadsafe_iter_3(f(*args, **kwargs))
+    return g
+
+
 class Workflow(object):
     """Represents an immutable workflow."""
 
@@ -98,7 +137,7 @@ def format_command(params):
             return params.split(" ")
         return params
 
-    @pu.threadsafe_generator
+    @threadsafe_generator
     def get_stages(self):
         """Generator of stages. A stages is a list of steps that can be
         executed in parallel.