From 4e14a648f3317581bee7c2228db7725b968b3ff7 Mon Sep 17 00:00:00 2001 From: mvdbeek Date: Wed, 10 Apr 2019 13:48:19 +0200 Subject: [PATCH] Add hostname to task message queues This prevents processes with the same server_name on different hosts to execute processes not destined for them. This time I think it closes https://github.com/galaxyproject/galaxy/issues/5601 --- lib/galaxy/model/__init__.py | 3 ++- lib/galaxy/model/mapping.py | 5 +++-- .../migrate/versions/0151_add_worker_process.py | 7 +++++-- lib/galaxy/queue_worker.py | 4 ++-- lib/galaxy/queues.py | 16 ++++++++-------- lib/galaxy/web/stack/database_heartbeat.py | 8 ++++++-- 6 files changed, 26 insertions(+), 17 deletions(-) diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py index 6edef04fddf..933ac424cc0 100644 --- a/lib/galaxy/model/__init__.py +++ b/lib/galaxy/model/__init__.py @@ -219,8 +219,9 @@ class UsesCreateAndUpdateTime(object): class WorkerProcess(UsesCreateAndUpdateTime): - def __init__(self, server_name): + def __init__(self, server_name, hostname): self.server_name = server_name + self.hostname = hostname def cached_id(galaxy_model_object): diff --git a/lib/galaxy/model/mapping.py b/lib/galaxy/model/mapping.py index 54ec5ef59ee..3ca43d4d913 100644 --- a/lib/galaxy/model/mapping.py +++ b/lib/galaxy/model/mapping.py @@ -50,11 +50,12 @@ metadata = MetaData() model.WorkerProcess.table = Table( 'worker_process', metadata, - Column('server_name', Text, primary_key=True), + Column("id", Integer, primary_key=True), + Column("server_name", Text, index=True), + Column("hostname", Text), Column("update_time", DateTime, default=now, onupdate=now), ) - model.User.table = Table( "galaxy_user", metadata, Column("id", Integer, primary_key=True), diff --git a/lib/galaxy/model/migrate/versions/0151_add_worker_process.py b/lib/galaxy/model/migrate/versions/0151_add_worker_process.py index c16ed1a63ca..8bb2e3b7037 100644 --- a/lib/galaxy/model/migrate/versions/0151_add_worker_process.py +++ b/lib/galaxy/model/migrate/versions/0151_add_worker_process.py @@ -8,9 +8,10 @@ import logging from sqlalchemy import ( Column, DateTime, + Integer, MetaData, Table, - TEXT, + Text, ) from galaxy.model.migrate.versions.util import create_table, drop_table @@ -23,7 +24,9 @@ metadata = MetaData() WorkerProcess_table = Table( 'worker_process', metadata, - Column('server_name', TEXT, primary_key=True), + Column("id", Integer, primary_key=True), + Column("server_name", Text, index=True), + Column("hostname", Text), Column("update_time", DateTime, default=now, onupdate=now), ) diff --git a/lib/galaxy/queue_worker.py b/lib/galaxy/queue_worker.py index 9760a30d8cb..7d5b3302858 100644 --- a/lib/galaxy/queue_worker.py +++ b/lib/galaxy/queue_worker.py @@ -34,7 +34,7 @@ def send_local_control_task(app, task, kwargs={}): log.info("Queuing async task %s for %s." % (task, app.config.server_name)) payload = {'task': task, 'kwargs': kwargs} - routing_key = 'control.%s' % app.config.server_name + routing_key = 'control.%s@%s' % (app.config.server_name, socket.gethostname()) control_task = ControlTask(app.control_worker) control_task.send_task(payload, routing_key, local=True, get_response=False) @@ -317,7 +317,7 @@ class GalaxyQueueWorker(ConsumerProducerMixin, threading.Thread): @property def declare_queues(self): # dynamically produce queues, allows addressing all known processes at a given time - return galaxy.queues.all_control_queues_for_declare(self.app.config, self.app.application_stack) + return galaxy.queues.all_control_queues_for_declare(self.app.application_stack) def bind_and_start(self): # This is post-forking, so we got the correct sever name diff --git a/lib/galaxy/queues.py b/lib/galaxy/queues.py index 56c08753c4a..abb9b993027 100644 --- a/lib/galaxy/queues.py +++ b/lib/galaxy/queues.py @@ -3,6 +3,7 @@ All message queues used by Galaxy """ +import socket from kombu import ( Connection, @@ -14,17 +15,14 @@ ALL_CONTROL = "control.*" galaxy_exchange = Exchange('galaxy_core_exchange', type='topic') -def all_control_queues_for_declare(config, application_stack): +def all_control_queues_for_declare(application_stack): """ For in-memory routing (used by sqlalchemy-based transports), we need to be able to build the entire routing table in producers. """ # Get all active processes and construct queues for each process - if application_stack and application_stack.app: - server_names = (p.server_name for p in application_stack.app.database_heartbeat.get_active_processes()) - else: - server_names = config.server_names - return [Queue("control.%s" % server_name, galaxy_exchange, routing_key='control.*') for server_name in server_names] + process_names = ("{p.server_name}@{p.hostname}".format(p=p) for p in application_stack.app.database_heartbeat.get_active_processes()) + return [Queue("control.%s" % server_name, galaxy_exchange, routing_key='control.*') for server_name in process_names] def control_queues_from_config(config): @@ -32,8 +30,10 @@ def control_queues_from_config(config): Returns a Queue instance with the correct name and routing key for this galaxy process's config """ - exchange_queue = Queue("control.%s" % config.server_name, galaxy_exchange, routing_key='control.%s' % config.server_name) - non_exchange_queue = Queue("control.%s" % config.server_name, routing_key='control.%s' % config.server_name) + hostname = socket.gethostname() + process_name = "{server_name}@{hostname}".format(server_name=config.server_name, hostname=hostname) + exchange_queue = Queue("control.%s" % process_name, galaxy_exchange, routing_key='control.%s' % process_name) + non_exchange_queue = Queue("control.%s" % process_name, routing_key='control.%s' % process_name) return exchange_queue, non_exchange_queue diff --git a/lib/galaxy/web/stack/database_heartbeat.py b/lib/galaxy/web/stack/database_heartbeat.py index 483e8a5d0ea..97767a643a1 100644 --- a/lib/galaxy/web/stack/database_heartbeat.py +++ b/lib/galaxy/web/stack/database_heartbeat.py @@ -1,4 +1,5 @@ import datetime +import socket import threading from galaxy.model import WorkerProcess @@ -10,6 +11,7 @@ class DatabaseHeartbeat(object): def __init__(self, application_stack, heartbeat_interval=60): self.application_stack = application_stack self.heartbeat_interval = heartbeat_interval + self.hostname = socket.gethostname() self.exit = threading.Event() self.thread = None self.active = False @@ -47,9 +49,11 @@ class DatabaseHeartbeat(object): if self.active: while not self.exit.isSet(): worker_process = self.sa_session.query(WorkerProcess).filter_by( - server_name=self.server_name).first() + server_name=self.server_name, + hostname=self.hostname, + ).first() if not worker_process: - worker_process = WorkerProcess(server_name=self.server_name) + worker_process = WorkerProcess(server_name=self.server_name, hostname=self.hostname) worker_process.update_time = now() self.sa_session.add(worker_process) self.sa_session.flush()