Implement subset of the Common Workflow Language tool and workflow formats.

This should support a subset of [draft-3](http://www.commonwl.org/draft-3/) and [v1.0](http://www.commonwl.org/v1.0/) tools.

CWL Support (Tools):
--------------------

- Implemented integer, long, float, double, boolean, string, and File parameters, and arrays of files, as well as `["null", <simple_type>]` union parameters and Any-type parameters, and simple records. More complex unions and lists of datatypes may stil ube nsupported (unions of two or more non-null parameters, unions of `["null", Any]`, etc...).
- ``InlineJavascriptRequirement`` are support to define output files (see ``test_cat3`` test case).
- ``EnvVarRequirement``s are supported (see the ``test_env_tool1`` and ``test_env_tool2`` test cases).
- Expression tools are supported (see ``parseInt-tool`` test case).
- Shell tools are also support (see record output test case).
- Default File values are very un-Galaxy and have been hacked into work with Tools.

CWL Support (Workflows):
------------------------

- Simple connections and tool execution.
- MultipleInputFeatureRequirements to glue together multiple file inputs into a File[] or multiple File[] into a single flat File[]. (nested merge is still a TODO).
- Simple scatter semantics when they match Galaxy's implicit scatters (e.g. count-lines3) and simple scatters over non-File parameters.
- Simple subworkflows (e.g. count-lines10).

Previously Worked:
-------------------

The following things worked with Draft 3 and a previous version of cwltool but need to be updated.

- Draft 3 `CreateFileRequirement`s were supported (see the `test_rename` test case).
- Secondary files were supported at least partially, see the `index1` and `showindex1` CWL tools created to verify this as well as the `test_index1` test case.
- Docker integration is only partial (simple docker pull is supported) - so `cat3-tool.cwl` works for example. Full semantics of CWL docker support has yet to be implemented. The remaining work is straight-forward and trackd in the meta-issue #1684.

Remaining Work
---------------------------------

The work remaining is vast and will be tracked at https://github.com/common-workflow-language/galaxy/issues for the time being.

Implementation Notes:
----------------------

Tools:

 - Non-File CWL outputs are represented as ``expression.json`` files. Traditionally Galaxy hasn't supported non-File outputs from tools but CWL Galaxy has work in progress on bringing native Galaxy support for such outputs common-workflow-lab#27.
 - CWL secondary files are stored in ``__secondary_files__`` directory in the dataset's extra_files_path directory.
 - The tool execution API has been extended to add a ``inputs_representation`` parameter that can be set to "cwl" now. The ``cwl`` representation for running tools corresonding to the CWL job json format with {class: "File: path: "/path/to/file"} inputs replaced with {"src": "hda", "id": "<dataset_id>"}. Code for building these requests for CWL job json is available in the test class.
 - Since the CWL <-> Galaxy parameter translation may change over time, for instance if Galaxy develops or refines parameter classes - CWL state and CWL state version is tracked in the database and hopefully for reruns, etc... we could update the Galaxy state from an older version to a new one.
 - CWL allows output parameters to be either ``File`` or non-``File`` and determined at runtime, so ``galaxy.json`` is used to dynamically adjust output extension as needed for non-``File`` parameters.

Workflows:

- This work serializes embedded and referenced tools into the database - this will allow reuse and tracing without require the path to exist forever on the filesystem - this will have problems with default file references in workflows.
- Implements re-mapping CWL workflow connections to Galaxy input connections.
- Fix tool serialization for jobs for path-less tools (such as embedded tools).
- Hack tool state during workflow import for CWL.
- The sort of dynamic shaping of inputs CWL allows has required enhancing Galaxy's map/reduce stuff to allow mapping over dynamic collections that don't yet exist at the time of tool execution and need to be created on the fly. This commit creates them as HDCAs - but likely they should be something else that doesn't appear in the history panel.
- Multi-input scattering but only scatterMethod == "dotproduct" is currently support. Other scatter methods (nested_crossproduct and flatcross_product) are not used by workflows in GA4GH challenge.

Implementation Description:
-----------------------------

The reference implementation Python library (mainly developed by Peter Amstutz - https://github.com/common-workflow-language/common-workflow-language/tree/master/reference) is used to load tool files ending with ``.json`` or ``.cwl`` and proxy objects are created to adapt these tools to Galaxy representations. In particular input and output descriptions are loaded from the tool.

When the tool is submitted, a special specialized tool class is used to build a cwltool compatible job description from the supplied Galaxy inputs and the CWL reference implementation is used to generate a CWL reference implementation Job object. A command-line is generated from this Job object.

As a result of this - Galaxy largely does not need to worry about the details of command-line adapters, expressions, etc....

Galaxy writes a description of the CWL job that it can reload to the job working directory. After the process is complete (on the Galaxy compute server, but outside the Docker container) this representation is reloaded and the dynamic outputs are discovered and moved to fixed locations as expected by Galaxy. CWL allows for much more expressive output locations than Galaxy, for better or worse, and this step uses cwltool to adapt CWL to Galaxy outputs.

Currently all ``File`` outputs are sniffed to determined a Galaxy datatype, CWL allows refinement on this and this remains work to be done.

  1) CWL should support EDAM declaration of types and Galaxy should provide a mapping to core datasets to skip sniffing is types are found.
  2) For finer grain control within Galaxy, extensions to CWL should allow setting actual Galaxy output types on outputs. (Distinction between fastq and fastqsanger in Galaxy is very important for instance.)

Implementation Links:
----------------------

Hundreds of commits have been rebased into this one and so the details of individual parts of the implementation and how they built on each other are not enitrely clear. To see the original ideas behind individual features - here are some relevant links:

- Implement merge_nested link semantics for workflow steps (common-workflow-lab@a903abd).
- Implement subworkflows in CWL (common-workflow-lab@9933c3c)
- MultipleInputFeatureRequirements:
  - Second attempt: common-workflow-lab@ed8307f
  - First attempt: common-workflow-lab@ae11f56
- Basic, implicit dotproduct scattering of workflows - common-workflow-lab@d1ad64e.
- Simple input StepInputExpressionRequirements - common-workflow-lab@819a27b
- StepInputExpressionRequirements for multiple inputs - common-workflow-lab@5e7f622
- Record Types in CWL - common-workflow-lab@e6be28a
- Rework original approach at mapping CWL state to tool state - common-workflow-lab@669ea55
- Rework approach at mapping CWL state to tool state again to use "FieldTypeToolParameter"s - implements default values, optional parameters, and union types for workflow inputs. common-workflow-lab@d1ca22f

Testing:
---------------------

    % git clone https://github.com/common-workflow-language/galaxy.git
    % cd galaxy
    % git checkout cwl-1.0

Start Galaxy.

    % GALAXY_RUN_WITH_TEST_TOOLS=1 sh run.sh

Open http://localhost:8080/ and see CWL test tools (along with all Galaxy test tools) in left hand tool panel.

To go a step further and actually run CWL jobs within their designated Docker containers, copy the following minimal Galaxy job configuration file to ``config/job_conf.xml``. (Adjust the ``docker_sudo`` parameter based on how you execute Docker).

https://gist.github.com/jmchilton/3997fa471d1b4c556966

Run API tests demonstrating the various CWL demo tools with the following command.

```
./run_tests.sh -api test/api/test_tools_cwl.py
./run_tests.sh -api test/api/test_workflows_cwl.py
./run_tests.sh -api test/api/test_cwl_conformance_v1_0.py
```

The first two execute various tool and workflow test cases manually crafted during implementation of this work. The third is an auto-generate test case class that contains Python tests for every CWL conformance test found with the reference specification.

Issues and Contact
---------------------------------

Report issues at https://github.com/common-workflow-language/galaxy/issues and feel free ping jmchilton on the CWL [Gitter channel](https://gitter.im/common-workflow-language/common-workflow-language).

Author: jmchilton

lib/galaxy/config.py

@@ -324,9 +324,10 @@ def __init__( self, **kwargs ):
         # These are not even beta - just experiments - don't use them unless
         # you want yours tools to be broken in the future.
         self.enable_beta_tool_formats = string_as_bool( kwargs.get( 'enable_beta_tool_formats', 'False' ) )
+        # Should CWL artifacts be loaded with strict validation enabled.
+        self.strict_cwl_validation = string_as_bool( kwargs.get( 'strict_cwl_validation', 'True') )
         # Beta containers interface used by GIEs
         self.enable_beta_containers_interface = string_as_bool( kwargs.get( 'enable_beta_containers_interface', 'False' ) )
-
         # Certain modules such as the pause module will automatically cause
         # workflows to be scheduled in job handlers the way all workflows will
         # be someday - the following two properties can also be used to force this

lib/galaxy/dataset_collections/matching.py

@@ -16,8 +16,10 @@ class CollectionsToMatch( object ):
 
     def __init__( self ):
         self.collections = {}
+        self.uses_ephemeral_collections = False
 
     def add( self, input_name, hdca, subcollection_type=None, linked=True ):
+        self.uses_ephemeral_collections = self.uses_ephemeral_collections or not hasattr( hdca, "hid" )
         self.collections[ input_name ] = bunch.Bunch(
             hdca=hdca,
             subcollection_type=subcollection_type,
@@ -45,6 +47,7 @@ def __init__( self ):
         self.linked_structure = None
         self.unlinked_structures = []
         self.collections = {}
+        self.uses_ephemeral_collections = False
 
     def __attempt_add_to_linked_match( self, input_name, hdca, collection_type_description, subcollection_type ):
         structure = get_structure( hdca, collection_type_description, leaf_subcollection_type=subcollection_type )
@@ -69,12 +72,21 @@ def structure( self ):
         effective_structure = effective_structure.multiply( linked_structure )
         return None if effective_structure.is_leaf else effective_structure
 
+    @property
+    def implicit_inputs( self ):
+        if not self.uses_ephemeral_collections:
+            # Consider doing something smarter here.
+            return list( self.collections.items() )
+        else:
+            return []
+
     @staticmethod
     def for_collections( collections_to_match, collection_type_descriptions ):
         if not collections_to_match.has_collections():
             return None
 
         matching_collections = MatchingCollections()
+        matching_collections.uses_ephemeral_collections = collections_to_match.uses_ephemeral_collections
         for input_key, to_match in collections_to_match.items():
             hdca = to_match.hdca
             collection_type_description = collection_type_descriptions.for_collection_type( hdca.collection.collection_type )

lib/galaxy/dependencies/pinned-requirements.txt

@@ -72,3 +72,6 @@ pysam==0.8.4+gx5
 
 # Chronos client
 chronos-python==0.38.0
+
+# For CWL support.
+cwltool==1.0.20170727112954

lib/galaxy/jobs/__init__.py

@@ -782,6 +782,10 @@ def can_split( self ):
         # Should the job handler split this job up?
         return self.app.config.use_tasked_jobs and self.tool.parallelism
 
+    @property
+    def is_cwl_job( self ):
+        return self.tool.tool_type == "cwl"
+
     def get_job_runner_url( self ):
         log.warning('(%s) Job runner URLs are deprecated, use destinations instead.' % self.job_id)
         return self.job_destination.url
@@ -886,10 +890,13 @@ def get_special( ):
         # if the server was stopped and restarted before the job finished
         job.command_line = unicodify(self.command_line)
         job.dependencies = self.tool.dependencies
+        param_dict = tool_evaluator.param_dict
+        job.cwl_command_state = param_dict.get('__cwl_command_state', None)
+        job.cwl_command_state_version = param_dict.get('__cwl_command_state_version', None)
         self.sa_session.add( job )
         self.sa_session.flush()
         # Return list of all extra files
-        self.param_dict = tool_evaluator.param_dict
+        self.param_dict = param_dict
         version_string_cmd_raw = self.tool.version_string_cmd
         if version_string_cmd_raw:
             version_command_template = string.Template(version_string_cmd_raw)

lib/galaxy/jobs/command_factory.py

@@ -6,6 +6,7 @@
 )
 
 from galaxy import util
+from galaxy.util import bunch
 from galaxy.jobs.runners.util.job_script import (
     check_script_integrity,
     INTEGRITY_INJECTION,
@@ -175,8 +176,13 @@ def __handle_work_dir_outputs(commands_builder, job_wrapper, runner, remote_comm
     if 'working_directory' in remote_command_params:
         work_dir_outputs_kwds['job_working_directory'] = remote_command_params['working_directory']
     work_dir_outputs = runner.get_work_dir_outputs( job_wrapper, **work_dir_outputs_kwds )
-    if work_dir_outputs:
+    if work_dir_outputs or job_wrapper.is_cwl_job:
         commands_builder.capture_return_code()
+        if job_wrapper.is_cwl_job:
+            metadata_script_file = join(job_wrapper.working_directory, "relocate_dynamic_outputs.py")
+            relocate_contents = 'from galaxy_ext.cwl.handle_outputs import relocate_dynamic_outputs; relocate_dynamic_outputs()'
+            write_script(metadata_script_file, relocate_contents, bunch.Bunch(check_job_script_integrity=False))
+            commands_builder.append_command("python %s" % metadata_script_file)
         copy_commands = map(__copy_if_exists_command, work_dir_outputs)
         commands_builder.append_commands(copy_commands)
 

lib/galaxy/managers/collections.py

@@ -71,8 +71,10 @@ def create( self, trans, parent, name, collection_type, element_identifiers=None
                 name=name,
             )
             if implicit_collection_info:
-                for input_name, input_collection in implicit_collection_info[ "implicit_inputs" ]:
-                    dataset_collection_instance.add_implicit_input_collection( input_name, input_collection )
+                implicit_inputs = implicit_collection_info[ "implicit_inputs" ]
+                if implicit_inputs:
+                    for input_name, input_collection in implicit_inputs:
+                        dataset_collection_instance.add_implicit_input_collection( input_name, input_collection )
                 for output_dataset in implicit_collection_info.get( "outputs" ):
                     if output_dataset not in trans.sa_session:
                         output_dataset = trans.sa_session.query( type( output_dataset ) ).get( output_dataset.id )

lib/galaxy/managers/tools.py

@@ -62,6 +62,13 @@ def create_tool(self, tool_payload):
             if tool_id is None:
                 tool_id = str(uuid)
 
+            tool_version = representation.get("version", None)
+            tool_hash = build_tool_hash(representation)
+            value = representation
+        elif tool_format in ["CommandLineTool", "ExpressionTool"]:
+            # CWL tools
+            uuid = None
+            tool_id = representation.get("id", None)
             tool_version = representation.get("version", None)
             tool_hash = build_tool_hash(representation)
             value = representation

lib/galaxy/managers/workflows.py

@@ -22,6 +22,7 @@
 from galaxy.util.sanitize_html import sanitize_html
 from galaxy.workflow.steps import attach_ordered_steps
 from galaxy.workflow.modules import module_factory, is_tool_module_type, ToolModule, WorkflowModuleInjector
+from galaxy.tools.cwl import workflow_proxy
 from galaxy.tools.parameters.basic import DataToolParameter, DataCollectionToolParameter, RuntimeValue, workflow_building_modes
 from galaxy.tools.parameters import visit_input_values, params_to_incoming
 from galaxy.jobs.actions.post import ActionBox
@@ -196,6 +197,18 @@ def build_workflow_from_dict(
     ):
         # Put parameters in workflow mode
         trans.workflow_building_mode = True
+        if data and "src" in data and data["src"] == "from_path":
+            from galaxy.tools.cwl import workflow_proxy
+            wf_proxy = workflow_proxy(data["path"])
+            tool_reference_proxies = wf_proxy.tool_reference_proxies()
+            for tool_reference_proxy in tool_reference_proxies:
+                # TODO: Namespace IDS in workflows.
+                # TODO: Don't duplicately load these tools.
+                self.app.dynamic_tool_manager.create_tool({
+                    "representation": tool_reference_proxy.to_persistent_representation(),
+                })
+            data = wf_proxy.to_dict()
+
         # If there's a source, put it in the workflow name.
         if source:
             name = "%s (imported from %s)" % ( data['name'], source )
@@ -277,6 +290,10 @@ def update_workflow_from_dict(self, trans, stored_workflow, workflow_data):
     def _workflow_from_dict(self, trans, data, name, **kwds):
         if isinstance(data, string_types):
             data = json.loads(data)
+        if "src" in data:
+            assert data["src"] == "path"
+            wf_proxy = workflow_proxy(data["path"])
+            data = wf_proxy.to_dict()
 
         # Create new workflow from source data
         workflow = model.Workflow()
@@ -834,7 +851,7 @@ def __module_from_dict( self, trans, steps, steps_by_external_id, step_dict, **k
         """
         step = model.WorkflowStep()
         # TODO: Consider handling position inside module.
-        step.position = step_dict['position']
+        step.position = step_dict.get('position', {"left": 0, "top": 0})
         if step_dict.get("uuid", None) and step_dict['uuid'] != "None":
             step.uuid = step_dict["uuid"]
         if "label" in step_dict:
@@ -858,6 +875,15 @@ def __module_from_dict( self, trans, steps, steps_by_external_id, step_dict, **k
         # Stick this in the step temporarily
         step.temp_input_connections = step_dict['input_connections']
 
+        if "inputs" in step_dict:
+            for input_dict in step_dict["inputs"]:
+                step_input = model.WorkflowStepInput()
+                step_input.name = input_dict["name"]
+                step_input.merge_type = input_dict.get("merge_type", step_input.default_merge_type)
+                step_input.scatter_type = input_dict.get("scatter_type", step_input.default_scatter_type)
+                step_input.value_from = input_dict.get("value_from", None)
+                step.inputs.append(step_input)
+
         # Create the model class for the step
         steps.append( step )
         steps_by_external_id[ step_dict[ 'id' ] ] = step

lib/galaxy/model/__init__.py

@@ -767,6 +767,8 @@ def to_dict( self, view='collection', system_details=False ):
             # System level details that only admins should have.
             rval['external_id'] = self.job_runner_external_id
             rval['command_line'] = self.command_line
+            rval['cwl_command_state'] = self.cwl_command_state
+            rval['cwl_command_state_version'] = self.cwl_command_state_version
 
         if view == 'element':
             param_dict = dict( [ ( p.name, p.value ) for p in self.parameters ] )
@@ -3229,6 +3231,10 @@ def __init__(
         if not populated:
             self.populated_state = DatasetCollection.populated_states.NEW
 
+    @property
+    def allow_implicit_mapping(self):
+        return self.collection_type != "record"
+
     @property
     def populated( self ):
         top_level_populated = self.populated_state == DatasetCollection.populated_states.OK
@@ -3822,12 +3828,14 @@ def __init__( self ):
         self.tool_inputs = None
         self.tool_errors = None
         self.position = None
+        self.inputs = []
         self.input_connections = []
         self.config = None
         self.label = None
         self.uuid = uuid4()
         self.workflow_outputs = []
         self._input_connections_by_name = None
+        self._inputs_by_name = None
 
     @property
     def unique_workflow_outputs(self):
@@ -3863,6 +3871,12 @@ def input_connections_by_name(self):
             self.setup_input_connections_by_name()
         return self._input_connections_by_name
 
+    @property
+    def inputs_by_name(self):
+        if self._inputs_by_name is None:
+            self.setup_inputs_by_name()
+        return self._inputs_by_name
+
     def setup_input_connections_by_name(self):
         # Ensure input_connections has already been set.
 
@@ -3875,6 +3889,17 @@ def setup_input_connections_by_name(self):
             input_connections_by_name[input_name].append(conn)
         self._input_connections_by_name = input_connections_by_name
 
+    def setup_inputs_by_name(self):
+        # Ensure input_connections has already been set.
+
+        # Make connection information available on each step by input name.
+        inputs_by_name = {}
+        for step_input in self.inputs:
+            input_name = step_input.name
+            assert input_name not in inputs_by_name
+            inputs_by_name[input_name] = step_input
+        self._inputs_by_name = inputs_by_name
+
     def create_or_update_workflow_output(self, output_name, label, uuid):
         output = self.workflow_output_for(output_name)
         if output is None:
@@ -3929,6 +3954,19 @@ def log_str(self):
         return "WorkflowStep[index=%d,type=%s]" % (self.order_index, self.type)
 
 
+class WorkflowStepInput( object ):
+
+    default_merge_type = "merge_flattened"
+    default_scatter_type = "dotproduct"
+
+    def __init__( self ):
+        self.id = None
+        self.name = None
+        self.default_value = None
+        self.merge_type = self.default_merge_type
+        self.scatter_type = self.default_scatter_type
+
+
 class WorkflowStepConnection( object ):
     # Constant used in lieu of output_name and input_name to indicate an
     # implicit connection between two steps that is not dependent on a dataset

lib/galaxy/model/mapping.py

@@ -498,6 +498,8 @@
     Column( "object_store_id", TrimmedString( 255 ), index=True ),
     Column( "imported", Boolean, default=False, index=True ),
     Column( "params", TrimmedString( 255 ), index=True ),
+    Column( "cwl_command_state", JSONType, nullable=True ),
+    Column( "cwl_command_state_version", Integer, default=1 ),
     Column( "handler", TrimmedString( 255 ), index=True ) )
 
 model.JobStateHistory.table = Table(
@@ -839,6 +841,19 @@
     # Column( "input_connections", JSONType ),
     Column( "label", Unicode( 255 ) ) )
 
+
+model.WorkflowStepInput.table = Table(
+    "workflow_step_input", metadata,
+    Column( "id", Integer, primary_key=True ),
+    Column( "workflow_step_id", Integer, ForeignKey( "workflow_step.id" ), index=True ),
+    Column( "name", Unicode( 255 ) ),
+    Column( "merge_type", TEXT ),
+    Column( "scatter_type", TEXT ),
+    Column( "value_from", JSONType ),
+    Column( "value_from_type", TEXT ),
+    Column( "default_value", JSONType ) )
+
+
 model.WorkflowRequestStepState.table = Table(
     "workflow_request_step_states", metadata,
     Column( "id", Integer, primary_key=True ),
@@ -2275,7 +2290,12 @@ def simple_mapping( model, **kwds ):
         backref="workflow_steps" ),
     annotations=relation( model.WorkflowStepAnnotationAssociation,
         order_by=model.WorkflowStepAnnotationAssociation.table.c.id,
-        backref="workflow_steps" )
+        backref="workflow_steps"),
+) )
+
+mapper( model.WorkflowStepInput, model.WorkflowStepInput.table, properties=dict(
+    workflow_step=relation( model.WorkflowStep,
+        backref="inputs"),
 ) )
 
 mapper( model.WorkflowOutput, model.WorkflowOutput.table, properties=dict(
@@ -2352,8 +2372,6 @@ def simple_mapping( model, **kwds ):
     input_step_parameters=relation( model.WorkflowRequestInputStepParmeter ),
     input_datasets=relation( model.WorkflowRequestToInputDatasetAssociation ),
     input_dataset_collections=relation( model.WorkflowRequestToInputDatasetCollectionAssociation ),
-    #output_datasets=relation( model.WorkflowInvocationOutputDatasetAssociation ),
-    #output_dataset_collections=relation( model.WorkflowInvocationOutputDatasetCollectionAssociation ),
     subworkflow_invocations=relation( model.WorkflowInvocationToSubworkflowInvocationAssociation,
         primaryjoin=( ( model.WorkflowInvocationToSubworkflowInvocationAssociation.table.c.workflow_invocation_id == model.WorkflowInvocation.table.c.id ) ),
         backref=backref("parent_workflow_invocation", uselist=False),