Add new option [messaging].prefix to configure prefix of RabbitMQ e…

…xchange/queue names (#6282)

CHANGELOG.rst

@@ -69,7 +69,7 @@ Added
   working on StackStorm, improve our security posture, and improve CI reliability thanks in part
   to pants' use of PEX lockfiles. This is not a user-facing addition.
   #6118 #6141 #6133 #6120 #6181 #6183 #6200 #6237 #6229 #6240 #6241 #6244 #6251 #6253
-  #6254 #6258 #6259 #6260 #6269 #6275 #6279 #6278
+  #6254 #6258 #6259 #6260 #6269 #6275 #6279 #6278 #6282
   Contributed by @cognifloyd
 * Build of ST2 EL9 packages #6153
   Contributed by @amanda11
@@ -93,6 +93,12 @@ Added
   If you experience any issues when using this experimental feature, please file an issue. #6277
   Contributed by @cognifloyd
 
+* Add new option `[messaging].prefix` to configure the prefix used in RabbitMQ exchanges and queues.
+  The default is `st2` (resulting in exchange names like `st2.execution` and `st2.sensor`).
+  This is primarily designed to support safely running tests in parallel where creating a vhost for
+  each parallel test run would be a maintenance burden. #6282
+  Contributed by @cognifloyd
+
 3.8.1 - December 13, 2023
 -------------------------
 Fixed

conf/st2.conf.sample

@@ -230,6 +230,8 @@ connection_retries = 10
 connection_retry_wait = 10000
 # Login method to use (AMQPLAIN, PLAIN, EXTERNAL, etc.).
 login_method = None
+# Prefix for all exchange and queue names.
+prefix = st2
 # Use SSL / TLS to connect to the messaging server. Same as appending "?ssl=true" at the end of the connection URL string.
 ssl = False
 # ca_certs file contains a set of concatenated CA certificates, which are used to validate certificates passed from RabbitMQ.

conf/st2.dev.conf

@@ -103,6 +103,7 @@ ssh_key_file = /home/vagrant/.ssh/stanley_rsa
 
 [messaging]
 url = amqp://guest:[email protected]:5672/
+prefix = st2dev
 # Uncomment to test SSL options
 #url = amqp://guest:[email protected]:5671/
 #ssl = True

pants-plugins/uses_services/rabbitmq_rules.py

@@ -15,6 +15,7 @@
 
 from dataclasses import dataclass
 from textwrap import dedent
+from typing import Tuple
 
 from pants.backend.python.goals.pytest_runner import (
     PytestPluginSetupRequest,
@@ -27,6 +28,8 @@
     VenvPexProcess,
     rules as pex_rules,
 )
+from pants.core.goals.test import TestExtraEnv
+from pants.engine.env_vars import EnvironmentVars
 from pants.engine.fs import CreateDigest, Digest, FileContent
 from pants.engine.rules import collect_rules, Get, MultiGet, rule
 from pants.engine.process import FallibleProcessResult, ProcessCacheScope
@@ -54,13 +57,17 @@ class UsesRabbitMQRequest:
     # These config opts for integration tests are in:
     #   conf/st2.tests*.conf st2tests/st2tests/fixtures/conf/st2.tests*.conf
     #       (changed by setting ST2_CONFIG_PATH env var inside the tests)
-    # TODO: for unit tests: modify code to pull mq connect settings from env vars
-    # TODO: for int tests: modify st2.tests*.conf on the fly to set the per-pantsd-slot vhost
-    #                      and either add env vars for mq connect settings or modify conf files as well
+    # These can also be updated via the ST2_MESSAGING_* env vars (which oslo_config reads).
+    # Integration tests should pass these changes onto subprocesses via the same env vars.
 
-    #   with our version of oslo.config (newer are slower) we can't directly override opts w/ environment variables.
+    mq_urls: Tuple[str] = ("amqp://guest:[email protected]:5672//",)
 
-    mq_urls: tuple[str] = ("amqp://guest:[email protected]:5672//",)
+    @classmethod
+    def from_env(cls, env: EnvironmentVars) -> UsesRabbitMQRequest:
+        default = cls()
+        url = env.get("ST2_MESSAGING__URL", None)
+        mq_urls = (url,) if url else default.mq_urls
+        return UsesRabbitMQRequest(mq_urls=mq_urls)
 
 
 @dataclass(frozen=True)
@@ -83,9 +90,12 @@ def is_applicable(cls, target: Target) -> bool:
 )
 async def rabbitmq_is_running_for_pytest(
     request: PytestUsesRabbitMQRequest,
+    test_extra_env: TestExtraEnv,
 ) -> PytestPluginSetup:
     # this will raise an error if rabbitmq is not running
-    _ = await Get(RabbitMQIsRunning, UsesRabbitMQRequest())
+    _ = await Get(
+        RabbitMQIsRunning, UsesRabbitMQRequest.from_env(env=test_extra_env.env)
+    )
 
     return PytestPluginSetup()
 
@@ -167,6 +177,16 @@ async def rabbitmq_is_running(
                 """
             ),
             service_start_cmd_generic="systemctl start rabbitmq-server",
+            env_vars_hint=dedent(
+                """\
+                You can also export the ST2_MESSAGING__URL env var to automatically use any
+                RabbitMQ host, local or remote, while running unit and integration tests.
+                If needed, you can also override the default exchange/queue name prefix
+                by exporting ST2_MESSAGING__PREFIX. Note that tests always add a numeric
+                suffix to the exchange/queue name prefix so that tests can safely run
+                in parallel.
+                """
+            ),
         ),
     )
 

pants-plugins/uses_services/rabbitmq_rules_test.py

@@ -51,7 +51,14 @@ def run_rabbitmq_is_running(
             "--backend-packages=uses_services",
             *(extra_args or ()),
         ],
-        env_inherit={"PATH", "PYENV_ROOT", "HOME"},
+        env_inherit={
+            "PATH",
+            "PYENV_ROOT",
+            "HOME",
+            "ST2_MESSAGING__URL",
+            "ST2_MESSAGING__PREFIX",
+            "ST2TESTS_PARALLEL_SLOT",
+        },
     )
     result = rule_runner.request(
         RabbitMQIsRunning,
@@ -62,7 +69,7 @@ def run_rabbitmq_is_running(
 
 # Warning this requires that rabbitmq be running
 def test_rabbitmq_is_running(rule_runner: RuleRunner) -> None:
-    request = UsesRabbitMQRequest()
+    request = UsesRabbitMQRequest.from_env(env=rule_runner.environment)
     mock_platform = platform(os="TestMock")
 
     # we are asserting that this does not raise an exception

pants.toml

@@ -247,6 +247,9 @@ extra_env_vars = [
   "ST2_DATABASE__CONNECTION_TIMEOUT",
   "ST2_DATABASE__USERNAME",
   "ST2_DATABASE__PASSWORD",
+  # Use these to override RabbitMQ connection details
+  "ST2_MESSAGING__URL",
+  "ST2_MESSAGING__PREFIX", # Tests will modify this to be "{prefix}{ST2TESTS_PARALLEL_SLOT}"
   # Use these to override the redis host and port
   "ST2TESTS_REDIS_HOST",
   "ST2TESTS_REDIS_PORT",

st2actions/tests/integration/test_actions_queue_consumer.py

@@ -17,11 +17,10 @@
 import random
 import eventlet
 
-from kombu import Exchange
-from kombu import Queue
 from unittest import TestCase
 
 from st2common.transport.consumers import ActionsQueueConsumer
+from st2common.transport.kombu import Exchange, Queue
 from st2common.transport.publishers import PoolPublisher
 from st2common.transport import utils as transport_utils
 from st2common.models.db.liveaction import LiveActionDB
@@ -35,7 +34,7 @@ class ActionsQueueConsumerTestCase(TestCase):
 
     def test_stop_consumption_on_shutdown(self):
         exchange = Exchange("st2.execution.test", type="topic")
-        queue_name = "test-" + str(random.randint(1, 10000))
+        queue_name = f"st2.test-{random.randint(1, 10000)}"
         queue = Queue(
             name=queue_name, exchange=exchange, routing_key="#", auto_delete=True
         )

st2common/benchmarks/micro/test_publisher_compression.py

@@ -16,7 +16,6 @@
 
 monkey_patch()
 
-from kombu import Exchange
 from kombu.serialization import pickle
 
 import os
@@ -27,6 +26,7 @@
 
 from st2common.models.db.liveaction import LiveActionDB
 from st2common.transport import publishers
+from st2common.transport.kombu import Exchange
 
 from common import FIXTURES_DIR
 from common import PYTEST_FIXTURE_FILE_PARAM_DECORATOR

st2common/st2common/config.py

@@ -404,6 +404,11 @@ def register_opts(ignore_errors=False):
             help="Compression algorithm to use for compressing the payloads which are sent over "
             "the message bus. Defaults to no compression.",
         ),
+        cfg.StrOpt(
+            "prefix",
+            default="st2",
+            help="Prefix for all exchange and queue names.",
+        ),
     ]
 
     do_register_opts(messaging_opts, "messaging", ignore_errors)

st2common/st2common/stream/listener.py

@@ -58,9 +58,15 @@ def get_consumers(self, consumer, channel):
         raise NotImplementedError("get_consumers() is not implemented")
 
     def processor(self, model=None):
+        exchange_prefix = cfg.CONF.messaging.prefix
+
         def process(body, message):
             meta = message.delivery_info
-            event_name = "%s__%s" % (meta.get("exchange"), meta.get("routing_key"))
+            event_prefix = meta.get("exchange", "")
+            if exchange_prefix != "st2" and event_prefix.startswith(exchange_prefix):
+                # use well-known event names over configurable exchange names
+                event_prefix = event_prefix.replace(f"{exchange_prefix}.", "st2.", 1)
+            event_name = f"{event_prefix}__{meta.get('routing_key')}"
 
             try:
                 if model:

st2common/st2common/transport/actionalias.py

@@ -15,8 +15,9 @@
 # All Exchanges and Queues related to liveaction.
 
 from __future__ import absolute_import
-from kombu import Exchange, Queue
+
 from st2common.transport import publishers
+from st2common.transport.kombu import Exchange, Queue
 
 __all__ = [
     "ActionAliasPublisher",

st2common/st2common/transport/actionexecutionstate.py

@@ -17,9 +17,8 @@
 
 from __future__ import absolute_import
 
-from kombu import Exchange, Queue
-
 from st2common.transport import publishers
+from st2common.transport.kombu import Exchange, Queue
 
 __all__ = ["ActionExecutionStatePublisher"]
 

st2common/st2common/transport/announcement.py

@@ -15,12 +15,11 @@
 
 from __future__ import absolute_import
 
-from kombu import Exchange, Queue
-
 from st2common import log as logging
 from st2common.constants.trace import TRACE_CONTEXT
 from st2common.models.api.trace import TraceContext
 from st2common.transport import publishers
+from st2common.transport.kombu import Exchange, Queue
 
 __all__ = ["AnnouncementPublisher", "AnnouncementDispatcher", "get_queue"]
 

st2common/st2common/transport/bootstrap_utils.py

@@ -31,7 +31,7 @@
 from st2common.transport.actionexecutionstate import ACTIONEXECUTIONSTATE_XCHG
 from st2common.transport.announcement import ANNOUNCEMENT_XCHG
 from st2common.transport.connection_retry_wrapper import ConnectionRetryWrapper
-from st2common.transport.execution import EXECUTION_XCHG
+from st2common.transport.execution import EXECUTION_XCHG, EXECUTION_OUTPUT_XCHG
 from st2common.transport.liveaction import LIVEACTION_XCHG, LIVEACTION_STATUS_MGMT_XCHG
 from st2common.transport.reactor import SENSOR_CUD_XCHG
 from st2common.transport.reactor import TRIGGER_CUD_XCHG, TRIGGER_INSTANCE_XCHG
@@ -67,6 +67,7 @@
     ACTIONEXECUTIONSTATE_XCHG,
     ANNOUNCEMENT_XCHG,
     EXECUTION_XCHG,
+    EXECUTION_OUTPUT_XCHG,
     LIVEACTION_XCHG,
     LIVEACTION_STATUS_MGMT_XCHG,
     TRIGGER_CUD_XCHG,

st2common/st2common/transport/execution.py

@@ -16,8 +16,9 @@
 # All Exchanges and Queues related to liveaction.
 
 from __future__ import absolute_import
-from kombu import Exchange, Queue
+
 from st2common.transport import publishers
+from st2common.transport.kombu import Exchange, Queue
 
 __all__ = [
     "ActionExecutionPublisher",

st2common/st2common/transport/kombu.py

@@ -0,0 +1,36 @@
+# Copyright 2024 The StackStorm Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import absolute_import
+
+import kombu
+from oslo_config import cfg
+
+
+class Exchange(kombu.Exchange):
+    def __call__(self, *args, **kwargs):
+        # update exchange name with prefix just before binding (as late as possible).
+        prefix = cfg.CONF.messaging.prefix
+        if self.name and prefix != "st2":
+            self.name = self.name.replace("st2.", f"{prefix}.", 1)
+        return super().__call__(*args, **kwargs)
+
+
+class Queue(kombu.Queue):
+    def __call__(self, *args, **kwargs):
+        # update queue name with prefix just before binding (as late as possible).
+        prefix = cfg.CONF.messaging.prefix
+        if self.name and prefix != "st2":
+            self.name = self.name.replace("st2.", f"{prefix}.", 1)
+        return super().__call__(*args, **kwargs)

st2common/st2common/transport/liveaction.py

@@ -17,9 +17,8 @@
 
 from __future__ import absolute_import
 
-from kombu import Exchange, Queue
-
 from st2common.transport import publishers
+from st2common.transport.kombu import Exchange, Queue
 
 __all__ = ["LiveActionPublisher", "get_queue", "get_status_management_queue"]
 

st2common/st2common/transport/publishers.py

@@ -71,10 +71,11 @@ def do_publish(connection, channel):
                     # completely invalidating this ConnectionPool. Also, a ConnectionPool for
                     # producer does not really solve any problems for us so better to create a
                     # Producer for each publish.
-                    producer = Producer(channel)
+                    # passing exchange to Producer __init__ allows auto_declare to declare
+                    # anything that's missing (especially useful for tests).
+                    producer = Producer(channel, exchange=exchange)
                     kwargs = {
                         "body": payload,
-                        "exchange": exchange,
                         "routing_key": routing_key,
                         "serializer": "pickle",
                         "compression": compression,

st2common/st2common/transport/queues.py

@@ -22,8 +22,6 @@
 
 from __future__ import absolute_import
 
-from kombu import Queue
-
 from st2common.constants import action as action_constants
 from st2common.transport import actionalias
 from st2common.transport import actionexecutionstate
@@ -33,6 +31,7 @@
 from st2common.transport import publishers
 from st2common.transport import reactor
 from st2common.transport import workflow
+from st2common.transport.kombu import Queue
 
 __all__ = [
     "ACTIONSCHEDULER_REQUEST_QUEUE",

st2common/st2common/transport/reactor.py

@@ -14,12 +14,12 @@
 # limitations under the License.
 
 from __future__ import absolute_import
-from kombu import Exchange, Queue
 
 from st2common import log as logging
 from st2common.constants.trace import TRACE_CONTEXT
 from st2common.models.api.trace import TraceContext
 from st2common.transport import publishers
+from st2common.transport.kombu import Exchange, Queue
 
 __all__ = [
     "TriggerCUDPublisher",