Config: add the warnings.rabbitmq_version option (#5415)

This option can be used to suppress the warning that is emitted when a
profile storage is loaded that is configured with a version of RabbitMQ
that is known to be able to cause problems.

The `check_rabbitmq_version` and `check_version` free functions are
moved to `Manager`.

The `is_rabbitmq_version_supported` function is removed as it is no
longer used.

Cherry-pick: 9dd216f

aiida/cmdline/commands/cmd_status.py

@@ -57,7 +57,6 @@ def verdi_status(print_traceback, no_rmq):
     # pylint: disable=broad-except,too-many-statements,too-many-branches,too-many-locals
     from aiida.cmdline.utils.daemon import get_daemon_status, delete_stale_pid_file
     from aiida.common.utils import Capturing
-    from aiida.manage.configuration import get_rabbitmq_version, is_rabbitmq_version_supported
     from aiida.manage.configuration.settings import AIIDA_CONFIG_FOLDER
     from aiida.manage.manager import get_manager
 
@@ -113,16 +112,15 @@ def verdi_status(print_traceback, no_rmq):
         try:
             with Capturing(capture_stderr=True):
                 with override_log_level():  # temporarily suppress noisy logging
-                    comm = manager.create_communicator(with_orm=False)
-                    comm.close()
+                    comm = manager.get_communicator()
         except Exception as exc:
             message = f'Unable to connect to rabbitmq with URL: {profile.get_rmq_url()}'
             print_status(ServiceStatus.ERROR, 'rabbitmq', message, exception=exc, print_traceback=print_traceback)
             exit_code = ExitCode.CRITICAL
         else:
-            version = get_rabbitmq_version()
+            version, supported = manager.check_rabbitmq_version(comm)
             connection = f'Connected to RabbitMQ v{version} as {profile.get_rmq_url()}'
-            if is_rabbitmq_version_supported():
+            if supported:
                 print_status(ServiceStatus.UP, 'rabbitmq', connection)
             else:
                 print_status(ServiceStatus.WARNING, 'rabbitmq', 'Incompatible RabbitMQ version detected! ' + connection)

aiida/manage/configuration/__init__.py

@@ -28,38 +28,6 @@
 )
 
 
-def is_rabbitmq_version_supported():
-    """Return whether the version of RabbitMQ configured for the current profile is supported.
-
-    Versions 3.8 and above are not compatible with AiiDA with default configuration.
-
-    :return: boolean whether the current RabbitMQ version is supported.
-    """
-    from packaging.version import parse
-    return get_rabbitmq_version() < parse('3.8.15')
-
-
-def get_rabbitmq_version():
-    """Return the version of the RabbitMQ server that the current profile connects to.
-
-    :return: :class:`packaging.version.Version`
-    """
-    from packaging.version import parse
-
-    from aiida.manage.manager import get_manager
-    communicator = get_manager().get_communicator()
-    return parse(communicator.server_properties['version'].decode('utf-8'))
-
-
-def check_rabbitmq_version():
-    """Check the version of RabbitMQ that is being connected to and emit warning if the version is not compatible."""
-    from aiida.cmdline.utils import echo
-    if not is_rabbitmq_version_supported():
-        echo.echo_warning(f'RabbitMQ v{get_rabbitmq_version()} is not supported and will cause unexpected problems!')
-        echo.echo_warning('It can cause long-running workflows to crash and jobs to be submitted multiple times.')
-        echo.echo_warning('See https://github.com/aiidateam/aiida-core/wiki/RabbitMQ-version-to-use for details.')
-
-
 def load_profile(profile=None):
     """Load a profile.
 
@@ -97,9 +65,6 @@ def load_profile(profile=None):
     # instead be done lazily in `Manager._load_backend`.
     configure_logging()
 
-    # Check whether a compatible version of RabbitMQ is being used.
-    check_rabbitmq_version()
-
     return PROFILE
 
 

aiida/manage/configuration/schema/config-v5.schema.json

@@ -164,6 +164,11 @@
                     "default": true,
                     "description": "Whether to print AiiDA deprecation warnings"
                 },
+                "warnings.rabbitmq_version": {
+                  "type": "boolean",
+                  "default": true,
+                  "description": "Whether to print a warning when an incompatible version of RabbitMQ is configured"
+                },
                 "transport.task_retry_initial_interval": {
                     "type": "integer",
                     "default": 20,

aiida/manage/manager.py

@@ -250,7 +250,7 @@ def create_communicator(
             encoder = functools.partial(json.dumps, encoding='utf-8')
             decoder = json.loads
 
-        return kiwipy.rmq.RmqThreadCommunicator.connect(
+        communicator = kiwipy.rmq.RmqThreadCommunicator.connect(
             connection_params={'url': profile.get_rmq_url()},
             message_exchange=rmq.get_message_exchange_name(prefix),
             encoder=encoder,
@@ -264,6 +264,11 @@ def create_communicator(
             testing_mode=profile.is_test_profile,
         )
 
+        # Check whether a compatible version of RabbitMQ is being used.
+        self.check_rabbitmq_version(communicator)
+
+        return communicator
+
     def get_daemon_client(self) -> 'DaemonClient':
         """Return the daemon client for the current profile.
 
@@ -374,6 +379,37 @@ def create_daemon_runner(self, loop: Optional[asyncio.AbstractEventLoop] = None)
 
         return runner
 
+    def check_rabbitmq_version(self, communicator: 'RmqThreadCommunicator'):
+        """Check the version of RabbitMQ that is being connected to and emit warning if the version is not compatible.
+
+        Versions 3.8.15 and above are not compatible with AiiDA with default configuration.
+        """
+        from packaging.version import parse
+
+        from aiida.cmdline.utils import echo
+
+        config = self.get_config()
+        profile = self.get_profile()
+        show_warning = config.get_option('warnings.rabbitmq_version', profile.name if profile else None)
+        version = get_rabbitmq_version(communicator)
+
+        if show_warning and version >= parse('3.8.15'):
+            echo.echo_warning(f'RabbitMQ v{version} is not supported and will cause unexpected problems!')
+            echo.echo_warning('It can cause long-running workflows to crash and jobs to be submitted multiple times.')
+            echo.echo_warning('See https://github.com/aiidateam/aiida-core/wiki/RabbitMQ-version-to-use for details.')
+            return version, False
+
+        return version, True
+
+
+def get_rabbitmq_version(communicator: 'RmqThreadCommunicator'):
+    """Return the version of the RabbitMQ server that the current profile connects to.
+
+    :return: :class:`packaging.version.Version`
+    """
+    from packaging.version import parse
+    return parse(communicator.server_properties['version'].decode('utf-8'))
+
 
 MANAGER: Optional[Manager] = None
 

tests/cmdline/commands/test_run.py

@@ -63,7 +63,7 @@ def wf():
             self.assertClickResultNoException(result)
 
             # Try to load the function calculation node from the printed pk in the output
-            pk = int(result.output)
+            pk = int(result.output.strip().split('\n')[-1])
             node = load_node(pk)
 
             # Verify that the node has the correct function name and content
@@ -285,7 +285,7 @@ def test_autogroup_filter_class(self):  # pylint: disable=too-many-locals
                 result = self.cli_runner.invoke(cmd_run.run, options)
                 self.assertClickResultNoException(result)
 
-                pk1_str, pk2_str, pk3_str, pk4_str, pk5_str, pk6_str = result.output.split()
+                pk1_str, pk2_str, pk3_str, pk4_str, pk5_str, pk6_str = result.output.strip().split('\n')[-6:]
                 pk1 = int(pk1_str)
                 pk2 = int(pk2_str)
                 pk3 = int(pk3_str)

tests/manage/configuration/test_options.py

@@ -22,7 +22,7 @@ class TestConfigurationOptions(AiidaTestCase):
     def test_get_option_names(self):
         """Test `get_option_names` function."""
         self.assertIsInstance(get_option_names(), list)
-        self.assertEqual(len(get_option_names()), 26)
+        self.assertEqual(len(get_option_names()), 27)
 
     def test_get_option(self):
         """Test `get_option` function."""