adding support for lc processing on existing users and buckets passed from the config

Signed-off-by: Hemanth Sai Maheswarla <hemanthsaimaheswarla@Hemanths-MacBook-Pro.local>

adding support for setting configs on all rgw daemons and restarting all rgw daemons

Signed-off-by: Hemanth Sai Maheswarla <hemanthsaimaheswarla@Hemanths-MacBook-Pro.local>

adding rgw_lc_max_wp_worker config parameter

Signed-off-by: Hemanth Sai Maheswarla <hemanthsaimaheswarla@Hemanths-MacBook-Pro.local>

Author: Hemanth Sai Maheswarla

rgw/v2/lib/resource_op.py

@@ -68,7 +68,9 @@ def resource_op(exec_info):
         return False
 
 
-def create_users(no_of_users_to_create, user_names=None, cluster_name="ceph"):
+def create_users(
+    no_of_users_to_create, user_names=None, cluster_name="ceph", config=None
+):
     """
     This function is to create n users on the cluster
 
@@ -83,11 +85,11 @@ def create_users(no_of_users_to_create, user_names=None, cluster_name="ceph"):
     all_users_details = []
     primary = utils.is_cluster_primary()
     user_detail_file = os.path.join(lib_dir, "user_details.json")
-    if primary:
+    if primary or (config and config.user_names):
         for i in range(no_of_users_to_create):
             if user_names:
                 user_details = admin_ops.create_admin_user(
-                    user_id=user_names,
+                    user_id=user_names[i],
                     displayname=user_names,
                     cluster_name=cluster_name,
                 )
@@ -308,6 +310,7 @@ def read(self, ssh_con=None):
             "rgw_lifecycle_work_time", "00:00-06:00"
         )
         self.rgw_lc_max_worker = self.doc["config"].get("rgw_lc_max_worker", 10)
+        self.rgw_lc_max_wp_worker = self.doc["config"].get("rgw_lc_max_wp_worker", 10)
         self.parallel_lc = self.doc["config"].get("parallel_lc", False)
         self.multiple_delete_marker_check = self.doc["config"].get(
             "multiple_delete_marker_check", False
@@ -440,6 +443,8 @@ def read(self, ssh_con=None):
         )
         self.user_conflict_write_ops = self.doc["config"].get("user_conflict_write_ops")
         self.permutation_count = self.doc["config"].get("permutation_count")
+        self.user_names = self.doc["config"].get("user_names")
+        self.bucket_names = self.doc["config"].get("bucket_names")
         ceph_version_id, ceph_version_name = utils.get_ceph_version()
         # todo: improve Frontend class
         if ceph_version_name in ["luminous", "nautilus"]:

rgw/v2/lib/rgw_config_opts.py

@@ -30,6 +30,7 @@ def __init__(self):
     rgw_enable_lc_threads = "rgw_enable_lc_threads"
     rgw_lifecycle_work_time = "rgw_lifecycle_work_time"
     rgw_lc_max_worker = "rgw_lc_max_worker"
+    rgw_lc_max_wp_worker = "rgw_lc_max_wp_worker"
     debug_rgw = "debug_rgw"
     log_to_file = "log_to_file"
     rgw_crypt_require_ssl = "rgw_crypt_require_ssl"
@@ -126,12 +127,17 @@ def set_to_ceph_conf_file(self, section, option, value=None, ssh_con=None):
 
 
 class CephConfigSet:
-    def set_to_ceph_cli(self, key, value, set_to_all=False):
+    def set_to_ceph_cli(self, key, value, set_to_all=False, remote_ssh_con=None):
         log.info("setting key and value using ceph config set cli")
         self.prefix = "sudo ceph config set"
 
         cmd_ps = "ceph orch ps --daemon_type rgw -f json"
-        out_ps = utils.exec_shell_cmd(cmd_ps)
+        if remote_ssh_con:
+            out_ps = utils.remote_exec_shell_cmd(
+                remote_ssh_con, cmd_ps, return_output=True
+            )
+        else:
+            out_ps = utils.exec_shell_cmd(cmd_ps)
         out = json.loads(out_ps)
         daemon_name_list = []
         for node in out:
@@ -146,7 +152,12 @@ def set_to_ceph_cli(self, key, value, set_to_all=False):
             log.info(f"got value: {value}")
             cmd_list = [self.prefix, self.who, key, str(value)]
             cmd = " ".join(cmd_list)
-            config_set = utils.exec_shell_cmd(cmd)
+            if remote_ssh_con:
+                config_set = utils.remote_exec_shell_cmd(
+                    remote_ssh_con, cmd, return_output=False
+                )
+            else:
+                config_set = utils.exec_shell_cmd(cmd)
             if config_set is False:
                 raise InvalidCephConfigOption("Invalid ceph config options")
             if not set_to_all:
@@ -158,7 +169,13 @@ def __init__(self, ssh_con=None) -> None:
         super().__init__(ssh_con)
 
     def set_to_ceph_conf(
-        self, section, option, value=None, ssh_con=None, set_to_all=False
+        self,
+        section,
+        option,
+        value=None,
+        ssh_con=None,
+        set_to_all=False,
+        remote_ssh_con=None,
     ):
         version_id, version_name = utils.get_ceph_version()
         log.info(f"ceph version id: {version_id}")
@@ -174,4 +191,6 @@ def set_to_ceph_conf(
             log.info("using ceph config cli to set the config values")
             log.info(option)
             log.info(value)
-            self.set_to_ceph_cli(option, value, set_to_all=set_to_all)
+            self.set_to_ceph_cli(
+                option, value, set_to_all=set_to_all, remote_ssh_con=remote_ssh_con
+            )

rgw/v2/lib/s3/lifecycle_validation.py

@@ -176,7 +176,6 @@ def validate_prefix_rule(bucket, config):
                         f"Lifecycle cloud transition validation failed with retain_head_object_false for bucket {bucket.name}"
                     )
         else:
-
             log.info("Start the validation of LC pool transition")
             curr = 0
             ncurr = 0
@@ -244,21 +243,28 @@ def validate_prefix_rule(bucket, config):
                 )
 
 
-def validate_prefix_rule_non_versioned(bucket, config):
+def validate_prefix_rule_non_versioned(bucket, config, ssh_con=None):
     log.info("verification starts")
     objects_count = config.objects_count
     if config.test_lc_transition:
-        cmd = utils.exec_shell_cmd(f"radosgw-admin bucket list --bucket {bucket.name}")
+        cmd = utils.exec_shell_cmd(
+            f"radosgw-admin bucket list --bucket {bucket.name}{f' --max-entries {config.objects_count}' if  config.objects_count > 1000 else ''}"
+        )
         json_doc = json.loads(cmd)
-        for i in range(0, objects_count):
+        obj_count = (
+            config.objects_count if config.objects_count < 1000 else len(json_doc)
+        )
+        for i in range(0, obj_count):
             storage_class = json_doc[i]["meta"]["storage_class"]
             log.info(f"object has transitioned to {storage_class}")
             if storage_class != config.storage_class:
                 raise AssertionError("lc validation for object transition failed")
     else:
-        op = utils.exec_shell_cmd(
-            "radosgw-admin bucket stats --bucket=%s" % bucket.name
-        )
+        bkt_stats_cmd = f"radosgw-admin bucket stats --bucket {bucket.name}"
+        if ssh_con:
+            op = utils.remote_exec_shell_cmd(ssh_con, bkt_stats_cmd, return_output=True)
+        else:
+            op = utils.exec_shell_cmd(bkt_stats_cmd)
         json_doc = json.loads(op)
         objects = json_doc["usage"]["rgw.main"]["num_objects"]
         if objects != 0:
@@ -276,13 +282,18 @@ def validate_and_rule(bucket, config):
     log.info("verification starts")
     op = utils.exec_shell_cmd("radosgw-admin bucket stats --bucket=%s" % bucket.name)
     json_doc = json.loads(op)
-    op2 = utils.exec_shell_cmd(f"radosgw-admin bucket list --bucket {bucket.name}")
+    op2 = utils.exec_shell_cmd(
+        f"radosgw-admin bucket list --bucket {bucket.name}{f' --max-entries {config.objects_count}' if  config.objects_count > 1000 else ''}"
+    )
     json_doc2 = json.loads(op2)
     objects = json_doc["usage"]["rgw.main"]["num_objects"]
     if config.test_lc_transition and not config.test_ops.get(
         "conflict_btw_exp_transition"
     ):
-        for i in range(0, config.objects_count):
+        obj_count = (
+            config.objects_count if config.objects_count < 1000 else len(json_doc2)
+        )
+        for i in range(0, obj_count):
             storage_class = json_doc2[i]["meta"]["storage_class"]
             if storage_class != config.storage_class:
                 raise AssertionError("LC transition for AND filters failed")

rgw/v2/tests/s3_swift/reusable.py

@@ -1058,7 +1058,7 @@ def put_get_bucket_lifecycle_test(
     for rule in config.lifecycle_conf:
         if rule.get("Expiration", {}).get("Date", False):
             # todo: need to get the interval value from yaml file
-            log.info("wait for 60 seconds")
+            log.info(f"wait for 60 seconds")
             time.sleep(60)
         else:
             while time.time() < time_limit:
@@ -1071,10 +1071,11 @@ def put_get_bucket_lifecycle_test(
                     time.sleep(config.rgw_lc_debug_interval)
                 else:
                     raise TestExecError("Objects expired before the expected days")
+    lc_grace_time = config.test_ops.get("lc_grace_time", 90)
     log.info(
-        f"sleeping for {time_diff + 90} seconds so that all objects gets expired/transitioned"
+        f"sleeping for {time_diff + lc_grace_time} seconds so that all objects gets expired/transitioned"
     )
-    time.sleep(time_diff + 90)
+    time.sleep(time_diff + lc_grace_time)
 
     if config.test_ops.get("conflict_exp_days"):
         bucket_stats_op = utils.exec_shell_cmd(
@@ -1095,10 +1096,14 @@ def put_get_bucket_lifecycle_test(
     for i, entry in enumerate(json_doc):
         print(i)
         print(entry["status"])
-        if entry["status"] == "COMPLETE" or entry["status"] == "PROCESSING":
-            log.info("LC is applied on the bucket")
-        else:
-            raise TestExecError("LC is not applied")
+        if bucket.name in entry["bucket"]:
+            if entry["status"] == "COMPLETE" or entry["status"] == "PROCESSING":
+                log.info("LC is applied on the bucket")
+            else:
+                raise TestExecError("LC is not applied")
+            break
+    else:
+        raise TestExecError("bucket not listed in lc list")
     if config.test_ops.get("tenant_name"):
         tenant_name = config.test_ops.get("tenant_name")
         op_lc_get = utils.exec_shell_cmd(

rgw/v2/tests/s3_swift/test_bucket_lifecycle_object_expiration_transition.py

@@ -82,19 +82,22 @@ def test_exec(config, ssh_con):
         ConfigOpts.rgw_lc_debug_interval,
         str(config.rgw_lc_debug_interval),
         ssh_con,
+        set_to_all=config.test_ops.get("set_ceph_configs_to_all_daemons", False),
     )
     if not config.rgw_enable_lc_threads:
         ceph_conf.set_to_ceph_conf(
             "global",
             ConfigOpts.rgw_enable_lc_threads,
             str(config.rgw_enable_lc_threads),
             ssh_con,
+            set_to_all=config.test_ops.get("set_ceph_configs_to_all_daemons", False),
         )
     ceph_conf.set_to_ceph_conf(
         "global",
         ConfigOpts.rgw_lifecycle_work_time,
         str(config.rgw_lifecycle_work_time),
         ssh_con,
+        set_to_all=config.test_ops.get("set_ceph_configs_to_all_daemons", False),
     )
     _, version_name = utils.get_ceph_version()
     if "nautilus" in version_name:
@@ -110,6 +113,72 @@ def test_exec(config, ssh_con):
             option=ConfigOpts.rgw_lc_max_worker,
             value=str(config.rgw_lc_max_worker),
             ssh_con=ssh_con,
+            set_to_all=config.test_ops.get("set_ceph_configs_to_all_daemons", False),
+        )
+    ceph_conf.set_to_ceph_conf(
+        "global",
+        ConfigOpts.rgw_lc_max_wp_worker,
+        str(config.rgw_lc_max_wp_worker),
+        ssh_con,
+        set_to_all=config.test_ops.get("set_ceph_configs_to_all_daemons", False),
+    )
+    log.info(f"test_lc_on_other_site: {config.test_ops.get('test_lc_on_other_site')}")
+    if config.test_ops.get("test_lc_on_other_site", False) is True:
+        log.info(f"other_site_master: {config.test_ops.get('other_site_master')}")
+        other_site_rgw_ip = utils.get_rgw_ip(
+            config.test_ops.get("other_site_master", False)
+        )
+        log.info(f"other_site_rgw_ip: {other_site_rgw_ip}")
+        other_site_ssh_con = utils.connect_remote(other_site_rgw_ip)
+        log.info(f"other_site_ssh_con: {other_site_ssh_con}")
+        ceph_conf.set_to_ceph_conf(
+            "global",
+            ConfigOpts.rgw_lc_debug_interval,
+            str(config.rgw_lc_debug_interval),
+            remote_ssh_con=other_site_ssh_con,
+            set_to_all=config.test_ops.get("set_ceph_configs_to_all_daemons", False),
+        )
+        if not config.rgw_enable_lc_threads:
+            ceph_conf.set_to_ceph_conf(
+                "global",
+                ConfigOpts.rgw_enable_lc_threads,
+                str(config.rgw_enable_lc_threads),
+                remote_ssh_con=other_site_ssh_con,
+                set_to_all=config.test_ops.get(
+                    "set_ceph_configs_to_all_daemons", False
+                ),
+            )
+        ceph_conf.set_to_ceph_conf(
+            "global",
+            ConfigOpts.rgw_lifecycle_work_time,
+            str(config.rgw_lifecycle_work_time),
+            remote_ssh_con=other_site_ssh_con,
+            set_to_all=config.test_ops.get("set_ceph_configs_to_all_daemons", False),
+        )
+        _, version_name = utils.get_ceph_version()
+        if "nautilus" in version_name:
+            ceph_conf.set_to_ceph_conf(
+                "global",
+                ConfigOpts.rgw_lc_max_worker,
+                str(config.rgw_lc_max_worker),
+                remote_ssh_con=other_site_ssh_con,
+            )
+        else:
+            ceph_conf.set_to_ceph_conf(
+                section=None,
+                option=ConfigOpts.rgw_lc_max_worker,
+                value=str(config.rgw_lc_max_worker),
+                remote_ssh_con=other_site_ssh_con,
+                set_to_all=config.test_ops.get(
+                    "set_ceph_configs_to_all_daemons", False
+                ),
+            )
+        ceph_conf.set_to_ceph_conf(
+            "global",
+            ConfigOpts.rgw_lc_max_wp_worker,
+            str(config.rgw_lc_max_wp_worker),
+            ssh_con,
+            set_to_all=config.test_ops.get("set_ceph_configs_to_all_daemons", False),
         )
 
     if config.test_lc_transition:
@@ -120,6 +189,12 @@ def test_exec(config, ssh_con):
 
     log.info("trying to restart services")
     srv_restarted = rgw_service.restart(ssh_con)
+
+    if config.test_ops.get("set_ceph_configs_to_all_daemons", False):
+        utils.restart_rgw(restart_all=True)
+        if config.test_ops.get("test_lc_on_other_site", False):
+            utils.restart_rgw(restart_all=True, ssh_con=other_site_ssh_con)
+
     time.sleep(30)
     if srv_restarted is False:
         raise TestExecError("RGW service restart failed")
@@ -143,7 +218,9 @@ def test_exec(config, ssh_con):
         )
     else:
         log.info(f"user count is {config.user_count}")
-        user_info = s3lib.create_users(config.user_count)
+        user_info = s3lib.create_users(
+            config.user_count, config.user_names, config=config
+        )
     log.info(f"print user info {user_info}")
 
     if config.test_ops.get("send_bucket_notifications", False) is True:
@@ -161,9 +238,12 @@ def test_exec(config, ssh_con):
 
         log.info("no of buckets to create: %s" % config.bucket_count)
         for bc in range(config.bucket_count):
-            bucket_name = utils.gen_bucket_name_from_userid(
-                each_user["user_id"], rand_no=bc
-            )
+            if config.bucket_names:
+                bucket_name = config.bucket_names[bc]
+            else:
+                bucket_name = utils.gen_bucket_name_from_userid(
+                    each_user["user_id"], rand_no=bc
+                )
             obj_list = []
             obj_tag = "suffix1=WMV1"
             bucket = reusable.create_bucket(bucket_name, rgw_conn, each_user)
@@ -691,10 +771,19 @@ def test_exec(config, ssh_con):
                 reusable.put_bucket_lifecycle(
                     bucket, rgw_conn, rgw_conn2, life_cycle_rule
                 )
+            if config.test_ops.get("lc_grace_time"):
+                log.info(
+                    f"sleeping for lc_grace_time {config.test_ops.get('lc_grace_time')} seconds"
+                )
+                time.sleep(config.test_ops.get("lc_grace_time"))
             time.sleep(60)
             for bucket in buckets:
                 if not config.test_ops.get("enable_versioning", False) is True:
                     lc_ops.validate_prefix_rule_non_versioned(bucket, config)
+                    if config.test_ops.get("test_lc_on_other_site", False):
+                        lc_ops.validate_prefix_rule_non_versioned(
+                            bucket, config, other_site_ssh_con
+                        )
                 else:
                     lc_ops.validate_prefix_rule(bucket, config)
                 if config.test_ops.get("send_bucket_notifications", False) is True:
@@ -706,9 +795,14 @@ def test_exec(config, ssh_con):
             )
             rgw_service.restart()
             time.sleep(30)
+
+        if config.test_ops.get("delete_bucket_object", False):
+            for bucket in buckets:
+                reusable.delete_bucket(bucket)
+
         if config.test_ops.get("test_via_rgw_accounts", False) is True:
             log.info("do not remove user")
-        else:
+        elif config.user_remove:
             reusable.remove_user(each_user)
         # check for any crashes during the execution
         crash_info = reusable.check_for_crash()