diff --git a/config/workflow_schedulers_conf.xml.sample b/config/workflow_schedulers_conf.xml.sample index c7572d9da6e..e719453389e 100644 --- a/config/workflow_schedulers_conf.xml.sample +++ b/config/workflow_schedulers_conf.xml.sample @@ -11,4 +11,15 @@ require any external dependencies. --> + + + diff --git a/lib/galaxy/jobs/handler.py b/lib/galaxy/jobs/handler.py index 38b39cb1252..0d7547421e9 100644 --- a/lib/galaxy/jobs/handler.py +++ b/lib/galaxy/jobs/handler.py @@ -87,7 +87,8 @@ class JobHandlerQueue(object): # Start the queue self.monitor_thread.start() # The stack code is initialized in the application - self.app.application_stack.register_message_handler(self.handle_msg, name=JobHandlerMessage.target) + JobHandlerMessage().bind_default_handler(self, '_handle_message') + self.app.application_stack.register_message_handler(self._handle_message, name=JobHandlerMessage.target) log.info("job handler queue started") def job_wrapper(self, job, use_persisted_destination=False): @@ -644,17 +645,19 @@ class JobHandlerQueue(object): else: log.warning("(%s) Handler '%s' received setup message but handler '%s' is already assigned, ignoring", job.id, self.app.config.server_name, job.handler) - def handle_msg(self, msg): - try: - getattr(self, '_handle_%s_msg' % msg.task)(**msg.params) - except: - log.exception( "Exception in mule message handling" ) - def put(self, job_id, tool_id): """Add a job to the queue (by job identifier)""" 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/web/stack/message.py b/lib/galaxy/web/stack/message.py index 111794689c3..e972572133b 100644 --- a/lib/galaxy/web/stack/message.py +++ b/lib/galaxy/web/stack/message.py @@ -4,6 +4,7 @@ from __future__ import absolute_import import json import logging +from types import MethodType log = logging.getLogger(__name__) @@ -45,15 +46,52 @@ class ApplicationStackMessageDispatcher(object): class ApplicationStackMessage(dict): target = None + default_handler = None + _validate_kwargs = ('target',) - def __init__(self, target=None, params=None, **kwargs): + def __init__(self, target=None, **kwargs): self['target'] = target or self.__class__.target - self['params'] = params or {} - for k, v in kwargs.items(): - self['params'][k] = v + self._merge_class_tuples() + + def _merge_class_tuples(self): + """Locates any class-level tuples beginning with a single (but not double) underscore in the MRO and creates a + property on the instance with the same name (without the leading underscore) that will return the union of + those tuples. + """ + names = set() + for cls in reversed(self.__class__.mro()): + names.update([x for x in dir(cls) if x.startswith('_') and not x.startswith('__') and type(getattr(cls, x)) == tuple]) + for name in names: + setattr(self.__class__, name.lstrip('_'), property(lambda self, name=name: self._get_list_from_mro(name))) + + def _get_list_from_mro(self, name): + """Locate all class-level tuples with the given `name` in the MRO and return their union. + """ + r = set() + for cls in reversed(self.__class__.mro()): + r.update(getattr(cls, name, [])) + return r + + def _validate_items(self, obj, items, name): + for item in items: + assert item in obj, "Missing '%s' message %" % (item, name) def validate(self): - assert self['target'] is not None, "Missing 'target' parameter" + self._validate_items(self, self.validate_kwargs, 'argument') + + def encode(self): + self['__classname__'] = self.__class__.__name__ + return json.dumps(self) + + def bind_default_handler(self, obj, name): + """Bind the default handler method to `obj` as attribute `name`. + + This could also be implemented as a mixin class. + """ + assert self.default_handler is not None, '%s has no default handler method, cannot bind' % self.__class__.__name__ + setattr(obj, name, MethodType(self.default_handler, obj, obj.__class__)) + log.debug("Bound default message handler '%s.%s' to %s", self.__class__.__name__, self.default_handler.__name__, + getattr(obj, name)) @property def target(self): @@ -63,37 +101,62 @@ class ApplicationStackMessage(dict): def set_target(self, target): self['target'] = target + +class ParamMessage(ApplicationStackMessage): + _validate_kwargs = ('params',) + _validate_params = () + _exclude_params = () + + def __init__(self, target=None, params=None, **kwargs): + super(ParamMessage, self).__init__(target=target) + self['params'] = params or {} + for k, v in kwargs.items(): + self['params'][k] = v + + def validate(self): + super(ParamMessage, self).validate() + self._validate_items(self['params'], self.validate_params, 'parameters') + @property def params(self): - return self['params'] + d = self['params'].copy() + for key in self.exclude_params: + d.pop(key, None) + return d @params.setter def set_params(self, params): self['params'] = params - def encode(self): - self['__classname__'] = self.__class__.__name__ - return json.dumps(self) +class TaskMessage(ParamMessage): + _validate_params = ('task',) + _exclude_params = ('task',) -# TODO: when additional messages are added we should refactor and improve validation and param/task separation (but not all msgs may use task) -class JobHandlerMessage(ApplicationStackMessage): - target = 'job_handler' - - def validate(self): - super(JobHandlerMessage, self).validate() - for param in ('task', 'job_id'): - assert param in self['params'], "Missing required parameter '%s'" % param + @staticmethod + def default_handler(self, msg): + """Can be bound to an instance of any class that has message handling methods named like `_handle_{task}_method` + """ + name = '_handle_{task}_msg'.format(task=msg.task) + assert name in dir(self), "{cls} has no method _handle_{task}_msg, cannot handle message: {msg}".format( + cls=self.__class__.__name__, + task=msg.task, + msg=msg) + getattr(self, '_handle_%s_msg' % msg.task)(**msg.params) @property def task(self): return self['params']['task'] - @property - def params(self): - d = self['params'].copy() - del d['task'] - return d + +class JobHandlerMessage(TaskMessage): + target = 'job_handler' + _validate_params = ('job_id',) + + +class WorkflowSchedulingMessage(TaskMessage): + target = 'workflow_scheduling' + _validate_params = ('workflow_invocation_id',) def decode(msg_str): diff --git a/lib/galaxy/workflow/scheduling_manager.py b/lib/galaxy/workflow/scheduling_manager.py index 5b6d1ee52dd..24b47b61d5f 100644 --- a/lib/galaxy/workflow/scheduling_manager.py +++ b/lib/galaxy/workflow/scheduling_manager.py @@ -8,6 +8,7 @@ import galaxy.workflow.schedulers from galaxy import model from galaxy.util import plugin_config from galaxy.util.handlers import ConfiguresHandlers +from galaxy.web.stack.message import WorkflowSchedulingMessage log = logging.getLogger(__name__) @@ -31,6 +32,10 @@ class WorkflowSchedulingManager(object, ConfiguresHandlers): 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 + 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 @@ -45,9 +50,14 @@ class WorkflowSchedulingManager(object, ConfiguresHandlers): self.__start_schedulers() if self.active_workflow_schedulers: self.__start_request_monitor() + if self.__use_stack_messages: + WorkflowSchedulingMessage().bind_default_handler(self, '_handle_message') + self.app.application_stack.register_message_handler( + self._handle_message, + name=WorkflowSchedulingMessage.target) else: - # Process should not schedule workflows - do nothing. - pass + # 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 @@ -57,6 +67,29 @@ class WorkflowSchedulingManager(object, ConfiguresHandlers): 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) + + def _handle_setup_msg(self, workflow_invocation_id=None): + sa_session = self.app.model.context + workflow_invocation = sa_session.query(model.WorkflowInvocation).get(workflow_invocation_id) + if workflow_invocation.handler is None: + workflow_invocation.handler = self.app.config.server_name + sa_session.add(workflow_invocation) + sa_session.flush() + else: + log.warning("(%s) Handler '%s' received setup message for workflow invocation but handler '%s' is" + " already assigned, ignoring", workflow_invocation.id, self.app.config.server_name, + workflow_invocation.handler) + def _is_workflow_handler(self): # If we have explicitly configured handlers, check them. # Else just make sure we are a job handler. @@ -90,14 +123,24 @@ class WorkflowSchedulingManager(object, ConfiguresHandlers): 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) - log.info("Queueing workflow invocation for handler [%s]" % handler) + 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 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) + return workflow_invocation def __start_schedulers(self):