diff --git a/client/galaxy/scripts/mvc/upload/upload-row.js b/client/galaxy/scripts/mvc/upload/upload-row.js index fde0545dd0f..5889d3fec57 100644 --- a/client/galaxy/scripts/mvc/upload/upload-row.js +++ b/client/galaxy/scripts/mvc/upload/upload-row.js @@ -118,7 +118,7 @@ return Backbone.View.extend({ .on('mousedown', function(e) { e.preventDefault(); }); // handle text editing event - it.find('#text-content').on('keyup', function(e) { + it.find('#text-content').on('change input', function(e) { self.model.set('url_paste', $(e.target).val()); self.model.set('file_size', $(e.target).val().length); }); diff --git a/config/job_conf.xml.sample_advanced b/config/job_conf.xml.sample_advanced index 444b7d2787e..08df8099af7 100644 --- a/config/job_conf.xml.sample_advanced +++ b/config/job_conf.xml.sample_advanced @@ -54,6 +54,19 @@ galaxy_infrastructure_url is set in galaxy.ini. --> http://localhost:8080 + + + + + + diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py index edd816a89cd..0acac1f0edc 100644 --- a/lib/galaxy/jobs/__init__.py +++ b/lib/galaxy/jobs/__init__.py @@ -1036,6 +1036,10 @@ class JobWrapper( object ): def change_state( self, state, info=False ): job = self.get_job() self.sa_session.refresh( job ) + if job.state in model.Job.terminal_states: + log.warning( "(%s) Ignoring state change from '%s' to '%s' for job " + "that is already terminal", job.id, job.state, state ) + return for dataset_assoc in job.output_datasets + job.output_library_datasets: dataset = dataset_assoc.dataset self.sa_session.refresh( dataset ) diff --git a/lib/galaxy/jobs/runners/pulsar.py b/lib/galaxy/jobs/runners/pulsar.py index 8067b3528cb..eaf24a0903a 100644 --- a/lib/galaxy/jobs/runners/pulsar.py +++ b/lib/galaxy/jobs/runners/pulsar.py @@ -61,6 +61,14 @@ PULSAR_PARAM_SPECS = dict( map=specs.to_str_or_none, default=None, ), + persistence_directory=dict( + map=specs.to_str_or_none, + default=None, + ), + amqp_acknowledge=dict( + map=specs.to_bool_or_none, + default=None + ), amqp_consumer_timeout=dict( map=lambda val: None if val == "None" else float(val), default=None, @@ -145,7 +153,7 @@ class PulsarJobRunner( AsynchronousJobRunner ): def __init_client_manager( self ): client_manager_kwargs = {} - for kwd in 'manager', 'cache', 'transport': + for kwd in 'manager', 'cache', 'transport', 'persistence_directory': client_manager_kwargs[ kwd ] = self.runner_params[ kwd ] for kwd in self.runner_params.keys(): if kwd.startswith( 'amqp_' ): @@ -611,6 +619,7 @@ class PulsarMQJobRunner( PulsarJobRunner ): # This is a message queue driven runner, don't monitor # just setup required callback. self.client_manager.ensure_has_status_update_callback(self.__async_update) + self.client_manager.ensure_has_ack_consumers() def __async_update( self, full_status ): job_id = None diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py index 44c5d6c0fed..2bd9788899f 100644 --- a/lib/galaxy/model/__init__.py +++ b/lib/galaxy/model/__init__.py @@ -317,6 +317,9 @@ class Job( object, HasJobMetrics, Dictifiable ): PAUSED = 'paused', DELETED = 'deleted', DELETED_NEW = 'deleted_new' ) + terminal_states = [ states.OK, + states.ERROR, + states.DELETED ] # Please include an accessor (get/set pair) for any new columns/members. def __init__( self ): self.session_id = None diff --git a/lib/galaxy/webapps/tool_shed/controllers/admin.py b/lib/galaxy/webapps/tool_shed/controllers/admin.py index 12f3ded8c5c..1008ba96fcc 100644 --- a/lib/galaxy/webapps/tool_shed/controllers/admin.py +++ b/lib/galaxy/webapps/tool_shed/controllers/admin.py @@ -222,6 +222,7 @@ class AdminController( BaseUIController, Admin ): @web.expose @web.require_admin def edit_category( self, trans, **kwd ): + '''Handle requests to edit TS category name or description''' message = escape( kwd.get( 'message', '' ) ) status = kwd.get( 'status', 'done' ) id = kwd.get( 'id', None ) @@ -248,19 +249,19 @@ class AdminController( BaseUIController, Admin ): else: category.name = new_name flush_needed = True - if original_category_description != new_description: - category.description = new_description - if not flush_needed: - flush_needed = True - if flush_needed: - trans.sa_session.add( category ) - trans.sa_session.flush() - if original_category_name != new_name: - # Update the Tool Shed's repository registry. - trans.app.repository_registry.edit_category_entry( original_category_name, new_name ) - message = "The information has been saved for category '%s'" % escape( category.name ) - status = 'done' - return trans.response.send_redirect( web.url_for( controller='admin', + if original_category_description != new_description: + category.description = new_description + if not flush_needed: + flush_needed = True + if flush_needed: + trans.sa_session.add( category ) + trans.sa_session.flush() + if original_category_name != new_name: + # Update the Tool Shed's repository registry. + trans.app.repository_registry.edit_category_entry( original_category_name, new_name ) + message = "The information has been saved for category '%s'" % escape( category.name ) + status = 'done' + return trans.response.send_redirect( web.url_for( controller='admin', action='manage_categories', message=message, status=status ) ) diff --git a/lib/pulsar/client/action_mapper.py b/lib/pulsar/client/action_mapper.py index e81b4cce9fb..7d19eb55054 100644 --- a/lib/pulsar/client/action_mapper.py +++ b/lib/pulsar/client/action_mapper.py @@ -83,7 +83,9 @@ class FileActionMapper(object): ... f.close() ... 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 + ... as_dict = config=mapper.to_dict() + ... # print(as_dict["paths"]) + ... mapper = FileActionMapper(config=as_dict) # Serialize and deserialize it to make sure still works ... unlink(f.name) ... return mapper >>> mapper = mapper_for(default_action='none', config_contents=json_string) @@ -171,7 +173,7 @@ class FileActionMapper(object): ssh_user=self.ssh_user, ssh_port=self.ssh_port, ssh_host=self.ssh_host, - paths=map(lambda m: m.to_dict(), self.mappers) + paths=list(map(lambda m: m.to_dict(), self.mappers)) ) def __client_to_config(self, client): @@ -261,7 +263,7 @@ class BaseAction(object): if self.staging_needed: # To ensure uniqueness, prepend unique prefix to each name prefix = unique_path_prefix(self.path) - for path, name in unstructured_map.iteritems(): + for path, name in unstructured_map.items(): unstructured_map[path] = join(prefix, name) else: path_rewrites = {} @@ -280,6 +282,16 @@ class BaseAction(object): def staging_action_local(self): return self.staging == STAGING_ACTION_LOCAL + def to_dict(self): + return dict(action_type=self.action_type) + + def __str__(self): + as_dict = self.to_dict() + attribute_str = "" + for key, value in as_dict.items(): + attribute_str += "%s=%s" % (key, value) + return "FileAction[%s]" % attribute_str + class NoneAction(BaseAction): """ This action indicates the corresponding path does not require any @@ -446,7 +458,7 @@ class PubkeyAuthenticatedTransferAction(BaseAction): def __serialize_ssh_key(self): f = tempfile.NamedTemporaryFile(delete=False) if self.ssh_key is not None: - f.write(self.ssh_key) + f.write(self.ssh_key.encode("utf-8")) else: raise Exception("SSH_KEY not available") return f.name @@ -644,7 +656,7 @@ MAPPER_CLASS_DICT = dict(map(lambda c: (c.match_type, c), MAPPER_CLASSES)) def mappers_from_dicts(mapper_def_list): - return map(lambda m: _mappper_from_dict(m), mapper_def_list) + return list(map(lambda m: _mappper_from_dict(m), mapper_def_list)) def _mappper_from_dict(mapper_dict): diff --git a/lib/pulsar/client/amqp_exchange.py b/lib/pulsar/client/amqp_exchange.py index ae22c82f5af..31085583f1a 100644 --- a/lib/pulsar/client/amqp_exchange.py +++ b/lib/pulsar/client/amqp_exchange.py @@ -3,7 +3,7 @@ import uuid import socket import logging import threading -from time import sleep +from time import sleep, time try: import kombu @@ -26,6 +26,14 @@ DEFAULT_RECONNECT_CONSUMER_WAIT = 1 DEFAULT_HEARTBEAT_WAIT = 1 DEFAULT_HEARTBEAT_JOIN_TIMEOUT = 10 +ACK_QUEUE_SUFFIX = "_ack" +ACK_UUID_KEY = 'acknowledge_uuid' +ACK_QUEUE_KEY = 'acknowledge_queue' +ACK_UUID_RESPONSE_KEY = 'acknowledge_uuid_response' +ACK_FORCE_NOACK_KEY = 'force_noack' +DEFAULT_ACK_MANAGER_SLEEP = 15 +DEFAULT_REPUBLISH_TIME = 30 + class PulsarExchange(object): """ Utility for publishing and consuming structured Pulsar queues using kombu. @@ -47,6 +55,9 @@ class PulsarExchange(object): connect_ssl=None, timeout=DEFAULT_TIMEOUT, publish_kwds={}, + publish_uuid_store=None, + consume_uuid_store=None, + republish_time=DEFAULT_REPUBLISH_TIME, ): """ """ @@ -57,27 +68,46 @@ class PulsarExchange(object): self.__connect_ssl = connect_ssl self.__exchange = kombu.Exchange(DEFAULT_EXCHANGE_NAME, DEFAULT_EXCHANGE_TYPE) self.__timeout = timeout + self.__republish_time = republish_time # Be sure to log message publishing failures. if publish_kwds.get("retry", False): if "retry_policy" not in publish_kwds: publish_kwds["retry_policy"] = {} - if "errback" not in publish_kwds["retry_policy"]: - publish_kwds["retry_policy"]["errback"] = self.__publish_errback self.__publish_kwds = publish_kwds + self.publish_uuid_store = publish_uuid_store + self.consume_uuid_store = consume_uuid_store + self.publish_ack_lock = threading.Lock() + + @staticmethod + def __publish_errback(exc, interval, publish_log_prefix=""): + log.error("%sConnection error while publishing: %r", publish_log_prefix, exc, exc_info=1) + log.info("%sRetrying in %s seconds", publish_log_prefix, interval) @property def url(self): return self.__url + @property + def acks_enabled(self): + return self.publish_uuid_store is not None + def consume(self, queue_name, callback, check=True, connection_kwargs={}): queue = self.__queue(queue_name) log.debug("Consuming queue '%s'", queue) + callbacks = [self.__ack_callback] + if callback is not None: + callbacks.append(callback) while check: heartbeat_thread = None try: with self.connection(self.__url, heartbeat=DEFAULT_HEARTBEAT, **connection_kwargs) as connection: - with kombu.Consumer(connection, queues=[queue], callbacks=[callback], accept=['json']): + with kombu.Consumer(connection, queues=[queue], callbacks=callbacks, accept=['json']): heartbeat_thread = self.__start_heartbeat(queue_name, connection) + # Ack manager should sleep before checking for + # repbulishes, but if that changes, need to drain the + # queue once before the ack manager starts doing its + # thing + self.__start_ack_manager(queue_name) while check and connection.connected: try: connection.drain_events(timeout=self.__timeout) @@ -90,6 +120,41 @@ class PulsarExchange(object): raise log.info("Done consuming queue %s" % queue_name) + def __ack_callback(self, body, message): + if ACK_UUID_KEY in body: + # The consumer of a normal queue has received a message requiring + # acknowledgement + ack_uuid = body[ACK_UUID_KEY] + ack_queue = body[ACK_QUEUE_KEY] + response = {ACK_UUID_RESPONSE_KEY: ack_uuid} + log.debug('Acknowledging UUID %s on queue %s', ack_uuid, ack_queue) + self.publish(ack_queue, response) + if self.consume_uuid_store is None: + log.warning('Received an ack request (UUID: %s, response queue: ' + '%s) but ack UUID persistence is not enabled, check ' + 'your config', ack_uuid, ack_queue) + elif ack_uuid not in self.consume_uuid_store: + # This message has not been seen before, store the uuid so it + # is not operated on more than once + self.consume_uuid_store[ack_uuid] = time() + else: + # This message has been seen before, prevent downstream + # callbacks from processing normally by acknowledging it here, + # still send the ack reply + log.warning('Message with UUID %s on queue %s has already ' + 'been performed, skipping callback', ack_uuid, ack_queue) + message.ack() + elif ACK_UUID_RESPONSE_KEY in body: + # The consumer of an ack queue has received an ack, remove it from the store + ack_uuid = body[ACK_UUID_RESPONSE_KEY] + log.debug('Got acknowledgement for UUID %s, will remove from store', ack_uuid) + try: + with self.publish_ack_lock: + del self.publish_uuid_store[ack_uuid] + except KeyError: + log.warning('Cannot remove UUID %s from store, already removed', ack_uuid) + message.ack() + def __handle_io_error(self, exc, heartbeat_thread): # In testing, errno is None log.warning('Got %s, will retry: %s', exc.__class__.__name__, exc) @@ -120,6 +185,16 @@ class PulsarExchange(object): key = self.__queue_name(name) publish_log_prefix = self.__publish_log_prefex(transaction_uuid) log.debug("%sBegin publishing to key %s", publish_log_prefix, key) + if (self.acks_enabled and not name.endswith(ACK_QUEUE_SUFFIX) + and ACK_FORCE_NOACK_KEY not in payload): + # Publishing a message on a normal queue and it's not a republish + # (or explicitly forced do-not-ack), so add ack keys + ack_uuid = str(transaction_uuid) + ack_queue = name + ACK_QUEUE_SUFFIX + payload[ACK_UUID_KEY] = ack_uuid + payload[ACK_QUEUE_KEY] = ack_queue + self.publish_uuid_store[ack_uuid] = payload + log.debug('Requesting acknowledgement of UUID %s on queue %s', ack_uuid, ack_queue) with self.connection(self.__url) as connection: with pools.producers[connection].acquire() as producer: log.debug("%sHave producer for publishing to key %s", publish_log_prefix, key) @@ -134,21 +209,36 @@ class PulsarExchange(object): ) log.debug("%sPublished to key %s", publish_log_prefix, key) + def ack_manager(self, queue_name): + log.debug('Acknowledgement manager thread alive') + resubmit_queue = queue_name[:-len(ACK_QUEUE_SUFFIX)] + try: + while True: + sleep(DEFAULT_ACK_MANAGER_SLEEP) + with self.publish_ack_lock: + for unack_uuid in self.publish_uuid_store.keys(): + if self.publish_uuid_store.get_time(unack_uuid) < time() - self.__republish_time: + log.debug('UUID %s has not been acknowledged, republishing original message', unack_uuid) + payload = self.publish_uuid_store[unack_uuid] + payload[ACK_FORCE_NOACK_KEY] = True + self.publish(resubmit_queue, payload) + self.publish_uuid_store.set_time(unack_uuid) + except: + log.exception("Problem with acknowledgement manager, leaving ack_manager method in problematic state!") + raise + log.debug('Acknowledgedment manager thread exiting') + def __prepare_publish_kwds(self, publish_log_prefix): if "retry_policy" in self.__publish_kwds: publish_kwds = copy.deepcopy(self.__publish_kwds) def errback(exc, interval): - return self.__publish_errback(exc, interval, publish_log_prefix) + return PulsarExchange.__publish_errback(exc, interval, publish_log_prefix) publish_kwds["retry_policy"]["errback"] = errback else: publish_kwds = self.__publish_kwds return publish_kwds - def __publish_errback(self, exc, interval, publish_log_prefix=""): - log.error("%sConnection error while publishing: %r", publish_log_prefix, exc, exc_info=1) - log.info("%sRetrying in %s seconds", publish_log_prefix, interval) - def __publish_log_prefex(self, transaction_uuid=None): prefix = "" if transaction_uuid: @@ -182,3 +272,11 @@ class PulsarExchange(object): thread = threading.Thread(name=thread_name, target=self.heartbeat, args=(connection,)) thread.start() return thread + + def __start_ack_manager(self, queue_name): + if self.acks_enabled and queue_name.endswith(ACK_QUEUE_SUFFIX): + thread_name = "acknowledgement-manager-%s" % (self.__queue_name(queue_name)) + thread = threading.Thread(name=thread_name, target=self.ack_manager, args=(queue_name,)) + thread.daemon = True + thread.start() + return thread diff --git a/lib/pulsar/client/amqp_exchange_factory.py b/lib/pulsar/client/amqp_exchange_factory.py index df26f8f5853..72f917860c7 100644 --- a/lib/pulsar/client/amqp_exchange_factory.py +++ b/lib/pulsar/client/amqp_exchange_factory.py @@ -1,5 +1,5 @@ from .amqp_exchange import PulsarExchange -from .util import filter_destination_params +from .util import filter_destination_params, MessageQueueUUIDStore def get_exchange(url, manager_name, params): @@ -9,6 +9,8 @@ def get_exchange(url, manager_name, params): connect_ssl=connect_ssl, publish_kwds=parse_amqp_publish_kwds(params) ) + if params.get('amqp_acknowledge', False): + exchange_kwds.update(parse_ack_kwds(params, manager_name)) timeout = params.get('amqp_consumer_timeout', False) if timeout is not False: exchange_kwds['timeout'] = timeout @@ -39,3 +41,16 @@ def parse_amqp_publish_kwds(params): if retry_policy_params: all_publish_params["retry_policy"] = retry_policy_params return all_publish_params + + +def parse_ack_kwds(params, manager_name): + ack_params = {} + persistence_directory = params.get('persistence_directory', None) + if persistence_directory: + subdirs = ['amqp_ack-%s' % manager_name] + ack_params['publish_uuid_store'] = MessageQueueUUIDStore(persistence_directory, subdirs=subdirs + ['publish']) + ack_params['consume_uuid_store'] = MessageQueueUUIDStore(persistence_directory, subdirs=subdirs + ['consume']) + republish_time = params.get('amqp_ack_republish_time', None) + if republish_time: + ack_params['republish_time'] = int(republish_time) + return ack_params diff --git a/lib/pulsar/client/client.py b/lib/pulsar/client/client.py index 27bb22ab37b..d7f3439bac9 100644 --- a/lib/pulsar/client/client.py +++ b/lib/pulsar/client/client.py @@ -1,12 +1,14 @@ import os -from json import dumps -from json import loads + +from six import string_types from .destination import submit_params from .setup_handler import build as build_setup_handler from .job_directory import RemoteJobDirectory from .decorators import parseJson from .decorators import retry +from .util import json_dumps +from .util import json_loads from .util import copy from .util import ensure_directory from .util import to_base64_json @@ -109,19 +111,19 @@ class JobClient(BaseJobClient): launch_params = dict(command_line=command_line, job_id=self.job_id) submit_params_dict = submit_params(self.destination_params) if submit_params_dict: - launch_params['params'] = dumps(submit_params_dict) + launch_params['params'] = json_dumps(submit_params_dict) if dependencies_description: - launch_params['dependencies_description'] = dumps(dependencies_description.to_dict()) + launch_params['dependencies_description'] = json_dumps(dependencies_description.to_dict()) if env: - launch_params['env'] = dumps(env) + launch_params['env'] = json_dumps(env) if remote_staging: - launch_params['remote_staging'] = dumps(remote_staging) + launch_params['remote_staging'] = json_dumps(remote_staging) if job_config and self.setup_handler.local: # Setup not yet called, job properties were inferred from # destination arguments. Hence, must have Pulsar setup job # before queueing. setup_params = _setup_params_from_job_config(job_config) - launch_params["setup_params"] = dumps(setup_params) + launch_params["setup_params"] = json_dumps(setup_params) return self._raw_execute("submit", launch_params) def full_status(self): @@ -174,10 +176,12 @@ class JobClient(BaseJobClient): # action type == 'message' should either copy or transfer # depending on default not just fallback to transfer. if action_type in ['transfer', 'message']: + if isinstance(contents, string_types): + contents = contents.encode("utf-8") return self._upload_file(args, contents, input_path) elif action_type == 'copy': path_response = self._raw_execute('path', args) - pulsar_path = loads(path_response)['path'] + pulsar_path = json_loads(path_response)['path'] copy(path, pulsar_path) return {'path': pulsar_path} diff --git a/lib/pulsar/client/config_util.py b/lib/pulsar/client/config_util.py index ab5453d0c8f..21103bff746 100644 --- a/lib/pulsar/client/config_util.py +++ b/lib/pulsar/client/config_util.py @@ -1,12 +1,14 @@ """ Generic interface for reading YAML/INI/JSON config files into nested dictionaries. """ +import codecs try: from galaxy import eggs eggs.require('PyYAML') except Exception: # If not in Galaxy, ignore this. pass + try: import yaml except ImportError: @@ -70,8 +72,9 @@ def __read_ini(path): def __read_json(path): + reader = codecs.getreader("utf-8") with open(path, "rb") as f: - return json.load(f) + return json.load(reader(f)) EXT_READERS = { CONFIG_TYPE_JSON: __read_json, diff --git a/lib/pulsar/client/decorators.py b/lib/pulsar/client/decorators.py index 94035f5ba1b..659b5c633d4 100644 --- a/lib/pulsar/client/decorators.py +++ b/lib/pulsar/client/decorators.py @@ -1,5 +1,6 @@ import time -import json + +from .util import json_loads MAX_RETRY_COUNT = 5 RETRY_SLEEP_TIME = 0.1 @@ -10,7 +11,7 @@ class parseJson(object): def __call__(self, func): def replacement(*args, **kwargs): response = func(*args, **kwargs) - return json.loads(response) + return json_loads(response) return replacement diff --git a/lib/pulsar/client/interface.py b/lib/pulsar/client/interface.py index af13a02e4f9..9d625f72c60 100644 --- a/lib/pulsar/client/interface.py +++ b/lib/pulsar/client/interface.py @@ -2,10 +2,7 @@ from abc import ABCMeta from abc import abstractmethod from string import Template -try: - from StringIO import StringIO as BytesIO -except ImportError: - from io import BytesIO +from six import BytesIO try: from six import text_type except ImportError: @@ -145,7 +142,7 @@ class LocalPulsarInterface(PulsarInterface): def __build_body(self, data, input_path): if data is not None: - return BytesIO(data.encode('utf-8')) + return BytesIO(data) elif input_path is not None: return open(input_path, 'rb') else: diff --git a/lib/pulsar/client/manager.py b/lib/pulsar/client/manager.py index f403fd17250..ff4e4e9164a 100644 --- a/lib/pulsar/client/manager.py +++ b/lib/pulsar/client/manager.py @@ -88,9 +88,14 @@ class MessageQueueClientManager(object): self.status_cache = {} self.callback_lock = threading.Lock() self.callback_thread = None + self.ack_consumer_threads = {} self.active = True def callback_wrapper(self, callback, body, message): + if message.acknowledged: + log.info("Message is already acknowledged (by an upstream " + "callback?), Pulsar client will not handle this message") + return if not self.active: log.debug("Obtained update message for inactive client manager, attempting requeue.") try: @@ -117,9 +122,12 @@ class MessageQueueClientManager(object): try: self.exchange.consume("status_update", callback_wrapper, check=self) except Exception: - log.exception("Exception while handling status update messages, this shouldn't really happen. Handler should be restarted.") + log.exception("Exception while handling status update messages, " + "this shouldn't really happen. Handler should be " + "restarted.") finally: - log.debug("Leaving Pulsar client status update thread, no additional Pulsar updates will be processed.") + log.debug("Leaving Pulsar client status update thread, no " + "additional Pulsar updates will be processed.") def ensure_has_status_update_callback(self, callback): with self.callback_lock: @@ -136,14 +144,45 @@ class MessageQueueClientManager(object): thread.start() self.callback_thread = thread + def ack_consumer(self, queue_name): + try: + self.exchange.consume(queue_name + '_ack', None, check=self) + except Exception: + log.exception("Exception while handling %s acknowledgement " + "messages, this shouldn't really happen. Handler " + "should be restarted.", queue_name) + finally: + log.debug("Leaving Pulsar client %s acknowledgement thread, no " + "additional acknowledgements will be processed.", + queue_name) + + def ensure_has_ack_consumers(self): + with self.callback_lock: + for name in ('setup', 'kill'): + if name in self.ack_consumer_threads: + return + + run = functools.partial(self.ack_consumer, name) + thread = threading.Thread( + name="pulsar_client_%s_%s_ack" % (self.manager_name, name), + target=run + ) + thread.daemon = False # Lets not interrupt processing of this. + thread.start() + self.ack_consumer_threads[name] = thread + def shutdown(self, ensure_cleanup=False): self.active = False if ensure_cleanup: self.callback_thread.join() + for v in self.ack_consumer_threads.values(): + v.join() def __nonzero__(self): return self.active + __bool__ = __nonzero__ # Both needed Py2 v 3 + def get_client(self, destination_params, job_id, **kwargs): if job_id is None: raise Exception("Cannot generate Pulsar client for empty job_id.") diff --git a/lib/pulsar/client/staging/__init__.py b/lib/pulsar/client/staging/__init__.py index 7b0a47c3c9b..c121bf28be7 100644 --- a/lib/pulsar/client/staging/__init__.py +++ b/lib/pulsar/client/staging/__init__.py @@ -1,3 +1,4 @@ +import re from os.path import basename from os.path import join from os.path import dirname @@ -6,6 +7,7 @@ from os import sep from ..util import PathHelper COMMAND_VERSION_FILENAME = "COMMAND_VERSION" +DEFAULT_DYNAMIC_COLLECTION_PATTERN = [r"primary_.*|galaxy.json|metadata_.*|dataset_\d+\.dat|__instrument_.*|dataset_\d+_files.+"] class ClientJobDescription(object): @@ -50,12 +52,12 @@ class ClientJobDescription(object): def __init__( self, - tool, command_line, - config_files, - input_files, - client_outputs, - working_directory, + tool=None, + config_files=[], + input_files=[], + client_outputs=None, + working_directory=None, # More sensible default? dependencies_description=None, env=[], arbitrary_files=None, @@ -65,7 +67,7 @@ class ClientJobDescription(object): self.command_line = command_line self.config_files = config_files self.input_files = input_files - self.client_outputs = client_outputs + self.client_outputs = client_outputs or ClientOutputs() self.working_directory = working_directory self.dependencies_description = dependencies_description self.env = env @@ -95,18 +97,28 @@ class ClientOutputs(object): runner client. """ - def __init__(self, working_directory, output_files, work_dir_outputs=None, version_file=None): + def __init__( + self, + working_directory=None, + output_files=[], + work_dir_outputs=None, + version_file=None, + dynamic_outputs=None + ): self.working_directory = working_directory - self.work_dir_outputs = work_dir_outputs - self.output_files = output_files + self.work_dir_outputs = work_dir_outputs or [] + self.output_files = output_files or [] self.version_file = version_file + self.dynamic_outputs = dynamic_outputs or DEFAULT_DYNAMIC_COLLECTION_PATTERN + self.__dynamic_patterns = list(map(re.compile, self.dynamic_outputs)) def to_dict(self): return dict( working_directory=self.working_directory, work_dir_outputs=self.work_dir_outputs, output_files=self.output_files, - version_file=self.version_file + version_file=self.version_file, + dynamic_outputs=self.dynamic_outputs, ) @staticmethod @@ -116,8 +128,12 @@ class ClientOutputs(object): work_dir_outputs=config_dict.get('work_dir_outputs'), output_files=config_dict.get('output_files'), version_file=config_dict.get('version_file'), + dynamic_outputs=config_dict.get('dynamic_outputs'), ) + def dynamic_match(self, filename): + return any(map(lambda pattern: pattern.match(filename), self.__dynamic_patterns)) + class PulsarOutputs(object): """ Abstraction describing the output files PRODUCED by the remote Pulsar diff --git a/lib/pulsar/client/staging/down.py b/lib/pulsar/client/staging/down.py index cfb901d5eb0..37f2e4965c3 100644 --- a/lib/pulsar/client/staging/down.py +++ b/lib/pulsar/client/staging/down.py @@ -94,7 +94,7 @@ class ResultsCollector(object): if output_generated: self._attempt_collect_output('output', output_file) - for galaxy_path, pulsar in self.pulsar_outputs.output_extras(output_file).iteritems(): + for galaxy_path, pulsar in self.pulsar_outputs.output_extras(output_file).items(): self._attempt_collect_output('output', path=galaxy_path, name=pulsar) # else not output generated, do not attempt download. @@ -110,7 +110,8 @@ class ResultsCollector(object): for name in self.working_directory_contents: if name in self.downloaded_working_directory_files: continue - if COPY_FROM_WORKING_DIRECTORY_PATTERN.match(name): + if self.client_outputs.dynamic_match(name): + log.debug("collecting dynamic output %s" % name) output_file = join(working_directory, self.pulsar_outputs.path_helper.local_name(name)) if self._attempt_collect_output(output_type='output_workdir', path=output_file, name=name): self.downloaded_working_directory_files.append(name) @@ -128,6 +129,7 @@ class ResultsCollector(object): return collected def _collect_output(self, output_type, action, name): + log.info("collecting output %s with action %s" % (name, action)) return self.output_collector.collect_output(self, output_type, action, name) diff --git a/lib/pulsar/client/staging/up.py b/lib/pulsar/client/staging/up.py index 6fd01136aca..46ebe319290 100644 --- a/lib/pulsar/client/staging/up.py +++ b/lib/pulsar/client/staging/up.py @@ -65,9 +65,14 @@ class FileStager(object): self.config_files = client_job_description.config_files self.input_files = client_job_description.input_files self.output_files = client_job_description.output_files - self.tool_id = client_job_description.tool.id - self.tool_version = client_job_description.tool.version - self.tool_dir = abspath(client_job_description.tool.tool_dir) + if client_job_description.tool is not None: + self.tool_id = client_job_description.tool.id + self.tool_version = client_job_description.tool.version + self.tool_dir = abspath(client_job_description.tool.tool_dir) + else: + self.tool_id = None + self.tool_version = None + self.tool_dir = None self.working_directory = client_job_description.working_directory self.version_file = client_job_description.version_file self.arbitrary_files = client_job_description.arbitrary_files @@ -142,7 +147,7 @@ class FileStager(object): for path in paths: if path not in referenced_arbitrary_path_mappers: referenced_arbitrary_path_mappers[path] = mapper - for path, mapper in referenced_arbitrary_path_mappers.iteritems(): + for path, mapper in referenced_arbitrary_path_mappers.items(): action = self.action_mapper.action(path, path_type.UNSTRUCTURED, mapper) unstructured_map = action.unstructured_map(self.path_helper) self.arbitrary_files.update(unstructured_map) @@ -152,7 +157,7 @@ class FileStager(object): self.transfer_tracker.handle_transfer(referenced_tool_file, path_type.TOOL) def __upload_arbitrary_files(self): - for path, name in self.arbitrary_files.iteritems(): + for path, name in self.arbitrary_files.items(): self.transfer_tracker.handle_transfer(path, path_type.UNSTRUCTURED, name=name) def __upload_input_files(self): @@ -180,11 +185,17 @@ class FileStager(object): def __upload_working_directory_files(self): # Task manager stages files into working directory, these need to be # uploaded if present. - working_directory_files = listdir(self.working_directory) if exists(self.working_directory) else [] + working_directory_files = self.__working_directory_files() for working_directory_file in working_directory_files: path = join(self.working_directory, working_directory_file) self.transfer_tracker.handle_transfer(path, path_type.WORKDIR) + def __working_directory_files(self): + if self.working_directory and exists(self.working_directory): + return listdir(self.working_directory) + else: + return [] + def __initialize_version_file_rename(self): version_file = self.version_file if version_file: @@ -298,6 +309,9 @@ class JobInputs(object): Full path to directory to search. """ + if directory is None: + return [] + pattern = r"(%s%s\S+)" % (directory, sep) return self.find_pattern_references(pattern) diff --git a/lib/pulsar/client/transport/curl.py b/lib/pulsar/client/transport/curl.py index 983b9786f0b..d2f0f50cf52 100644 --- a/lib/pulsar/client/transport/curl.py +++ b/lib/pulsar/client/transport/curl.py @@ -1,7 +1,13 @@ +import logging + try: - from cStringIO import StringIO + from galaxy import eggs + eggs.require("six") except ImportError: - from io import StringIO + pass + +from six import string_types +from six import BytesIO try: from pycurl import Curl, HTTP_CODE @@ -19,6 +25,8 @@ NO_SUCH_FILE_MESSAGE = "Attempt to post file %s to URL %s, but file does not exi POST_FAILED_MESSAGE = "Failed to post_file properly for url %s, remote server returned status code of %s." GET_FAILED_MESSAGE = "Failed to get_file properly for url %s, remote server returned status code of %s." +log = logging.getLogger(__name__) + class PycurlTransport(object): @@ -36,7 +44,7 @@ class PycurlTransport(object): c.setopt(c.INFILESIZE, filesize) if data: c.setopt(c.POST, 1) - if type(data).__name__ == 'unicode': + if isinstance(data, string_types): data = data.encode('UTF-8') c.setopt(c.POSTFIELDS, data) c.perform() @@ -62,21 +70,31 @@ def post_file(url, path): def get_file(url, path): - buf = _open_output(path) + if path and os.path.exists(path): + buf = _open_output(path, 'ab') + size = os.path.getsize(path) + success_codes = (200, 206) + else: + buf = _open_output(path) + size = 0 + success_codes = (200,) try: c = _new_curl_object_for_url(url) c.setopt(c.WRITEFUNCTION, buf.write) + if size > 0: + log.info('transfer of %s will resume at %s bytes', url, size) + c.setopt(c.RESUME_FROM, size) c.perform() - status_code = c.getinfo(HTTP_CODE) - if int(status_code) != 200: + status_code = int(c.getinfo(HTTP_CODE)) + if status_code not in success_codes: message = GET_FAILED_MESSAGE % (url, status_code) raise Exception(message) finally: buf.close() -def _open_output(output_path): - return open(output_path, 'wb') if output_path else StringIO() +def _open_output(output_path, mode='wb'): + return open(output_path, mode) if output_path else BytesIO() def _new_curl_object_for_url(url): diff --git a/lib/pulsar/client/util.py b/lib/pulsar/client/util.py index 04908db6256..35696befdc1 100644 --- a/lib/pulsar/client/util.py +++ b/lib/pulsar/client/util.py @@ -1,19 +1,46 @@ +from functools import wraps from threading import Lock, Event from weakref import WeakValueDictionary from os import walk from os import curdir +from os import listdir +from os import makedirs +from os import unlink from os.path import relpath from os.path import join +from os.path import abspath +from os.path import exists +from errno import ENOENT, EEXIST import os.path import hashlib import shutil import json -import base64 +import sys + +from six import binary_type + +# Variant of base64 compat layer inspired by BSD code from Bcfg2 +# https://github.com/Bcfg2/bcfg2/blob/maint/src/lib/Bcfg2/Compat.py +if sys.version_info >= (3, 0): + from base64 import b64encode as _b64encode, b64decode as _b64decode + + @wraps(_b64encode) + def b64encode(val, **kwargs): + try: + return _b64encode(val, **kwargs) + except TypeError: + return _b64encode(val.encode('UTF-8'), **kwargs).decode('UTF-8') + + @wraps(_b64decode) + def b64decode(val, **kwargs): + return _b64decode(val.encode('UTF-8'), **kwargs).decode('UTF-8') +else: + from base64 import b64encode, b64decode def unique_path_prefix(path): m = hashlib.md5() - m.update(path) + m.update(path.encode('utf-8')) return m.hexdigest() @@ -75,15 +102,17 @@ def filter_destination_params(destination_params, prefix): def to_base64_json(data): """ - >>> x = from_base64_json(to_base64_json(dict(a=5))) - >>> x["a"] + >>> enc = to_base64_json(dict(a=5)) + >>> dec = from_base64_json(enc) + >>> dec["a"] 5 """ - return base64.b64encode(json.dumps(data)) + dumped = json_dumps(data) + return b64encode(dumped) def from_base64_json(data): - return json.loads(base64.b64decode(data)) + return json.loads(b64decode(data)) class PathHelper(object): @@ -175,3 +204,78 @@ class EventHolder(object): def fail(self): self.failed = True + + +def json_loads(obj): + if isinstance(obj, binary_type): + obj = obj.decode("utf-8") + return json.loads(obj) + + +def json_dumps(obj): + if isinstance(obj, binary_type): + obj = obj.decode("utf-8") + return json.dumps(obj, cls=ClientJsonEncoder) + + +class ClientJsonEncoder(json.JSONEncoder): + + def default(self, obj): + if isinstance(obj, binary_type): + return obj.decode("utf-8") + return json.JSONEncoder.default(self, obj) + + +class MessageQueueUUIDStore(object): + """Persistent dict-like object for persisting message queue UUIDs that are + awaiting acknowledgement or that have been operated on. + """ + + def __init__(self, persistence_directory, subdirs=None): + if subdirs is None: + subdirs = ['acknowledge_uuids'] + self.__store = abspath(join(persistence_directory, *subdirs)) + try: + makedirs(self.__store) + except (OSError, IOError) as exc: + if exc.errno != EEXIST: + raise + + def __path(self, item): + return join(self.__store, item) + + def __contains__(self, item): + return exists(self.__path(item)) + + def __setitem__(self, key, value): + open(self.__path(key), 'w').write(json.dumps(value)) + + def __getitem__(self, key): + return json.loads(open(self.__path(key)).read()) + + def __delitem__(self, key): + try: + unlink(self.__path(key)) + except (OSError, IOError) as exc: + if exc.errno == ENOENT: + raise KeyError(key) + raise + + def keys(self): + return iter(listdir(self.__store)) + + def get_time(self, key): + try: + return os.stat(self.__path(key)).st_mtime + except (OSError, IOError) as exc: + if exc.errno == ENOENT: + raise KeyError(key) + raise + + def set_time(self, key): + try: + os.utime(self.__path(key), None) + except (OSError, IOError) as exc: + if exc.errno == ENOENT: + raise KeyError(key) + raise diff --git a/static/maps/mvc/upload/upload-row.js.map b/static/maps/mvc/upload/upload-row.js.map index ed09a46fb9e..4923eeca7ee 100644 --- a/static/maps/mvc/upload/upload-row.js.map +++ b/static/maps/mvc/upload/upload-row.js.map @@ -1 +1 @@ -{"version":3,"file":"upload-row.js","sources":["../../../src/mvc/upload/upload-row.js"],"names":["define","Utils","UploadModel","UploadSettings","Popover","Select","Backbone","View","extend","options","padding","status_classes","init","queued","running","success","error","settings","select_genome","select_extension","initialize","app","this","self","model","Model","setElement","_template","it","$el","title","container","find","placement","default_genome","value","css","onchange","genome","set","updateGenome","data","list_genomes","default_extension","extension","updateExtension","list_extensions","on","_removeRow","e","showExtensionInfo","$","target","text","preventDefault","_showSettings","val","length","prop","_refreshPercentage","_refreshStatus","_refreshInfo","_refreshGenome","_refreshExtension","_refreshFileSize","remove","collection","render","file_name","get","file_size","file_mode","html","bytesToString","removeClass","addClass","width","height","show","prototype","apply","info","hide","percentage","parseInt","status","status_class","sy","enable","attr","disable","count","visible","empty","append","id"],"mappings":"AACAA,QAAQ,cACA,0BACA,6BACA,oBACA,oBAEA,SAAYC,EACAC,EACAC,EACAC,EACAC,GAIpB,MAAOC,UAASC,KAAKC,QAEjBC,SACIC,QAAU,GAIdC,gBACIC,KAAU,mCACVC,OAAU,oCACVC,QAAU,oCACVC,QAAU,iCACVC,MAAU,iDAIdC,SAAU,KAGVC,cAAgB,KAGhBC,iBAAmB,KAGnBC,WAAY,SAASC,EAAKZ,GAEtBa,KAAKD,IAAMA,CAGX,IAAIE,GAAOD,IAGXA,MAAKE,MAAQ,GAAItB,GAAYuB,MAAMhB,GAGnCa,KAAKI,WAAWJ,KAAKK,UAAUlB,GAG/B,IAAImB,GAAKN,KAAKO,GAGdP,MAAKL,SAAW,GAAIb,GAAQG,MACxBuB,MAAc,uBACdC,UAAcH,EAAGI,KAAK,aACtBC,UAAc,UAIlB,IAAIC,GAAiBZ,KAAKD,IAAIH,cAAciB,OAG5Cb,MAAKJ,cAAgB,GAAIb,GAAOE,MAC5B6B,IAAK,SACLC,SAAW,SAASC,GAChBf,EAAKC,MAAMe,IAAI,SAAUD,GACzBf,EAAKF,IAAImB,aAAaF,GAAQ,IAElCG,KAAMlB,EAAKF,IAAIqB,aACfX,UAAWH,EAAGI,KAAK,WACnBG,MAAOD,IAIXZ,KAAKE,MAAMe,IAAI,SAAUL,EAGzB,IAAIS,GAAoBrB,KAAKD,IAAIF,iBAAiBgB,OAGlDb,MAAKH,iBAAmB,GAAId,GAAOE,MAC/B6B,IAAK,YACLC,SAAW,SAASO,GAChBrB,EAAKC,MAAMe,IAAI,YAAaK,GAC5BrB,EAAKF,IAAIwB,gBAAgBD,GAAW,IAExCH,KAAMlB,EAAKF,IAAIyB,gBACff,UAAWH,EAAGI,KAAK,cACnBG,MAAOQ,IAIXrB,KAAKE,MAAMe,IAAI,YAAaI,GAO5Bf,EAAGI,KAAK,WAAWe,GAAG,QAAS,WAAaxB,EAAKyB,eAGjDpB,EAAGI,KAAK,mBAAmBe,GAAG,QAAU,SAASE,GAC7C1B,EAAKF,IAAI6B,mBACLrB,IAAcsB,EAAEF,EAAEG,QAClBtB,MAAcP,EAAKJ,iBAAiBkC,OACpCT,UAAcrB,EAAKJ,iBAAiBgB,YAEzCY,GAAG,YAAa,SAASE,GAAKA,EAAEK,mBAGnC1B,EAAGI,KAAK,aAAae,GAAG,QAAU,WAAcxB,EAAKgC,kBAChCR,GAAG,YAAa,SAASE,GAAKA,EAAEK,mBAGrD1B,EAAGI,KAAK,iBAAiBe,GAAG,QAAS,SAASE,GAC1C1B,EAAKC,MAAMe,IAAI,YAAaY,EAAEF,EAAEG,QAAQI,OACxCjC,EAAKC,MAAMe,IAAI,YAAaY,EAAEF,EAAEG,QAAQI,MAAMC,UAIlD7B,EAAGI,KAAK,kBAAkBe,GAAG,SAAU,SAASE,GAC5C1B,EAAKC,MAAMe,IAAI,gBAAiBY,EAAEF,EAAEG,QAAQM,KAAK,cAMrDpC,KAAKE,MAAMuB,GAAG,oBAAqB,WAC/BxB,EAAKoC,uBAETrC,KAAKE,MAAMuB,GAAG,gBAAiB,WAC3BxB,EAAKqC,mBAETtC,KAAKE,MAAMuB,GAAG,cAAe,WACzBxB,EAAKsC,iBAETvC,KAAKE,MAAMuB,GAAG,gBAAiB,WAC3BxB,EAAKuC,mBAETxC,KAAKE,MAAMuB,GAAG,mBAAoB,WAC9BxB,EAAKwC,sBAETzC,KAAKE,MAAMuB,GAAG,mBAAoB,WAC9BxB,EAAKyC,qBAET1C,KAAKE,MAAMuB,GAAG,SAAU,WACpBxB,EAAK0C,WAET3C,KAAKD,IAAI6C,WAAWnB,GAAG,QAAS,WAC5BxB,EAAK0C,YAKbE,OAAQ,WAEJ,GAAIC,GAAc9C,KAAKE,MAAM6C,IAAI,aAC7BC,EAAchD,KAAKE,MAAM6C,IAAI,aAC7BE,EAAcjD,KAAKE,MAAM6C,IAAI,aAG7BzC,EAAKN,KAAKO,GAcd,IAXAD,EAAGI,KAAK,UAAUwC,KAAKJ,GAGvBxC,EAAGI,KAAK,SAASwC,KAAKvE,EAAMwE,cAAeH,IAG3C1C,EAAGI,KAAK,SACH0C,cACAC,SAAS,QAGG,OAAbJ,EAAoB,CAEpB,GAAIlB,GAAOzB,EAAGI,KAAK,SAGftB,EAAUY,KAAKb,QAAQC,QAGvBkE,EAAQhD,EAAGgD,QAAU,EAAIlE,EACzBmE,EAASjD,EAAGiD,SAAWnE,CAG3B2C,GAAKjB,IAAI,QAASwC,EAAQ,MAC1BvB,EAAKjB,IAAI,MAAOyC,EAAS,MACzBjD,EAAGiD,OAAOA,EAASxB,EAAKwB,SAAW,EAAInE,GAGvC2C,EAAKyB,OAGLlD,EAAGI,KAAK,SAAS2C,SAAS,gBAIb,SAAbJ,GAEA3C,EAAGI,KAAK,SAAS2C,SAAS,gBAIb,OAAbJ,GAEA3C,EAAGI,KAAK,SAAS2C,SAAS,oBAKlCV,OAAQ,WAEJ3C,KAAKJ,cAAc+C,SACnB3C,KAAKH,iBAAiB8C,SAGtB3D,SAASC,KAAKwE,UAAUd,OAAOe,MAAM1D,OAQzCyC,kBAAmB,WACfzC,KAAKH,iBAAiBgB,MAAMb,KAAKE,MAAM6C,IAAI,eAI/CP,eAAgB,WACZxC,KAAKJ,cAAciB,MAAMb,KAAKE,MAAM6C,IAAI,YAI5CR,aAAc,WAEV,GAAIoB,GAAO3D,KAAKE,MAAM6C,IAAI,OACtBY,GACA3D,KAAKO,IAAIG,KAAK,SAASwC,KAAK,4BAA8BS,GAAMH,OAEhExD,KAAKO,IAAIG,KAAK,SAASkD,QAK/BvB,mBAAqB,WACjB,GAAIwB,GAAaC,SAAS9D,KAAKE,MAAM6C,IAAI,cACzC/C,MAAKO,IAAIG,KAAK,iBAAiBI,KAAMwC,MAAQO,EAAa,MAEtD7D,KAAKO,IAAIG,KAAK,eAAewC,KADf,KAAdW,EACkCA,EAAa,IAEb,yBAI1CvB,eAAiB,WAEb,GAAIhC,GAAKN,KAAKO,IAGVwD,EAAS/D,KAAKE,MAAM6C,IAAI,UACxBiB,EAAehE,KAAKX,eAAe0E,GAGnCE,EAAKjE,KAAKO,IAAIG,KAAK,UACvBuD,GAAGb,cAGHa,EAAGZ,SAASW,GAGE,QAAVD,GAEA/D,KAAKJ,cAAcsE,SACnBlE,KAAKH,iBAAiBqE,SAGtB5D,EAAGI,KAAK,iBAAiByD,KAAK,YAAY,GAC1C7D,EAAGI,KAAK,kBAAkByD,KAAK,YAAY,KAG3CnE,KAAKJ,cAAcwE,UACnBpE,KAAKH,iBAAiBuE,UAGtB9D,EAAGI,KAAK,iBAAiByD,KAAK,YAAY,GAC1C7D,EAAGI,KAAK,kBAAkByD,KAAK,YAAY,IAIjC,WAAVJ,IACAzD,EAAG+C,SAAS,WACZ/C,EAAGI,KAAK,eAAewC,KAAK,SAIlB,SAAVa,IACAzD,EAAG+C,SAAS,UACZ/C,EAAGI,KAAK,aAAaiC,WAK7BD,iBAAkB,WACd,GAAI2B,GAAQrE,KAAKE,MAAM6C,IAAI,YAC3B/C,MAAKO,IAAIG,KAAK,SAASwC,KAAKvE,EAAMwE,cAAekB,KAQrD3C,WAAY,WAER,GAAIqC,GAAS/D,KAAKE,MAAM6C,IAAI,WAGd,QAAVgB,GAA8B,WAAVA,GAAiC,SAAVA,IAC3C/D,KAAKD,IAAI6C,WAAWD,OAAO3C,KAAKE,QAKxC+B,cAAgB,WAEPjC,KAAKL,SAAS2E,QAOftE,KAAKL,SAASiE,QALd5D,KAAKL,SAAS4E,QACdvE,KAAKL,SAAS6E,OAAO,GAAK3F,GAAemB,MAAOO,KAChDP,KAAKL,SAAS6D,SAQtBnD,UAAW,SAASlB,GAChB,MAAQ,uBAAyBA,EAAQsF,GAAK,k4BA+BDzE,KAAKX,eAAeC,KAAO"} \ No newline at end of file +{"version":3,"file":"upload-row.js","sources":["../../../src/mvc/upload/upload-row.js"],"names":["define","Utils","UploadModel","UploadSettings","Popover","Select","Backbone","View","extend","options","padding","status_classes","init","queued","running","success","error","settings","select_genome","select_extension","initialize","app","this","self","model","Model","setElement","_template","it","$el","title","container","find","placement","default_genome","value","css","onchange","genome","set","updateGenome","data","list_genomes","default_extension","extension","updateExtension","list_extensions","on","_removeRow","e","showExtensionInfo","$","target","text","preventDefault","_showSettings","val","length","prop","_refreshPercentage","_refreshStatus","_refreshInfo","_refreshGenome","_refreshExtension","_refreshFileSize","remove","collection","render","file_name","get","file_size","file_mode","html","bytesToString","removeClass","addClass","width","height","show","prototype","apply","info","hide","percentage","parseInt","status","status_class","sy","enable","attr","disable","count","visible","empty","append","id"],"mappings":"AACAA,QAAQ,cACA,0BACA,6BACA,oBACA,oBAEA,SAAYC,EACAC,EACAC,EACAC,EACAC,GAIpB,MAAOC,UAASC,KAAKC,QAEjBC,SACIC,QAAU,GAIdC,gBACIC,KAAU,mCACVC,OAAU,oCACVC,QAAU,oCACVC,QAAU,iCACVC,MAAU,iDAIdC,SAAU,KAGVC,cAAgB,KAGhBC,iBAAmB,KAGnBC,WAAY,SAASC,EAAKZ,GAEtBa,KAAKD,IAAMA,CAGX,IAAIE,GAAOD,IAGXA,MAAKE,MAAQ,GAAItB,GAAYuB,MAAMhB,GAGnCa,KAAKI,WAAWJ,KAAKK,UAAUlB,GAG/B,IAAImB,GAAKN,KAAKO,GAGdP,MAAKL,SAAW,GAAIb,GAAQG,MACxBuB,MAAc,uBACdC,UAAcH,EAAGI,KAAK,aACtBC,UAAc,UAIlB,IAAIC,GAAiBZ,KAAKD,IAAIH,cAAciB,OAG5Cb,MAAKJ,cAAgB,GAAIb,GAAOE,MAC5B6B,IAAK,SACLC,SAAW,SAASC,GAChBf,EAAKC,MAAMe,IAAI,SAAUD,GACzBf,EAAKF,IAAImB,aAAaF,GAAQ,IAElCG,KAAMlB,EAAKF,IAAIqB,aACfX,UAAWH,EAAGI,KAAK,WACnBG,MAAOD,IAIXZ,KAAKE,MAAMe,IAAI,SAAUL,EAGzB,IAAIS,GAAoBrB,KAAKD,IAAIF,iBAAiBgB,OAGlDb,MAAKH,iBAAmB,GAAId,GAAOE,MAC/B6B,IAAK,YACLC,SAAW,SAASO,GAChBrB,EAAKC,MAAMe,IAAI,YAAaK,GAC5BrB,EAAKF,IAAIwB,gBAAgBD,GAAW,IAExCH,KAAMlB,EAAKF,IAAIyB,gBACff,UAAWH,EAAGI,KAAK,cACnBG,MAAOQ,IAIXrB,KAAKE,MAAMe,IAAI,YAAaI,GAO5Bf,EAAGI,KAAK,WAAWe,GAAG,QAAS,WAAaxB,EAAKyB,eAGjDpB,EAAGI,KAAK,mBAAmBe,GAAG,QAAU,SAASE,GAC7C1B,EAAKF,IAAI6B,mBACLrB,IAAcsB,EAAEF,EAAEG,QAClBtB,MAAcP,EAAKJ,iBAAiBkC,OACpCT,UAAcrB,EAAKJ,iBAAiBgB,YAEzCY,GAAG,YAAa,SAASE,GAAKA,EAAEK,mBAGnC1B,EAAGI,KAAK,aAAae,GAAG,QAAU,WAAcxB,EAAKgC,kBAChCR,GAAG,YAAa,SAASE,GAAKA,EAAEK,mBAGrD1B,EAAGI,KAAK,iBAAiBe,GAAG,eAAgB,SAASE,GACjD1B,EAAKC,MAAMe,IAAI,YAAaY,EAAEF,EAAEG,QAAQI,OACxCjC,EAAKC,MAAMe,IAAI,YAAaY,EAAEF,EAAEG,QAAQI,MAAMC,UAIlD7B,EAAGI,KAAK,kBAAkBe,GAAG,SAAU,SAASE,GAC5C1B,EAAKC,MAAMe,IAAI,gBAAiBY,EAAEF,EAAEG,QAAQM,KAAK,cAMrDpC,KAAKE,MAAMuB,GAAG,oBAAqB,WAC/BxB,EAAKoC,uBAETrC,KAAKE,MAAMuB,GAAG,gBAAiB,WAC3BxB,EAAKqC,mBAETtC,KAAKE,MAAMuB,GAAG,cAAe,WACzBxB,EAAKsC,iBAETvC,KAAKE,MAAMuB,GAAG,gBAAiB,WAC3BxB,EAAKuC,mBAETxC,KAAKE,MAAMuB,GAAG,mBAAoB,WAC9BxB,EAAKwC,sBAETzC,KAAKE,MAAMuB,GAAG,mBAAoB,WAC9BxB,EAAKyC,qBAET1C,KAAKE,MAAMuB,GAAG,SAAU,WACpBxB,EAAK0C,WAET3C,KAAKD,IAAI6C,WAAWnB,GAAG,QAAS,WAC5BxB,EAAK0C,YAKbE,OAAQ,WAEJ,GAAIC,GAAc9C,KAAKE,MAAM6C,IAAI,aAC7BC,EAAchD,KAAKE,MAAM6C,IAAI,aAC7BE,EAAcjD,KAAKE,MAAM6C,IAAI,aAG7BzC,EAAKN,KAAKO,GAcd,IAXAD,EAAGI,KAAK,UAAUwC,KAAKJ,GAGvBxC,EAAGI,KAAK,SAASwC,KAAKvE,EAAMwE,cAAeH,IAG3C1C,EAAGI,KAAK,SACH0C,cACAC,SAAS,QAGG,OAAbJ,EAAoB,CAEpB,GAAIlB,GAAOzB,EAAGI,KAAK,SAGftB,EAAUY,KAAKb,QAAQC,QAGvBkE,EAAQhD,EAAGgD,QAAU,EAAIlE,EACzBmE,EAASjD,EAAGiD,SAAWnE,CAG3B2C,GAAKjB,IAAI,QAASwC,EAAQ,MAC1BvB,EAAKjB,IAAI,MAAOyC,EAAS,MACzBjD,EAAGiD,OAAOA,EAASxB,EAAKwB,SAAW,EAAInE,GAGvC2C,EAAKyB,OAGLlD,EAAGI,KAAK,SAAS2C,SAAS,gBAIb,SAAbJ,GAEA3C,EAAGI,KAAK,SAAS2C,SAAS,gBAIb,OAAbJ,GAEA3C,EAAGI,KAAK,SAAS2C,SAAS,oBAKlCV,OAAQ,WAEJ3C,KAAKJ,cAAc+C,SACnB3C,KAAKH,iBAAiB8C,SAGtB3D,SAASC,KAAKwE,UAAUd,OAAOe,MAAM1D,OAQzCyC,kBAAmB,WACfzC,KAAKH,iBAAiBgB,MAAMb,KAAKE,MAAM6C,IAAI,eAI/CP,eAAgB,WACZxC,KAAKJ,cAAciB,MAAMb,KAAKE,MAAM6C,IAAI,YAI5CR,aAAc,WAEV,GAAIoB,GAAO3D,KAAKE,MAAM6C,IAAI,OACtBY,GACA3D,KAAKO,IAAIG,KAAK,SAASwC,KAAK,4BAA8BS,GAAMH,OAEhExD,KAAKO,IAAIG,KAAK,SAASkD,QAK/BvB,mBAAqB,WACjB,GAAIwB,GAAaC,SAAS9D,KAAKE,MAAM6C,IAAI,cACzC/C,MAAKO,IAAIG,KAAK,iBAAiBI,KAAMwC,MAAQO,EAAa,MAEtD7D,KAAKO,IAAIG,KAAK,eAAewC,KADf,KAAdW,EACkCA,EAAa,IAEb,yBAI1CvB,eAAiB,WAEb,GAAIhC,GAAKN,KAAKO,IAGVwD,EAAS/D,KAAKE,MAAM6C,IAAI,UACxBiB,EAAehE,KAAKX,eAAe0E,GAGnCE,EAAKjE,KAAKO,IAAIG,KAAK,UACvBuD,GAAGb,cAGHa,EAAGZ,SAASW,GAGE,QAAVD,GAEA/D,KAAKJ,cAAcsE,SACnBlE,KAAKH,iBAAiBqE,SAGtB5D,EAAGI,KAAK,iBAAiByD,KAAK,YAAY,GAC1C7D,EAAGI,KAAK,kBAAkByD,KAAK,YAAY,KAG3CnE,KAAKJ,cAAcwE,UACnBpE,KAAKH,iBAAiBuE,UAGtB9D,EAAGI,KAAK,iBAAiByD,KAAK,YAAY,GAC1C7D,EAAGI,KAAK,kBAAkByD,KAAK,YAAY,IAIjC,WAAVJ,IACAzD,EAAG+C,SAAS,WACZ/C,EAAGI,KAAK,eAAewC,KAAK,SAIlB,SAAVa,IACAzD,EAAG+C,SAAS,UACZ/C,EAAGI,KAAK,aAAaiC,WAK7BD,iBAAkB,WACd,GAAI2B,GAAQrE,KAAKE,MAAM6C,IAAI,YAC3B/C,MAAKO,IAAIG,KAAK,SAASwC,KAAKvE,EAAMwE,cAAekB,KAQrD3C,WAAY,WAER,GAAIqC,GAAS/D,KAAKE,MAAM6C,IAAI,WAGd,QAAVgB,GAA8B,WAAVA,GAAiC,SAAVA,IAC3C/D,KAAKD,IAAI6C,WAAWD,OAAO3C,KAAKE,QAKxC+B,cAAgB,WAEPjC,KAAKL,SAAS2E,QAOftE,KAAKL,SAASiE,QALd5D,KAAKL,SAAS4E,QACdvE,KAAKL,SAAS6E,OAAO,GAAK3F,GAAemB,MAAOO,KAChDP,KAAKL,SAAS6D,SAQtBnD,UAAW,SAASlB,GAChB,MAAQ,uBAAyBA,EAAQsF,GAAK,k4BA+BDzE,KAAKX,eAAeC,KAAO"} \ No newline at end of file diff --git a/static/scripts/mvc/upload/upload-row.js b/static/scripts/mvc/upload/upload-row.js index 2f77c021d5b..823daa9ad71 100644 --- a/static/scripts/mvc/upload/upload-row.js +++ b/static/scripts/mvc/upload/upload-row.js @@ -1,2 +1,2 @@ -define(["utils/utils","mvc/upload/upload-model","mvc/upload/upload-settings","mvc/ui/ui-popover","mvc/ui/ui-select"],function(a,b,c,d,e){return Backbone.View.extend({options:{padding:8},status_classes:{init:"upload-icon-button fa fa-trash-o",queued:"upload-icon fa fa-spinner fa-spin",running:"upload-icon fa fa-spinner fa-spin",success:"upload-icon-button fa fa-check",error:"upload-icon-button fa fa-exclamation-triangle"},settings:null,select_genome:null,select_extension:null,initialize:function(a,c){this.app=a;var f=this;this.model=new b.Model(c),this.setElement(this._template(c));var g=this.$el;this.settings=new d.View({title:"Upload configuration",container:g.find("#settings"),placement:"bottom"});var h=this.app.select_genome.value();this.select_genome=new e.View({css:"genome",onchange:function(a){f.model.set("genome",a),f.app.updateGenome(a,!0)},data:f.app.list_genomes,container:g.find("#genome"),value:h}),this.model.set("genome",h);var i=this.app.select_extension.value();this.select_extension=new e.View({css:"extension",onchange:function(a){f.model.set("extension",a),f.app.updateExtension(a,!0)},data:f.app.list_extensions,container:g.find("#extension"),value:i}),this.model.set("extension",i),g.find("#symbol").on("click",function(){f._removeRow()}),g.find("#extension-info").on("click",function(a){f.app.showExtensionInfo({$el:$(a.target),title:f.select_extension.text(),extension:f.select_extension.value()})}).on("mousedown",function(a){a.preventDefault()}),g.find("#settings").on("click",function(){f._showSettings()}).on("mousedown",function(a){a.preventDefault()}),g.find("#text-content").on("keyup",function(a){f.model.set("url_paste",$(a.target).val()),f.model.set("file_size",$(a.target).val().length)}),g.find("#space_to_tabs").on("change",function(a){f.model.set("space_to_tabs",$(a.target).prop("checked"))}),this.model.on("change:percentage",function(){f._refreshPercentage()}),this.model.on("change:status",function(){f._refreshStatus()}),this.model.on("change:info",function(){f._refreshInfo()}),this.model.on("change:genome",function(){f._refreshGenome()}),this.model.on("change:extension",function(){f._refreshExtension()}),this.model.on("change:file_size",function(){f._refreshFileSize()}),this.model.on("remove",function(){f.remove()}),this.app.collection.on("reset",function(){f.remove()})},render:function(){var b=this.model.get("file_name"),c=this.model.get("file_size"),d=this.model.get("file_mode"),e=this.$el;if(e.find("#title").html(b),e.find("#size").html(a.bytesToString(c)),e.find("#mode").removeClass().addClass("mode"),"new"==d){var f=e.find("#text"),g=this.options.padding,h=e.width()-2*g,i=e.height()-g;f.css("width",h+"px"),f.css("top",i+"px"),e.height(i+f.height()+2*g),f.show(),e.find("#mode").addClass("fa fa-pencil")}"local"==d&&e.find("#mode").addClass("fa fa-laptop"),"ftp"==d&&e.find("#mode").addClass("fa fa-code-fork")},remove:function(){this.select_genome.remove(),this.select_extension.remove(),Backbone.View.prototype.remove.apply(this)},_refreshExtension:function(){this.select_extension.value(this.model.get("extension"))},_refreshGenome:function(){this.select_genome.value(this.model.get("genome"))},_refreshInfo:function(){var a=this.model.get("info");a?this.$el.find("#info").html("Failed: "+a).show():this.$el.find("#info").hide()},_refreshPercentage:function(){var a=parseInt(this.model.get("percentage"));this.$el.find(".progress-bar").css({width:a+"%"}),this.$el.find("#percentage").html(100!=a?a+"%":"Adding to history...")},_refreshStatus:function(){var a=this.$el,b=this.model.get("status"),c=this.status_classes[b],d=this.$el.find("#symbol");d.removeClass(),d.addClass(c),"init"==b?(this.select_genome.enable(),this.select_extension.enable(),a.find("#text-content").attr("disabled",!1),a.find("#space_to_tabs").attr("disabled",!1)):(this.select_genome.disable(),this.select_extension.disable(),a.find("#text-content").attr("disabled",!0),a.find("#space_to_tabs").attr("disabled",!0)),"success"==b&&(a.addClass("success"),a.find("#percentage").html("100%")),"error"==b&&(a.addClass("danger"),a.find(".progress").remove())},_refreshFileSize:function(){var b=this.model.get("file_size");this.$el.find("#size").html(a.bytesToString(b))},_removeRow:function(){var a=this.model.get("status");("init"==a||"success"==a||"error"==a)&&this.app.collection.remove(this.model)},_showSettings:function(){this.settings.visible?this.settings.hide():(this.settings.empty(),this.settings.append(new c(this).$el),this.settings.show())},_template:function(a){return'
You can tell Galaxy to download data from web by entering URL in this box (one per line). You can also directly paste the contents of a file.