Merge pull request #55424 from ClickHouse/update-rabbitq

Updated RabbitMQ image and fixed log retrieval in failed tests.

docker/test/integration/runner/compose/docker_compose_rabbitmq.yml

@@ -2,15 +2,13 @@ version: '2.3'
 
 services:
     rabbitmq1:
-        image: rabbitmq:3.8-management-alpine
+        image: rabbitmq:3.12.6-management-alpine
         hostname: rabbitmq1
         expose:
             - ${RABBITMQ_PORT:-5672}
-        environment:
-            RABBITMQ_DEFAULT_USER: "root"
-            RABBITMQ_DEFAULT_PASS: "clickhouse"
-            RABBITMQ_LOG_BASE: /rabbitmq_logs/
         volumes:
             - type: ${RABBITMQ_LOGS_FS:-tmpfs}
               source: ${RABBITMQ_LOGS:-}
               target: /rabbitmq_logs/
+            - "${RABBITMQ_COOKIE_FILE}:/var/lib/rabbitmq/.erlang.cookie"
+            - /misc/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf

docker/test/integration/runner/misc/rabbitmq.conf

@@ -0,0 +1,8 @@
+loopback_users.guest = false
+listeners.tcp.default = 5672
+default_pass = clickhouse
+default_user = root
+management.tcp.port = 15672
+
+log.file = /rabbitmq_logs/rabbit.log
+log.file.level = debug

tests/integration/helpers/cluster.py

@@ -239,15 +239,71 @@ def check_postgresql_java_client_is_available(postgresql_java_client_id):
     return p.returncode == 0
 
 
-def check_rabbitmq_is_available(rabbitmq_id):
+def check_rabbitmq_is_available(rabbitmq_id, cookie):
     p = subprocess.Popen(
-        ("docker", "exec", "-i", rabbitmq_id, "rabbitmqctl", "await_startup"),
+        (
+            "docker",
+            "exec",
+            "-e",
+            f"RABBITMQ_ERLANG_COOKIE={cookie}",
+            "-i",
+            rabbitmq_id,
+            "rabbitmqctl",
+            "await_startup",
+        ),
         stdout=subprocess.PIPE,
     )
     p.communicate()
     return p.returncode == 0
 
 
+def rabbitmq_debuginfo(rabbitmq_id, cookie):
+    p = subprocess.Popen(
+        (
+            "docker",
+            "exec",
+            "-e",
+            f"RABBITMQ_ERLANG_COOKIE={cookie}",
+            "-i",
+            rabbitmq_id,
+            "rabbitmq-diagnostics",
+            "status",
+        ),
+        stdout=subprocess.PIPE,
+    )
+    p.communicate()
+
+    p = subprocess.Popen(
+        (
+            "docker",
+            "exec",
+            "-e",
+            f"RABBITMQ_ERLANG_COOKIE={cookie}",
+            "-i",
+            rabbitmq_id,
+            "rabbitmq-diagnostics",
+            "listeners",
+        ),
+        stdout=subprocess.PIPE,
+    )
+    p.communicate()
+
+    p = subprocess.Popen(
+        (
+            "docker",
+            "exec",
+            "-e",
+            f"RABBITMQ_ERLANG_COOKIE={cookie}",
+            "-i",
+            rabbitmq_id,
+            "rabbitmq-diagnostics",
+            "environment",
+        ),
+        stdout=subprocess.PIPE,
+    )
+    p.communicate()
+
+
 async def check_nats_is_available(nats_port, ssl_ctx=None):
     nc = await nats_connect_ssl(
         nats_port, user="click", password="house", ssl_ctx=ssl_ctx
@@ -271,11 +327,13 @@ async def nats_connect_ssl(nats_port, user, password, ssl_ctx=None):
     return nc
 
 
-def enable_consistent_hash_plugin(rabbitmq_id):
+def enable_consistent_hash_plugin(rabbitmq_id, cookie):
     p = subprocess.Popen(
         (
             "docker",
             "exec",
+            "-e",
+            f"RABBITMQ_ERLANG_COOKIE={cookie}",
             "-i",
             rabbitmq_id,
             "rabbitmq-plugins",
@@ -527,7 +585,9 @@ def __init__(
         self.rabbitmq_ip = None
         self.rabbitmq_port = 5672
         self.rabbitmq_dir = p.abspath(p.join(self.instances_dir, "rabbitmq"))
+        self.rabbitmq_cookie_file = os.path.join(self.rabbitmq_dir, "erlang.cookie")
         self.rabbitmq_logs_dir = os.path.join(self.rabbitmq_dir, "logs")
+        self.rabbitmq_cookie = self.get_instance_docker_id(self.rabbitmq_host)
 
         self.nats_host = "nats1"
         self.nats_port = 4444
@@ -1250,6 +1310,8 @@ def setup_rabbitmq_cmd(self, instance, env_variables, docker_compose_yml_dir):
         env_variables["RABBITMQ_PORT"] = str(self.rabbitmq_port)
         env_variables["RABBITMQ_LOGS"] = self.rabbitmq_logs_dir
         env_variables["RABBITMQ_LOGS_FS"] = "bind"
+        env_variables["RABBITMQ_COOKIE_FILE"] = self.rabbitmq_cookie_file
+        env_variables["RABBITMQ_COOKIE_FILE_FS"] = "bind"
 
         self.base_cmd.extend(
             ["--file", p.join(docker_compose_yml_dir, "docker_compose_rabbitmq.yml")]
@@ -2222,25 +2284,36 @@ def wait_postgresql_java_client(self, timeout=180):
                 time.sleep(0.5)
         raise Exception("Cannot wait PostgreSQL Java Client container")
 
-    def wait_rabbitmq_to_start(self, timeout=180, throw=True):
+    def wait_rabbitmq_to_start(self, timeout=30):
+        self.print_all_docker_pieces()
         self.rabbitmq_ip = self.get_instance_ip(self.rabbitmq_host)
 
         start = time.time()
         while time.time() - start < timeout:
             try:
-                if check_rabbitmq_is_available(self.rabbitmq_docker_id):
+                if check_rabbitmq_is_available(
+                    self.rabbitmq_docker_id, self.rabbitmq_cookie
+                ):
                     logging.debug("RabbitMQ is available")
-                    if enable_consistent_hash_plugin(self.rabbitmq_docker_id):
+                    if enable_consistent_hash_plugin(
+                        self.rabbitmq_docker_id, self.rabbitmq_cookie
+                    ):
                         logging.debug("RabbitMQ consistent hash plugin is available")
-                        return True
+                    return True
                 time.sleep(0.5)
             except Exception as ex:
                 logging.debug("Can't connect to RabbitMQ " + str(ex))
                 time.sleep(0.5)
 
-        if throw:
-            raise Exception("Cannot wait RabbitMQ container")
-        return False
+        try:
+            with open(os.path.join(self.rabbitmq_dir, "docker.log"), "w+") as f:
+                subprocess.check_call(  # STYLE_CHECK_ALLOW_SUBPROCESS_CHECK_CALL
+                    self.base_rabbitmq_cmd + ["logs"], stdout=f
+                )
+            rabbitmq_debuginfo(self.rabbitmq_docker_id, self.rabbitmq_cookie)
+        except Exception as e:
+            logging.debug("Unable to get logs from docker.")
+        raise Exception("Cannot wait RabbitMQ container")
 
     def wait_nats_is_available(self, max_retries=5):
         retries = 0
@@ -2749,15 +2822,18 @@ def start(self):
                 os.makedirs(self.rabbitmq_logs_dir)
                 os.chmod(self.rabbitmq_logs_dir, stat.S_IRWXU | stat.S_IRWXO)
 
-                for i in range(5):
-                    subprocess_check_call(
-                        self.base_rabbitmq_cmd + common_opts + ["--renew-anon-volumes"]
-                    )
-                    self.up_called = True
-                    self.rabbitmq_docker_id = self.get_instance_docker_id("rabbitmq1")
-                    logging.debug(f"RabbitMQ checking container try: {i}")
-                    if self.wait_rabbitmq_to_start(throw=(i == 4)):
-                        break
+                with open(self.rabbitmq_cookie_file, "w") as f:
+                    f.write(self.rabbitmq_cookie)
+                os.chmod(self.rabbitmq_cookie_file, stat.S_IRUSR)
+
+                subprocess_check_call(
+                    self.base_rabbitmq_cmd + common_opts + ["--renew-anon-volumes"]
+                )
+                self.up_called = True
+                self.rabbitmq_docker_id = self.get_instance_docker_id("rabbitmq1")
+                time.sleep(2)
+                logging.debug(f"RabbitMQ checking container try")
+                self.wait_rabbitmq_to_start()
 
             if self.with_nats and self.base_nats_cmd:
                 logging.debug("Setup NATS")