Merge latest galaxy-central-lwr changes.

job_conf.xml.sample_advanced

@@ -7,7 +7,11 @@
         <plugin id="local" type="runner" load="galaxy.jobs.runners.local:LocalJobRunner"/>
         <plugin id="pbs" type="runner" load="galaxy.jobs.runners.pbs:PBSJobRunner" workers="2"/>
         <plugin id="drmaa" type="runner" load="galaxy.jobs.runners.drmaa:DRMAAJobRunner"/>
-        <plugin id="lwr" type="runner" load="galaxy.jobs.runners.lwr:LwrJobRunner" /> <!-- https://lwr.readthedocs.org -->
+        <plugin id="lwr" type="runner" load="galaxy.jobs.runners.lwr:LwrJobRunner">
+          <!-- More information on LWR can be found at https://lwr.readthedocs.org -->
+          <!-- Uncomment following line to use libcurl to perform HTTP calls (defaults to urllib) -->
+          <!-- <param id="transport">curl</param> -->
+        </plugin>
         <plugin id="cli" type="runner" load="galaxy.jobs.runners.cli:ShellJobRunner" />
         <plugin id="condor" type="runner" load="galaxy.jobs.runners.condor:CondorJobRunner" />
     </plugins>
@@ -46,6 +50,10 @@
             <param id="url">https://windowshost.examle.com:8913/</param>
             <!-- If set, private_token must match token remote LWR server configured with. -->
             <param id="private_token">123456789changeme</param>
+            <!-- Uncomment the following statement to disable file staging (e.g.
+                 if there is a shared file system between Galaxy and the LWR 
+                 server). -->
+            <!-- <param id="default_file_action">none</param> -->
         </destination>
         <destination id="ssh_torque" runner="cli">
             <param id="shell_plugin">SecureShell</param>

lib/galaxy/jobs/runners/lwr.py

@@ -3,12 +3,15 @@ import logging
 from galaxy import model
 from galaxy.jobs.runners import AsynchronousJobState, AsynchronousJobRunner
 from galaxy.jobs import JobDestination
+from galaxy.util import string_as_bool_or_none
 
 import errno
 from time import sleep
 import os
 
-from lwr_client import FileStager, Client, url_to_destination_params
+from .lwr_client import ClientManager, url_to_destination_params
+from .lwr_client import finish_job as lwr_finish_job
+from .lwr_client import submit_job as lwr_submit_job
 
 log = logging.getLogger( __name__ )
 
@@ -21,12 +24,13 @@ class LwrJobRunner( AsynchronousJobRunner ):
     """
     runner_name = "LWRRunner"
 
-    def __init__( self, app, nworkers, transport=None ):
+    def __init__( self, app, nworkers, transport=None, cache=None ):
         """Start the job runner """
         super( LwrJobRunner, self ).__init__( app, nworkers )
         self._init_monitor_thread()
         self._init_worker_threads()
-        self.transport_type = transport
+        client_manager_kwargs = {'transport_type': transport, 'cache': string_as_bool_or_none(cache)}
+        self.client_manager = ClientManager(**client_manager_kwargs)
 
     def url_to_destination( self, url ):
         """Convert a legacy URL to a job destination"""
@@ -78,10 +82,9 @@ class LwrJobRunner( AsynchronousJobRunner ):
             input_files = job_wrapper.get_input_fnames()
             working_directory = job_wrapper.working_directory
             tool = job_wrapper.tool
-            file_stager = FileStager(client, tool, command_line, job_wrapper.extra_filenames, input_files, output_files, working_directory)
-            rebuilt_command_line = file_stager.get_rewritten_command_line()
-            job_id = file_stager.job_id
-            client.launch( rebuilt_command_line )
+            config_files = job_wrapper.extra_filenames
+            job_id = lwr_submit_job(client, tool, command_line, config_files, input_files, output_files, working_directory)
+            log.info("lwr job submitted with job_id %s" % job_id)
             job_wrapper.set_job_destination( job_destination, job_id )
             job_wrapper.change_state( model.Job.states.QUEUED )
         except:
@@ -113,42 +116,35 @@ class LwrJobRunner( AsynchronousJobRunner ):
         return self.get_client( job_destination_params, job_id )
 
     def get_client( self, job_destination_params, job_id ):
-        return Client( job_destination_params, job_id, transport_type=self.transport_type )
+        return self.client_manager.get_client( job_destination_params, job_id )
 
     def finish_job( self, job_state ):
-        stderr = stdout = command_line = ''
+        stderr = stdout = ''
         job_wrapper = job_state.job_wrapper
         try:
             client = self.get_client_from_state(job_state)
 
             run_results = client.raw_check_complete()
-            stdout = run_results['stdout']
-            stderr = run_results['stderr']
+            stdout = run_results.get('stdout', '')
+            stderr = run_results.get('stderr', '')
 
-            download_failure_exceptions = []
-            if job_wrapper.get_state() not in [ model.Job.states.ERROR, model.Job.states.DELETED ]:
-                work_dir_outputs = self.get_work_dir_outputs(job_wrapper)
-                output_files = self.get_output_files(job_wrapper)
-                for source_file, output_file in work_dir_outputs:
-                    try:
-                        client.download_work_dir_output(source_file, job_wrapper.working_directory, output_file)
-                    except Exception, e:
-                        download_failure_exceptions.append(e)
-                    # Remove from full output_files list so don't try to download directly.
-                    output_files.remove(output_file)
-                for output_file in output_files:
-                    try:
-                        client.download_output(output_file, working_directory=job_wrapper.working_directory)
-                    except Exception, e:
-                        download_failure_exceptions.append(e)
-            if download_failure_exceptions or self.app.config.cleanup_job == "always":
-                try:
-                    client.clean()
-                except:
-                    log.warn("Failed to cleanup remote LWR job")
-            if download_failure_exceptions:
+            # Use LWR client code to transfer/copy files back
+            # and cleanup job if needed.
+            completed_normally = \
+                job_wrapper.get_state() not in [ model.Job.states.ERROR, model.Job.states.DELETED ]
+            cleanup_job = self.app.config.cleanup_job
+            work_dir_outputs = self.get_work_dir_outputs( job_wrapper )
+            output_files = self.get_output_files( job_wrapper )
+            finish_args = dict( client=client,
+                                working_directory=job_wrapper.working_directory,
+                                job_completed_normally=completed_normally,
+                                cleanup_job=cleanup_job,
+                                work_dir_outputs=work_dir_outputs,
+                                output_files=output_files )
+            failed = lwr_finish_job( **finish_args )
+
+            if failed:
                 job_wrapper.fail("Failed to find or download one or more job outputs from remote server.", exception=True)
-            log.debug('execution finished: %s' % command_line)
         except:
             message = "Failed to communicate with remote job server."
             job_wrapper.fail( message, exception=True )

lib/galaxy/jobs/runners/lwr_client/__init__.py

@@ -6,8 +6,9 @@ This module contains logic for interfacing with an external LWR server.
 
 """
 
-from .stager import FileStager
-from .client import Client
+from .stager import submit_job, finish_job
+from .client import OutputNotFoundException
+from .manager import ClientManager
 from .destination import url_to_destination_params
 
-__all__ = [Client, FileStager, url_to_destination_params]
+__all__ = [ClientManager, OutputNotFoundException, url_to_destination_params, finish_job, submit_job]

lib/galaxy/jobs/runners/lwr_client/action_mapper.py

@@ -0,0 +1,122 @@
+from simplejson import load
+from os.path import abspath
+from fnmatch import fnmatch
+from re import compile
+
+
+DEFAULT_MAPPED_ACTION = 'transfer'  # Not really clear to me what this should be, exception?
+DEFAULT_PATH_MAPPER_TYPE = 'prefix'
+
+
+class FileActionMapper(object):
+    """
+    Objects of this class define how paths are mapped to actions.
+
+    >>> json_string = r'''{"paths": [ \
+      {"path": "/opt/galaxy", "action": "none"}, \
+      {"path": "/galaxy/data", "action": "transfer"}, \
+      {"path": "/cool/bamfiles/**/*.bam", "action": "copy", "type": "glob"}, \
+      {"path": ".*/dataset_\\\\d+.dat", "action": "copy", "type": "regex"} \
+    ]}'''
+    >>> from tempfile import NamedTemporaryFile
+    >>> from os import unlink
+    >>> f = NamedTemporaryFile(delete=False)
+    >>> f.write(json_string)
+    >>> f.close()
+    >>> class MockClient():
+    ...     default_file_action = 'none'
+    ...     action_config_path = f.name
+    ...
+    >>> mapper = FileActionMapper(MockClient())
+    >>> unlink(f.name)
+    >>> # Test first config line above, implicit path prefix mapper
+    >>> mapper.action('/opt/galaxy/tools/filters/catWrapper.py', 'input')
+    ('none',)
+    >>> # Test another (2nd) mapper, this one with a different action
+    >>> mapper.action('/galaxy/data/files/000/dataset_1.dat', 'input')
+    ('transfer',)
+    >>> # Always at least copy work_dir outputs.
+    >>> mapper.action('/opt/galaxy/database/working_directory/45.sh', 'work_dir')
+    ('copy',)
+    >>> # Test glob mapper (matching test)
+    >>> mapper.action('/cool/bamfiles/projectABC/study1/patient3.bam', 'input')
+    ('copy',)
+    >>> # Test glob mapper (non-matching test)
+    >>> mapper.action('/cool/bamfiles/projectABC/study1/patient3.bam.bai', 'input')
+    ('none',)
+    >>> # Regex mapper test.
+    >>> mapper.action('/old/galaxy/data/dataset_10245.dat', 'input')
+    ('copy',)
+    """
+
+    def __init__(self, client):
+        self.default_action = client.default_file_action
+        action_config_path = client.action_config_path
+        self.mappers = []
+        if action_config_path:
+            self.__load_action_config(action_config_path)
+
+    def __load_action_config(self, path):
+        config = load(open(path, 'rb'))
+        for path_config in config.get('paths', []):
+            map_type = path_config.get('type', DEFAULT_PATH_MAPPER_TYPE)
+            self.mappers.append(mappers[map_type](path_config))
+
+    def action(self, path, type):
+        action = self.default_action
+        normalized_path = abspath(path)
+        for mapper in self.mappers:
+            if mapper.matches(normalized_path):
+                action = mapper.action
+                break
+        if type in ["work_dir", "output_task"] and action == "none":
+            ## We are changing the working_directory relative to what
+            ## Galaxy would use, these need to be copied over.
+            action = "copy"
+        return (action,)
+
+
+class BasePathMapper(object):
+
+    def __init__(self, config):
+        self.action = config.get('action', DEFAULT_MAPPED_ACTION)
+
+
+class PrefixPathMapper(BasePathMapper):
+
+    def __init__(self, config):
+        super(PrefixPathMapper, self).__init__(config)
+        self.prefix_path = abspath(config['path'])
+
+    def matches(self, path):
+        return path.startswith(self.prefix_path)
+
+
+class GlobPathMapper(BasePathMapper):
+
+    def __init__(self, config):
+        super(GlobPathMapper, self).__init__(config)
+        self.glob_path = config['path']
+
+    def matches(self, path):
+        return fnmatch(path, self.glob_path)
+
+
+class RegexPathMapper(BasePathMapper):
+
+    def __init__(self, config):
+        super(RegexPathMapper, self).__init__(config)
+        self.pattern = compile(config['path'])
+
+    def matches(self, path):
+        return self.pattern.match(path) is not None
+
+
+mappers = {
+    'prefix': PrefixPathMapper,
+    'glob': GlobPathMapper,
+    'regex': RegexPathMapper,
+}
+
+
+__all__ = [FileActionMapper]

lib/galaxy/jobs/runners/lwr_client/client.py

@@ -1,17 +1,18 @@
 import os
-import time
-import urllib
+import shutil
 import simplejson
+from simplejson import dumps
+from time import sleep
 
-from .transport import get_transport
-from .destination import url_to_destination_params
+from .destination import submit_params
+
+CACHE_WAIT_SECONDS = 3
+MAX_RETRY_COUNT = 5
+RETRY_SLEEP_TIME = 0.1
 
 
 class parseJson(object):
 
-    def __init__(self):
-        pass
-
     def __call__(self, func):
         def replacement(*args, **kwargs):
             response = func(*args, **kwargs)
@@ -19,6 +20,27 @@ class parseJson(object):
         return replacement
 
 
+class retry(object):
+
+    def __call__(self, func):
+
+        def replacement(*args, **kwargs):
+            max_count = MAX_RETRY_COUNT
+            count = 0
+            while True:
+                count += 1
+                try:
+                    return func(*args, **kwargs)
+                except:
+                    if count >= max_count:
+                        raise
+                    else:
+                        sleep(RETRY_SLEEP_TIME)
+                        continue
+
+        return replacement
+
+
 class OutputNotFoundException(Exception):
 
     def __init__(self, path):
@@ -40,103 +62,106 @@ class Client(object):
         Galaxy job/task id.
     """
 
-    def __init__(self, destination_params, job_id, transport_type=None):
-        if isinstance(destination_params, str) or isinstance(destination_params, unicode):
-            destination_params = url_to_destination_params(destination_params)
-        self.remote_host = destination_params.get("url")
-        assert self.remote_host != None, "Failed to determine url for LWR client."
-        self.private_key = destination_params.get("private_token", None)
+    def __init__(self, destination_params, job_id, job_manager_interface):
+        self.job_manager_interface = job_manager_interface
+        self.destination_params = destination_params
         self.job_id = job_id
-        self.transport = get_transport(transport_type)
 
-    def __build_url(self, command, args):
-        if self.private_key:
-            args["private_key"] = self.private_key
-        data = urllib.urlencode(args)
-        url = self.remote_host + command + "?" + data
-        return url
+        self.default_file_action = self.destination_params.get("default_file_action", "transfer")
+        self.action_config_path = self.destination_params.get("file_action_config", None)
 
-    def __raw_execute(self, command, args={}, data=None, input_path=None, output_path=None):
-        url = self.__build_url(command, args)
-        response = self.transport.execute(url, data=data, input_path=input_path, output_path=output_path)
-        return response
+    def _raw_execute(self, command, args={}, data=None, input_path=None, output_path=None):
+        return self.job_manager_interface.execute(command, args, data, input_path, output_path)
+
+    @property
+    def _submit_params(self):
+        return submit_params(self.destination_params)
 
     @parseJson()
-    def __upload_file(self, action, path, name=None, contents=None):
+    def input_path(self, path, input_type, name=None):
+        args = {"job_id": self.job_id, "name": name, "input_type": input_type}
+        return self._raw_execute('input_path', args)
+
+    def put_file(self, path, input_type, name=None, contents=None, action='transfer'):
         if not name:
             name = os.path.basename(path)
-        args = {"job_id": self.job_id, "name": name}
+        args = {"job_id": self.job_id, "name": name, "input_type": input_type}
         input_path = path
         if contents:
             input_path = None
-        return self.__raw_execute(action, args, contents, input_path)
+        if action == 'transfer':
+            return self._upload_file(args, contents, input_path)
+        elif action == 'copy':
+            lwr_path = self._raw_execute('input_path', args)
+            self._copy(path, lwr_path)
+            return {'path': lwr_path}
 
-    def upload_tool_file(self, path):
-        """
-        Upload a tool related file (e.g. wrapper) required to run job.
+    @parseJson()
+    def _upload_file(self, args, contents, input_path):
+        return self._raw_execute(self._upload_file_action(args), args, contents, input_path)
 
-        **Parameters**
-
-        path : str
-            Local path tool.
-        """
-        return self.__upload_file("upload_tool_file", path)
-
-    def upload_input(self, path):
-        """
-        Upload input dataset to remote server.
-
-        **Parameters**
-
-        path : str
-            Local path of input dataset.
-        """
-        return self.__upload_file("upload_input", path)
-
-    def upload_extra_input(self, path, relative_name):
-        """
-        Upload extra input file to remote server.
-
-        **Parameters**
-
-        path : str
-            Extra files path of input dataset corresponding to this input.
-        relative_name : str
-            Relative path of extra file to upload relative to inputs extra files path.
-        """
-        return self.__upload_file("upload_extra_input", path, name=relative_name)
-
-    def upload_config_file(self, path, contents):
-        """
-        Upload a job's config file to the remote server.
-
-        **Parameters**
-
-        path : str
-            Local path to the original config file.
-        contents : str
-            Rewritten contents of the config file to upload.
-        """
-        return self.__upload_file("upload_config_file", path, contents=contents)
-
-    def upload_working_directory_file(self, path):
-        """
-        Upload the supplied file (path) from a job's working directory
-        to remote server.
-
-        **Parameters**
-
-        path : str
-            Path to file to upload.
-        """
-        return self.__upload_file("upload_working_directory_file", path)
+    def _upload_file_action(self, args):
+        ## Hack for backward compatibility, instead of using new upload_file
+        ## path. Use old paths.
+        input_type = args['input_type']
+        action = {
+            'input': 'upload_input',
+            'input_extra': 'upload_extra_input',
+            'config': 'upload_config_file',
+            'work_dir': 'upload_working_directory_file',
+            'tool': 'upload_tool_file'
+        }[input_type]
+        del args['input_type']
+        return action
 
     @parseJson()
     def _get_output_type(self, name):
-        return self.__raw_execute("get_output_type", {"name": name,
-                                                      "job_id": self.job_id})
+        return self._raw_execute("get_output_type", {"name": name,
+                                                     "job_id": self.job_id})
 
-    def download_work_dir_output(self, source, working_directory, output_path):
+    def fetch_output(self, path, working_directory, action='transfer'):
+        """
+        Download an output dataset from the remote server.
+
+        **Parameters**
+
+        path : str
+            Local path of the dataset.
+        working_directory : str
+            Local working_directory for the job.
+        """
+        name = os.path.basename(path)
+        output_type = self._get_output_type(name)
+
+        if output_type == "none":
+            # Just make sure the file was created.
+            if not os.path.exists(path):
+                raise OutputNotFoundException(path)
+            return
+
+        output_path = self.__output_path(path, name, working_directory, output_type)
+        self.__populate_output_path(name, output_path, output_type, action)
+
+    def __populate_output_path(self, name, output_path, output_type, action):
+        if action == 'transfer':
+            self.__raw_download_output(name, self.job_id, output_type, output_path)
+        elif action == 'copy':
+            lwr_path = self._output_path(name, self.job_id, output_type)['path']
+            self._copy(lwr_path, output_path)
+
+    def __output_path(self, path, name, working_directory, output_type):
+        """
+        Preconditions: output_type is not 'none'.
+        """
+        if output_type == "direct":
+            output_path = path
+        elif output_type == "task":
+            output_path = os.path.join(working_directory, name)
+        else:
+            raise Exception("Unknown output_type returned from LWR server %s" % output_type)
+        return output_path
+
+    def fetch_work_dir_output(self, source, working_directory, output_path, action='transfer'):
         """
         Download an output dataset specified with from_work_dir from the
         remote server.
@@ -152,35 +177,27 @@ class Client(object):
         """
         output = open(output_path, "wb")
         name = os.path.basename(source)
-        self.__raw_download_output(name, self.job_id, "work_dir", output)
+        if action == 'transfer':
+            self.__raw_download_output(name, self.job_id, "work_dir", output)
+        elif action == 'copy':
+            lwr_path = self._output_path(name, self.job_id, 'work_dir')['path']
+            self._copy(lwr_path, output_path)
 
-    def download_output(self, path, working_directory):
-        """
-        Download an output dataset from the remote server.
-
-        **Parameters**
-
-        path : str
-            Local path of the dataset.
-        working_directory : str
-            Local working_directory for the job.
-        """
-        name = os.path.basename(path)
-        output_type = self._get_output_type(name)
-        if output_type == "direct":
-            output_path = path
-        elif output_type == "task":
-            output_path = os.path.join(working_directory, name)
-        else:
-            raise OutputNotFoundException(path)
-        self.__raw_download_output(name, self.job_id, output_type, output_path)
+    @parseJson()
+    def _output_path(self, name, job_id, output_type):
+        return self._raw_execute("output_path",
+                                 {"name": name,
+                                  "job_id": self.job_id,
+                                  "output_type": output_type})
 
+    @retry()
     def __raw_download_output(self, name, job_id, output_type, output_path):
-        self.__raw_execute("download_output",
-                           {"name": name,
-                            "job_id": self.job_id,
-                            "output_type": output_type},
-                           output_path=output_path)
+        output_params = {
+            "name": name,
+            "job_id": self.job_id,
+            "output_type": output_type
+        }
+        self._raw_execute("download_output", output_params, output_path=output_path)
 
     def launch(self, command_line):
         """
@@ -192,14 +209,17 @@ class Client(object):
         command_line : str
             Command to execute.
         """
-        return self.__raw_execute("launch", {"command_line": command_line,
-                                             "job_id": self.job_id})
+        launch_params = dict(command_line=command_line, job_id=self.job_id)
+        submit_params = self._submit_params
+        if submit_params:
+            launch_params['params'] = dumps(submit_params)
+        return self._raw_execute("launch", launch_params)
 
     def kill(self):
         """
         Cancel remote job, either removing from the queue or killing it.
         """
-        return self.__raw_execute("kill", {"job_id": self.job_id})
+        return self._raw_execute("kill", {"job_id": self.job_id})
 
     def wait(self):
         """
@@ -209,24 +229,25 @@ class Client(object):
             complete_response = self.raw_check_complete()
             if complete_response["complete"] == "true":
                 return complete_response
-            time.sleep(1)
+            sleep(1)
 
     @parseJson()
     def raw_check_complete(self):
         """
         Get check_complete response from the remote server.
         """
-        check_complete_response = self.__raw_execute("check_complete", {"job_id": self.job_id})
+        check_complete_response = self._raw_execute("check_complete", {"job_id": self.job_id})
         return check_complete_response
 
     def check_complete(self, response=None):
         """
         Return boolean indicating whether the job is complete.
         """
-        if response == None:
+        if response is None:
             response = self.raw_check_complete()
         return response["complete"] == "true"
 
+    @retry()
     def get_status(self):
         check_complete_response = self.raw_check_complete()
         # Older LWR instances won't set status so use 'complete', at some
@@ -243,7 +264,7 @@ class Client(object):
         """
         Cleanup the remote job.
         """
-        self.__raw_execute("clean", {"job_id": self.job_id})
+        self._raw_execute("clean", {"job_id": self.job_id})
 
     @parseJson()
     def setup(self, tool_id=None, tool_version=None):
@@ -255,4 +276,53 @@ class Client(object):
             setup_args["tool_id"] = tool_id
         if tool_version:
             setup_args["tool_version"] = tool_version
-        return self.__raw_execute("setup", setup_args)
+        return self._raw_execute("setup", setup_args)
+
+    def _copy(self, source, destination):
+        source = os.path.abspath(source)
+        destination = os.path.abspath(destination)
+        if source != destination:
+            shutil.copyfile(source, destination)
+
+
+class InputCachingClient(Client):
+    """
+    Beta client that cache's staged files to prevent duplication.
+    """
+
+    def __init__(self, destination_params, job_id, job_manager_interface, client_cacher):
+        super(InputCachingClient, self).__init__(destination_params, job_id, job_manager_interface)
+        self.client_cacher = client_cacher
+
+    @parseJson()
+    def _upload_file(self, args, contents, input_path):
+        action = self._upload_file_action(args)
+        if contents:
+            input_path = None
+            return self._raw_execute(action, args, contents, input_path)
+        else:
+            event_holder = self.client_cacher.acquire_event(input_path)
+            cache_required = self.cache_required(input_path)
+            if cache_required:
+                self.client_cacher.queue_transfer(self, input_path)
+            while not event_holder.failed:
+                available = self.file_available(input_path)
+                if available['ready']:
+                    token = available['token']
+                    args["cache_token"] = token
+                    return self._raw_execute(action, args)
+                event_holder.event.wait(30)
+            if event_holder.failed:
+                raise Exception("Failed to transfer file %s" % input_path)
+
+    @parseJson()
+    def cache_required(self, path):
+        return self._raw_execute("cache_required", {"path": path})
+
+    @parseJson()
+    def cache_insert(self, path):
+        return self._raw_execute("cache_insert", {"path": path}, None, path)
+
+    @parseJson()
+    def file_available(self, path):
+        return self._raw_execute("file_available", {"path": path})

lib/galaxy/jobs/runners/lwr_client/destination.py

@@ -1,6 +1,8 @@
 
 from re import match
 
+SUBMIT_PREFIX = "submit_"
+
 
 def url_to_destination_params(url):
     """Convert a legacy runner URL to a job destination
@@ -42,3 +44,16 @@ def url_to_destination_params(url):
                         "private_token": private_token}
 
     return destination_args
+
+
+def submit_params(destination_params):
+    """
+
+    >>> destination_params = {"private_token": "12345", "submit_native_specification": "-q batch"}
+    >>> result = submit_params(destination_params)
+    >>> result.items()
+    [('native_specification', '-q batch')]
+    """
+    return dict([(key[len(SUBMIT_PREFIX):], value)
+                 for key, value in (destination_params or {}).iteritems()
+                 if key.startswith(SUBMIT_PREFIX)])

lib/galaxy/jobs/runners/lwr_client/manager.py

@@ -0,0 +1,191 @@
+from abc import ABCMeta, abstractmethod
+try:
+    from Queue import Queue
+except ImportError:
+    from queue import Queue
+from threading import Thread
+from os import getenv
+from urllib import urlencode
+from StringIO import StringIO
+
+from .client import Client, InputCachingClient
+from .transport import get_transport
+from .util import TransferEventManager
+from .destination import url_to_destination_params
+
+
+from logging import getLogger
+log = getLogger(__name__)
+
+DEFAULT_TRANSFER_THREADS = 2
+
+
+class ClientManager(object):
+    """
+    Factory to create LWR clients, used to manage potential shared
+    state between multiple client connections.
+    """
+    def __init__(self, **kwds):
+        if 'job_manager' in kwds:
+            self.job_manager_interface_class = LocalJobManagerInterface
+            self.job_manager_interface_args = dict(job_manager=kwds['job_manager'], file_cache=kwds['file_cache'])
+        else:
+            self.job_manager_interface_class = HttpJobManagerInterface
+            transport_type = kwds.get('transport_type', None)
+            transport = get_transport(transport_type)
+            self.job_manager_interface_args = dict(transport=transport)
+        cache = kwds.get('cache', None)
+        if cache is None:
+            cache = _environ_default_int('LWR_CACHE_TRANSFERS')
+        if cache:
+            log.info("Setting LWR client class to caching variant.")
+            self.client_cacher = ClientCacher(**kwds)
+            self.client_class = InputCachingClient
+            self.extra_client_kwds = {"client_cacher": self.client_cacher}
+        else:
+            log.info("Setting LWR client class to standard, non-caching variant.")
+            self.client_class = Client
+            self.extra_client_kwds = {}
+
+    def get_client(self, destination_params, job_id):
+        destination_params = self.__parse_destination_params(destination_params)
+        job_manager_interface_class = self.job_manager_interface_class
+        job_manager_interface_args = dict(destination_params=destination_params, **self.job_manager_interface_args)
+        job_manager_interface = job_manager_interface_class(**job_manager_interface_args)
+        return self.client_class(destination_params, job_id, job_manager_interface, **self.extra_client_kwds)
+
+    def __parse_destination_params(self, destination_params):
+        if isinstance(destination_params, str) or isinstance(destination_params, unicode):
+            destination_params = url_to_destination_params(destination_params)
+        return destination_params
+
+
+class JobManagerInteface(object):
+    """
+    Abstract base class describes how client communicates with remote job
+    manager.
+    """
+    __metaclass__ = ABCMeta
+
+    @abstractmethod
+    def execute(self, command, args={}, data=None, input_path=None, output_path=None):
+        """
+        Execute the correspond command against configured LWR job manager. Arguments are
+        method parameters and data or input_path describe essentially POST bodies. If command
+        results in a file, resulting path should be specified as output_path.
+        """
+
+
+class HttpJobManagerInterface(object):
+
+    def __init__(self, destination_params, transport):
+        self.transport = transport
+        self.remote_host = destination_params.get("url")
+        assert self.remote_host is not None, "Failed to determine url for LWR client."
+        self.private_key = destination_params.get("private_token", None)
+
+    def execute(self, command, args={}, data=None, input_path=None, output_path=None):
+        url = self.__build_url(command, args)
+        response = self.transport.execute(url, data=data, input_path=input_path, output_path=output_path)
+        return response
+
+    def __build_url(self, command, args):
+        if self.private_key:
+            args["private_key"] = self.private_key
+        data = urlencode(args)
+        url = self.remote_host + command + "?" + data
+        return url
+
+
+class LocalJobManagerInterface(object):
+
+    def __init__(self, destination_params, job_manager, file_cache):
+        self.job_manager = job_manager
+        self.file_cache = file_cache
+
+    def __app_args(self):
+        ## Arguments that would be specified from LwrApp if running
+        ## in web server.
+        return {
+            'manager': self.job_manager,
+            'file_cache': self.file_cache,
+            'ip': None
+        }
+
+    def execute(self, command, args={}, data=None, input_path=None, output_path=None):
+        from lwr import routes
+        from lwr.framework import build_func_args
+        controller = getattr(routes, command)
+        action = controller.func
+        body_args = dict(body=self.__build_body(data, input_path))
+        args = build_func_args(action, args.copy(), self.__app_args(), body_args)
+        result = action(**args)
+        if controller.response_type != 'file':
+            return controller.body(result)
+        else:
+            from lwr.util import copy_to_path
+            with open(result, 'rb') as result_file:
+                copy_to_path(result_file, output_path)
+
+    def __build_body(self, data, input_path):
+        if data is not None:
+            return StringIO(data)
+        elif input_path is not None:
+            return open(input_path, 'r')
+        else:
+            return None
+
+
+class ClientCacher(object):
+
+    def __init__(self, **kwds):
+        self.event_manager = TransferEventManager()
+        default_transfer_threads = _environ_default_int('LWR_CACHE_THREADS', DEFAULT_TRANSFER_THREADS)
+        num_transfer_threads = int(kwds.get('transfer_threads', default_transfer_threads))
+        self.__init_transfer_threads(num_transfer_threads)
+
+    def queue_transfer(self, client, path):
+        self.transfer_queue.put((client, path))
+
+    def acquire_event(self, input_path):
+        return self.event_manager.acquire_event(input_path)
+
+    def _transfer_worker(self):
+        while True:
+            transfer_info = self.transfer_queue.get()
+            try:
+                self.__perform_transfer(transfer_info)
+            except BaseException as e:
+                log.warn("Transfer failed.")
+                log.exception(e)
+                pass
+            self.transfer_queue.task_done()
+
+    def __perform_transfer(self, transfer_info):
+        (client, path) = transfer_info
+        event_holder = self.event_manager.acquire_event(path, force_clear=True)
+        failed = True
+        try:
+            client.cache_insert(path)
+            failed = False
+        finally:
+            event_holder.failed = failed
+            event_holder.release()
+
+    def __init_transfer_threads(self, num_transfer_threads):
+        self.num_transfer_threads = num_transfer_threads
+        self.transfer_queue = Queue()
+        for i in range(num_transfer_threads):
+            t = Thread(target=self._transfer_worker)
+            t.daemon = True
+            t.start()
+
+
+def _environ_default_int(variable, default="0"):
+    val = getenv(variable, default)
+    int_val = int(default)
+    if str(val).isdigit():
+        int_val = int(val)
+    return int_val
+
+__all__ = [ClientManager, HttpJobManagerInterface]

lib/galaxy/jobs/runners/lwr_client/stager.py

@@ -1,7 +1,12 @@
-
-import os
+from os.path import abspath, basename, join, exists
+from os import listdir, sep
 from re import findall
 
+from .action_mapper import FileActionMapper
+
+from logging import getLogger
+log = getLogger(__name__)
+
 
 class JobInputs(object):
     """
@@ -61,7 +66,7 @@ class JobInputs(object):
             Full path to directory to search.
 
         """
-        pattern = r"(%s%s\S+)" % (directory, os.sep)
+        pattern = r"(%s%s\S+)" % (directory, sep)
         referenced_files = set()
         for input_contents in self.__items():
             referenced_files.update(findall(pattern, input_contents))
@@ -96,6 +101,52 @@ class JobInputs(object):
         return items
 
 
+class TransferTracker(object):
+
+    def __init__(self, client, job_inputs):
+        self.client = client
+        self.action_mapper = FileActionMapper(client)
+        self.job_inputs = job_inputs
+        self.file_renames = {}
+
+    def handle_transfer(self, path, type, name=None, contents=None):
+        if contents:
+            # If contents loaded in memory, no need to write out file and copy,
+            # just transfer.
+            action = ('transfer', )
+        else:
+            if not exists(path):
+                message = "handle_tranfer called on non-existent file - [%s]" % path
+                log.warn(message)
+                raise Exception(message)
+            action = self.__action(path, type)
+
+        if action[0] in ['transfer', 'copy']:
+            response = self.client.put_file(path, type, name=name, contents=contents)
+            self.register_rewrite(path, response['path'], type, force=True)
+        elif action[0] == 'none':
+            # No action for this file.
+            pass
+        else:
+            raise Exception("Unknown action type (%s) encountered for path (%s)" % (action[0], path))
+
+    def register_rewrite(self, local_path, remote_path, type, force=False):
+        action = self.__action(local_path, type)
+        if action[0] in ['transfer', 'copy'] or force:
+            self.file_renames[local_path] = remote_path
+
+    def rewrite_input_paths(self):
+        """
+        For each file that has been transferred and renamed, updated
+        command_line and configfiles to reflect that rewrite.
+        """
+        for local_path, remote_path in self.file_renames.iteritems():
+            self.job_inputs.rewrite_paths(local_path, remote_path)
+
+    def __action(self, path, type):
+        return self.action_mapper.action(path, type)
+
+
 class FileStager(object):
     """
     Objects of the FileStager class interact with an LWR client object to
@@ -130,14 +181,14 @@ class FileStager(object):
         self.output_files = output_files
         self.tool_id = tool.id
         self.tool_version = tool.version
-        self.tool_dir = os.path.abspath(tool.tool_dir)
+        self.tool_dir = abspath(tool.tool_dir)
         self.working_directory = working_directory
 
         # Setup job inputs, these will need to be rewritten before
         # shipping off to remote LWR server.
         self.job_inputs = JobInputs(self.command_line, self.config_files)
 
-        self.file_renames = {}
+        self.transfer_tracker = TransferTracker(client, self.job_inputs)
 
         self.__handle_setup()
         self.__initialize_referenced_tool_files()
@@ -173,8 +224,7 @@ class FileStager(object):
 
     def __upload_tool_files(self):
         for referenced_tool_file in self.referenced_tool_files:
-            tool_upload_response = self.client.upload_tool_file(referenced_tool_file)
-            self.file_renames[referenced_tool_file] = tool_upload_response['path']
+            self.transfer_tracker.handle_transfer(referenced_tool_file, 'tool')
 
     def __upload_input_files(self):
         for input_file in self.input_files:
@@ -183,61 +233,58 @@ class FileStager(object):
 
     def __upload_input_file(self, input_file):
         if self.job_inputs.path_referenced(input_file):
-            input_upload_response = self.client.upload_input(input_file)
-            self.file_renames[input_file] = input_upload_response['path']
+            if exists(input_file):
+                self.transfer_tracker.handle_transfer(input_file, 'input')
+            else:
+                message = "LWR: __upload_input_file called on empty or missing dataset." + \
+                          " So such file: [%s]" % input_file
+                log.debug(message)
 
     def __upload_input_extra_files(self, input_file):
         # TODO: Determine if this is object store safe and what needs to be
         # done if it is not.
         files_path = "%s_files" % input_file[0:-len(".dat")]
-        if os.path.exists(files_path) and self.job_inputs.path_referenced(files_path):
-            for extra_file in os.listdir(files_path):
-                extra_file_path = os.path.join(files_path, extra_file)
-                relative_path = os.path.basename(files_path)
-                extra_file_relative_path = os.path.join(relative_path, extra_file)
-                response = self.client.upload_extra_input(extra_file_path, extra_file_relative_path)
-                self.file_renames[extra_file_path] = response['path']
+        if exists(files_path) and self.job_inputs.path_referenced(files_path):
+            for extra_file in listdir(files_path):
+                extra_file_path = join(files_path, extra_file)
+                relative_path = basename(files_path)
+                extra_file_relative_path = join(relative_path, extra_file)
+                self.transfer_tracker.handle_transfer(extra_file_path, 'input_extra', name=extra_file_relative_path)
 
     def __upload_working_directory_files(self):
         # Task manager stages files into working directory, these need to be
         # uploaded if present.
-        for working_directory_file in os.listdir(self.working_directory):
-            path = os.path.join(self.working_directory, working_directory_file)
-            working_file_response = self.client.upload_working_directory_file(path)
-            self.file_renames[path] = working_file_response['path']
+        for working_directory_file in listdir(self.working_directory):
+            path = join(self.working_directory, working_directory_file)
+            self.transfer_tracker.handle_transfer(path, 'work_dir')
 
     def __initialize_output_file_renames(self):
         for output_file in self.output_files:
-            self.file_renames[output_file] = r'%s%s%s' % (self.new_outputs_directory,
-                                                         self.remote_path_separator,
-                                                         os.path.basename(output_file))
+            remote_path = r'%s%s%s' % (self.new_outputs_directory, self.remote_path_separator, basename(output_file))
+            self.transfer_tracker.register_rewrite(output_file, remote_path, 'output')
 
     def __initialize_task_output_file_renames(self):
         for output_file in self.output_files:
-            name = os.path.basename(output_file)
-            self.file_renames[os.path.join(self.working_directory, name)] = r'%s%s%s' % (self.new_working_directory,
-                                                                                         self.remote_path_separator,
-                                                                                         name)
+            name = basename(output_file)
+            task_file = join(self.working_directory, name)
+            remote_path = r'%s%s%s' % (self.new_working_directory, self.remote_path_separator, name)
+            self.transfer_tracker.register_rewrite(task_file, remote_path, 'output_task')
 
     def __initialize_config_file_renames(self):
         for config_file in self.config_files:
-            self.file_renames[config_file] = r'%s%s%s' % (self.new_configs_drectory,
-                                                         self.remote_path_separator,
-                                                         os.path.basename(config_file))
-
-    def __rewrite_paths(self, contents):
-        new_contents = contents
-        for local_path, remote_path in self.file_renames.iteritems():
-            new_contents = new_contents.replace(local_path, remote_path)
-        return new_contents
+            remote_path = r'%s%s%s' % (self.new_configs_drectory, self.remote_path_separator, basename(config_file))
+            self.transfer_tracker.register_rewrite(config_file, remote_path, 'config')
 
     def __handle_rewrites(self):
-        for local_path, remote_path in self.file_renames.iteritems():
-            self.job_inputs.rewrite_paths(local_path, remote_path)
+        """
+        For each file that has been transferred and renamed, updated
+        command_line and configfiles to reflect that rewrite.
+        """
+        self.transfer_tracker.rewrite_input_paths()
 
     def __upload_rewritten_config_files(self):
         for config_file, new_config_contents in self.job_inputs.rewritten_config_files.iteritems():
-            self.client.upload_config_file(config_file, new_config_contents)
+            self.client.put_file(config_file, input_type='config', contents=new_config_contents)
 
     def get_rewritten_command_line(self):
         """
@@ -247,6 +294,49 @@ class FileStager(object):
         return self.job_inputs.rewritten_command_line
 
 
+def finish_job(client, cleanup_job, job_completed_normally, working_directory, work_dir_outputs, output_files):
+    """
+    """
+    action_mapper = FileActionMapper(client)
+    download_failure_exceptions = []
+    if job_completed_normally:
+        for source_file, output_file in work_dir_outputs:
+            try:
+                action = action_mapper.action(output_file, 'output')
+                client.fetch_work_dir_output(source_file, working_directory, output_file, action[0])
+            except Exception, e:
+                download_failure_exceptions.append(e)
+            # Remove from full output_files list so don't try to download directly.
+            output_files.remove(output_file)
+        for output_file in output_files:
+            try:
+                action = action_mapper.action(output_file, 'output')
+                client.fetch_output(output_file, working_directory=working_directory, action=action[0])
+            except Exception, e:
+                download_failure_exceptions.append(e)
+    return __clean(download_failure_exceptions, cleanup_job, client)
+
+
+def __clean(download_failure_exceptions, cleanup_job, client):
+    failed = (len(download_failure_exceptions) > 0)
+    if not failed or cleanup_job == "always":
+        try:
+            client.clean()
+        except:
+            log.warn("Failed to cleanup remote LWR job")
+    return failed
+
+
+def submit_job(client, tool, command_line, config_files, input_files, output_files, working_directory):
+    """
+    """
+    file_stager = FileStager(client, tool, command_line, config_files, input_files, output_files, working_directory)
+    rebuilt_command_line = file_stager.get_rewritten_command_line()
+    job_id = file_stager.job_id
+    client.launch(rebuilt_command_line)
+    return job_id
+
+
 def _read(path):
     """
     Utility method to quickly read small files (config files and tool
@@ -257,3 +347,5 @@ def _read(path):
         return input.read()
     finally:
         input.close()
+
+__all__ = [submit_job, finish_job]

lib/galaxy/jobs/runners/lwr_client/transport/__init__.py

@@ -1,5 +1,5 @@
-from standard import Urllib2Transport
-from curl import PycurlTransport
+from .standard import Urllib2Transport
+from .curl import PycurlTransport
 import os
 
 

lib/galaxy/jobs/runners/lwr_client/util.py

@@ -0,0 +1,35 @@
+from threading import Lock, Event
+from weakref import WeakValueDictionary
+
+
+class TransferEventManager(object):
+
+    def __init__(self):
+        self.events = WeakValueDictionary(dict())
+        self.events_lock = Lock()
+
+    def acquire_event(self, path, force_clear=False):
+        with self.events_lock:
+            if path in self.events:
+                event_holder = self.events[path]
+            else:
+                event_holder = EventHolder(Event(), path, self)
+                self.events[path] = event_holder
+        if force_clear:
+            event_holder.event.clear()
+        return event_holder
+
+
+class EventHolder(object):
+
+    def __init__(self, event, path, condition_manager):
+        self.event = event
+        self.path = path
+        self.condition_manager = condition_manager
+        self.failed = False
+
+    def release(self):
+        self.event.set()
+
+    def fail(self):
+        self.failed = True