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

@@ -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):

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),

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),
 )
 

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

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
 
 

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()