From e2e3e5f7d12a2f862595b380bf9da12b3b201333 Mon Sep 17 00:00:00 2001 From: Nate Coraor Date: Thu, 27 Jul 2017 13:28:46 -0400 Subject: [PATCH] WIP job handlers as uWSGI mules from GCC 2017 Hackathon. --- lib/galaxy/config.py | 23 +-- lib/galaxy/jobs/handler.py | 39 ++--- lib/galaxy/jobs/manager.py | 29 +++- lib/galaxy/model/__init__.py | 9 +- lib/galaxy/util/handlers.py | 11 +- lib/galaxy/web/stack/__init__.py | 61 +++++++ lib/galaxy/web/stack/uwsgi_mule/handler.py | 179 +++++++++++++++++++++ scripts/get_uwsgi_args.py | 10 +- 8 files changed, 305 insertions(+), 56 deletions(-) create mode 100644 lib/galaxy/web/stack/uwsgi_mule/handler.py diff --git a/lib/galaxy/config.py b/lib/galaxy/config.py index 146cd033518..9e1b4072bcf 100644 --- a/lib/galaxy/config.py +++ b/lib/galaxy/config.py @@ -20,10 +20,6 @@ from datetime import timedelta from six import string_types from six.moves import configparser -try: - import uwsgi -except: - uwsgi = None from galaxy.containers import parse_containers_config from galaxy.exceptions import ConfigurationError @@ -503,17 +499,14 @@ class Configuration(object): if self.heartbeat_log is None: self.heartbeat_log = 'heartbeat_{server_name}.log' # Determine which 'server:' this is - if uwsgi and uwsgi.mule_id() > 0: - self.server_name = 'mule%d' % uwsgi.mule_id() - else: - self.server_name = 'main' - for arg in sys.argv: - # Crummy, but PasteScript does not give you a way to determine this - if arg.lower().startswith('--server-name='): - self.server_name = arg.split('=', 1)[-1] - # Allow explicit override of server name in confg params - if "server_name" in kwargs: - self.server_name = kwargs.get("server_name") + self.server_name = 'main' + for arg in sys.argv: + # Crummy, but PasteScript does not give you a way to determine this + if arg.lower().startswith('--server-name='): + self.server_name = arg.split('=', 1)[-1] + # Allow explicit override of server name in confg params + if "server_name" in kwargs: + self.server_name = kwargs.get("server_name") # Store all configured server names self.server_names = [] for section in global_conf_parser.sections(): diff --git a/lib/galaxy/jobs/handler.py b/lib/galaxy/jobs/handler.py index b3702116504..38f8be4b525 100644 --- a/lib/galaxy/jobs/handler.py +++ b/lib/galaxy/jobs/handler.py @@ -11,10 +11,6 @@ import threading from Queue import Queue, Empty from sqlalchemy.sql.expression import and_, or_, select, func, true, null -try: - import uwsgi -except ImportError: - uwsgi = None from galaxy import model from galaxy.util.sleeper import Sleeper @@ -81,9 +77,6 @@ class JobHandlerQueue(object): self.running = True self.monitor_thread = threading.Thread(name="JobHandlerQueue.monitor_thread", target=self.__monitor) self.monitor_thread.setDaemon(True) - if uwsgi: - self.mule_thread = threading.Thread( name="JobHandlerQueue.mule_thread", target=self.__mule ) - self.mule_thread.setDaemon( True ) def start(self): """ @@ -93,8 +86,7 @@ class JobHandlerQueue(object): self.__check_jobs_at_startup() # Start the queue self.monitor_thread.start() - if uwsgi: - self.mule_thread.start() + self.app.application_stack.initialize_handler(self.app) log.info("job handler queue started") def job_wrapper(self, job, use_persisted_destination=False): @@ -181,19 +173,6 @@ class JobHandlerQueue(object): job_wrapper.job_runner_mapper.cached_job_destination = job_destination return job_wrapper - def __mule( self ): - while self.running: - try: - msg = uwsgi.mule_get_msg() - job_dict = json.loads(msg) - job = self.sa_session.query( model.Job ).get( job_dict['job_id'] ) - job.handler = self.app.config.server_name - self.sa_session.add( job ) - self.sa_session.flush() - except: - log.exception( "Exception in mule message handling" ) - time.sleep( 1 ) - def __monitor(self): """ Continually iterate the waiting jobs, checking is each is ready to @@ -653,6 +632,21 @@ class JobHandlerQueue(object): return JOB_WAIT return JOB_READY + def __handle_setup_msg(self, job_dict): + job = self.sa_session.query(model.Job).get(job_dict['job_id']) + job.handler = self.app.config.server_name + self.sa_session.add(job) + self.sa_session.flush() + + def handle_msg(self, msg): + try: + job_dict = json.loads(msg) + assert 'msg_type' in job_dict, "Missing required 'msg_type' message parameter" + getattr(self, '__handle_%s_msg' % job_dict['msg_type'])(job_dict) + except: + log.exception( "Exception in mule message handling" ) + time.sleep( 1 ) + def put(self, job_id, tool_id): """Add a job to the queue (by job identifier)""" if not self.track_jobs_in_database: @@ -667,6 +661,7 @@ class JobHandlerQueue(object): else: log.info("sending stop signal to worker thread") self.running = False + self.app.application_stack.terminate_handler() if not self.app.config.track_jobs_in_database: self.queue.put(self.STOP_SIGNAL) self.sleeper.wake() diff --git a/lib/galaxy/jobs/manager.py b/lib/galaxy/jobs/manager.py index 811cb637f5e..2dc54c40993 100644 --- a/lib/galaxy/jobs/manager.py +++ b/lib/galaxy/jobs/manager.py @@ -2,9 +2,11 @@ Top-level Galaxy job manager, moves jobs to handler(s) """ +import json import logging from galaxy.jobs import handler, NoopQueue +from galaxy.model import Job log = logging.getLogger(__name__) @@ -22,11 +24,15 @@ class JobManager(object): if self.app.is_job_handler(): log.debug("Starting job handler") self.job_handler = handler.JobHandler(app) - self.job_queue = self.job_handler.job_queue self.job_stop_queue = self.job_handler.job_stop_queue + elif app.application_stack.setup_jobs_with_msg: + self.job_handler = NoopHandler() + self.job_handler.job_queue = MessageJobQueue(app) + self.job_stop_queue = NoopQueue() else: self.job_handler = NoopHandler() - self.job_queue = self.job_stop_queue = NoopQueue() + self.job_stop_queue = NoopQueue + self.job_queue = self.job_handler.job_queue self.job_lock = False def start(self): @@ -46,3 +52,22 @@ class NoopHandler(object): def shutdown(self, *args): pass + + +class MessageJobQueue( object ): + """ + 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): + # FIXME: uwsgi farm name hardcoded here + # TODO: probably need a single class that encodes and decodes messages + self.app.application_stack.send_msg(json.dumps({'msg_type': 'setup', 'job_id': job_id, 'state': Job.states.NEW}), 'handlers') + + def put_stop(self, *args): + return + + def shutdown(self): + return diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py index ebd7f384a40..9b06e090d68 100644 --- a/lib/galaxy/model/__init__.py +++ b/lib/galaxy/model/__init__.py @@ -23,10 +23,6 @@ from sqlalchemy import (and_, func, join, not_, or_, select, true, type_coerce, types) from sqlalchemy.ext import hybrid from sqlalchemy.orm import aliased, joinedload, object_session -try: - import uwsgi -except ImportError: - uwsgi = None import galaxy.model.metadata import galaxy.model.orm.now @@ -654,10 +650,7 @@ class Job(object, JobLike, Dictifiable): self.imported = imported def set_handler(self, handler): - if uwsgi and handler is None: - uwsgi.mule_msg(json.dumps({'job_id': self.id, 'state': Job.states.NEW})) - else: - self.handler = handler + self.handler = handler def set_params(self, params): self.params = params diff --git a/lib/galaxy/util/handlers.py b/lib/galaxy/util/handlers.py index 9dee5d23761..ccd825b9953 100644 --- a/lib/galaxy/util/handlers.py +++ b/lib/galaxy/util/handlers.py @@ -8,10 +8,7 @@ import logging import os import random -try: - import uwsgi -except ImportError: - uwsgi = None +from galaxy.web.stack import process_is_handler log = logging.getLogger(__name__) @@ -106,8 +103,10 @@ class ConfiguresHandlers: :return: bool """ - if uwsgi and self.app.config.server_name.startswith('mule'): - return True + stack_handler = process_is_handler(self.app) + if stack_handler is not None: + # Handlers started as uWSGI mules do not require configuration in the job conf + return stack_handler for collection in self.handlers.values(): if server_name in collection: return True diff --git a/lib/galaxy/web/stack/__init__.py b/lib/galaxy/web/stack/__init__.py index 3b7689be4d5..0b7a323bcb1 100644 --- a/lib/galaxy/web/stack/__init__.py +++ b/lib/galaxy/web/stack/__init__.py @@ -5,6 +5,7 @@ from __future__ import print_function import inspect import logging import os +import threading # The uwsgi module is automatically injected by the parent uwsgi # process and only exists that way. If anything works, this is a @@ -25,6 +26,8 @@ except: "HINT:\n {venv}/bin/pip install uwsgidecorators".format( venv=os.environ.get('VIRTUAL_ENV', '/path/to/venv'))) +from galaxy import model + log = logging.getLogger(__name__) @@ -32,6 +35,7 @@ log = logging.getLogger(__name__) class ApplicationStack(object): name = None prohibited_middleware = frozenset() + setup_jobs_with_msg = False @classmethod def register_postfork_function(cls, f, *args, **kwargs): @@ -45,6 +49,15 @@ class ApplicationStack(object): middleware = middleware.__name__ return middleware not in self.prohibited_middleware + def process_is_handler(self, app): + return None + + def initialize_handler(self, app): + return None + + def send_msg(self, msg, dest): + pass + def set_postfork_server_name(self, app): pass @@ -55,16 +68,60 @@ class UWSGIApplicationStack(ApplicationStack): 'wrap_in_static', 'EvalException', ]) + setup_jobs_with_msg = True postfork_functions = [] + handler_prefix = 'mule-handler-' @classmethod def register_postfork_function(cls, f, *args, **kwargs): cls.postfork_functions.append((f, args, kwargs)) + def __handle_msgs(self): + while self.running: + try: + log.debug('################ waiting for mule farm msg with mule %s', uwsgi.mule_id()) + # Hopefully farm_get_msg blocks like mule_get_msg does + #self.msg_handler(uwsgi.farm_get_msg()) + #msg = uwsgi.farm_get_msg() + msg = uwsgi.mule_get_msg() + log.debug('################ received msg: %s', msg) + self.msg_handler(msg) + except: + log.exception( "Exception in mule message handling" ) + + def __init__(self): + super(UWSGIApplicationStack, self).__init__() + self.app = None + self.running = False + self.msg_handler = None + self.msg_thread = threading.Thread(name="UWSGIApplicationStack.mule_msg_thread", target=self.__handle_msgs) + self.msg_thread.daemon = True + def workers(self): return uwsgi.workers() + def process_is_handler(self, app): + if app.config.server_name.startswith(UWSGIApplicationStack.handler_prefix): + return True + return False + + def initialize_handler(self, app): + self.app = app + self.running = True + self.msg_handler = self.app.job_manager.job_queue.handle_msg + self.msg_thread.start() + + def send_msg(self, msg, dest): + # TODO: the handler farm name should be configurable + log.debug('################## sending message from mule %s: %s', uwsgi.mule_id(), msg) + #uwsgi.farm_msg(msg, dest) + uwsgi.mule_msg(msg) + log.debug('################## message sent') + + def terminate_handler(self): + self.running = False + def set_postfork_server_name(self, app): app.config.server_name += ".%d" % uwsgi.worker_id() @@ -100,6 +157,10 @@ def register_postfork_function(f, *args, **kwargs): application_stack_class().register_postfork_function(f, *args, **kwargs) +def process_is_handler(app): + return application_stack_instance().process_is_handler(app) + + @uwsgi_postfork def _do_postfork(): for f, args, kwargs in [t for t in UWSGIApplicationStack.postfork_functions]: diff --git a/lib/galaxy/web/stack/uwsgi_mule/handler.py b/lib/galaxy/web/stack/uwsgi_mule/handler.py new file mode 100644 index 00000000000..4abd90f7876 --- /dev/null +++ b/lib/galaxy/web/stack/uwsgi_mule/handler.py @@ -0,0 +1,179 @@ +#!/usr/bin/env python +""" Entry point for starting a Galaxy job handler as a uWSGI mule + +Example Usage: + +uwsgi ... --mule=lib/galaxy/web/stack/uwsgi_mule/handler.py --mule=lib/galaxy/web/stack/uwsgi_mule/handler.py --farm=handlers:1,2 +""" +import logging +import os +import sys +import time +from logging.config import fileConfig + +import uwsgi + +try: + import ConfigParser as configparser +except ImportError: + import configparser + + +SERVER_NAME_TEMPLATE = 'mule-handler-{mule_id}' + +log = logging.getLogger(__name__) + +real_file = os.path.realpath(__file__) +GALAXY_ROOT_DIR = os.path.abspath(os.path.join(os.path.dirname(real_file), os.pardir, os.pardir, os.pardir, os.pardir, os.pardir)) +GALAXY_LIB_DIR = os.path.join(GALAXY_ROOT_DIR, "lib") +DEFAULT_INI_APP = "main" +DEFAULT_INIS = ["config/galaxy.ini", "universe_wsgi.ini", "config/galaxy.ini.sample"] + + +def load_galaxy_app( + config_builder, + config_env=False, + log=None, + **kwds +): + # Allow specification of log so daemon can reuse properly configured one. + if log is None: + log = logging.getLogger(__name__) + + # If called in daemon mode, set the ROOT directory and ensure Galaxy is on + # sys.path. + if config_env: + try: + os.chdir(GALAXY_ROOT_DIR) + except Exception: + log.exception("Failed to chdir") + raise + try: + sys.path.insert(1, GALAXY_LIB_DIR) + except Exception: + log.exception("Failed to add Galaxy to sys.path") + raise + + config_builder.setup_logging() + from galaxy.util.properties import load_app_properties + kwds = config_builder.app_kwds() + kwds = load_app_properties(**kwds) + from galaxy.app import UniverseApplication + app = UniverseApplication( + global_conf={"__file__": config_builder.ini_path}, + **kwds + ) + app.control_worker.bind_and_start() + return app + + +def app_loop(ini_path, log): + try: + config_builder = GalaxyConfigBuilder(ini_path=ini_path) + galaxy_app = load_galaxy_app( + config_builder, + config_env=True, + log=log, + ) + except BaseException: + log.exception("Failed to initialize Galaxy application") + raise + sleep = True + while sleep: + try: + time.sleep(5) + except KeyboardInterrupt: + sleep = False + except SystemExit: + sleep = False + except Exception: + pass + try: + galaxy_app.shutdown() + except Exception: + log.exception("Failed to shutdown Galaxy application") + raise + + +def absolute_config_path(path, galaxy_root): + if path and not os.path.isabs(path): + path = os.path.join(galaxy_root, path) + return path + + +def find_ini(supplied_ini, galaxy_root): + if supplied_ini: + return supplied_ini + + # If not explicitly supplied an ini, check server.ini and then + # just restort to sample if that has not been configured. + for guess in DEFAULT_INIS: + ini_path = os.path.join(galaxy_root, guess) + if os.path.exists(ini_path): + return ini_path + + return guess + + +class GalaxyConfigBuilder(object): + """ Generate paste-like configuration from supplied command-line arguments. + """ + + def __init__(self, **kwds): + ini_path = kwds.get("ini_path", None) + # If given app_conf_path - use that - else we need to ensure we have an + # ini path. + if not ini_path: + galaxy_root = kwds.get("galaxy_root", GALAXY_ROOT_DIR) + ini_path = find_ini(ini_path, galaxy_root) + ini_path = absolute_config_path(ini_path, galaxy_root=galaxy_root) + self.ini_path = ini_path + self.app_name = kwds.get("app") or DEFAULT_INI_APP + + def app_kwds(self): + config = dict( + ini_file=self.ini_path, + ini_section="app:%s" % self.app_name, + ) + return config + + def setup_logging(self): + # Galaxy will attempt to setup logging if loggers is not present in + # ini config file - this handles that loggers block however if present + # (the way paste normally would) + if not self.ini_path: + return + raw_config = configparser.ConfigParser() + raw_config.read([self.ini_path]) + if raw_config.has_section('loggers'): + config_file = os.path.abspath(self.ini_path) + fileConfig( + config_file, + dict(__file__=config_file, here=os.path.dirname(config_file)) + ) + + +def galaxy_config_from_uwsgi(): + if 'ini-paste' in uwsgi.opt: + return uwsgi.opt['ini-paste'] + elif 'set' in uwsgi.opt: + for k, v in [x.split('=', 1) for x in uwsgi.opt['set']]: + if k == 'galaxy_config_file': + return v + return None + + +def main(): + # FIXME: need file logging... set=galaxy_mule_handler_log_dir=/path ?? + #if args.log_file: + # os.environ["GALAXY_CONFIG_LOG_DESTINATION"] = os.path.abspath(args.log_file) + os.environ["GALAXY_CONFIG_SERVER_NAME"] = SERVER_NAME_TEMPLATE.format(mule_id=uwsgi.mule_id()) + ini_path = galaxy_config_from_uwsgi() + + log.setLevel(logging.DEBUG) + log.propagate = False + app_loop(ini_path, log) + + +if __name__ == "__main__": + main() diff --git a/scripts/get_uwsgi_args.py b/scripts/get_uwsgi_args.py index 439f0d0c2d6..0abe50d3807 100644 --- a/scripts/get_uwsgi_args.py +++ b/scripts/get_uwsgi_args.py @@ -1,6 +1,6 @@ from __future__ import print_function -import os.path +import os import sys sys.path.insert(1, os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir, 'lib'))) @@ -18,9 +18,13 @@ DESCRIPTION = "Script to determine uWSGI command line arguments." def _get_uwsgi_args(args, kwargs): - print('--ini-paste {galaxy_ini}{mule}'.format( + handlerct = int(kwargs.get('job_handler_count', 1)) + virtualenv = os.environ.get('VIRTUAL_ENV', None) + print('{virtualenv}--ini-paste {galaxy_ini}{mule}{farm}'.format( + virtualenv='--virtualenv %s ' % virtualenv if virtualenv else '', galaxy_ini=args.config_file, - mule=' --mule=lib/galaxy/main.py' * int(kwargs.get('job_handler_count', 1)))) + mule=' --mule=lib/galaxy/main.py' * handlerct, + farm=' --farm=handlers:%s' % ','.join([str(x) for x in range(1, handlerct + 1)]))) ACTIONS = {