Add an optional relay topic prefix

Author: mvdbeek

app.yml.sample

@@ -83,6 +83,18 @@
 ## Python library is available and start the Pulsar normally.
 #message_queue_url: amqp://guest:guest@localhost:5672//
 
+## Alternatively, use pulsar-relay (experimental) with HTTP/HTTPS instead of AMQP.
+## This does not require kombu or RabbitMQ.
+#message_queue_url: https://relay-server.example.org:9000
+#message_queue_username: admin
+#message_queue_password: changeme
+
+## Optional topic prefix for relay messages. This allows multiple independent
+## Galaxy/Pulsar instance pairs to share the same relay server by using different
+## prefixes (e.g., "production", "staging"). Must match the relay_topic_prefix
+## configured in Galaxy's job_conf.yml if set.
+#relay_topic_prefix: production
+
 ## Pulsar loops over waiting for queue messages for a short time before checking
 ## to see if it has been instructed to shut down. By default this is 0.2
 ## seconds. This value is used as the value of the 'timeout' parameter to

docs/configure.rst

@@ -267,6 +267,16 @@ To configure Pulsar to use pulsar-relay, set the ``message_queue_url`` in
 The ``http://`` / ``https://`` prefix tells Pulsar to use the proxy communication mode instead
 of AMQP.
 
+**Optional Topic Prefix**
+
+You can optionally set a ``relay_topic_prefix`` to namespace your topics. This is useful
+when multiple independent Galaxy/Pulsar instance pairs share the same relay::
+
+    message_queue_url: http://proxy-server.example.org:9000
+    message_queue_username: admin
+    message_queue_password: your_secure_password
+    relay_topic_prefix: production
+
 .. note::
 
     Unlike AMQP mode, the pulsar-relay mode does **not** require the ``kombu``
@@ -286,6 +296,8 @@ with proxy parameters::
         proxy_url: http://proxy-server.example.org:9000
         proxy_username: your_username
         proxy_password: your_secure_password
+        # Optional topic prefix (must match Pulsar configuration)
+        # relay_topic_prefix: production
 
 
     execution:
@@ -298,6 +310,11 @@ with proxy parameters::
           # Remote job staging directory
           jobs_directory: /data/pulsar/staging
 
+.. note::
+
+    The ``relay_topic_prefix`` must match on both Galaxy and Pulsar sides.
+    If set on one side but not the other, messages will not be routed correctly.
+
 
 Authentication
 ``````````````
@@ -368,18 +385,105 @@ In Galaxy's job configuration, route jobs to specific clusters using the
           manager: cluster_b
           # ... other settings
 
+Multiple Galaxy/Pulsar Instance Pairs
+``````````````````````````````````````
+
+You can have multiple independent Galaxy and Pulsar instance pairs all sharing
+the same relay by using different topic prefixes. This is useful for:
+
+* Running separate production and staging environments
+* Supporting multiple research groups with isolated instances
+* Multi-tenant deployments
+
+**Example: Production and Staging Environments**
+
+**Production Pulsar** (``app.yml``)::
+
+    message_queue_url: https://shared-relay:9000
+    message_queue_username: admin
+    message_queue_password: password
+    relay_topic_prefix: production
+    managers:
+      cluster_a:
+        type: queued_slurm
+
+**Staging Pulsar** (``app.yml``)::
+
+    message_queue_url: https://shared-relay:9000
+    message_queue_username: admin
+    message_queue_password: password
+    relay_topic_prefix: staging
+    managers:
+      cluster_a:
+        type: queued_slurm
+
+**Production Galaxy** (``job_conf.yml``)::
+
+    runners:
+      pulsar:
+        load: galaxy.jobs.runners.pulsar:PulsarMQJobRunner
+        proxy_url: https://shared-relay:9000
+        proxy_username: admin
+        proxy_password: password
+        relay_topic_prefix: production
+
+    execution:
+      environments:
+        pulsar_jobs:
+          runner: pulsar
+          manager: cluster_a
+          # ... other settings
+
+**Staging Galaxy** (``job_conf.yml``)::
+
+    runners:
+      pulsar:
+        load: galaxy.jobs.runners.pulsar:PulsarMQJobRunner
+        proxy_url: https://shared-relay:9000
+        proxy_username: admin
+        proxy_password: password
+        relay_topic_prefix: staging
+
+    execution:
+      environments:
+        pulsar_jobs:
+          runner: pulsar
+          manager: cluster_a
+          # ... other settings
+
+In this setup, the topics will be completely isolated:
+
+* **Production**: ``production_job_setup_cluster_a``, ``production_job_status_update_cluster_a``
+* **Staging**: ``staging_job_setup_cluster_a``, ``staging_job_status_update_cluster_a``
+
 Topic Naming
 ````````````
 
-Messages are organized by topic with automatic naming based on the manager name:
+Messages are organized by topic with automatic naming based on the optional prefix
+and manager name:
+
+* Job setup: ``job_setup`` (default manager, no prefix)
+* Job setup: ``job_setup_{manager_name}`` (named manager, no prefix)
+* Job setup: ``{prefix}_job_setup`` (default manager, with prefix)
+* Job setup: ``{prefix}_job_setup_{manager_name}`` (named manager, with prefix)
+
+The same pattern applies to other message types:
+
+* Status requests: ``job_status_request``, ``job_status_request_{manager_name}``
+* Kill commands: ``job_kill``, ``job_kill_{manager_name}``
+* Status updates: ``job_status_update``, ``job_status_update_{manager_name}``
+
+When a ``relay_topic_prefix`` is configured, it is prepended to all topic names:
+
+* ``production_job_setup``
+* ``production_job_setup_cluster_a``
+* ``production_job_status_update_cluster_a``
 
-* Job setup: ``job_setup_{manager_name}`` or ``job_setup`` (for default manager)
-* Status requests: ``job_status_request_{manager_name}``
-* Kill commands: ``job_kill_{manager_name}``
-* Status updates: ``job_status_update_{manager_name}``
+This allows:
 
-This allows multiple Pulsar instances to share the same proxy without message
-conflicts.
+* Multiple Pulsar instances to share the same relay (using different manager names)
+* Multiple independent Galaxy/Pulsar instance pairs to share the same relay
+  (using different topic prefixes)
 
 Comparison with AMQP Mode
 ``````````````````````````

pulsar/client/client.py

@@ -604,7 +604,7 @@ def launch(self, command_line, dependencies_description=None, env=None, remote_s
 
         # Determine topic name based on manager
         manager_name = self.client_manager.manager_name
-        topic = f"job_setup_{manager_name}" if manager_name != "_default_" else "job_setup"
+        topic = self.client_manager._make_topic_name("job_setup", manager_name)
 
         # Post message to relay
         self.client_manager.relay_transport.post_message(topic, launch_params)
@@ -618,7 +618,7 @@ def get_status(self):
             Cached status if available, None otherwise
         """
         manager_name = self.client_manager.manager_name
-        topic = f"job_status_request_{manager_name}" if manager_name != "_default_" else "job_status_request"
+        topic = self.client_manager._make_topic_name("job_status_request", manager_name)
 
         status_params = {
             'job_id': self.job_id,
@@ -633,7 +633,7 @@ def get_status(self):
     def kill(self):
         """Kill a job by posting a kill message to the relay."""
         manager_name = self.client_manager.manager_name
-        topic = f"job_kill_{manager_name}" if manager_name != "_default_" else "job_kill"
+        topic = self.client_manager._make_topic_name("job_kill", manager_name)
 
         kill_params = {'job_id': self.job_id}
         self.client_manager.relay_transport.post_message(topic, kill_params)

pulsar/client/manager.py

@@ -264,14 +264,15 @@ class RelayClientManager(BaseRemoteConfiguredJobClientManager):
     """
     status_cache: Dict[str, Any]
 
-    def __init__(self, relay_url: str, relay_username: str, relay_password: str, **kwds: Dict[str, Any]):
+    def __init__(self, relay_url: str, relay_username: str, relay_password: str, relay_topic_prefix: str = '', **kwds: Dict[str, Any]):
         super().__init__(**kwds)
 
         if not relay_url:
             raise Exception("relay_url is required for RelayClientManager")
 
         # Initialize relay transport
         self.relay_transport = RelayTransport(relay_url, relay_username, relay_password)
+        self.relay_topic_prefix = relay_topic_prefix
         self.status_cache = {}
         self.callback_lock = threading.Lock()
         self.callback_thread = None
@@ -298,7 +299,7 @@ def callback_wrapper(self, callback, message_data):
     def status_consumer(self, callback_wrapper):
         """Long-poll the relay for status update messages."""
         manager_name = self.manager_name
-        topic = f"job_status_update_{manager_name}" if manager_name != "_default_" else "job_status_update"
+        topic = self._make_topic_name("job_status_update", manager_name)
 
         log.info("Starting relay status consumer for topic '%s'", topic)
 
@@ -341,6 +342,31 @@ def ensure_has_ack_consumers(self):
         """No-op for relay client manager, as acknowledgements are handled via HTTP."""
         pass
 
+    def _make_topic_name(self, base_topic: str, manager_name: str) -> str:
+        """Create a topic name with optional prefix and manager suffix.
+
+        Args:
+            base_topic: Base topic name (e.g., 'job_setup', 'job_status_update')
+            manager_name: Manager name (e.g., '_default_', 'cluster_a')
+
+        Returns:
+            Fully qualified topic name
+        """
+        parts = []
+
+        # Add prefix if provided
+        if self.relay_topic_prefix:
+            parts.append(self.relay_topic_prefix)
+
+        # Add base topic
+        parts.append(base_topic)
+
+        # Add manager name if not default
+        if manager_name != "_default_":
+            parts.append(manager_name)
+
+        return "_".join(parts)
+
     def shutdown(self, ensure_cleanup: bool = False):
         """Shutdown the client manager and cleanup resources."""
         self.active = False
@@ -391,6 +417,7 @@ def build_client_manager(
     relay_url: Optional[str] = None,
     relay_username: Optional[str] = None,
     relay_password: Optional[str] = None,
+    relay_topic_prefix: Optional[str] = None,
     amqp_url: Optional[str] = None,
     k8s_enabled: Optional[bool] = None,
     tes_enabled: Optional[bool] = None,
@@ -401,7 +428,13 @@ def build_client_manager(
         return ClientManager(job_manager=job_manager, **kwargs)  # TODO: Consider more separation here.
     elif relay_url:
         assert relay_password and relay_username, "relay_url set, but relay_username and relay_password must also be set"
-        return RelayClientManager(relay_url=relay_url, relay_username=relay_username, relay_password=relay_password, **kwargs)
+        return RelayClientManager(
+            relay_url=relay_url,
+            relay_username=relay_username,
+            relay_password=relay_password,
+            relay_topic_prefix=relay_topic_prefix or '',
+            **kwargs
+        )
     elif amqp_url:
         return MessageQueueClientManager(amqp_url=amqp_url, **kwargs)
     elif k8s_enabled or tes_enabled or gcp_batch_enabled:

pulsar/messaging/bind_relay.py

@@ -34,6 +34,9 @@ def bind_manager_to_relay(manager, relay_state: RelayState, relay_url, conf):
     if not password:
         raise Exception("message_queue_password is required for relay communication")
 
+    # Extract optional relay topic prefix
+    relay_topic_prefix = conf.get('relay_topic_prefix', '')
+
     # Create relay transport
     relay_transport = RelayTransport(relay_url, username, password)
 
@@ -42,11 +45,11 @@ def bind_manager_to_relay(manager, relay_state: RelayState, relay_url, conf):
     process_kill_messages = functools.partial(__process_kill_message, manager)
     process_status_messages = functools.partial(__process_status_message, manager)
 
-    # Determine topics based on manager name
-    setup_topic = f"job_setup_{manager_name}" if manager_name != "_default_" else "job_setup"
-    status_request_topic = f"job_status_request_{manager_name}" if manager_name != "_default_" else "job_status_request"
-    kill_topic = f"job_kill_{manager_name}" if manager_name != "_default_" else "job_kill"
-    status_update_topic = f"job_status_update_{manager_name}" if manager_name != "_default_" else "job_status_update"
+    # Determine topics based on manager name and optional prefix
+    setup_topic = __make_topic_name(relay_topic_prefix, "job_setup", manager_name)
+    status_request_topic = __make_topic_name(relay_topic_prefix, "job_status_request", manager_name)
+    kill_topic = __make_topic_name(relay_topic_prefix, "job_kill", manager_name)
+    status_update_topic = __make_topic_name(relay_topic_prefix, "job_status_update", manager_name)
 
     # Start consumer threads if message_queue_consume is enabled
     if conf.get("message_queue_consume", True):
@@ -206,3 +209,36 @@ def __client_job_id_from_body(body):
     """
     job_id = body.get("job_id", None)
     return job_id
+
+
+def __make_topic_name(prefix, base_topic, manager_name):
+    """Create a topic name with optional prefix and manager suffix.
+
+    Args:
+        prefix: Optional prefix string (e.g., 'galaxy1', 'prod')
+        base_topic: Base topic name (e.g., 'job_setup', 'job_status_update')
+        manager_name: Manager name (e.g., '_default_', 'cluster_a')
+
+    Returns:
+        Fully qualified topic name
+
+    Examples:
+        __make_topic_name('', 'job_setup', '_default_') -> 'job_setup'
+        __make_topic_name('', 'job_setup', 'cluster_a') -> 'job_setup_cluster_a'
+        __make_topic_name('prod', 'job_setup', '_default_') -> 'prod_job_setup'
+        __make_topic_name('prod', 'job_setup', 'cluster_a') -> 'prod_job_setup_cluster_a'
+    """
+    parts = []
+
+    # Add prefix if provided
+    if prefix:
+        parts.append(prefix)
+
+    # Add base topic
+    parts.append(base_topic)
+
+    # Add manager name if not default
+    if manager_name != "_default_":
+        parts.append(manager_name)
+
+    return "_".join(parts)