diff --git a/lib/galaxy/jobs/runners/lwr_client/action_mapper.py b/lib/galaxy/jobs/runners/lwr_client/action_mapper.py index 68255bbb7f2..a5096b45c68 100644 --- a/lib/galaxy/jobs/runners/lwr_client/action_mapper.py +++ b/lib/galaxy/jobs/runners/lwr_client/action_mapper.py @@ -1,4 +1,6 @@ from json import load +from os import makedirs +from os.path import exists from os.path import abspath from os.path import dirname from os.path import join @@ -11,6 +13,9 @@ import galaxy.util from galaxy.util.bunch import Bunch from .util import directory_files from .util import unique_path_prefix +from .transport import get_file +from .transport import post_file + DEFAULT_MAPPED_ACTION = 'transfer' # Not really clear to me what this should be, exception? DEFAULT_PATH_MAPPER_TYPE = 'prefix' @@ -67,7 +72,7 @@ class FileActionMapper(object): ... f = NamedTemporaryFile(delete=False) ... f.write(config_contents.encode('UTF-8')) ... f.close() - ... mock_client = Bunch(default_file_action=default_action, action_config_path=f.name) + ... mock_client = Bunch(default_file_action=default_action, action_config_path=f.name, files_endpoint=None) ... mapper = FileActionMapper(mock_client) ... mapper = FileActionMapper(config=mapper.to_dict()) # Serialize and deserialize it to make sure still works ... unlink(f.name) @@ -125,10 +130,12 @@ class FileActionMapper(object): config = self.__client_to_config(client) self.default_action = config.get("default_action", "transfer") self.mappers = mappers_from_dicts(config.get("paths", [])) + self.files_endpoint = config.get("files_endpoint", None) def to_dict(self): return dict( default_action=self.default_action, + files_endpoint=self.files_endpoint, paths=map(lambda m: m.to_dict(), self.mappers) ) @@ -139,6 +146,7 @@ class FileActionMapper(object): else: config = dict() config["default_action"] = client.default_file_action + config["files_endpoint"] = client.files_endpoint return config def __load_action_config(self, path): @@ -166,7 +174,9 @@ class FileActionMapper(object): message_template = "Unknown action_type encountered %s while trying to map path %s" message_args = (action_type, path) raise Exception(message_template % message_args) - return action_class(path, file_lister=file_lister) + action = action_class(path, file_lister=file_lister) + self.__process_action(action, type) + return action def unstructured_mappers(self): """ Return mappers that will map 'unstructured' files (i.e. go beyond @@ -174,6 +184,20 @@ class FileActionMapper(object): """ return filter(lambda m: path_type.UNSTRUCTURED in m.path_types, self.mappers) + def __process_action(self, action, file_type): + """ Extension point to populate extra action information after an + action has been created. + """ + if action.action_type == "remote_transfer": + url_base = self.files_endpoint + if not url_base: + raise Exception("Attempted to use remote_transfer action with defining a files_endpoint") + if "?" not in url_base: + url_base = "%s?" % url_base + # TODO: URL encode path. + url = "%s&path=%s&file_type=%s" % (url_base, action.path, file_type) + action.url = url + class BaseAction(object): @@ -233,7 +257,7 @@ class RemoteCopyAction(BaseAction): staging = STAGING_ACTION_REMOTE def to_dict(self): - return dict(path=self.path, action_type=RemoteCopyAction.action_type) + return dict(path=self.path, action_type=self.action_type) @classmethod def from_dict(cls, action_dict): @@ -242,6 +266,41 @@ class RemoteCopyAction(BaseAction): def write_to_path(self, path): galaxy.util.copy_to_path(open(self.path, "rb"), path) + def write_from_path(self, lwr_path): + destination = self.path + parent_directory = dirname(destination) + if not exists(parent_directory): + makedirs(parent_directory) + with open(lwr_path, "rb") as f: + galaxy.util.copy_to_path(f, destination) + + +class RemoteTransferAction(BaseAction): + """ This action indicates the LWR server should copy the file before + execution via direct file system copy. This is like a CopyAction, but + it indicates the action should occur on the LWR server instead of on + the client. + """ + action_type = "remote_transfer" + staging = STAGING_ACTION_REMOTE + + def __init__(self, path, file_lister=None, url=None): + super(RemoteTransferAction, self).__init__(path, file_lister=file_lister) + self.url = url + + def to_dict(self): + return dict(path=self.path, action_type=self.action_type, url=self.url) + + @classmethod + def from_dict(cls, action_dict): + return RemoteTransferAction(path=action_dict["path"], url=action_dict["url"]) + + def write_to_path(self, path): + get_file(self.url, path) + + def write_from_path(self, lwr_path): + post_file(self.url, lwr_path) + class MessageAction(object): """ Sort of pseudo action describing "files" store in memory and @@ -275,7 +334,8 @@ class MessageAction(object): def write_to_path(self, path): open(path, "w").write(self.contents) -DICTIFIABLE_ACTION_CLASSES = [RemoteCopyAction, MessageAction] + +DICTIFIABLE_ACTION_CLASSES = [RemoteCopyAction, RemoteTransferAction, MessageAction] def from_dict(action_dict): @@ -402,8 +462,20 @@ class FileLister(object): DEFAULT_FILE_LISTER = FileLister(dict(depth=0)) -ACTION_CLASSES = [NoneAction, TransferAction, CopyAction, RemoteCopyAction] +ACTION_CLASSES = [ + NoneAction, + TransferAction, + CopyAction, + RemoteCopyAction, + RemoteTransferAction +] actions = dict([(clazz.action_type, clazz) for clazz in ACTION_CLASSES]) -__all__ = [FileActionMapper, path_type, from_dict, MessageAction] +__all__ = [ + FileActionMapper, + path_type, + from_dict, + MessageAction, + RemoteTransferAction, # For testing +] diff --git a/lib/galaxy/jobs/runners/lwr_client/client.py b/lib/galaxy/jobs/runners/lwr_client/client.py index 1ab21f8de35..fd5a65433b8 100644 --- a/lib/galaxy/jobs/runners/lwr_client/client.py +++ b/lib/galaxy/jobs/runners/lwr_client/client.py @@ -39,6 +39,7 @@ class BaseJobClient(object): ) else: job_directory = None + self.files_endpoint = destination_params.get("files_endpoint", None) self.job_directory = job_directory self.default_file_action = self.destination_params.get("default_file_action", "transfer") diff --git a/lib/galaxy/jobs/runners/lwr_client/manager.py b/lib/galaxy/jobs/runners/lwr_client/manager.py index c20c333d81c..9052d2d9389 100644 --- a/lib/galaxy/jobs/runners/lwr_client/manager.py +++ b/lib/galaxy/jobs/runners/lwr_client/manager.py @@ -59,8 +59,9 @@ class ClientManager(object): self.client_class = JobClient self.extra_client_kwds = {} - def get_client(self, destination_params, job_id): + def get_client(self, destination_params, job_id, **kwargs): destination_params = _parse_destination_params(destination_params) + destination_params.update(**kwargs) 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) @@ -112,8 +113,9 @@ class MessageQueueClientManager(object): def __nonzero__(self): return self.active - def get_client(self, destination_params, job_id): + def get_client(self, destination_params, job_id, **kwargs): destination_params = _parse_destination_params(destination_params) + destination_params.update(**kwargs) return MessageJobClient(destination_params, job_id, self) diff --git a/lib/galaxy/jobs/runners/lwr_client/staging/up.py b/lib/galaxy/jobs/runners/lwr_client/staging/up.py index 23f678e2b03..251ee65c566 100644 --- a/lib/galaxy/jobs/runners/lwr_client/staging/up.py +++ b/lib/galaxy/jobs/runners/lwr_client/staging/up.py @@ -381,7 +381,7 @@ class TransferTracker(object): def register_rewrite(self, local_path, remote_path, type, force=False): action = self.__action(local_path, type) - if action.action_type in ['transfer', 'copy'] or force: + if action.staging_needed or force: self.file_renames[local_path] = remote_path def rewrite_input_paths(self): diff --git a/lib/galaxy/jobs/runners/lwr_client/transport/__init__.py b/lib/galaxy/jobs/runners/lwr_client/transport/__init__.py index 9071d7b7a8b..5e0a62ecf4d 100644 --- a/lib/galaxy/jobs/runners/lwr_client/transport/__init__.py +++ b/lib/galaxy/jobs/runners/lwr_client/transport/__init__.py @@ -23,4 +23,9 @@ def __get_transport_type(transport_type, os_module): transport_type = 'curl' return transport_type -__all__ = [get_transport] +# TODO: Provide urllib implementation if these unavailable, +# also explore a requests+poster option. +from .curl import get_file +from .curl import post_file + +__all__ = [get_transport, get_file, post_file] diff --git a/lib/galaxy/jobs/runners/lwr_client/transport/curl.py b/lib/galaxy/jobs/runners/lwr_client/transport/curl.py index 72ee1d92d59..99f6788f488 100644 --- a/lib/galaxy/jobs/runners/lwr_client/transport/curl.py +++ b/lib/galaxy/jobs/runners/lwr_client/transport/curl.py @@ -4,7 +4,7 @@ except ImportError: from io import StringIO try: from pycurl import Curl -except: +except ImportError: pass from os.path import getsize @@ -16,9 +16,9 @@ PYCURL_UNAVAILABLE_MESSAGE = \ class PycurlTransport(object): def execute(self, url, data=None, input_path=None, output_path=None): - buf = self._open_output(output_path) + buf = _open_output(output_path) try: - c = self._new_curl_object() + c = _new_curl_object() c.setopt(c.URL, url.encode('ascii')) c.setopt(c.WRITEFUNCTION, buf.write) if input_path: @@ -37,11 +37,33 @@ class PycurlTransport(object): finally: buf.close() - def _new_curl_object(self): - try: - return Curl() - except NameError: - raise ImportError(PYCURL_UNAVAILABLE_MESSAGE) - def _open_output(self, output_path): - return open(output_path, 'wb') if output_path else StringIO() +def post_file(url, path): + c = _new_curl_object() + c.setopt(c.URL, url.encode('ascii')) + c.setopt(c.HTTPPOST, [("file", (c.FORM_FILE, path.encode('ascii')))]) + c.perform() + + +def get_file(url, path): + buf = _open_output(path) + try: + c = _new_curl_object() + c.setopt(c.URL, url.encode('ascii')) + c.setopt(c.WRITEFUNCTION, buf.write) + c.perform() + finally: + buf.close() + + +def _open_output(output_path): + return open(output_path, 'wb') if output_path else StringIO() + + +def _new_curl_object(): + try: + return Curl() + except NameError: + raise ImportError(PYCURL_UNAVAILABLE_MESSAGE) + +___all__ = [PycurlTransport, post_file, get_file]