diff --git a/lib/galaxy/app.py b/lib/galaxy/app.py index 8871407a942..c2005ac1329 100644 --- a/lib/galaxy/app.py +++ b/lib/galaxy/app.py @@ -212,6 +212,11 @@ class UniverseApplication(config.ConfiguresGalaxyMixin): # Must be initialized after job_config. self.workflow_scheduling_manager = scheduling_manager.WorkflowSchedulingManager(self) + # Must be initialized after any component that might make use of stack messaging is configured. Alternatively if + # it becomes more commonly needed we could create a prefork function registration method like we do with + # postfork functions. + self.application_stack.init_late_prefork() + self.containers = {} if self.config.enable_beta_containers_interface: self.containers = build_container_interfaces( diff --git a/lib/galaxy/config.py b/lib/galaxy/config.py index 4f9b628a5ac..13003a7c2c5 100644 --- a/lib/galaxy/config.py +++ b/lib/galaxy/config.py @@ -154,7 +154,7 @@ def find_root(kwargs): class Configuration(object): - deprecated_options = ('database_file', ) + deprecated_options = ('database_file', 'track_jobs_in_database') def __init__(self, **kwargs): self.config_dict = kwargs diff --git a/lib/galaxy/exceptions/__init__.py b/lib/galaxy/exceptions/__init__.py index db59cd80d74..be9a1d5dee1 100644 --- a/lib/galaxy/exceptions/__init__.py +++ b/lib/galaxy/exceptions/__init__.py @@ -185,6 +185,15 @@ class InternalServerError(MessageException): err_code = error_codes.INTERNAL_SERVER_ERROR +class ToolExecutionError(MessageException): + status_code = 500 + err_code = error_codes.TOOL_EXECUTION_ERROR + + def __init__(self, err_msg, type="error", job=None): + super(ToolExecutionError, self).__init__(err_msg, type) + self.job = job + + class NotImplemented(MessageException): status_code = 501 err_code = error_codes.NOT_IMPLEMENTED @@ -221,3 +230,9 @@ class ContainerRunError(Exception): super(ContainerRunError, self).__init__(msg, **kwargs) self.image = image self.command = command + + +class HandlerAssignmentError(Exception): + def __init__(self, msg=None, obj=None, **kwargs): + super(HandlerAssignmentError, self).__init__(msg, **kwargs) + self.obj = obj diff --git a/lib/galaxy/exceptions/error_codes.json b/lib/galaxy/exceptions/error_codes.json index aa790ec16b8..11908515a1d 100644 --- a/lib/galaxy/exceptions/error_codes.json +++ b/lib/galaxy/exceptions/error_codes.json @@ -144,6 +144,11 @@ "code": 500003, "message": "Error in a configuration file." }, + { + "name": "TOOL_EXECUTION_ERROR", + "code": 500004, + "message": "Tool execution failed due to an internal server error." + }, { "name": "NOT_IMPLEMENTED", "code": 501001, diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py index 3b2dee083f4..e4b92978113 100644 --- a/lib/galaxy/jobs/__init__.py +++ b/lib/galaxy/jobs/__init__.py @@ -115,6 +115,8 @@ class JobConfiguration(ConfiguresHandlers): These features are configured in the job configuration, by default, ``job_conf.xml`` """ + DEFAULT_BASE_HANDLER_POOLS = ('job-handlers',) + DEFAULT_NWORKERS = 4 JOB_RESOURCE_CONDITIONAL_XML = """ @@ -135,6 +137,8 @@ class JobConfiguration(ConfiguresHandlers): self.handlers = {} self.handler_runner_plugins = {} self.default_handler_id = None + self.handler_assignment_methods = None + self.handler_assignment_methods_configured = False self.destinations = {} self.destination_tags = {} self.default_destination_id = None @@ -150,7 +154,6 @@ class JobConfiguration(ConfiguresHandlers): output_size=None, destination_user_concurrent_jobs={}, destination_total_concurrent_jobs={}) - self._is_handler = None default_resubmits = [] default_resubmit_condition = self.app.config.default_job_resubmission_condition @@ -212,23 +215,12 @@ class JobConfiguration(ConfiguresHandlers): # Parse handlers handlers_conf = root.find('handlers') + self._init_handler_assignment_methods(handlers_conf) self._init_handlers(handlers_conf) - - # Determine the default handler(s) - try: - self.default_handler_id = self._get_default(self.app.config, handlers_conf, list(self.handlers.keys())) - except Exception: - pass - # For tets, this may not exist - try: - base_server_name = self.app.config.base_server_name - except AttributeError: - base_server_name = self.app.config.get('base_server_name', None) - if (self.default_handler_id is None - or (len(self.handlers) == 1 and base_server_name == next(iter(self.handlers.keys())))): - # Shortcut for compatibility with existing job confs that use the default handlers block, - # there are no defined handlers, or there's only one handler and it's this server - self.__set_default_job_handler() + self._set_default_handler_assignment_methods() + log.info("Job handler assignment methods set to: %s", ', '.join(self.handler_assignment_methods)) + for tag, handlers in [(t, h) for t, h in self.handlers.items() if isinstance(h, list)]: + log.info("Tag [%s] handlers: %s", tag, ', '.join(handlers)) # Parse destinations destinations = root.find('destinations') @@ -269,7 +261,8 @@ class JobConfiguration(ConfiguresHandlers): self.destinations[tag].append(job_destination) # Determine the default destination - self.default_destination_id = self._get_default(self.app.config, destinations, list(self.destinations.keys())) + self.default_destination_id = self._get_default( + self.app.config, destinations, list(self.destinations.keys()), auto=True) # Parse resources... resources = root.find('resources') @@ -346,32 +339,14 @@ class JobConfiguration(ConfiguresHandlers): if self.app.config.use_tasked_jobs: self.runner_plugins.append(dict(id='tasks', load='tasks', workers=DEFAULT_LOCAL_WORKERS)) # Set the handlers - self.__set_default_job_handler() + self._init_handler_assignment_methods() + self._init_handlers() + self._set_default_handler_assignment_methods() # Set the destination self.default_destination_id = 'local' self.destinations['local'] = [JobDestination(id='local', runner='local')] log.debug('Done loading job configuration') - def __set_default_job_handler(self): - # Called when self.default_handler_id is None - if self.app.application_stack.has_pool(self.app.application_stack.pools.JOB_HANDLERS): - if (self.app.application_stack.in_pool(self.app.application_stack.pools.JOB_HANDLERS)): - log.info("Found job handler pool managed by application stack, this server (%s) is a member of pool: " - "%s", self.app.config.server_name, self.app.application_stack.pools.JOB_HANDLERS) - self._is_handler = True - else: - log.info("Found job handler pool managed by application stack, this server (%s) will submit jobs to " - "pool: %s", self.app.config.server_name, self.app.application_stack.pools.JOB_HANDLERS) - else: - log.info('Did not find job handler pool managed by application stack, this server (%s) will handle jobs ' - 'submitted to it', self.app.config.server_name) - self._is_handler = True - self.app.application_stack.register_postfork_function(self.make_self_default_handler) - - def make_self_default_handler(self): - self.default_handler_id = self.app.config.server_name - self.handlers[self.app.config.server_name] = [self.app.config.server_name] - def get_tool_resource_xml(self, tool_id, tool_type): """ Given a tool id, return XML elements describing parameters to insert into job resources. @@ -484,7 +459,7 @@ class JobConfiguration(ConfiguresHandlers): :returns: JobToolConfiguration -- a representation of a element that uses the default handler and destination """ - return JobToolConfiguration(id='default', handler=self.default_handler_id, destination=self.default_destination_id) + return JobToolConfiguration(id='_default_', handler=self.default_handler_id, destination=self.default_destination_id) # Called upon instantiation of a Tool object def get_job_tool_configurations(self, ids): diff --git a/lib/galaxy/jobs/handler.py b/lib/galaxy/jobs/handler.py index 2e4b35107c0..1233b2c912e 100644 --- a/lib/galaxy/jobs/handler.py +++ b/lib/galaxy/jobs/handler.py @@ -743,14 +743,6 @@ class JobHandlerQueue(Monitors): if not self.track_jobs_in_database: self.queue.put((job_id, tool_id)) self.sleeper.wake() - else: - # Workflow invocations farmed out to workers will submit jobs through here. If a handler is unassigned, we - # will submit for one, or else claim it ourself. TODO: This should be moved to a higher level as it's now - # implemented here and in MessageJobQueue - job = self.sa_session.query(model.Job).get(job_id) - if job.handler is None and self.app.application_stack.has_pool(self.app.application_stack.pools.JOB_HANDLERS): - msg = JobHandlerMessage(task='setup', job_id=job_id) - self.app.application_stack.send_message(self.app.application_stack.pools.JOB_HANDLERS, msg) def shutdown(self): """Attempts to gracefully shut down the worker thread""" diff --git a/lib/galaxy/jobs/manager.py b/lib/galaxy/jobs/manager.py index 2cb68405502..325e7e26f12 100644 --- a/lib/galaxy/jobs/manager.py +++ b/lib/galaxy/jobs/manager.py @@ -1,11 +1,12 @@ """ Top-level Galaxy job manager, moves jobs to handler(s) """ - import logging +from functools import partial from sqlalchemy.sql.expression import null +from galaxy.exceptions import HandlerAssignmentError, ToolExecutionError from galaxy.jobs import handler, NoopQueue from galaxy.model import Job from galaxy.web.stack.message import JobHandlerMessage @@ -16,35 +17,86 @@ log = logging.getLogger(__name__) class JobManager(object): """ Highest level interface to job management. - - TODO: Currently the app accesses "job_queue" and "job_stop_queue" directly. - This should be decoupled. """ - def __init__(self, app): self.app = app self.job_lock = False if self.app.is_job_handler: log.debug("Initializing job handler") self.job_handler = handler.JobHandler(app) - self.job_stop_queue = self.job_handler.job_stop_queue - elif app.application_stack.has_pool(app.application_stack.pools.JOB_HANDLERS): - log.debug("Initializing job handler messaging interface") - self.job_handler = MessageJobHandler(app) - self.job_stop_queue = NoopQueue() else: self.job_handler = NoopHandler() - self.job_stop_queue = NoopQueue() - self.job_queue = self.job_handler.job_queue + self.__check_jobs_at_startup() + + def __check_jobs_at_startup(self): + """ + TODO: It should be documented that starting two Galaxy uWSGI master processes simultaneously would result in a race condition that *could* cause two handlers to pick up the same job. + + The recommended config for now will be webless handlers if running more than one uWSGI (master) process + """ + # FIXME: test + if self.app.job_config.use_messaging: + jobs_at_startup = self.app.model.context.query(Job).enable_eagerloads(False) \ + .filter((Job.state == Job.states.NEW) & (Job.handler == null())).all() + if jobs_at_startup: + log.info( + 'No handler assigned at startup for the following jobs, will dispatch via message: %s', + ', '.join([str(j.id) for j in jobs_at_startup])) + for job in jobs_at_startup: + tool = self.app.toolbox.get_tool(job.tool_id, job.tool_version, exact=True) + self.enqueue(job, tool) def start(self): self.job_handler.start() + def _queue_callback(self, job, tool_id): + self.job_handler.job_queue.put(job.id, tool_id) + + def _message_callback(self, job): + return JobHandlerMessage(task='setup', job_id=job.id) + + def enqueue(self, job, tool=None): + tool_id = None + configured_handler = None + if tool: + tool_id = tool.id + configured_handler = tool.get_configured_job_handler(job.params) + if configured_handler is not None: + p = " (with job params: %s)" % str(job.params) if job.params else "" + log.debug("(%s) Configured job handler for tool '%s'%s is: %s", job.log_str(), tool_id, p, configured_handler) + queue_callback = partial(self._queue_callback, job, tool_id) + message_callback = partial(self._message_callback, job) + try: + return self.app.job_config.assign_handler( + job, configured=configured_handler, queue_callback=queue_callback, message_callback=message_callback) + except HandlerAssignmentError as exc: + raise ToolExecutionError(exc.args[0], job=exc.obj) + + def stop(self, job, message=None): + self.job_handler.job_stop_queue.put(job.id, error_msg=message) + def shutdown(self): self.job_handler.shutdown() +class NoopManager(object): + """ + Implements the JobManager interface but does nothing + """ + def __init__(self, *args, **kwargs): + self.job_handler = NoopHandler() + + def enqueue(self, *args, **kwargs): + pass + + def stop(self, *args, **kwargs): + pass + + class NoopHandler(object): + """ + Implements the JobHandler interface but does nothing + """ def __init__(self, *args, **kwargs): self.job_queue = NoopQueue() self.job_stop_queue = NoopQueue() @@ -54,36 +106,3 @@ class NoopHandler(object): def shutdown(self, *args): pass - - -class MessageJobHandler(NoopHandler): - """ - Implements the JobHandler interface but just to send setup messages on startup - - TODO: It should be documented that starting two Galaxy uWSGI master processes simultaneously would result in a race condition that *could* cause two handlers to pick up the same job. - - The recommended config for now will be webless handlers if running more than one uWSGI (master) process - """ - def __init__(self, app): - # This runs in the web (main) process pre-fork - self.app = app - self.job_queue = MessageJobQueue(app) - self.job_stop_queue = NoopQueue() - jobs_at_startup = self.app.model.context.query(Job).enable_eagerloads(False) \ - .filter((Job.state == Job.states.NEW) & (Job.handler == null())).all() - if jobs_at_startup: - log.info('No handler assigned at startup for the following jobs, will dispatch via message: %s', ', '.join([str(j.id) for j in jobs_at_startup])) - for job in jobs_at_startup: - self.job_queue.put(job.id, job.tool_id) - - -class MessageJobQueue(NoopQueue): - """ - Implements the JobQueue / JobStopQueue interface but only sends messages to the actual job queue - """ - def __init__(self, app): - self.app = app - - def put(self, job_id, tool_id): - msg = JobHandlerMessage(task='setup', job_id=job_id) - self.app.application_stack.send_message(self.app.application_stack.pools.JOB_HANDLERS, msg) diff --git a/lib/galaxy/managers/datasets.py b/lib/galaxy/managers/datasets.py index 7c78ea8f85e..b83d6f4337f 100644 --- a/lib/galaxy/managers/datasets.py +++ b/lib/galaxy/managers/datasets.py @@ -296,7 +296,7 @@ class DatasetAssociationManager(base.ModelManager, # Are *all* of the job's other output datasets deleted? if job.check_if_output_datasets_deleted(): job.mark_deleted(self.app.config.track_jobs_in_database) - self.app.job_manager.job_stop_queue.put(job.id) + self.app.job_manager.stop(job) return True return False diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py index 1130b9e93e1..a2a2537b4c9 100644 --- a/lib/galaxy/model/__init__.py +++ b/lib/galaxy/model/__init__.py @@ -561,6 +561,7 @@ class Job(JobLike, UsesCreateAndUpdateTime, Dictifiable, RepresentById): RUNNING='running', OK='ok', ERROR='error', + FAILED='failed', PAUSED='paused', DELETED='deleted', DELETED_NEW='deleted_new') @@ -862,6 +863,22 @@ class Job(JobLike, UsesCreateAndUpdateTime, Dictifiable, RepresentById): dataset.peek = 'Job deleted' dataset.info = 'Job output deleted by user before job completed' + def mark_failed(self, info="Job execution failed", blurb=None, peek=None): + """ + Mark this job as failed, and mark any output datasets as errored. + """ + self.state = self.states.FAILED + self.info = info + for jtod in self.output_datasets: + jtod.dataset.state = jtod.dataset.states.ERROR + for hda in jtod.dataset.dataset.history_associations: + hda.state = hda.states.ERROR + if blurb: + hda.blurb = blurb + if peek: + hda.peek = peek + hda.info = info + def resume(self, flush=True): if self.state == self.states.PAUSED: self.set_state(self.states.NEW) @@ -4388,6 +4405,17 @@ class WorkflowInvocation(UsesCreateAndUpdateTime, Dictifiable, RepresentById): target_invocation_step = invocation_step return target_invocation_step + @staticmethod + def poll_unhandled_workflow_ids(sa_session): + and_conditions = [ + WorkflowInvocation.state == WorkflowInvocation.states.NEW, + WorkflowInvocation.handler.is_(None) + ] + query = sa_session.query( + WorkflowInvocation.id + ).filter(and_(*and_conditions)).order_by(WorkflowInvocation.table.c.id.asc()) + return [wid for wid in query.all()] + @staticmethod def poll_active_workflow_ids( sa_session, @@ -4536,6 +4564,18 @@ class WorkflowInvocation(UsesCreateAndUpdateTime, Dictifiable, RepresentById): return True return False + def set_handler(self, handler): + self.handler = handler + + def log_str(self): + extra = "" + safe_id = getattr(self, "id", None) + if safe_id is not None: + extra += "id=%s" % safe_id + else: + extra += "unflushed" + return "%s[%s]" % (self.__class__.__name__, extra) + class WorkflowInvocationToSubworkflowInvocationAssociation(Dictifiable, RepresentById): dict_collection_visible_keys = ['id', 'workflow_step_id', 'workflow_invocation_id', 'subworkflow_invocation_id'] diff --git a/lib/galaxy/tools/__init__.py b/lib/galaxy/tools/__init__.py index 3a56e10c82a..2c14fb86404 100755 --- a/lib/galaxy/tools/__init__.py +++ b/lib/galaxy/tools/__init__.py @@ -561,16 +561,18 @@ class Tool(Dictifiable): assert rval is not None, 'Could not get a job tool configuration for Tool %s with job_params %s, this is a bug' % (self.id, job_params) return rval - def get_job_handler(self, job_params=None): - """Get a suitable job handler for this `Tool` given the provided `job_params`. If multiple handlers are valid for combination of `Tool` and `job_params` (e.g. the defined handler is a handler tag), one will be selected at random. + def get_configured_job_handler(self, job_params=None): + """Get the configured job handler for this `Tool` given the provided `job_params`. + + Unlike the former ``get_job_handler()`` method, this does not perform "preassignment" (random selection of + a configured handler ID from a tag). :param job_params: Any params specific to this job (e.g. the job source) :type job_params: dict or None - :returns: str -- The id of a job handler for a job run of this `Tool` + :returns: str or None -- The configured handler for a job run of this `Tool` """ - # convert tag to ID if necessary - return self.app.job_config.get_handler(self.__get_job_tool_configuration(job_params=job_params).handler) + return self.__get_job_tool_configuration(job_params=job_params).handler def get_job_destination(self, job_params=None): """ @@ -1516,7 +1518,16 @@ class Tool(Dictifiable): `self.tool_action`. In general this will create a `Job` that when run will build the tool's outputs, e.g. `DefaultToolAction`. """ - return self.tool_action.execute(self, trans, incoming=incoming, set_output_hid=set_output_hid, history=history, **kwargs) + try: + return self.tool_action.execute(self, trans, incoming=incoming, set_output_hid=set_output_hid, history=history, **kwargs) + except exceptions.ToolExecutionError as exc: + job = exc.job + job_id = 'unknown' + if job is not None: + job.mark_failed(info=exc.err_msg, blurb=exc.err_code.default_error_message) + job_id = job.id + log.error("Tool execution failed for job: %s", job_id) + raise def params_to_strings(self, params, app, nested=False): return params_to_strings(self.inputs, params, app, nested) diff --git a/lib/galaxy/tools/actions/__init__.py b/lib/galaxy/tools/actions/__init__.py index da9abcfd274..0ed09692a53 100644 --- a/lib/galaxy/tools/actions/__init__.py +++ b/lib/galaxy/tools/actions/__init__.py @@ -517,7 +517,6 @@ class DefaultToolAction(object): job.object_store_id = object_store_populator.object_store_id if job_params: job.params = dumps(job_params) - job.set_handler(tool.get_job_handler(job_params)) if completed_job: job.set_copied_from_job_id(completed_job.id) trans.sa_session.add(job) @@ -529,11 +528,8 @@ class DefaultToolAction(object): rerun_remap_job_id=rerun_remap_job_id, current_job=job, out_data=out_data) - log.info("Setup for job %s complete, ready to flush %s" % (job.log_str(), job_setup_timer)) + log.info("Setup for job %s complete, ready to be enqueued %s" % (job.log_str(), job_setup_timer)) - job_flush_timer = ExecutionTimer() - trans.sa_session.flush() - log.info("Flushed transaction for job %s %s" % (job.log_str(), job_flush_timer)) # Some tools are not really executable, but jobs are still created for them ( for record keeping ). # Examples include tools that redirect to other applications ( epigraph ). These special tools must # include something that can be retrieved from the params ( e.g., REDIRECT_URL ) to keep the job @@ -555,8 +551,8 @@ class DefaultToolAction(object): trans.sa_session.flush() trans.response.send_redirect(url_for(controller='tool_runner', action='redirect', redirect_url=redirect_url)) else: - # Put the job in the queue if tracking in memory - app.job_manager.job_queue.put(job.id, job.tool_id) + # Dispatch to a job handler. enqueue() is responsible for flushing the job + app.job_manager.enqueue(job, tool=tool) trans.log_event("Added job to the job queue, id: %s" % str(job.id), tool_id=job.tool_id) return job, out_data diff --git a/lib/galaxy/tools/actions/history_imp_exp.py b/lib/galaxy/tools/actions/history_imp_exp.py index 512e9c7514a..1ed267d0495 100644 --- a/lib/galaxy/tools/actions/history_imp_exp.py +++ b/lib/galaxy/tools/actions/history_imp_exp.py @@ -56,11 +56,9 @@ class ImportHistoryToolAction(ToolAction): job.add_parameter(name, value) job.state = start_job_state # job inputs have been configured, restore initial job state - job.set_handler(tool.get_job_handler(None)) - trans.sa_session.flush() # Queue the job for execution - trans.app.job_manager.job_queue.put(job.id, tool.id) + trans.app.job_manager.enqueue(job, tool=tool) trans.log_event("Added import history job to the job queue, id: %s" % str(job.id), tool_id=job.tool_id) return job, odict() @@ -136,11 +134,9 @@ class ExportHistoryToolAction(ToolAction): job.add_parameter(name, value) job.state = start_job_state # job inputs have been configured, restore initial job state - job.set_handler(tool.get_job_handler(None)) - trans.sa_session.flush() # Queue the job for execution - trans.app.job_manager.job_queue.put(job.id, tool.id) + trans.app.job_manager.enqueue(job, tool=tool) trans.log_event("Added export history job to the job queue, id: %s" % str(job.id), tool_id=job.tool_id) return job, odict() diff --git a/lib/galaxy/tools/actions/metadata.py b/lib/galaxy/tools/actions/metadata.py index e60cd9fd5f8..6f8d1146828 100644 --- a/lib/galaxy/tools/actions/metadata.py +++ b/lib/galaxy/tools/actions/metadata.py @@ -63,7 +63,6 @@ class SetMetadataToolAction(ToolAction): except AttributeError: job.tool_version = "1.0.1" job.state = job.states.WAITING # we need to set job state to something other than NEW, or else when tracking jobs in db it will be picked up before we have added input / output parameters - job.set_handler(tool.get_job_handler(job_params)) sa_session.add(job) sa_session.flush() # ensure job.id is available @@ -104,7 +103,7 @@ class SetMetadataToolAction(ToolAction): sa_session.flush() # Queue the job for execution - app.job_manager.job_queue.put(job.id, tool.id) + app.job_manager.enqueue(job, tool=tool) # FIXME: need to add event logging to app and log events there rather than trans. # trans.log_event( "Added set external metadata job to the job queue, id: %s" % str(job.id), tool_id=job.tool_id ) diff --git a/lib/galaxy/tools/actions/upload_common.py b/lib/galaxy/tools/actions/upload_common.py index 5becfe0a775..839c5a42f73 100644 --- a/lib/galaxy/tools/actions/upload_common.py +++ b/lib/galaxy/tools/actions/upload_common.py @@ -432,15 +432,13 @@ def create_job(trans, params, tool, json_file_path, outputs, folder=None, histor job.object_store_id = object_store_id job.set_state(job.states.NEW) - job.set_handler(tool.get_job_handler(None)) if job_params: for name, value in job_params.items(): job.add_parameter(name, value) trans.sa_session.add(job) - trans.sa_session.flush() # Queue the job for execution - trans.app.job_manager.job_queue.put(job.id, job.tool_id) + trans.app.job_manager.enqueue(job, tool=tool) trans.log_event("Added job to the job queue, id: %s" % str(job.id), tool_id=job.tool_id) output = odict() for i, v in enumerate(outputs): diff --git a/lib/galaxy/util/handlers.py b/lib/galaxy/util/handlers.py index cdd7f845d1e..ab31a21649c 100644 --- a/lib/galaxy/util/handlers.py +++ b/lib/galaxy/util/handlers.py @@ -8,12 +8,45 @@ from __future__ import absolute_import import logging import os import random +from collections import namedtuple +from sqlalchemy.orm import object_session + +from . import ( + ExecutionTimer, + listify +) +from ..exceptions import HandlerAssignmentError log = logging.getLogger(__name__) +_handler_assignment_methods = ('MEM_SELF', 'DB_SELF', 'DB_PREASSIGN', 'UWSGI_MULE_MESSAGE') +HANDLER_ASSIGNMENT_METHODS = namedtuple('JOB_HANDLER_ASSIGNMENT_METHODS', _handler_assignment_methods)( + *[x.lower().replace('_', '-') for x in _handler_assignment_methods] +) + + +class HandlerAssignmentSkip(Exception): + """Exception for handler assignment methods to raise if the next method should be tried. + """ + pass + class ConfiguresHandlers(object): + DEFAULT_HANDLER_TAG = '_default_' + DEFAULT_BASE_HANDLER_POOLS = () + UNSUPPORTED_HANDLER_ASSIGNMENT_METHODS = () + + def add_handler(self, handler_id, tags): + if handler_id in self.handlers: + log.error("Handler '%s' already added, ignoring", handler_id) + return + self.handlers[handler_id] = (handler_id,) + for tag in tags: + if tag in self.handlers: + self.handlers[tag].append(handler_id) + else: + self.handlers[tag] = [handler_id] def _init_handlers(self, config_element): # Parse handlers @@ -21,22 +54,70 @@ class ConfiguresHandlers(object): for handler in self._findall_with_required(config_element, 'handler'): handler_id = handler.get('id') if handler_id in self.handlers: - log.error("Handler '%s' overlaps handler with the same name, ignoring" % handler_id) + log.error("Handler '%s' overlaps handler with the same name, ignoring", handler_id) else: - log.debug("Read definition for handler '%s'" % handler_id) - self.handlers[handler_id] = (handler_id,) - self._parse_handler(handler_id, handler) - if handler.get('tags', None) is not None: - for tag in [x.strip() for x in handler.get('tags').split(',')]: - if tag in self.handlers: - self.handlers[tag].append(handler_id) - else: - self.handlers[tag] = [handler_id] + log.debug("Read definition for handler '%s'", handler_id) + self.add_handler( + handler_id, + [x.strip() for x in handler.get('tags', self.DEFAULT_HANDLER_TAG).split(',')] + ) + self.default_handler_id = self._get_default(self.app.config, config_element, list(self.handlers.keys())) + + def _init_handler_assignment_methods(self, config_element=None): + self.__is_handler = None + # This is set by the stack job handler init code + self.pool_for_tag = {} + self._handler_assignment_method_methods = { + HANDLER_ASSIGNMENT_METHODS.MEM_SELF: self._assign_mem_self_handler, + HANDLER_ASSIGNMENT_METHODS.DB_SELF: self._assign_db_self_handler, + HANDLER_ASSIGNMENT_METHODS.DB_PREASSIGN: self._assign_db_preassign_handler, + HANDLER_ASSIGNMENT_METHODS.UWSGI_MULE_MESSAGE: self._assign_uwsgi_mule_message_handler, + } + if config_element is not None: + for method in listify(config_element.attrib.get('assign_with', []), do_strip=True): + method = method.lower() + assert method in HANDLER_ASSIGNMENT_METHODS, \ + "Invalid job handler assignment method '%s', must be one of: %s" % ( + method, ', '.join(HANDLER_ASSIGNMENT_METHODS)) + try: + self.handler_assignment_methods.append(method) + except AttributeError: + self.handler_assignment_methods_configured = True + self.handler_assignment_methods = [method] + if self.handler_assignment_methods == [HANDLER_ASSIGNMENT_METHODS.MEM_SELF]: + self.app.config.track_jobs_in_database = False + + def _set_default_handler_assignment_methods(self): + if not self.handler_assignment_methods_configured: + if not self.app.config.track_jobs_in_database and \ + HANDLER_ASSIGNMENT_METHODS.MEM_SELF not in self.UNSUPPORTED_HANDLER_ASSIGNMENT_METHODS: + # DEPRECATED: You should just set mem_self as the only method if you want this + # FIXME: does setting MEM_SELF still disable track_jobs_in_database like it used to? + # FIXME: test MEM_SELF via all config routes: + # 1. no assign config, track_jobs_in_databse = False in config + # 2. assign_with='mem-self', track_jobs_in_databse = False in config + # 2. assign_with='mem-self', track_jobs_in_databse = True in config + log.warning("The `track_jobs_in_database` option is deprecated, please set `%s` as the job" + " handler assignment method in the job handler configuration", + HANDLER_ASSIGNMENT_METHODS.MEM_SELF) + self.handler_assignment_methods = [HANDLER_ASSIGNMENT_METHODS.MEM_SELF] + elif not self.handlers: + # No handlers defined, default is for processes to handle the jobs they create + self.handler_assignment_methods = [HANDLER_ASSIGNMENT_METHODS.DB_SELF] + else: + # Handlers are defined, default is the value if the default attribute, or any untagged handler if + # default attribute is unset + self.handler_assignment_methods = [HANDLER_ASSIGNMENT_METHODS.DB_PREASSIGN] + # If the stack has handler pools it can override these defaults + self.app.application_stack.init_job_handling(self) + log.info("%s: No job handler assignment method is set, defaulting to '%s', set the `assign_with` attribute" + " on to override the default", self.__class__.__name__, + self.handler_assignment_methods[0]) def _parse_handler(self, handler_id, handler_def): pass - def _get_default(self, config, parent, names): + def _get_default(self, config, parent, names, auto=False): """ Returns the default attribute set in a parent tag like or , or return the ID of the child, if there is no explicit @@ -46,6 +127,8 @@ class ConfiguresHandlers(object): :type parent: ``xml.etree.ElementTree.Element`` :param names: The list of destination or handler IDs or tags that were loaded. :type names: list of str + :param auto: Automatically set a default if there is no default in the parent tag and there is only one child. + :type auto: bool :returns: str -- id or tag representing the default. """ @@ -60,14 +143,12 @@ class ConfiguresHandlers(object): if rval is not None: # If the parent element has a 'default' attribute, use the id or tag in that attribute - if rval not in names: + if self.deterministic_handler_assignment and rval not in names: raise Exception("<%s> default attribute '%s' does not match a defined id or tag in a child element" % (parent.tag, rval)) log.debug("<%s> default set to child with id or tag '%s'" % (parent.tag, rval)) - elif len(names) == 1: + elif auto and len(names) == 1: log.info("Setting <%s> default to child with id '%s'" % (parent.tag, names[0])) rval = names[0] - else: - raise Exception("No <%s> default specified, please specify a valid id or tag with the 'default' attribute" % parent.tag) return rval def _findall_with_required(self, parent, match, attribs=None): @@ -95,21 +176,38 @@ class ConfiguresHandlers(object): return rval @property - def is_handler(self): - """Indicate whether the current server is a handler. + def use_messaging(self): + # The job manager uses this to determine whether the messaging job queue should be used + return HANDLER_ASSIGNMENT_METHODS.UWSGI_MULE_MESSAGE in self.handler_assignment_methods - :param server_name: The name to check - :type server_name: str + @property + def deterministic_handler_assignment(self): + return self.handler_assignment_methods and all( + filter(lambda x: x in ( + HANDLER_ASSIGNMENT_METHODS.UWSGI_MULE_MESSAGE, + HANDLER_ASSIGNMENT_METHODS.DB_PREASSIGN, + ), self.handler_assignment_methods)) + + def _get_is_handler(self): + """Indicate whether the current server is configured as a handler. :return: bool """ - if self._is_handler is not None: - return self._is_handler + if self.__is_handler is not None: + return self.__is_handler + if (HANDLER_ASSIGNMENT_METHODS.DB_SELF in self.handler_assignment_methods + or HANDLER_ASSIGNMENT_METHODS.MEM_SELF in self.handler_assignment_methods): + return True for collection in self.handlers.values(): if self.app.config.server_name in collection: return True return False + def _set_is_handler(self, value): + self.__is_handler = value + + is_handler = property(_get_is_handler, _set_is_handler) + def _get_single_item(self, collection, index=None): """Given a collection of handlers or destinations, return one item from the collection at random. """ @@ -121,19 +219,116 @@ class ConfiguresHandlers(object): else: return collection[index % len(collection)] - # This is called by Tool.get_job_handler() - def get_handler(self, id_or_tag, index=None): - """Given a handler ID or tag, return a handler matching it. + # If these get to be any more complex we should probably modularize them, or at least move to a separate class + + def _assign_handler_direct(self, obj, configured): + if self.app.config.track_jobs_in_database and configured: + try: + resolved = self.get_preassigned_handler(configured) + except KeyError: + resolved = None + if resolved == configured: + obj.set_handler(configured) + _timed_flush_obj(obj) + return True + return False + + def _assign_mem_self_handler(self, obj, method, configured, queue_callback=None, **kwargs): + assert queue_callback is not None, \ + "Cannot perform '%s' handler assignment: `queue_callback` is None" % HANDLER_ASSIGNMENT_METHODS.MEM_SELF + if configured: + log.warning("(%s) Ignoring handler assignment to '%s' because configured handler assignment method" + " '' overrides per-tool handler assignment", obj.log_str(), + HANDLER_ASSIGNMENT_METHODS.MEM_SELF, configured) + _timed_flush_obj(obj) + queue_callback() + return self.app.config.server_name + + def _assign_db_self_handler(self, obj, method, configured, **kwargs): + if configured: + return self.handler_assignment_method_methods[HANDLER_ASSIGNMENT_METHODS.DB_PREASSIGN]( + obj, method, configured, **kwargs + ) + obj.set_handler(self.app.config.server_name) + _timed_flush_obj(obj) + return self.app.config.server_name + + def _assign_db_preassign_handler(self, obj, method, configured, index=None, **kwargs): + """Given a handler ID or tag, return a handler matching it, of those handlers that are statically configured in + the job configuration, or known via preconfigured pools. - :param id_or_tag: A handler ID or tag. - :type id_or_tag: str :param index: Generate "consistent" "random" handlers with this index if specified. :type index: int :returns: str -- A valid job handler ID. """ - if id_or_tag is None and self.default_handler_id is None: - return None - if id_or_tag is None: - id_or_tag = self.default_handler_id - return self._get_single_item(self.handlers[id_or_tag], index=index) + # FIXME: test a combo of pool handlers and defined handlers w/ DB_PREASSIGN as the only method + handler = configured + if handler is None: + handler = self.default_handler_id or self.DEFAULT_HANDLER_TAG + # Get a random handler ID from the possible handlers. If the admin has configured a tool with a handler tag that + # does not exist, or if there are no default handlers and configured is None, this will raise KeyError to + # assign_handler, which should log it, try the next method (if any), and if no other methods succeed, raise + # HandlerAssigmentError. + handler_id = self._get_single_item(self.handlers[handler], index=index) + if handler != handler_id: + log.debug("(%s) Selected handler '%s' by random choice from handler tag '%s'", obj.log_str(), + handler_id, handler) + obj.set_handler(handler_id) + _timed_flush_obj(obj) + return handler_id + + def _assign_uwsgi_mule_message_handler(self, obj, method, configured, message_callback=None, **kwargs): + assert message_callback is not None, \ + "Cannot perform '%s' handler assignment: `message_callback` is None" \ + % HANDLER_ASSIGNMENT_METHODS.UWSGI_MULE_MESSAGE + tag = configured or self.DEFAULT_HANDLER_TAG + pool = self.pool_for_tag.get(tag) + if pool is None: + log.debug("(%s) No handler pool (uWSGI farm) for '%s' found", obj.log_str(), tag) + raise HandlerAssignmentSkip() + else: + _timed_flush_obj(obj) + message = message_callback() + self.app.application_stack.send_message(pool, message) + return pool + + def assign_handler(self, obj, configured=None, **kwargs): + """Set a job handler, flush obj + + :param obj: Object to assign to (must be a model object with ``handler`` attribute and ``log_str`` callable) + :param configured: A handler ID or tag/pool. + + :returns: bool -- True on successful assignment, False otherwise. + """ + # It's a bit awkward that the method that actually hands off job execution is in the JobConfiguration, but + # that's currently the best place for it. It's worth noting that this method is also part of the + # WorkflowSchedulingManager, which acts like a combined JobConfiguration and JobManager. Combining those two + # classes would probably be reasonable (and would remove the need for the queue callback). + if self._assign_handler_direct(obj, configured): + log.info("(%s) Skipped handler assignment logic due to explicit configuration to a single handler: %s", + obj.log_str(), configured) + return True + for method in self.handler_assignment_methods: + try: + handler = self._handler_assignment_method_methods[method]( + obj, method, configured=configured, **kwargs) + log.info("(%s) Handler '%s' assigned using '%s' assignment method", obj.log_str(), handler, method) + break + except HandlerAssignmentSkip: + log.debug("(%s) Handler assignment method '%s' did not assign a handler, trying next method", + obj.log_str(), method) + except Exception: + log.exception("Caught exception in handler assignment method: %s", method) + else: + # Ideally we could just expunge the object from the SA session here, but in most cases, some of its related + # objects have already been created, so instead we'll just have to fail it. + log.error("(%s) Failed to select handler", obj.log_str()) + raise HandlerAssignmentError("Job handler assignment failed.", obj=obj) + + +def _timed_flush_obj(obj): + obj_flush_timer = ExecutionTimer() + sa_session = object_session(obj) + sa_session.flush() + log.info("Flushed transaction for %s %s" % (obj.log_str(), obj_flush_timer)) diff --git a/lib/galaxy/web/stack/__init__.py b/lib/galaxy/web/stack/__init__.py index 2a24b6d80a1..461e4a7d6a2 100644 --- a/lib/galaxy/web/stack/__init__.py +++ b/lib/galaxy/web/stack/__init__.py @@ -19,8 +19,8 @@ import yaml from six import string_types from galaxy.util import unicodify -from galaxy.util.bunch import Bunch from galaxy.util.facts import get_facts +from galaxy.util.handlers import HANDLER_ASSIGNMENT_METHODS from galaxy.util.path import has_ext from galaxy.util.properties import nice_config_parser from .message import ApplicationStackMessage, ApplicationStackMessageDispatcher @@ -54,12 +54,6 @@ class ApplicationStack(object): server_name_template = '{server_name}' default_app_name = 'main' - # used both to route jobs to a pool with this name and indicate whether or - # not a stack is using messaging for handler assignment - pools = Bunch( - JOB_HANDLERS='job-handlers', - ) - @classmethod def log_filter(cls): return cls.log_filter_class() @@ -77,6 +71,12 @@ class ApplicationStack(object): self.config = config or (app and app.config) self.running = False + def init_job_handling(self, job_config): + pass + + def init_late_prefork(self): + pass + def log_startup(self): log.info("Galaxy server instance '%s' is running" % self.app.config.server_name) @@ -97,12 +97,18 @@ class ApplicationStack(object): # TODO: ideally jobs would be mappable to handlers by pool name return None + def has_base_pool(self, pool_name): + return False + def has_pool(self, pool_name): return False def in_pool(self, pool_name): return False + def pool_members(self, pool_name): + return None + @property def facts(self): facts = get_facts(config=self.config) @@ -129,12 +135,16 @@ class ApplicationStack(object): class MessageApplicationStack(ApplicationStack): def __init__(self, app=None, config=None): super(MessageApplicationStack, self).__init__(app=app, config=config) + self.use_messaging = False self.dispatcher = ApplicationStackMessageDispatcher() self.transport = self.transport_class(app, stack=self, dispatcher=self.dispatcher) + def init_late_prefork(self): + self.transport.init_late_prefork() + def start(self): super(MessageApplicationStack, self).start() - if not self.running: + if self.use_messaging and not self.running: self.transport.start() self.running = True @@ -157,7 +167,7 @@ class MessageApplicationStack(ApplicationStack): self.transport.send_message(msg.encode(), dest) def shutdown(self): - if self.running: + if self.use_messaging and self.running: log.info('Application stack interface shutting down') self.transport.shutdown() self.running = False @@ -282,8 +292,85 @@ class UWSGIApplicationStack(MessageApplicationStack): def __init__(self, app=None, config=None): self._farms_dict = None self._mules_list = None + self._farm_server_names_dict = None + + # If there's more than one farm *and* more than one farm will be using farm messaging, additional locks need to + # be configured by the admin. This allows us to keep track of how many such farms are configured. + self._lock_farms = set() + super(UWSGIApplicationStack, self).__init__(app=app, config=config) + def _set_default_job_handler_assignment_methods(self, job_config, pool): + # This is called once per job_config (job config or workflow scheduling manager), and only in the case that + # handler assignment is not explicitly configured by the admin. Disable DB_SELF if a valid farm (pool) is + # configured. Use mule messaging unless the job_config doesn't allow it (e.g. workflow scheduling manager), in + # which case, use DB_PREASSIGN. + # + # TODO MULTIPOOL: if there is no default in any base_pool (and no job_config.default_handler_id) then don't + # remove DB_SELF + conf_class_name = job_config.__class__.__name__ + remove_methods = [HANDLER_ASSIGNMENT_METHODS.DB_SELF] + if (HANDLER_ASSIGNMENT_METHODS.UWSGI_MULE_MESSAGE not in job_config.UNSUPPORTED_HANDLER_ASSIGNMENT_METHODS): + add_method = HANDLER_ASSIGNMENT_METHODS.UWSGI_MULE_MESSAGE + else: + add_method = HANDLER_ASSIGNMENT_METHODS.DB_PREASSIGN + remove_methods.append(HANDLER_ASSIGNMENT_METHODS.UWSGI_MULE_MESSAGE) + log.debug("%s: No job handler assignment methods were configured but a uWSGI farm named '%s' exists," + " automatically enabling the '%s' assignment method", conf_class_name, pool, add_method) + for m in remove_methods: + try: + job_config.handler_assignment_methods.remove(m) + log.debug("%s: Removed '%s' from handler assignment methods due to use of mules", conf_class_name, m) + except ValueError: + pass + if add_method not in job_config.handler_assignment_methods: + job_config.handler_assignment_methods.insert(0, add_method) + log.debug("%s: handler assignment methods updated to: %s", conf_class_name, + ', '.join(job_config.handler_assignment_methods)) + + def _init_job_handler_assignment_methods(self, job_config, pool): + if not job_config.handler_assignment_methods_configured: + self._set_default_job_handler_assignment_methods(job_config, pool) + if HANDLER_ASSIGNMENT_METHODS.UWSGI_MULE_MESSAGE in job_config.handler_assignment_methods: + self.use_messaging = True + + def _init_job_handler_subpools(self, job_config, pool): + # Set up members of "subpools" ("pool.*") as handlers (including the base pool itself, if it exists) + for farm in self._configured_farms: + if not (farm == pool or farm.startswith(pool + '.')): + continue + if job_config.use_messaging: + self._lock_farms.add(farm) + if farm == pool: + tag = job_config.DEFAULT_HANDLER_TAG + else: + tag = farm.replace(pool + '.', '', 1) + # Pools are hierarchical (so that you can have e.g. workflow schedulers use the job handlers pool if no + # workflow schedulers pool exists), so if a pool for a given tag has already been found higher in the + # hierarchy, don't add mules from a farm/pool lower in the hierarchy. + # FIXME: test + if tag not in job_config.pool_for_tag: + if self.in_pool(farm): + job_config.is_handler = True + for handler in self.pool_members(farm): + job_config.add_handler(handler, [tag]) + job_config.pool_for_tag[tag] = farm + + def init_job_handling(self, job_config): + """Automatically add uWSGI farms as pools if they are named per predefined names and there is not an explicit + job handler assignment configuration. + + Also automatically add mule messaging as the preferred assignment method, if supported. + """ + # Only do the stack handler assignment init once per job_config + stack_assignment_methods_configured = False + for pool in job_config.DEFAULT_BASE_HANDLER_POOLS: + if self.has_base_pool(pool): + if not stack_assignment_methods_configured: + self._init_job_handler_assignment_methods(job_config, pool) + stack_assignment_methods_configured = True + self._init_job_handler_subpools(job_config, pool) + @property def _configured_mules(self): if self._mules_list is None: @@ -314,20 +401,39 @@ class UWSGIApplicationStack(MessageApplicationStack): farms.append(farm) return farms - def _mule_index_in_farm(self, farm_name): + def _mule_index_in_farm(self, farm_name, mule_id=None): + mule_id = mule_id or uwsgi.mule_id() try: mules = self._configured_farms[farm_name] - return mules.index(uwsgi.mule_id()) + return mules.index(mule_id) except (KeyError, ValueError): return -1 @property def _farm_name(self): + # TODO: to allow mules to be in multiple farms you'll need to start here try: return self._farms[0] except IndexError: return None + @property + def _farm_server_names(self): + if self._farm_server_names_dict is None: + self._farm_server_names_dict = {} + for farm_name, mules in self._configured_farms.items(): + server_names = [] + facts = self.facts + for mule in mules: + facts.update({ + 'pool_name': farm_name, + 'server_id': mule, + 'instance_id': self._mule_index_in_farm(farm_name, mule) + 1, + }) + server_names.append(self.server_name_template.format(**facts)) + self._farm_server_names_dict[farm_name] = server_names + return self._farm_server_names_dict + @property def instance_id(self): if not self._is_mule: @@ -363,6 +469,9 @@ class UWSGIApplicationStack(MessageApplicationStack): os.environ['_GALAXY_UWSGI_FARM_NAMES'] = ','.join(self._farms) super(UWSGIApplicationStack, self).start() + def has_base_pool(self, pool_name): + return self.has_pool(pool_name) or any([farm.startswith(pool_name + '.') for farm in self._configured_farms]) + def has_pool(self, pool_name): return pool_name in self._configured_farms @@ -372,6 +481,9 @@ class UWSGIApplicationStack(MessageApplicationStack): else: return pool_name in self._farms + def pool_members(self, pool_name): + return self._farm_server_names.get(pool_name, None) + def workers(self): return uwsgi.workers() diff --git a/lib/galaxy/web/stack/transport.py b/lib/galaxy/web/stack/transport.py index 24930cf1ca1..233fed2b013 100644 --- a/lib/galaxy/web/stack/transport.py +++ b/lib/galaxy/web/stack/transport.py @@ -29,6 +29,9 @@ class ApplicationStackTransport(object): self.dispatcher = dispatcher self.dispatcher_thread = None + def init_late_prefork(self): + pass + def _dispatch_messages(self): pass @@ -70,13 +73,12 @@ class UWSGIFarmMessageTransport(ApplicationStackTransport): # Define any static lock names here, additional locks will be appended for each configured farm's message handler _locks = [] - def __initialize_locks(self): + def init_late_prefork(self): num = int(uwsgi.opt.get('locks', 0)) + 1 - farms = self.stack._configured_farms.keys() - need = len(farms) + need = len(self.stack._lock_farms) if num < need: raise RuntimeError('Need %i uWSGI locks but only %i exist(s): Set `locks = %i` in uWSGI configuration' % (need, num, need - 1)) - self._locks.extend(['RECV_MSG_FARM_' + x for x in farms]) + self._locks.extend(['RECV_MSG_FARM_' + x for x in sorted(self.stack._lock_farms)]) # this would be nice, but in my 2.0.15 uWSGI, the uwsgi module has no set_option function, and I don't know if it'd work even if the function existed as documented # if len(self.lock_map) > 1: # uwsgi.set_option('locks', len(self.lock_map)) @@ -84,7 +86,6 @@ class UWSGIFarmMessageTransport(ApplicationStackTransport): def __init__(self, app, stack, dispatcher=None): super(UWSGIFarmMessageTransport, self).__init__(app, stack, dispatcher=dispatcher) - self.__initialize_locks() def __lock(self, name_or_id): try: diff --git a/lib/galaxy/webapps/galaxy/api/jobs.py b/lib/galaxy/webapps/galaxy/api/jobs.py index ac51424c243..db25f4c479c 100644 --- a/lib/galaxy/webapps/galaxy/api/jobs.py +++ b/lib/galaxy/webapps/galaxy/api/jobs.py @@ -203,7 +203,7 @@ class JobController(BaseAPIController, UsesLibraryMixinItems): if not job.finished: job.mark_deleted(self.app.config.track_jobs_in_database) trans.sa_session.flush() - self.app.job_manager.job_stop_queue.put(job.id) + self.app.job_manager.stop(job) return True else: return False diff --git a/lib/galaxy/webapps/galaxy/config_schema.yml b/lib/galaxy/webapps/galaxy/config_schema.yml index 60fab94a475..854ad3f8170 100644 --- a/lib/galaxy/webapps/galaxy/config_schema.yml +++ b/lib/galaxy/webapps/galaxy/config_schema.yml @@ -2437,6 +2437,7 @@ mapping: failing jobs are just failed outright. track_jobs_in_database: + # FIXME: note deprecated, superceded by job_handler_assignment_method type: bool default: true required: false diff --git a/lib/galaxy/webapps/galaxy/controllers/admin.py b/lib/galaxy/webapps/galaxy/controllers/admin.py index a8d411be2e0..0b2a263a275 100644 --- a/lib/galaxy/webapps/galaxy/controllers/admin.py +++ b/lib/galaxy/webapps/galaxy/controllers/admin.py @@ -1648,7 +1648,7 @@ class AdminGalaxy(controller.JSAppLauncher, AdminActions, UsesQuotaMixin, QuotaP job.set_state(trans.app.model.Job.states.DELETED_NEW) trans.sa_session.add(job) else: - trans.app.job_manager.job_stop_queue.put(job_id, error_msg=error_msg) + trans.app.job_manager.stop(job, message=error_msg) deleted.append(str(job_id)) if deleted: msg = 'Queued job' diff --git a/lib/galaxy/webapps/galaxy/controllers/history.py b/lib/galaxy/webapps/galaxy/controllers/history.py index 238319c6634..64a766a38f0 100644 --- a/lib/galaxy/webapps/galaxy/controllers/history.py +++ b/lib/galaxy/webapps/galaxy/controllers/history.py @@ -360,7 +360,7 @@ class HistoryController(BaseUIController, SharableMixin, UsesAnnotations, UsesIt if job.history_id == history.id and not job.finished: # No need to check other outputs since the job's parent history is this history job.mark_deleted(trans.app.config.track_jobs_in_database) - trans.app.job_manager.job_stop_queue.put(job.id) + trans.app.job_manager.stop(job) trans.sa_session.flush() if n_deleted: part = "Deleted %d %s" % (n_deleted, iff(n_deleted != 1, "histories", "history")) diff --git a/lib/galaxy/workflow/scheduling_manager.py b/lib/galaxy/workflow/scheduling_manager.py index fc0b4c8e37b..c7801c99df6 100644 --- a/lib/galaxy/workflow/scheduling_manager.py +++ b/lib/galaxy/workflow/scheduling_manager.py @@ -1,11 +1,13 @@ import logging import os +from functools import partial from xml.etree import ElementTree import galaxy.workflow.schedulers from galaxy import model +from galaxy.exceptions import HandlerAssignmentError from galaxy.util import plugin_config -from galaxy.util.handlers import ConfiguresHandlers +from galaxy.util.handlers import ConfiguresHandlers, HANDLER_ASSIGNMENT_METHODS from galaxy.util.monitors import Monitors from galaxy.web.stack.message import WorkflowSchedulingMessage @@ -18,6 +20,7 @@ EXCEPTION_MESSAGE_SHUTDOWN = "Exception raised while attempting to shutdown work EXCEPTION_MESSAGE_NO_SCHEDULERS = "Failed to defined workflow schedulers - no workflow schedulers defined." EXCEPTION_MESSAGE_NO_DEFAULT_SCHEDULER = "Failed to defined workflow schedulers - no workflow scheduler found for default id '%s'." EXCEPTION_MESSAGE_DUPLICATE_SCHEDULERS = "Failed to defined workflow schedulers - workflow scheduling plugin id '%s' duplicated." +EXCEPTION_MESSAGE_SERIALIZE = "Parallelization is not desired but handler assignment methods are non-deterministic. Set DB_PREASSIGN in workflow_schedulers_conf.xml." class WorkflowSchedulingManager(ConfiguresHandlers): @@ -25,22 +28,24 @@ class WorkflowSchedulingManager(ConfiguresHandlers): ``galaxy.manager.JobManager``. Only schedules workflows on handler processes. """ + DEFAULT_BASE_HANDLER_POOLS = ('workflow-schedulers', 'job-handlers') + UNSUPPORTED_HANDLER_ASSIGNMENT_METHODS = ( + HANDLER_ASSIGNMENT_METHODS.UWSGI_MULE_MESSAGE, + ) def __init__(self, app): self.app = app self.__handlers_configured = False self.workflow_schedulers = {} self.active_workflow_schedulers = {} - # TODO: this should not hardcode the job handlers pool - self.__handler_pool = self.app.application_stack.pools.JOB_HANDLERS - # TODO: and we need a better way to indicate messaging should be used - self.__use_stack_messages = app.application_stack.has_pool(self.__handler_pool) # Passive workflow schedulers won't need to be monitored I guess. self.request_monitor = None self.handlers = {} - self._is_handler = None + self.handler_assignment_methods_configured = False + self.handler_assignment_methods = None + self.default_handler_id = None self.__plugin_classes = self.__plugins_dict() self.__init_schedulers() @@ -50,7 +55,17 @@ class WorkflowSchedulingManager(ConfiguresHandlers): self.__start_schedulers() if self.active_workflow_schedulers: self.__start_request_monitor() - if self.__use_stack_messages: + + # When assinging handlers to workflows being queued - use job_conf + # if not explicit workflow scheduling handlers have be specified or + # else use those explicit workflow scheduling handlers (on self). + if self.__handlers_configured: + self.__handlers_config = self + else: + self.__handlers_config = app.job_config + + if self._is_workflow_handler(): + if self.__handlers_config.use_messaging: WorkflowSchedulingMessage().bind_default_handler(self, '_handle_message') self.app.application_stack.register_message_handler( self._handle_message, @@ -59,24 +74,13 @@ class WorkflowSchedulingManager(ConfiguresHandlers): # Process should not schedule workflows but should check for any unassigned to handlers self.__startup_recovery() - # When assinging handlers to workflows being queued - use job_conf - # if not explicit workflow scheduling handlers have be specified or - # else use those explicit workflow scheduling handlers (on self). - if self.__handlers_configured: - self.__has_handlers = self - else: - self.__has_handlers = app.job_config - def __startup_recovery(self): sa_session = self.app.model.context - if self.__use_stack_messages: - for workflow_invocation in model.WorkflowInvocation.poll_active_workflow_ids( - sa_session, - handler=None): - log.info("(%s) Handler unassigned at startup, queueing workflow invocation via stack messaging for pool" - " [%s]", workflow_invocation.id, self.__handler_pool) - msg = WorkflowSchedulingMessage(task='setup', workflow_invocation_id=workflow_invocation.id) - self.app.application_stack.send_message(self.app.application_stack.pools.JOB_HANDLERS, msg) + for invocation_id in model.WorkflowInvocation.poll_unhandled_workflow_ids(sa_session): + log.info("(%s) Handler unassigned at startup, resubmitting workflow invocation for assignment", + invocation_id) + workflow_invocation = sa_session.query(model.WorkflowInvocation).get(invocation_id) + self._assign_handler(workflow_invocation) def _handle_setup_msg(self, workflow_invocation_id=None): sa_session = self.app.model.context @@ -99,13 +103,27 @@ class WorkflowSchedulingManager(ConfiguresHandlers): is_handler = self.app.is_job_handler return is_handler - def _get_handler(self, history_id): + def _queue_callback(self, workflow_invocation): + # There isn't an in-memory queue for workflow schedulers, so if MEM_SELF is used just assign with the DB + workflow_invocation.handler = self.app.config.server_name + sa_session = self.app.model.context + sa_session.add(workflow_invocation) + sa_session.flush() + + def _message_callback(self, workflow_invocation): + return WorkflowSchedulingMessage(task='setup', workflow_invocation_id=workflow_invocation.id) + + def _assign_handler(self, workflow_invocation): # Use random-ish integer history_id to produce a consistent index to pick # job handler with. - random_index = history_id + random_index = workflow_invocation.history_id + queue_callback = partial(self._queue_callback, workflow_invocation) + message_callback = partial(self._message_callback, workflow_invocation) if self.app.config.parallelize_workflow_scheduling_within_histories: random_index = None - return self.__has_handlers.get_handler(None, index=random_index) + return self.__handlers_config.assign_handler( + workflow_invocation, configured=None, index=random_index, queue_callback=queue_callback, + message_callback=message_callback) def shutdown(self): exception = None @@ -127,25 +145,16 @@ class WorkflowSchedulingManager(ConfiguresHandlers): def queue(self, workflow_invocation, request_params): workflow_invocation.state = model.WorkflowInvocation.states.NEW - scheduler = request_params.get("scheduler", None) or self.default_scheduler_id - handler = self._get_handler(workflow_invocation.history.id) - - if handler is None and not self.__use_stack_messages: - raise RuntimeError("Unable to set a handler for workflow invocation '%s'" % workflow_invocation.id) - - log.info("Queueing workflow invocation for handler [%s]", handler) - workflow_invocation.scheduler = scheduler - workflow_invocation.handler = handler - + workflow_invocation.scheduler = request_params.get("scheduler", None) or self.default_scheduler_id sa_session = self.app.model.context sa_session.add(workflow_invocation) - sa_session.flush() - if handler is None and self.__use_stack_messages: - log.info("(%s) Queueing workflow invocation via stack messaging for pool [%s]", - workflow_invocation.id, self.__handler_pool) - msg = WorkflowSchedulingMessage(task='setup', workflow_invocation_id=workflow_invocation.id) - self.app.application_stack.send_message(self.__handler_pool, msg) + # Assign handler (also performs the flush) + try: + self._assign_handler(workflow_invocation) + except HandlerAssignmentError: + # FIXME: test, raise new exception? + raise RuntimeError("Unable to set a handler for workflow invocation '%s'" % workflow_invocation.id) return workflow_invocation @@ -156,11 +165,17 @@ class WorkflowSchedulingManager(ConfiguresHandlers): def __plugins_dict(self): return plugin_config.plugins_dict(galaxy.workflow.schedulers, 'plugin_type') + @property + def __stack_has_pool(self): + # TODO: In the future it should be possible to map workflows to handlers based on workflow params. When that + # happens, we'll need to defer pool checks until execution time. + return any(map(self.app.application_stack.has_pool, self.DEFAULT_BASE_HANDLER_POOLS)) + def __init_schedulers(self): config_file = self.app.config.workflow_schedulers_config_file use_default_scheduler = False if not config_file: - log.info("Not workflow schedulers plugin config file defined, using default scheduler.") + log.info("No workflow schedulers plugin config file defined, using default scheduler.") use_default_scheduler = True elif not os.path.exists(config_file): log.info("Cannot find workflow schedulers plugin config file '%s', using default scheduler." % config_file) @@ -172,6 +187,11 @@ class WorkflowSchedulingManager(ConfiguresHandlers): plugins_element = ElementTree.parse(config_file).getroot() self.__init_schedulers_for_element(plugins_element) + if not self.__handlers_configured and self.__stack_has_pool: + # Stack has a pool for us so override inherited config and use the pool + self.__init_handler_assignment_methods() + self.__handlers_configured = True + def __init_default_scheduler(self): self.default_scheduler_id = DEFAULT_SCHEDULER_ID self.__init_plugin(DEFAULT_SCHEDULER_PLUGIN_TYPE) @@ -198,12 +218,27 @@ class WorkflowSchedulingManager(ConfiguresHandlers): raise Exception(EXCEPTION_MESSAGE_NO_SCHEDULERS) if self.default_scheduler_id not in self.workflow_schedulers: raise Exception(EXCEPTION_MESSAGE_NO_DEFAULT_SCHEDULER % self.default_scheduler_id) + if self.app.config.parallelize_workflow_scheduling_within_histories \ + and not self.deterministic_handler_assignment: + if self.__handlers_configured: + # There's an explicit configuration, the admin should fix it. + raise Exception(EXCEPTION_MESSAGE_SERIALIZE) + else: + log.warning(EXCEPTION_MESSAGE_SERIALIZE) def __init_handlers(self, config_element): assert not self.__handlers_configured + self.__init_handler_assignment_methods(config_element) self._init_handlers(config_element) self.__handlers_configured = True + def __init_handler_assignment_methods(self, config_element=None): + self._init_handler_assignment_methods(config_element) + self._set_default_handler_assignment_methods() + log.info("Workflow scheduling handler assignment method(s): %s", ', '.join(self.handler_assignment_methods)) + for tag, handlers in [(t, h) for t, h in self.handlers.items() if isinstance(h, list)]: + log.info("Tag [%s] handlers: %s", tag, ', '.join(handlers)) + def __init_plugin(self, plugin_type, workflow_scheduler_id=None, **kwds): workflow_scheduler_id = workflow_scheduler_id or self.default_scheduler_id @@ -241,6 +276,7 @@ class WorkflowRequestMonitor(Monitors): def __schedule(self, workflow_scheduler_id, workflow_scheduler): invocation_ids = self.__active_invocation_ids(workflow_scheduler_id) for invocation_id in invocation_ids: + log.debug("Attempting to schedule workflow invocation [%s]", invocation_id) self.__attempt_schedule(invocation_id, workflow_scheduler) if not self.monitor_running: return @@ -261,6 +297,7 @@ class WorkflowRequestMonitor(Monitors): if i.active and i.id < workflow_invocation.id: return False workflow_scheduler.schedule(workflow_invocation) + log.debug("Workflow invocation [%s] scheduled", workflow_invocation.id) except Exception: # TODO: eventually fail this - or fail it right away? log.exception("Exception raised while attempting to schedule workflow request.") diff --git a/test/unit/jobs/test_job_configuration.py b/test/unit/jobs/test_job_configuration.py index 2982cdb87cb..f10d5e9f800 100644 --- a/test/unit/jobs/test_job_configuration.py +++ b/test/unit/jobs/test_job_configuration.py @@ -25,6 +25,7 @@ class JobConfXmlParserTestCase(unittest.TestCase): job_resource_params_file="/tmp/fake_absent_path", config_dict={}, default_job_resubmission_condition="", + track_jobs_in_database=True, server_name="main", ) self.__write_config_from(SIMPLE_JOB_CONF) @@ -50,12 +51,10 @@ class JobConfXmlParserTestCase(unittest.TestCase): assert len(task_runners) == 1 assert task_runners[0]["workers"] == 5 - def test_load_simple_handler(self): - main_handler = self.job_config.handlers["main"] - assert main_handler[0] == "main" - - def test_if_one_handler_implicit_default(self): - assert self.job_config.default_handler_id == "main" + def test_if_no_handlers_implict_db_self(self): + assert self.job_config.default_handler_id is None + assert self.job_config.handlers == {} + assert self.job_config.handler_assignment_methods == ['db-self'] def test_explicit_handler_default(self): self.__with_advanced_config() diff --git a/test/unit/tools/test_actions.py b/test/unit/tools/test_actions.py index 502f65ea994..3c23507ab6b 100644 --- a/test/unit/tools/test_actions.py +++ b/test/unit/tools/test_actions.py @@ -12,8 +12,6 @@ from galaxy.tools.actions import ( from galaxy.tools.parser.output_objects import ToolOutput from .. import tools_support -TEST_HANDLER_NAME = "test_handler_1" - # I cannot think of a saner way to test if data is being wrapped than use a # data param in the output label - though you would probably never want to do @@ -73,7 +71,6 @@ class DefaultToolActionTestCase(unittest.TestCase, tools_support.UsesApp, tools_ self.app.model.context.flush() self.action = DefaultToolAction() self.app.config.len_file_path = "moocow" - self.app.job_config["get_handler"] = lambda h: TEST_HANDLER_NAME self.app.object_store = MockObjectStore() def test_output_created(self): @@ -114,10 +111,6 @@ class DefaultToolActionTestCase(unittest.TestCase, tools_support.UsesApp, tools_ # Again this is a stupid way to ensure data parameters are wrapped. self.assertEqual(output["out1"].name, "Output (%s)" % hda1.dataset.get_file_name()) - def test_handler_set(self): - job, _ = self._simple_execute() - assert job.handler == TEST_HANDLER_NAME - def test_inactive_user_job_create_failure(self): self.trans.user_is_active = False try: diff --git a/test/unit/unittest_utils/galaxy_mock.py b/test/unit/unittest_utils/galaxy_mock.py index 7c3bad26452..36f6563bd7b 100644 --- a/test/unit/unittest_utils/galaxy_mock.py +++ b/test/unit/unittest_utils/galaxy_mock.py @@ -11,7 +11,7 @@ from galaxy import ( quota ) from galaxy.datatypes import registry -from galaxy.jobs import NoopQueue +from galaxy.jobs.manager import NoopManager from galaxy.managers import tags from galaxy.model import mapping from galaxy.tools.deps.containers import NullContainerFinder @@ -76,7 +76,7 @@ class MockApp(object): self.container_finder = NullContainerFinder() self._toolbox_lock = MockLock() self.genome_builds = GenomeBuilds(self) - self.job_manager = Bunch(job_queue=NoopQueue()) + self.job_manager = NoopManager() self.application_stack = ApplicationStack() def init_datatypes(self):