diff --git a/job_conf.xml.sample_advanced b/job_conf.xml.sample_advanced
index 396b82b9c98..512f5ff6c30 100644
--- a/job_conf.xml.sample_advanced
+++ b/job_conf.xml.sample_advanced
@@ -7,7 +7,11 @@
-
+
+
+
+
+
@@ -46,6 +50,10 @@
https://windowshost.examle.com:8913/
123456789changeme
+
+
SecureShell
diff --git a/lib/galaxy/jobs/runners/lwr.py b/lib/galaxy/jobs/runners/lwr.py
index 70edc5d1ae0..83b9402da8b 100644
--- a/lib/galaxy/jobs/runners/lwr.py
+++ b/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 )
diff --git a/lib/galaxy/jobs/runners/lwr_client/__init__.py b/lib/galaxy/jobs/runners/lwr_client/__init__.py
index 15cdf261391..fbdb0a7d5f9 100644
--- a/lib/galaxy/jobs/runners/lwr_client/__init__.py
+++ b/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]
diff --git a/lib/galaxy/jobs/runners/lwr_client/action_mapper.py b/lib/galaxy/jobs/runners/lwr_client/action_mapper.py
new file mode 100644
index 00000000000..186b6aeea83
--- /dev/null
+++ b/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]
diff --git a/lib/galaxy/jobs/runners/lwr_client/client.py b/lib/galaxy/jobs/runners/lwr_client/client.py
index a4e6a3a7d23..324391900ef 100644
--- a/lib/galaxy/jobs/runners/lwr_client/client.py
+++ b/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})
diff --git a/lib/galaxy/jobs/runners/lwr_client/destination.py b/lib/galaxy/jobs/runners/lwr_client/destination.py
index 110227d88cf..6054b6e98fd 100644
--- a/lib/galaxy/jobs/runners/lwr_client/destination.py
+++ b/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)])
diff --git a/lib/galaxy/jobs/runners/lwr_client/manager.py b/lib/galaxy/jobs/runners/lwr_client/manager.py
new file mode 100644
index 00000000000..2a1d8c0693e
--- /dev/null
+++ b/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]
diff --git a/lib/galaxy/jobs/runners/lwr_client/stager.py b/lib/galaxy/jobs/runners/lwr_client/stager.py
index 51a93e8d303..2ceb30118f6 100644
--- a/lib/galaxy/jobs/runners/lwr_client/stager.py
+++ b/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]
diff --git a/lib/galaxy/jobs/runners/lwr_client/transport/__init__.py b/lib/galaxy/jobs/runners/lwr_client/transport/__init__.py
index 0505fac2115..9071d7b7a8b 100644
--- a/lib/galaxy/jobs/runners/lwr_client/transport/__init__.py
+++ b/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
diff --git a/lib/galaxy/jobs/runners/lwr_client/util.py b/lib/galaxy/jobs/runners/lwr_client/util.py
new file mode 100644
index 00000000000..6a610a58c02
--- /dev/null
+++ b/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