Fix flaky test_keeper_disks

antonio2368 · antonio2368 · commit 27cd800c54e9 · 2025-12-09T11:39:29.000+01:00
diff --git a/tests/integration/test_keeper_disks/configs/enable_keeper.xml b/tests/integration/test_keeper_disks/configs/enable_keeper.xml
@@ -37,7 +37,7 @@
             <operation_timeout_ms>5000</operation_timeout_ms>
             <session_timeout_ms>10000</session_timeout_ms>
             <raft_logs_level>trace</raft_logs_level>
-            <snapshot_distance>10</snapshot_distance>
+            <snapshot_distance>100000</snapshot_distance>
             <stale_log_gap>10</stale_log_gap>
             <reserved_log_items>1</reserved_log_items>
             <rotate_log_storage_interval>3</rotate_log_storage_interval>
diff --git a/tests/integration/test_keeper_disks/test.py b/tests/integration/test_keeper_disks/test.py
@@ -6,7 +6,7 @@
 
 import helpers.keeper_utils as keeper_utils
 from helpers.cluster import ClickHouseCluster, is_arm
-
+import re
 import time
 
 import logging
@@ -81,7 +81,7 @@ def stop_clickhouse(cluster, node, cleanup_disks):
     )
 
 
-def setup_storage(cluster, node, storage_config, cleanup_disks):
+def setup_storage(cluster, node, storage_config, cleanup_disks, before_start=None):
     stop_clickhouse(cluster, node, cleanup_disks)
     node.copy_file_to_container(
         os.path.join(CURRENT_TEST_DIR, "configs/enable_keeper.xml"),
@@ -92,6 +92,10 @@ def setup_storage(cluster, node, storage_config, cleanup_disks):
         "<!-- DISK DEFINITION PLACEHOLDER -->",
         storage_config,
     )
+
+    if before_start:
+        before_start(node)
+
     node.start_clickhouse()
     # complete readiness checks that the sessions can be established,
     # but it creates sesssion for this, which will create one more record in log,
@@ -144,11 +148,21 @@ def test_logs_with_disks(started_cluster):
         for _ in range(30):
             node_zk.create("/test/somenode", b"somedata", sequence=True)
 
-        node_logs.wait_for_log_line("Removed changelog changelog_25_27.bin because of compaction")
-
         stop_zk(node_zk)
 
-        previous_log_files = get_local_logs(node_logs)
+        previous_log_files = []
+
+        def collect_local_logs(node):
+            nonlocal previous_log_files
+            previous_log_files = get_local_logs(node)
+            # assert all filenames match the expected pattern
+            for name in previous_log_files:
+                assert re.match(
+                    r"^changelog_\d+_\d+\.bin$", name
+                ), f"Filename {name} does not match the expected pattern"
+
+            previous_log_files.sort(key=lambda name: int(name.split("_")[1]))
+            logging.info(f"Previous log files: {previous_log_files}")
 
         setup_storage(
             started_cluster,
@@ -157,10 +171,9 @@ def test_logs_with_disks(started_cluster):
             "<latest_log_storage_disk>log_local<\\/latest_log_storage_disk>"
             "<snapshot_storage_disk>snapshot_local<\\/snapshot_storage_disk>",
             cleanup_disks=False,
+            before_start=collect_local_logs,
         )
 
-        node_logs.wait_for_log_line("KeeperLogStore: Continue to write into changelog_34_36.bin")
-
         def get_single_local_log_file():
             local_log_files = get_local_logs(node_logs)
             start_time = time.time()
@@ -188,13 +201,16 @@ def get_single_local_log_file():
 
         stop_zk(node_zk)
 
-        local_log_files = get_single_local_log_file()
-        log_files = list_s3_objects(started_cluster, "logs/")
+        def collect_all_logs(node):
+            nonlocal previous_log_files
+            local_log_files = get_single_local_log_file()
+            log_files = list_s3_objects(started_cluster, "logs/")
 
-        log_files.extend(local_log_files)
-        assert set(log_files) != previous_log_files
+            log_files.extend(local_log_files)
+            assert set(log_files) != previous_log_files
 
-        previous_log_files = log_files
+            previous_log_files = log_files
+            logging.info(f"Previous log files: {previous_log_files}")
 
         setup_storage(
             started_cluster,
@@ -203,8 +219,19 @@ def get_single_local_log_file():
             "<log_storage_disk>log_local<\\/log_storage_disk>"
             "<snapshot_storage_disk>snapshot_local<\\/snapshot_storage_disk>",
             cleanup_disks=False,
+            before_start=collect_all_logs,
         )
 
+        s3_files = list_s3_objects(started_cluster, "logs/")
+        start_time = time.time()
+        while len(s3_files) != 0:
+            logging.debug(f"S3 log files: {s3_files}")
+            assert (
+                time.time() - start_time < 60
+            ), "s3_files size is not equal to 0 after 60s"
+            time.sleep(1)
+            s3_files = list_s3_objects(started_cluster, "logs/")
+
         local_log_files = get_local_logs(node_logs)
         assert set(local_log_files) == set(previous_log_files)
 
