Merge pull request #363 from douglasjacobsen/on_pipeline

Promote `on` to a proper pipeline

lib/ramble/docs/workspace.rst

@@ -232,7 +232,7 @@ Will only setup experiments that have less than 500 ranks, and:
 
 .. code-block:: console
 
-    $ ramble workspace setup --exclude-shere '"{application_name}" == "hostname"'
+    $ ramble workspace setup --exclude-where '"{application_name}" == "hostname"'
 
 Will exclude all experiments from the ``hostname`` application.
 
@@ -244,6 +244,7 @@ The commands that accept these filters are:
     $ ramble workspace archive
     $ ramble workspace mirror
     $ ramble workspace setup
+    $ ramble on
 
 **NOTE:** The exclusive filter takes precedence over the inclusive filter.
 
@@ -390,6 +391,31 @@ Once a workspace is set up, the experiments inside it can be executed using:
 
     $ ramble on
 
+^^^^^^^^^^^^^^^^
+Custom Executors
+^^^^^^^^^^^^^^^^
+
+When executing the experiments within a workspace, an executor is used.
+Executors are arbitrary strings which are expanded for each experiment, and
+then executed directly.
+
+The default executor is ``'{batch_submit}'`` as this is the variable that is
+used to generate the execution command in the ``all_experiments`` script.
+
+Custom executors can be defined using the ``--executor`` argument to ``ramble
+on`` as in:
+
+.. code-block:: console
+
+    $ ramble on --executor 'echo "{experiment_namespace}"'
+
+This executor will echo each experiment's fully qualified namespace instead of
+executing the experiment.
+
+The value of the executor will be expanded for each experiment, and executed
+independently. Custom executors can be used to have more control over what
+actions to perform with an experiment.
+
 ---------------------
 Analyzing a Workspace
 ---------------------

lib/ramble/ramble/application.py

@@ -35,6 +35,7 @@
 import ramble.keywords
 import ramble.repository
 import ramble.modifier
+import ramble.pipeline
 import ramble.util.executable
 import ramble.util.colors as rucolor
 import ramble.util.hashing
@@ -59,7 +60,7 @@ class ApplicationBase(object, metaclass=ApplicationMeta):
     _workload_exec_key = 'executables'
     _inventory_file_name = 'ramble_inventory.json'
     _status_file_name = 'ramble_status.json'
-    _pipelines = ['analyze', 'archive', 'mirror', 'setup', 'pushtocache']
+    _pipelines = ['analyze', 'archive', 'mirror', 'setup', 'pushtocache', 'execute']
     _language_classes = [ApplicationMeta, SharedMeta]
 
     #: Lists of strings which contains GitHub usernames of attributes.

lib/ramble/ramble/cmd/on.py

@@ -10,6 +10,10 @@
 
 import ramble.workspace
 import ramble.expander
+import ramble.pipeline
+import ramble.filters
+
+import ramble.cmd.common.arguments as arguments
 
 if sys.version_info >= (3, 3):
     from collections.abc import Sequence  # novm noqa: F401
@@ -24,18 +28,32 @@
 
 def setup_parser(subparser):
     subparser.add_argument(
-        '-w', '--workspace', metavar='workspace', dest='ramble_workspace',
-        help='name of workspace to `ramble on`',
+        '--executor', metavar='executor', dest='executor',
+        help='execution template for each experiment',
              required=False)
 
+    arguments.add_common_arguments(subparser, ['where', 'exclude_where'])
+
 
 def ramble_on(args):
-    ws = ramble.cmd.require_active_workspace(cmd_name='workspace info')
+    current_pipeline = ramble.pipeline.pipelines.execute
+    ws = ramble.cmd.require_active_workspace(cmd_name='on')
+
+    executor = args.executor if args.executor else '{batch_submit}'
+
+    filters = ramble.filters.Filters(
+        phase_filters=[],
+        include_where_filters=args.where,
+        exclude_where_filters=args.exclude_where
+    )
+
+    pipeline_cls = ramble.pipeline.pipeline_class(current_pipeline)
+    pipeline = pipeline_cls(ws, filters, executor=executor)
 
     with ws.write_transaction():
-        ws.run_experiments()
+        pipeline.run()
 
 
 def on(parser, args):
-    """Look for a function called environment_<name> and call it."""
+    """Execute `ramble_on` command"""
     ramble_on(args)

lib/ramble/ramble/pipeline.py

@@ -11,6 +11,7 @@
 import os
 import shutil
 import py.path
+import shlex
 
 import llnl.util.filesystem as fs
 import llnl.util.tty as tty
@@ -30,7 +31,7 @@
 from ramble.util.logger import logger
 
 import spack.util.spack_json as sjson
-from spack.util.executable import which
+from spack.util.executable import which, Executable
 
 if not ramble.config.get('config:disable_progress_bar', False):
     try:
@@ -402,17 +403,41 @@ def _complete(self):
         logger.msg(f'Pushed envs to spack cache {self.spack_cache_path}')
 
 
+class ExecutePipeline(Pipeline):
+    """class for the `execute` (`on`) pipeline"""
+
+    name = 'execute'
+
+    def __init__(self, workspace, filters, executor='{batch_submit}'):
+        super().__init__(workspace, filters)
+        self.action_string = 'Executing'
+        self.require_inventory = True
+        self.executor = executor
+
+    def _execute(self):
+        for exp, app_inst, idx in self._experiment_set.filtered_experiments(self.filters):
+            app_inst.add_expand_vars(self.workspace)
+            exec_str = app_inst.expander.expand_var(self.executor)
+            exec_parts = shlex.split(exec_str)
+            exec_name = exec_parts[0]
+            exec_args = exec_parts[1:]
+
+            executor = Executable(exec_name)
+            executor(' '.join(exec_args))
+
+
 pipelines = Enum('pipelines',
                  [AnalyzePipeline.name, ArchivePipeline.name, MirrorPipeline.name,
-                  SetupPipeline.name, PushToCachePipeline.name]
+                  SetupPipeline.name, PushToCachePipeline.name, ExecutePipeline.name]
                  )
 
 _pipeline_map = {
     pipelines.analyze: AnalyzePipeline,
     pipelines.archive: ArchivePipeline,
     pipelines.mirror: MirrorPipeline,
     pipelines.setup: SetupPipeline,
-    pipelines.pushtocache: PushToCachePipeline
+    pipelines.pushtocache: PushToCachePipeline,
+    pipelines.execute: ExecutePipeline
 }
 
 

lib/ramble/ramble/test/cmd/on.py

@@ -63,3 +63,37 @@ def test_execute_nothing(mutable_mock_workspace_path):
         assert os.path.exists(ws.root + '/all_experiments')
 
         ws.run_experiments()
+
+
+def test_on_where(mutable_mock_workspace_path):
+    ws_name = 'test'
+    workspace('create', ws_name)
+
+    with ramble.workspace.read('test') as ws:
+        ramble.test.cmd.workspace.add_basic(ws)
+        ramble.test.cmd.workspace.check_basic(ws)
+
+        workspace('concretize')
+        assert ws.is_concretized()
+
+        workspace('setup')
+        assert os.path.exists(ws.root + '/all_experiments')
+
+        on('--where', '"{experiment_index}" == "1"')
+
+
+def test_on_executor(mutable_mock_workspace_path):
+    ws_name = 'test'
+    workspace('create', ws_name)
+
+    with ramble.workspace.read('test') as ws:
+        ramble.test.cmd.workspace.add_basic(ws)
+        ramble.test.cmd.workspace.check_basic(ws)
+
+        workspace('concretize')
+        assert ws.is_concretized()
+
+        workspace('setup')
+        assert os.path.exists(ws.root + '/all_experiments')
+
+        on('--executor', 'echo "Index = {experiment_index}"')

lib/ramble/ramble/workspace/workspace.py

@@ -150,6 +150,7 @@ def default_config_yaml():
     workspace_template_extension
 
 template_execute_script = """\
+#!/bin/sh
 # This is a template execution script for
 # running the execute pipeline.
 #

share/ramble/ramble-completion.bash

@@ -543,7 +543,7 @@ _ramble_mods_info() {
 }
 
 _ramble_on() {
-    RAMBLE_COMPREPLY="-h --help -w --workspace"
+    RAMBLE_COMPREPLY="-h --help --executor --where --exclude-where"
 }
 
 _ramble_repo() {