Test bucket sync status in a multi-site environment.

rgw/v2/lib/resource_op.py

@@ -231,6 +231,7 @@ def read(self):
         self.ceph_conf = self.doc["config"].get("ceph_conf")
         self.gc_verification = self.doc["config"].get("gc_verification", False)
         self.bucket_sync_crash = self.doc["config"].get("bucket_sync_crash", False)
+        self.bucket_sync_status = self.doc["config"].get("bucket_sync_status", False)
         self.bucket_sync_run = self.doc["config"].get("bucket_sync_run", False)
         self.bucket_stats = self.doc["config"].get("bucket_stats", False)
         self.header_size = self.doc["config"].get("header_size", False)

rgw/v2/tests/s3_swift/multisite_configs/test_bucket_sync_status.yaml

@@ -0,0 +1,21 @@
+# upload type: non multipart
+# script: test_Mbuckets_with_Nobjects.py
+config:
+  user_count: 1
+  bucket_count: 1
+  objects_count: 10
+  objects_size_range:
+    min: 5M
+    max: 15M
+  bucket_sync_status: true
+  test_ops:
+    create_bucket: true
+    create_object: true
+    download_object: false
+    delete_bucket_object: false
+    sharding:
+      enable: false
+      max_shards: 0
+    compression:
+      enable: false
+      type: zlib

rgw/v2/tests/s3_swift/test_Mbuckets_with_Nobjects.py

@@ -338,6 +338,12 @@ def test_exec(config):
                             raise TestExecError(
                                 "Command is throwing error while running bucket sync run"
                             )
+
+                    if config.bucket_sync_status:
+                        out = utils.wait_till_bucket_synced(bucket.name)
+                        if not out:
+                            log.info("Bucket sync is not caught up with source.")
+
                     if config.bucket_sync_crash:
                         is_primary = utils.is_cluster_primary()
                         if is_primary is False:

rgw/v2/utils/utils.py

@@ -1,4 +1,5 @@
 import configparser
+import datetime
 import hashlib
 import json
 import logging
@@ -8,9 +9,11 @@
 import socket
 import string
 import subprocess
+import time
 from random import randint
 
 import yaml
+from v2.lib.exceptions import SyncFailedError
 
 BUCKET_NAME_PREFIX = "bucky" + "-" + str(random.randrange(1, 5000))
 S3_OBJECT_NAME_PREFIX = "key"
@@ -554,6 +557,18 @@ def check_bucket_sync(name):
     return out
 
 
+def wait_till_bucket_synced(name, timeout=120, interval=5):
+    """Wait until the bucket reports synchronized."""
+    cmd = f"radosgw-admin bucket sync status --bucket {name}"
+    end_time = datetime.datetime.now() + datetime.timedelta(seconds=timeout)
+    while end_time > datetime.datetime.now():
+        time.sleep(interval)
+        result = exec_shell_cmd(cmd)
+        if not "behind shards" in result:
+            return True
+    return False
+
+
 def get_hostname_ip():
     try:
         hostname = socket.gethostname()