From 0c74f5928efa3319a832e2fea18ebf28466906b5 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Tue, 20 Aug 2013 13:15:12 -0500 Subject: [PATCH 01/14] Allow LWR client to cache files when used with updated LWR servers. This synchronzies the LWR client to changeset LWR (f1feeb46845db15430b41af500af5544e73a8d38). --- lib/galaxy/jobs/runners/lwr.py | 14 +-- .../jobs/runners/lwr_client/__init__.py | 5 +- lib/galaxy/jobs/runners/lwr_client/client.py | 86 ++++++++++++++----- lib/galaxy/jobs/runners/lwr_client/manager.py | 53 ++++++++++++ lib/galaxy/jobs/runners/lwr_client/util.py | 31 +++++++ 5 files changed, 160 insertions(+), 29 deletions(-) create mode 100644 lib/galaxy/jobs/runners/lwr_client/manager.py create mode 100644 lib/galaxy/jobs/runners/lwr_client/util.py diff --git a/lib/galaxy/jobs/runners/lwr.py b/lib/galaxy/jobs/runners/lwr.py index 70edc5d1ae0..60b555e1e24 100644 --- a/lib/galaxy/jobs/runners/lwr.py +++ b/lib/galaxy/jobs/runners/lwr.py @@ -3,12 +3,13 @@ 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 import errno from time import sleep import os -from lwr_client import FileStager, Client, url_to_destination_params +from lwr_client import FileStager, ClientManager, url_to_destination_params log = logging.getLogger( __name__ ) @@ -21,12 +22,13 @@ class LwrJobRunner( AsynchronousJobRunner ): """ runner_name = "LWRRunner" - def __init__( self, app, nworkers, transport=None ): + def __init__( self, app, nworkers, transport=None, cache='false' ): """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': transport, 'cache': string_as_bool(cache)} + self.client_manager = ClientManager(**client_manager_kwargs) def url_to_destination( self, url ): """Convert a legacy URL to a job destination""" @@ -113,7 +115,7 @@ 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 = '' @@ -122,8 +124,8 @@ class LwrJobRunner( AsynchronousJobRunner ): 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 ]: diff --git a/lib/galaxy/jobs/runners/lwr_client/__init__.py b/lib/galaxy/jobs/runners/lwr_client/__init__.py index 15cdf261391..1d8390b3a9f 100644 --- a/lib/galaxy/jobs/runners/lwr_client/__init__.py +++ b/lib/galaxy/jobs/runners/lwr_client/__init__.py @@ -7,7 +7,8 @@ This module contains logic for interfacing with an external LWR server. """ from .stager import FileStager -from .client import Client +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, FileStager, url_to_destination_params] diff --git a/lib/galaxy/jobs/runners/lwr_client/client.py b/lib/galaxy/jobs/runners/lwr_client/client.py index a4e6a3a7d23..ac0b0637d93 100644 --- a/lib/galaxy/jobs/runners/lwr_client/client.py +++ b/lib/galaxy/jobs/runners/lwr_client/client.py @@ -1,11 +1,12 @@ import os -import time import urllib import simplejson +from time import sleep -from .transport import get_transport from .destination import url_to_destination_params +CACHE_WAIT_SECONDS = 3 + class parseJson(object): @@ -40,14 +41,14 @@ class Client(object): Galaxy job/task id. """ - def __init__(self, destination_params, job_id, transport_type=None): + def __init__(self, destination_params, job_id, client_manager): 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) self.job_id = job_id - self.transport = get_transport(transport_type) + self.client_manager = client_manager def __build_url(self, command, args): if self.private_key: @@ -56,20 +57,20 @@ class Client(object): url = self.remote_host + command + "?" + data return url - def __raw_execute(self, command, args={}, data=None, input_path=None, output_path=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) + response = self.client_manager.transport.execute(url, data=data, input_path=input_path, output_path=output_path) return response @parseJson() - def __upload_file(self, action, path, name=None, contents=None): + def _upload_file(self, action, path, name=None, contents=None): if not name: name = os.path.basename(path) args = {"job_id": self.job_id, "name": name} input_path = path if contents: input_path = None - return self.__raw_execute(action, args, contents, input_path) + return self._raw_execute(action, args, contents, input_path) def upload_tool_file(self, path): """ @@ -80,7 +81,7 @@ class Client(object): path : str Local path tool. """ - return self.__upload_file("upload_tool_file", path) + return self._upload_file("upload_tool_file", path) def upload_input(self, path): """ @@ -91,7 +92,7 @@ class Client(object): path : str Local path of input dataset. """ - return self.__upload_file("upload_input", path) + return self._upload_file("upload_input", path) def upload_extra_input(self, path, relative_name): """ @@ -104,7 +105,7 @@ class Client(object): 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) + return self._upload_file("upload_extra_input", path, name=relative_name) def upload_config_file(self, path, contents): """ @@ -117,7 +118,7 @@ class Client(object): contents : str Rewritten contents of the config file to upload. """ - return self.__upload_file("upload_config_file", path, contents=contents) + return self._upload_file("upload_config_file", path, contents=contents) def upload_working_directory_file(self, path): """ @@ -129,11 +130,11 @@ class Client(object): path : str Path to file to upload. """ - return self.__upload_file("upload_working_directory_file", path) + return self._upload_file("upload_working_directory_file", path) @parseJson() def _get_output_type(self, name): - return self.__raw_execute("get_output_type", {"name": name, + 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): @@ -176,7 +177,7 @@ class Client(object): self.__raw_download_output(name, self.job_id, output_type, output_path) def __raw_download_output(self, name, job_id, output_type, output_path): - self.__raw_execute("download_output", + self._raw_execute("download_output", {"name": name, "job_id": self.job_id, "output_type": output_type}, @@ -192,14 +193,14 @@ class Client(object): command_line : str Command to execute. """ - return self.__raw_execute("launch", {"command_line": command_line, + return self._raw_execute("launch", {"command_line": command_line, "job_id": self.job_id}) 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,14 +210,14 @@ 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): @@ -243,7 +244,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 +256,47 @@ 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) + + +class InputCachingClient(Client): + """ + Beta client that cache's staged files to prevent duplication. + """ + + def __init__(self, destination_params, job_id, client_manager): + super(InputCachingClient, self).__init__(destination_params, job_id, client_manager) + + @parseJson() + def _upload_file(self, action, path, name=None, contents=None): + if not name: + name = os.path.basename(path) + args = {"job_id": self.job_id, "name": name} + input_path = path + if contents: + input_path = None + return self._raw_execute(action, args, contents, input_path) + else: + event_holder = self.client_manager.event_manager.acquire_event(input_path) + cache_required = self.cache_required(input_path) + if cache_required: + self.client_manager.queue_transfer(self, input_path) + while True: + 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) + + @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/manager.py b/lib/galaxy/jobs/runners/lwr_client/manager.py new file mode 100644 index 00000000000..957a4c1a490 --- /dev/null +++ b/lib/galaxy/jobs/runners/lwr_client/manager.py @@ -0,0 +1,53 @@ +from Queue import Queue +from threading import Thread + +from .client import Client, InputCachingClient +from .transport import get_transport +from .util import TransferEventManager + +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): + self.transport = get_transport(kwds.get('transport_type', None)) + self.event_manager = TransferEventManager() + cache = kwds.get('cache', False) + if cache: + self.client_class = InputCachingClient + num_transfer_threads = int(kwds.get('transfer_threads', DEFAULT_TRANSFER_THREADS)) + self.__init_transfer_threads(num_transfer_threads) + else: + self.client_class = Client + + def _transfer_worker(self): + while True: + transfer_info = self.transfer_queue.get() + try: + self.__perform_transfer(transfer_info) + except: + 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) + client.cache_insert(path) + event_holder.event.set() + + def __init_transfer_threads(self, 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 queue_transfer(self, client, path): + self.transfer_queue.put((client, path)) + + def get_client(self, destination_params, job_id): + return self.client_class(destination_params, job_id, self) 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..001db19c2a9 --- /dev/null +++ b/lib/galaxy/jobs/runners/lwr_client/util.py @@ -0,0 +1,31 @@ +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 + + def release(self): + self.event.set() From d0fe6bda6782fd2ea4d366098f1211bcc7264504 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Tue, 27 Aug 2013 12:13:13 -0500 Subject: [PATCH 02/14] Allow LWR caching to be configured via environment variables for pre-job_conf.xml setups. Updates LWR client to LWR changesets: 2effb7f7b7ad. --- lib/galaxy/jobs/runners/lwr.py | 6 +++--- lib/galaxy/jobs/runners/lwr_client/manager.py | 21 +++++++++++++++++-- 2 files changed, 22 insertions(+), 5 deletions(-) diff --git a/lib/galaxy/jobs/runners/lwr.py b/lib/galaxy/jobs/runners/lwr.py index 60b555e1e24..e91c2e6e79c 100644 --- a/lib/galaxy/jobs/runners/lwr.py +++ b/lib/galaxy/jobs/runners/lwr.py @@ -3,7 +3,7 @@ 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 +from galaxy.util import string_as_bool_or_none import errno from time import sleep @@ -22,12 +22,12 @@ class LwrJobRunner( AsynchronousJobRunner ): """ runner_name = "LWRRunner" - def __init__( self, app, nworkers, transport=None, cache='false' ): + 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() - client_manager_kwargs = {'transport': transport, 'cache': string_as_bool(cache)} + client_manager_kwargs = {'transport': transport, 'cache': string_as_bool_or_none(cache)} self.client_manager = ClientManager(**client_manager_kwargs) def url_to_destination( self, url ): diff --git a/lib/galaxy/jobs/runners/lwr_client/manager.py b/lib/galaxy/jobs/runners/lwr_client/manager.py index 957a4c1a490..4ecca4c779b 100644 --- a/lib/galaxy/jobs/runners/lwr_client/manager.py +++ b/lib/galaxy/jobs/runners/lwr_client/manager.py @@ -1,10 +1,14 @@ from Queue import Queue from threading import Thread +from os import getenv from .client import Client, InputCachingClient from .transport import get_transport from .util import TransferEventManager +from logging import getLogger +log = getLogger(__name__) + DEFAULT_TRANSFER_THREADS = 2 @@ -16,12 +20,17 @@ class ClientManager(object): def __init__(self, **kwds): self.transport = get_transport(kwds.get('transport_type', None)) self.event_manager = TransferEventManager() - cache = kwds.get('cache', False) + 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_class = InputCachingClient - num_transfer_threads = int(kwds.get('transfer_threads', DEFAULT_TRANSFER_THREADS)) + num_transfer_threads = int(kwds.get('transfer_threads', _environ_default_int('LWR_CACHE_TRANSFERS', DEFAULT_TRANSFER_THREADS))) self.__init_transfer_threads(num_transfer_threads) else: + log.info("Setting LWR client class to standard, non-caching variant.") self.client_class = Client def _transfer_worker(self): @@ -51,3 +60,11 @@ class ClientManager(object): def get_client(self, destination_params, job_id): return self.client_class(destination_params, job_id, self) + + +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 From 09a79e45da17dd3f485db3db072e5a0cbcd60d77 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Tue, 27 Aug 2013 12:43:07 -0500 Subject: [PATCH 03/14] Fix small bug related to lwr client caching thread count. Updates LWR client through LWR changeset abac180. --- lib/galaxy/jobs/runners/lwr_client/manager.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/lib/galaxy/jobs/runners/lwr_client/manager.py b/lib/galaxy/jobs/runners/lwr_client/manager.py index 4ecca4c779b..d8ad177ba2f 100644 --- a/lib/galaxy/jobs/runners/lwr_client/manager.py +++ b/lib/galaxy/jobs/runners/lwr_client/manager.py @@ -27,7 +27,8 @@ class ClientManager(object): if cache: log.info("Setting LWR client class to caching variant.") self.client_class = InputCachingClient - num_transfer_threads = int(kwds.get('transfer_threads', _environ_default_int('LWR_CACHE_TRANSFERS', DEFAULT_TRANSFER_THREADS))) + 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) else: log.info("Setting LWR client class to standard, non-caching variant.") @@ -49,6 +50,7 @@ class ClientManager(object): event_holder.event.set() 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) From b530fd29a30e2612c4913c6914f853aa14673f74 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Thu, 29 Aug 2013 14:38:38 -0500 Subject: [PATCH 04/14] LWR client fixes for Python 3. Brings Galaxy client in line through LWR changeset 4ce6788. --- lib/galaxy/jobs/runners/lwr_client/manager.py | 5 ++++- lib/galaxy/jobs/runners/lwr_client/transport/__init__.py | 4 ++-- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/lib/galaxy/jobs/runners/lwr_client/manager.py b/lib/galaxy/jobs/runners/lwr_client/manager.py index d8ad177ba2f..2167ed14cf3 100644 --- a/lib/galaxy/jobs/runners/lwr_client/manager.py +++ b/lib/galaxy/jobs/runners/lwr_client/manager.py @@ -1,4 +1,7 @@ -from Queue import Queue +try: + from Queue import Queue +except ImportError: + from queue import Queue from threading import Thread from os import getenv 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 From a78ef01ed1c5332bad1f2ba6a0e7208faffc6d95 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Thu, 29 Aug 2013 14:43:24 -0500 Subject: [PATCH 05/14] In LWR client, retry certain calls to get around transient peer reset connection errors. See the following LWR changeset for full information: https://bitbucket.org/jmchilton/lwr/commits/95ce1503d53562a2e46cdbeb35d5c5d9cb35278c. --- lib/galaxy/jobs/runners/lwr_client/client.py | 28 +++++++++++++++++--- 1 file changed, 25 insertions(+), 3 deletions(-) diff --git a/lib/galaxy/jobs/runners/lwr_client/client.py b/lib/galaxy/jobs/runners/lwr_client/client.py index ac0b0637d93..30846f9a755 100644 --- a/lib/galaxy/jobs/runners/lwr_client/client.py +++ b/lib/galaxy/jobs/runners/lwr_client/client.py @@ -6,13 +6,12 @@ from time import sleep from .destination import url_to_destination_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) @@ -20,6 +19,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): @@ -176,6 +196,7 @@ class Client(object): raise OutputNotFoundException(path) self.__raw_download_output(name, self.job_id, output_type, output_path) + @retry() def __raw_download_output(self, name, job_id, output_type, output_path): self._raw_execute("download_output", {"name": name, @@ -228,6 +249,7 @@ class Client(object): 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 From fc66a2bd1b7ca3c62cf563184542e961223492b4 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Tue, 3 Sep 2013 09:23:15 -0500 Subject: [PATCH 06/14] Fix bug in LWR cleanup logic. --- lib/galaxy/jobs/runners/lwr.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/lib/galaxy/jobs/runners/lwr.py b/lib/galaxy/jobs/runners/lwr.py index e91c2e6e79c..62bb2718b31 100644 --- a/lib/galaxy/jobs/runners/lwr.py +++ b/lib/galaxy/jobs/runners/lwr.py @@ -143,12 +143,13 @@ class LwrJobRunner( AsynchronousJobRunner ): 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": + failed = len(download_failure_exceptions) > 0 + if not failed or self.app.config.cleanup_job == "always": try: client.clean() except: log.warn("Failed to cleanup remote LWR job") - if download_failure_exceptions: + 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: From 48757f6dca7e5a6efa21323a9014f6cf9dca6dd9 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Wed, 4 Sep 2013 12:12:52 -0500 Subject: [PATCH 07/14] Update LWR client through LWR changeset 80f90e94f780b361a27ca70a8c8cff6b7add27ca. In addition to small cleanups and bug fixes, this adds ability to disable file staging (for the most part) by adding default_file_action="none" to LWR destination. This is outlined in greater detail in the following LWR changeset: https://bitbucket.org/jmchilton/lwr/commits/1f214688c0f02cbb7b85fc033b96daf61d56a309 Setting default_file_action="none" requires a newer LWR server (1f214688c0f02cbb7b85fc033b96daf61d56a309 or better), transferring LWR jobs should continue to work with an older or newer LWR server. --- lib/galaxy/jobs/runners/lwr.py | 43 ++--- .../jobs/runners/lwr_client/__init__.py | 4 +- lib/galaxy/jobs/runners/lwr_client/client.py | 168 +++++++++--------- lib/galaxy/jobs/runners/lwr_client/manager.py | 3 +- lib/galaxy/jobs/runners/lwr_client/stager.py | 155 ++++++++++++---- 5 files changed, 219 insertions(+), 154 deletions(-) diff --git a/lib/galaxy/jobs/runners/lwr.py b/lib/galaxy/jobs/runners/lwr.py index 62bb2718b31..bb72553a9a8 100644 --- a/lib/galaxy/jobs/runners/lwr.py +++ b/lib/galaxy/jobs/runners/lwr.py @@ -9,7 +9,8 @@ import errno from time import sleep import os -from lwr_client import FileStager, ClientManager, url_to_destination_params +from .lwr_client import FileStager, ClientManager, url_to_destination_params +from .lwr_client import finish_job as lwr_finish_job log = logging.getLogger( __name__ ) @@ -118,7 +119,7 @@ class LwrJobRunner( AsynchronousJobRunner ): 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) @@ -127,31 +128,23 @@ class LwrJobRunner( AsynchronousJobRunner ): 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) - failed = len(download_failure_exceptions) > 0 - if not failed or self.app.config.cleanup_job == "always": - try: - client.clean() - except: - log.warn("Failed to cleanup remote LWR job") + # 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 1d8390b3a9f..0a2dd61bebb 100644 --- a/lib/galaxy/jobs/runners/lwr_client/__init__.py +++ b/lib/galaxy/jobs/runners/lwr_client/__init__.py @@ -6,9 +6,9 @@ This module contains logic for interfacing with an external LWR server. """ -from .stager import FileStager +from .stager import FileStager, finish_job from .client import OutputNotFoundException from .manager import ClientManager from .destination import url_to_destination_params -__all__ = [ClientManager, OutputNotFoundException, FileStager, url_to_destination_params] +__all__ = [ClientManager, OutputNotFoundException, FileStager, url_to_destination_params, finish_job] diff --git a/lib/galaxy/jobs/runners/lwr_client/client.py b/lib/galaxy/jobs/runners/lwr_client/client.py index 30846f9a755..7aea0621176 100644 --- a/lib/galaxy/jobs/runners/lwr_client/client.py +++ b/lib/galaxy/jobs/runners/lwr_client/client.py @@ -1,4 +1,5 @@ import os +import shutil import urllib import simplejson from time import sleep @@ -65,6 +66,7 @@ class Client(object): 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") + self.default_file_action = destination_params.get("default_file_action", "transfer") assert self.remote_host != None, "Failed to determine url for LWR client." self.private_key = destination_params.get("private_token", None) self.job_id = job_id @@ -83,99 +85,48 @@ class Client(object): return response @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}) - def download_work_dir_output(self, source, working_directory, output_path): - """ - Download an output dataset specified with from_work_dir from the - remote server. - - **Parameters** - - source : str - Path in job's working_directory to find output in. - working_directory : str - Local working_directory for the job. - output_path : str - Full path to output dataset. - """ - output = open(output_path, "wb") - name = os.path.basename(source) - self.__raw_download_output(name, self.job_id, "work_dir", output) - - def download_output(self, path, working_directory): + def fetch_output(self, path, working_directory, action='transfer'): """ Download an output dataset from the remote server. @@ -192,9 +143,49 @@ class Client(object): output_path = path elif output_type == "task": output_path = os.path.join(working_directory, name) + elif output_type == "none": + if action == "transfer": + raise OutputNotFoundException(path) else: - raise OutputNotFoundException(path) - self.__raw_download_output(name, self.job_id, output_type, output_path) + raise Exception("Unknown output_type returned from LWR server %s" % output_type) + if action == 'transfer': + self.__raw_download_output(name, self.job_id, output_type, output_path) + elif output_type == 'none': + # Just make sure the file was created. + if not os.path.exists(path): + raise OutputNotFoundException(path) + elif action == 'copy': + lwr_path = self._output_path(name, self.job_id, output_type)['path'] + self._copy(lwr_path, 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. + + **Parameters** + + source : str + Path in job's working_directory to find output in. + working_directory : str + Local working_directory for the job. + output_path : str + Full path to output dataset. + """ + output = open(output_path, "wb") + name = os.path.basename(source) + 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) + + @parseJson() + def _output_path(self, name, job_id, output_type): + 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): @@ -280,6 +271,12 @@ class Client(object): setup_args["tool_version"] = tool_version 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): """ @@ -290,11 +287,8 @@ class InputCachingClient(Client): super(InputCachingClient, self).__init__(destination_params, job_id, client_manager) @parseJson() - def _upload_file(self, action, path, name=None, contents=None): - if not name: - name = os.path.basename(path) - args = {"job_id": self.job_id, "name": name} - input_path = path + 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) diff --git a/lib/galaxy/jobs/runners/lwr_client/manager.py b/lib/galaxy/jobs/runners/lwr_client/manager.py index 2167ed14cf3..4b99dfdf22e 100644 --- a/lib/galaxy/jobs/runners/lwr_client/manager.py +++ b/lib/galaxy/jobs/runners/lwr_client/manager.py @@ -21,7 +21,8 @@ class ClientManager(object): state between multiple client connections. """ def __init__(self, **kwds): - self.transport = get_transport(kwds.get('transport_type', None)) + transport_type = kwds.get('transport_type', None) + self.transport = get_transport(transport_type) self.event_manager = TransferEventManager() cache = kwds.get('cache', None) if cache is None: diff --git a/lib/galaxy/jobs/runners/lwr_client/stager.py b/lib/galaxy/jobs/runners/lwr_client/stager.py index 51a93e8d303..44f6f4238f1 100644 --- a/lib/galaxy/jobs/runners/lwr_client/stager.py +++ b/lib/galaxy/jobs/runners/lwr_client/stager.py @@ -1,7 +1,10 @@ - -import os +from os.path import abspath, basename, join, exists +from os import listdir, sep from re import findall +from logging import getLogger +log = getLogger(__name__) + class JobInputs(object): """ @@ -61,7 +64,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 +99,59 @@ class JobInputs(object): return items +class FileActionMapper(object): + + def __init__(self, client): + self.default_action = client.default_file_action + + def action(self, path, type): + action = self.default_action + if type in ["work_dir", "output_task", "output"] 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 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: + action = ('transfer', ) + else: + 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] == "transfer" 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 +186,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 +229,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 +238,53 @@ 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'] + self.transfer_tracker.handle_transfer(input_file, 'input') 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,34 @@ 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) + + 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 _read(path): """ Utility method to quickly read small files (config files and tool @@ -257,3 +332,5 @@ def _read(path): return input.read() finally: input.close() + +__all__ = [FileStager, finish_job] From 13679a1fdc8f23cd30a480b0866084801c3716ed Mon Sep 17 00:00:00 2001 From: John Chilton Date: Wed, 4 Sep 2013 12:13:27 -0500 Subject: [PATCH 08/14] Document some additional LWR functionality in job_conf.xml.sample_advanced. --- job_conf.xml.sample_advanced | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/job_conf.xml.sample_advanced b/job_conf.xml.sample_advanced index 6a9a6c6e3e0..e326c42ff87 100644 --- a/job_conf.xml.sample_advanced +++ b/job_conf.xml.sample_advanced @@ -7,7 +7,11 @@ - + + + + + @@ -46,6 +50,9 @@ https://windowshost.examle.com:8913/ 123456789changeme + + SecureShell From 177776841008493cfd58d1f9946241ee6af2f1eb Mon Sep 17 00:00:00 2001 From: John Chilton Date: Wed, 4 Sep 2013 12:15:25 -0500 Subject: [PATCH 09/14] Fix bug (inability to correctly configure curl transport using new job_conf.xml syntax) I introduced in d9f874a. --- lib/galaxy/jobs/runners/lwr.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/galaxy/jobs/runners/lwr.py b/lib/galaxy/jobs/runners/lwr.py index bb72553a9a8..9956fb4b82a 100644 --- a/lib/galaxy/jobs/runners/lwr.py +++ b/lib/galaxy/jobs/runners/lwr.py @@ -28,7 +28,7 @@ class LwrJobRunner( AsynchronousJobRunner ): super( LwrJobRunner, self ).__init__( app, nworkers ) self._init_monitor_thread() self._init_worker_threads() - client_manager_kwargs = {'transport': transport, 'cache': string_as_bool_or_none(cache)} + 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 ): From ce02b61f85f9d8fdafd51a7e656a5ebecce6e64e Mon Sep 17 00:00:00 2001 From: John Chilton Date: Mon, 9 Sep 2013 13:59:20 -0500 Subject: [PATCH 10/14] Update the LWR client through LWR changeset 3602261cf9fced102dd019a0ce22cb25f9e656d1. Introduces syntax allowing LWR client to assign various actions for staging files on a per path basis - 'transfer', 'copy', or leave untouched ('none'). For full details see this LWR changeset: https://bitbucket.org/jmchilton/lwr/commits/b9471d7c3ccb61d56f558e055c997ff53a38d540. --- .../jobs/runners/lwr_client/action_mapper.py | 121 ++++++++++++++++++ lib/galaxy/jobs/runners/lwr_client/client.py | 20 +-- lib/galaxy/jobs/runners/lwr_client/stager.py | 21 +-- 3 files changed, 139 insertions(+), 23 deletions(-) create mode 100644 lib/galaxy/jobs/runners/lwr_client/action_mapper.py 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..2e69d512dfa --- /dev/null +++ b/lib/galaxy/jobs/runners/lwr_client/action_mapper.py @@ -0,0 +1,121 @@ +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 7aea0621176..f6f8a02ed26 100644 --- a/lib/galaxy/jobs/runners/lwr_client/client.py +++ b/lib/galaxy/jobs/runners/lwr_client/client.py @@ -2,6 +2,7 @@ import os import shutil import urllib import simplejson +from simplejson import dumps from time import sleep from .destination import url_to_destination_params @@ -67,6 +68,7 @@ class Client(object): destination_params = url_to_destination_params(destination_params) self.remote_host = destination_params.get("url") self.default_file_action = destination_params.get("default_file_action", "transfer") + self.action_config_path = destination_params.get("file_action_config", None) assert self.remote_host != None, "Failed to determine url for LWR client." self.private_key = destination_params.get("private_token", None) self.job_id = job_id @@ -124,7 +126,7 @@ class Client(object): @parseJson() def _get_output_type(self, name): return self._raw_execute("get_output_type", {"name": name, - "job_id": self.job_id}) + "job_id": self.job_id}) def fetch_output(self, path, working_directory, action='transfer'): """ @@ -182,10 +184,10 @@ class Client(object): @parseJson() def _output_path(self, name, job_id, output_type): - self._raw_execute("output_path", - {"name": name, - "job_id": self.job_id, - "output_type": 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): @@ -195,7 +197,7 @@ class Client(object): "output_type": output_type}, output_path=output_path) - def launch(self, command_line): + def launch(self, command_line, submit_params=None): """ Run or queue up the execution of the supplied `command_line` on the remote server. @@ -205,8 +207,10 @@ 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) + if submit_params: + launch_params['params'] = dumps(submit_params) + return self._raw_execute("launch", launch_params) def kill(self): """ diff --git a/lib/galaxy/jobs/runners/lwr_client/stager.py b/lib/galaxy/jobs/runners/lwr_client/stager.py index 44f6f4238f1..10968caee31 100644 --- a/lib/galaxy/jobs/runners/lwr_client/stager.py +++ b/lib/galaxy/jobs/runners/lwr_client/stager.py @@ -2,6 +2,8 @@ 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__) @@ -99,20 +101,6 @@ class JobInputs(object): return items -class FileActionMapper(object): - - def __init__(self, client): - self.default_action = client.default_file_action - - def action(self, path, type): - action = self.default_action - if type in ["work_dir", "output_task", "output"] 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 TransferTracker(object): def __init__(self, client, job_inputs): @@ -123,9 +111,12 @@ class TransferTracker(object): 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: 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) @@ -137,7 +128,7 @@ class TransferTracker(object): def register_rewrite(self, local_path, remote_path, type, force=False): action = self.__action(local_path, type) - if action[0] == "transfer" or force: + if action[0] in ['transfer', 'copy'] or force: self.file_renames[local_path] = remote_path def rewrite_input_paths(self): From 2c263be10816efa575bfa12bf0cce04315d84f7f Mon Sep 17 00:00:00 2001 From: John Chilton Date: Wed, 11 Sep 2013 07:53:44 -0500 Subject: [PATCH 11/14] Update LWR client through LWR changeset 430d650ba49a. These changes include small code cleanups and the ability to pass job submission parameters through from client to server. Specifically, if the remote LWR is configured using the DRMAA manager, the destination parameter "submit_native_specification" should be passed through and used on the server. In theory this should be able to be set from the job_conf.xml or via dynamic job destinations. --- job_conf.xml.sample_advanced | 5 +++-- lib/galaxy/jobs/runners/lwr.py | 9 ++++----- lib/galaxy/jobs/runners/lwr_client/__init__.py | 4 ++-- lib/galaxy/jobs/runners/lwr_client/client.py | 9 ++++++++- lib/galaxy/jobs/runners/lwr_client/destination.py | 15 +++++++++++++++ lib/galaxy/jobs/runners/lwr_client/stager.py | 14 +++++++++++++- 6 files changed, 45 insertions(+), 11 deletions(-) diff --git a/job_conf.xml.sample_advanced b/job_conf.xml.sample_advanced index e326c42ff87..bb21df534ab 100644 --- a/job_conf.xml.sample_advanced +++ b/job_conf.xml.sample_advanced @@ -50,8 +50,9 @@ https://windowshost.examle.com:8913/ 123456789changeme - + diff --git a/lib/galaxy/jobs/runners/lwr.py b/lib/galaxy/jobs/runners/lwr.py index 9956fb4b82a..64a596350e0 100644 --- a/lib/galaxy/jobs/runners/lwr.py +++ b/lib/galaxy/jobs/runners/lwr.py @@ -9,8 +9,9 @@ import errno from time import sleep import os -from .lwr_client import FileStager, ClientManager, 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__ ) @@ -81,10 +82,8 @@ 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) job_wrapper.set_job_destination( job_destination, job_id ) job_wrapper.change_state( model.Job.states.QUEUED ) except: diff --git a/lib/galaxy/jobs/runners/lwr_client/__init__.py b/lib/galaxy/jobs/runners/lwr_client/__init__.py index 0a2dd61bebb..fbdb0a7d5f9 100644 --- a/lib/galaxy/jobs/runners/lwr_client/__init__.py +++ b/lib/galaxy/jobs/runners/lwr_client/__init__.py @@ -6,9 +6,9 @@ This module contains logic for interfacing with an external LWR server. """ -from .stager import FileStager, finish_job +from .stager import submit_job, finish_job from .client import OutputNotFoundException from .manager import ClientManager from .destination import url_to_destination_params -__all__ = [ClientManager, OutputNotFoundException, FileStager, url_to_destination_params, finish_job] +__all__ = [ClientManager, OutputNotFoundException, url_to_destination_params, finish_job, submit_job] diff --git a/lib/galaxy/jobs/runners/lwr_client/client.py b/lib/galaxy/jobs/runners/lwr_client/client.py index f6f8a02ed26..a4ee845b710 100644 --- a/lib/galaxy/jobs/runners/lwr_client/client.py +++ b/lib/galaxy/jobs/runners/lwr_client/client.py @@ -6,6 +6,7 @@ from simplejson import dumps from time import sleep from .destination import url_to_destination_params +from .destination import submit_params CACHE_WAIT_SECONDS = 3 MAX_RETRY_COUNT = 5 @@ -69,11 +70,16 @@ class Client(object): self.remote_host = destination_params.get("url") self.default_file_action = destination_params.get("default_file_action", "transfer") self.action_config_path = destination_params.get("file_action_config", None) + self.destination_params = destination_params assert self.remote_host != None, "Failed to determine url for LWR client." self.private_key = destination_params.get("private_token", None) self.job_id = job_id self.client_manager = client_manager + @property + def _submit_params(self): + return submit_params(self.destination_params) + def __build_url(self, command, args): if self.private_key: args["private_key"] = self.private_key @@ -197,7 +203,7 @@ class Client(object): "output_type": output_type}, output_path=output_path) - def launch(self, command_line, submit_params=None): + def launch(self, command_line): """ Run or queue up the execution of the supplied `command_line` on the remote server. @@ -208,6 +214,7 @@ class Client(object): Command to execute. """ 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) diff --git a/lib/galaxy/jobs/runners/lwr_client/destination.py b/lib/galaxy/jobs/runners/lwr_client/destination.py index 110227d88cf..5c44ce76d10 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/stager.py b/lib/galaxy/jobs/runners/lwr_client/stager.py index 10968caee31..1203239907d 100644 --- a/lib/galaxy/jobs/runners/lwr_client/stager.py +++ b/lib/galaxy/jobs/runners/lwr_client/stager.py @@ -286,6 +286,8 @@ class FileStager(object): 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: @@ -313,6 +315,16 @@ def finish_job(client, cleanup_job, job_completed_normally, working_directory, w 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 @@ -324,4 +336,4 @@ def _read(path): finally: input.close() -__all__ = [FileStager, finish_job] +__all__ = [submit_job, finish_job] From c746d786e7eb07e45380471ec17f2616b0e13c06 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Mon, 16 Sep 2013 09:40:54 -0500 Subject: [PATCH 12/14] Make some attempt to deal with failed uploads while handling LWR caching. Updates LWR client through changeset https://bitbucket.org/jmchilton/lwr/commits/54df64df1a00579ea51d91ef7215d2046e94304b. --- lib/galaxy/jobs/runners/lwr_client/client.py | 4 +++- lib/galaxy/jobs/runners/lwr_client/manager.py | 14 ++++++++++---- lib/galaxy/jobs/runners/lwr_client/util.py | 4 ++++ 3 files changed, 17 insertions(+), 5 deletions(-) diff --git a/lib/galaxy/jobs/runners/lwr_client/client.py b/lib/galaxy/jobs/runners/lwr_client/client.py index a4ee845b710..0aef72a5977 100644 --- a/lib/galaxy/jobs/runners/lwr_client/client.py +++ b/lib/galaxy/jobs/runners/lwr_client/client.py @@ -308,13 +308,15 @@ class InputCachingClient(Client): cache_required = self.cache_required(input_path) if cache_required: self.client_manager.queue_transfer(self, input_path) - while True: + 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): diff --git a/lib/galaxy/jobs/runners/lwr_client/manager.py b/lib/galaxy/jobs/runners/lwr_client/manager.py index 4b99dfdf22e..cb74c602233 100644 --- a/lib/galaxy/jobs/runners/lwr_client/manager.py +++ b/lib/galaxy/jobs/runners/lwr_client/manager.py @@ -27,7 +27,6 @@ class ClientManager(object): 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_class = InputCachingClient @@ -43,15 +42,22 @@ class ClientManager(object): transfer_info = self.transfer_queue.get() try: self.__perform_transfer(transfer_info) - except: + 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) - client.cache_insert(path) - event_holder.event.set() + 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 diff --git a/lib/galaxy/jobs/runners/lwr_client/util.py b/lib/galaxy/jobs/runners/lwr_client/util.py index 001db19c2a9..6a610a58c02 100644 --- a/lib/galaxy/jobs/runners/lwr_client/util.py +++ b/lib/galaxy/jobs/runners/lwr_client/util.py @@ -26,6 +26,10 @@ class EventHolder(object): 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 From 815d1ef52d3cc91729b090b7e99a4ded399e7529 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Mon, 23 Sep 2013 07:25:04 -0500 Subject: [PATCH 13/14] In LWR, handle 'ok' datasets which do not corresponding to existing files (previous tool produced no output). Should Galaxy prevent these datasets/hdas to be 'ok'. Updates LWR client through LWR changeset: 1cc5f35ecb8b. --- lib/galaxy/jobs/runners/lwr.py | 1 + lib/galaxy/jobs/runners/lwr_client/stager.py | 11 ++++++++++- 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/lib/galaxy/jobs/runners/lwr.py b/lib/galaxy/jobs/runners/lwr.py index 64a596350e0..83b9402da8b 100644 --- a/lib/galaxy/jobs/runners/lwr.py +++ b/lib/galaxy/jobs/runners/lwr.py @@ -84,6 +84,7 @@ class LwrJobRunner( AsynchronousJobRunner ): tool = job_wrapper.tool 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: diff --git a/lib/galaxy/jobs/runners/lwr_client/stager.py b/lib/galaxy/jobs/runners/lwr_client/stager.py index 1203239907d..626f710793c 100644 --- a/lib/galaxy/jobs/runners/lwr_client/stager.py +++ b/lib/galaxy/jobs/runners/lwr_client/stager.py @@ -115,6 +115,10 @@ class TransferTracker(object): # 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']: @@ -229,7 +233,12 @@ class FileStager(object): def __upload_input_file(self, input_file): if self.job_inputs.path_referenced(input_file): - self.transfer_tracker.handle_transfer(input_file, 'input') + 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 From 5a3a39eb4e645c2e72b78e6ae98c3a3777166b10 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Mon, 21 Oct 2013 16:03:54 -0500 Subject: [PATCH 14/14] Update LWR client through LWR changeset 6c2dc7c8395b. This is mostly just PEP-8/formatting fixes, but also includes a refactoring of the client to potentially allow the LWR client to talk directly to an LWR via Python instead of going through HTTP. In the context of Galaxy, this would require embedded an LWR manager right inside of Galaxy, which is not yet possible, but should be doable someday. --- .../jobs/runners/lwr_client/action_mapper.py | 9 +- lib/galaxy/jobs/runners/lwr_client/client.py | 87 ++++++------ .../jobs/runners/lwr_client/destination.py | 4 +- lib/galaxy/jobs/runners/lwr_client/manager.py | 133 ++++++++++++++++-- lib/galaxy/jobs/runners/lwr_client/stager.py | 3 + 5 files changed, 173 insertions(+), 63 deletions(-) diff --git a/lib/galaxy/jobs/runners/lwr_client/action_mapper.py b/lib/galaxy/jobs/runners/lwr_client/action_mapper.py index 2e69d512dfa..186b6aeea83 100644 --- a/lib/galaxy/jobs/runners/lwr_client/action_mapper.py +++ b/lib/galaxy/jobs/runners/lwr_client/action_mapper.py @@ -112,10 +112,11 @@ class RegexPathMapper(BasePathMapper): return self.pattern.match(path) is not None -mappers = {'prefix': PrefixPathMapper, - 'glob': GlobPathMapper, - 'regex': RegexPathMapper, - } +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 0aef72a5977..324391900ef 100644 --- a/lib/galaxy/jobs/runners/lwr_client/client.py +++ b/lib/galaxy/jobs/runners/lwr_client/client.py @@ -1,11 +1,9 @@ import os import shutil -import urllib import simplejson from simplejson import dumps from time import sleep -from .destination import url_to_destination_params from .destination import submit_params CACHE_WAIT_SECONDS = 3 @@ -64,34 +62,21 @@ class Client(object): Galaxy job/task id. """ - def __init__(self, destination_params, job_id, client_manager): - 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") - self.default_file_action = destination_params.get("default_file_action", "transfer") - self.action_config_path = destination_params.get("file_action_config", None) + def __init__(self, destination_params, job_id, job_manager_interface): + self.job_manager_interface = job_manager_interface self.destination_params = destination_params - assert self.remote_host != None, "Failed to determine url for LWR client." - self.private_key = destination_params.get("private_token", None) self.job_id = job_id - self.client_manager = client_manager + + 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): + return self.job_manager_interface.execute(command, args, data, input_path, output_path) @property def _submit_params(self): return submit_params(self.destination_params) - 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 - - def _raw_execute(self, command, args={}, data=None, input_path=None, output_path=None): - url = self.__build_url(command, args) - response = self.client_manager.transport.execute(url, data=data, input_path=input_path, output_path=output_path) - return response - @parseJson() def input_path(self, path, input_type, name=None): args = {"job_id": self.job_id, "name": name, "input_type": input_type} @@ -147,24 +132,34 @@ class Client(object): """ 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) - elif output_type == "none": - if action == "transfer": - raise OutputNotFoundException(path) else: raise Exception("Unknown output_type returned from LWR server %s" % output_type) - if action == 'transfer': - self.__raw_download_output(name, self.job_id, output_type, output_path) - elif output_type == 'none': - # Just make sure the file was created. - if not os.path.exists(path): - raise OutputNotFoundException(path) - elif action == 'copy': - lwr_path = self._output_path(name, self.job_id, output_type)['path'] - self._copy(lwr_path, output_path) + return output_path def fetch_work_dir_output(self, source, working_directory, output_path, action='transfer'): """ @@ -197,11 +192,12 @@ class Client(object): @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): """ @@ -247,7 +243,7 @@ class Client(object): """ Return boolean indicating whether the job is complete. """ - if response == None: + if response is None: response = self.raw_check_complete() return response["complete"] == "true" @@ -294,8 +290,9 @@ class InputCachingClient(Client): Beta client that cache's staged files to prevent duplication. """ - def __init__(self, destination_params, job_id, client_manager): - super(InputCachingClient, self).__init__(destination_params, job_id, client_manager) + 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): @@ -304,10 +301,10 @@ class InputCachingClient(Client): input_path = None return self._raw_execute(action, args, contents, input_path) else: - event_holder = self.client_manager.event_manager.acquire_event(input_path) + event_holder = self.client_cacher.acquire_event(input_path) cache_required = self.cache_required(input_path) if cache_required: - self.client_manager.queue_transfer(self, input_path) + self.client_cacher.queue_transfer(self, input_path) while not event_holder.failed: available = self.file_available(input_path) if available['ready']: diff --git a/lib/galaxy/jobs/runners/lwr_client/destination.py b/lib/galaxy/jobs/runners/lwr_client/destination.py index 5c44ce76d10..6054b6e98fd 100644 --- a/lib/galaxy/jobs/runners/lwr_client/destination.py +++ b/lib/galaxy/jobs/runners/lwr_client/destination.py @@ -55,5 +55,5 @@ def submit_params(destination_params): [('native_specification', '-q batch')] """ return dict([(key[len(SUBMIT_PREFIX):], value) - for key, value in (destination_params or {}).iteritems() - if key.startswith(SUBMIT_PREFIX)]) + 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 index cb74c602233..2a1d8c0693e 100644 --- a/lib/galaxy/jobs/runners/lwr_client/manager.py +++ b/lib/galaxy/jobs/runners/lwr_client/manager.py @@ -1,13 +1,18 @@ +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__) @@ -21,21 +26,129 @@ class ClientManager(object): state between multiple client connections. """ def __init__(self, **kwds): - transport_type = kwds.get('transport_type', None) - self.transport = get_transport(transport_type) - self.event_manager = TransferEventManager() + 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 - 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) + 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: @@ -67,12 +180,6 @@ class ClientManager(object): t.daemon = True t.start() - def queue_transfer(self, client, path): - self.transfer_queue.put((client, path)) - - def get_client(self, destination_params, job_id): - return self.client_class(destination_params, job_id, self) - def _environ_default_int(variable, default="0"): val = getenv(variable, default) @@ -80,3 +187,5 @@ def _environ_default_int(variable, default="0"): 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 626f710793c..2ceb30118f6 100644 --- a/lib/galaxy/jobs/runners/lwr_client/stager.py +++ b/lib/galaxy/jobs/runners/lwr_client/stager.py @@ -314,7 +314,10 @@ def finish_job(client, cleanup_job, job_completed_normally, working_directory, w 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: