apply messaging.prefix for sensor_{wrapper,watcher}

st2reactor/st2reactor/container/sensor_wrapper.py

@@ -15,6 +15,8 @@
 
 from __future__ import absolute_import
 
+from st2common.transport.reactor import TRIGGER_CUD_XCHG, TRIGGER_INSTANCE_XCHG
+
 # Note: We need to perform monkey patching in the worker. If we do it in
 # the master process (gunicorn_config.py), it breaks tons of things
 # including shutdown
@@ -241,6 +243,12 @@ def __init__(
             ssl_match_hostname=cfg.CONF.database.ssl_match_hostname,
         )
 
+        # Apply prefix to RabbitMQ exchanges/queues (which are not bootstrapped in the wrapper)
+        prefix = cfg.CONF.messaging.prefix
+        if prefix != "st2":
+            for exchange in (TRIGGER_CUD_XCHG, TRIGGER_INSTANCE_XCHG):
+                exchange.name = exchange.name.replace("st2.", f"{prefix}.", 1)
+
         # 3. Instantiate the watcher
         self._trigger_watcher = TriggerWatcher(
             create_handler=self._handle_create_trigger,

tools/st2-inject-trigger-instances.py

@@ -37,7 +37,7 @@
 from st2common import config
 from st2common.util.monkey_patch import monkey_patch
 from st2common.util import date as date_utils
-from st2common.transport.reactor import TriggerDispatcher
+from st2common.transport.reactor import TRIGGER_INSTANCE_XCHG, TriggerDispatcher
 
 
 def do_register_cli_opts(opts, ignore_errors=False):
@@ -126,6 +126,12 @@ def main():
     do_register_cli_opts(cli_opts)
     config.parse_args()
 
+    # Apply prefix to RabbitMQ exchanges/queues (which have not been bootstrapped)
+    prefix = cfg.CONF.messaging.prefix
+    if prefix != "st2":
+        exchange = TRIGGER_INSTANCE_XCHG
+        exchange.name = exchange.name.replace("st2.", f"{prefix}.", 1)
+
     # Get config values
     triggers = cfg.CONF.triggers
     trigger_payload_schema = {}