Delete ibmcloud buckets

ocs_ci/cleanup/ibm/cleanup.py

@@ -0,0 +1,84 @@
+import argparse
+import logging
+import re
+from datetime import datetime, timedelta
+
+from ocs_ci.framework import config
+from ocs_ci.deployment.ibmcloud import IBMCloudIPI
+from ocs_ci.cleanup.ibm import defaults
+
+
+logger = logging.getLogger(__name__)
+
+
+def ibm_cleanup():
+    parser = argparse.ArgumentParser(
+        description="ibmcloud cleanup",
+        formatter_class=argparse.RawTextHelpFormatter,
+    )
+    bucket_group = parser.add_argument_group("S3 Bucket Sweeping Options")
+    bucket_group.add_argument(
+        "--sweep-buckets", action="store_true", help="Deleting S3 buckets."
+    )
+    parser.add_argument(
+        "--hours-buckets",
+        action="store",
+        required=False,
+        help="""
+            Running time for the buckets in hours.
+            Buckets older than to this will be deleted.
+        """,
+    )
+    args = parser.parse_args()
+    if args.sweep_buckets:
+        bucket_hours = (
+            args.hours_buckets
+            if args.hours_buckets is not None
+            else defaults.DEFAULT_TIME_BUCKETS
+        )
+        delete_buckets(bucket_hours)
+
+
+def delete_buckets(hours):
+    """ """
+    status = []
+    config.ENV_DATA["cluster_path"] = "/"
+    config.ENV_DATA["cluster_name"] = "cluster"
+    ibm_cloud_ipi_obj = IBMCloudIPI()
+    buckets = ibm_cloud_ipi_obj.get_bucket_list()
+    buckets_delete = buckets_to_delete(buckets, hours)
+    for bucket_delete in buckets_delete:
+        try:
+            ibm_cloud_ipi_obj.delete_bucket(bucket_delete)
+        except Exception as e:
+            log = f"Failed to delete {bucket_delete}\nerror: {e}"
+            logger.info(log)
+            status.append(log)
+    if len(status) > 0:
+        raise Exception(status)
+
+
+def buckets_to_delete(buckets, hours):
+    """
+    Buckets to Delete
+
+    Args:
+
+    """
+    buckets_delete = []
+    current_time = datetime.utcnow()
+    for bucket in buckets:
+        bucket_name = bucket["Name"]
+        creation_date = datetime.strptime(
+            bucket["CreationDate"], "%Y-%m-%dT%H:%M:%S.%fZ"
+        )
+        # Check if the bucket matches any prefix rule
+        hours_bucket = hours
+        for prefix, max_age_hours in defaults.BUCKET_PREFIXES_SPECIAL_RULES.items():
+            if re.match(prefix, bucket_name):
+                hours_bucket = max_age_hours
+        if hours_bucket == "never":
+            continue
+        if current_time - creation_date > timedelta(hours=int(hours_bucket)):
+            buckets_delete.append(bucket_name)
+    return buckets_delete[:10]

ocs_ci/cleanup/ibm/defaults.py

@@ -0,0 +1,22 @@
+"""
+Defaults module for IBM cleanup
+"""
+
+BUCKET_PREFIXES_SPECIAL_RULES = {
+    "jnk-pr": 100,  # keep it as first item before jnk prefix for fist match
+    "jnk": 200,
+    "j\\d\\d\\d": 60,
+    "j-\\d\\d\\d": 60,
+    "odf-qe": "never",
+    "Default": "never",
+    "dnd": "never",
+    "lr1": 24,
+    "lr2": 48,
+    "lr3": 72,
+    "lr4": 96,
+    "promptlab-donotdelete-": "never",
+    "ibmcos-uls": 987,
+}
+
+DEFAULT_TIME_BUCKETS = 100
+IBM_REGION = "us-south"

ocs_ci/deployment/ibmcloud.py

@@ -7,6 +7,7 @@
 import json
 import logging
 import os
+import re
 
 from ocs_ci.deployment.cloud import CloudDeploymentBase, IPIOCPDeployment
 from ocs_ci.deployment.ocp import OCPDeployment as BaseOCPDeployment
@@ -571,3 +572,41 @@ def prepare_custom_vpc_and_network(self):
                 ibmcloud.attach_subnet_to_public_gateway(
                     subnet_name, gateway_name, vpc_name
                 )
+
+    def get_bucket_list(self):
+        """
+        Get Buckets in 'ODF-QE-IBM-COS'
+
+        """
+        cmd = "ibmcloud resource service-instance 'ODF-QE-IBM-COS' --output json"
+        proc = exec_cmd(cmd)
+        instance_objs = json.loads(proc.stdout)
+        cmd = f"ibmcloud cos config crn --crn {instance_objs[0]['guid']}"
+        exec_cmd(cmd)
+        cmd = "ibmcloud cos buckets --output json"
+        proc = exec_cmd(cmd)
+        return json.loads(proc.stdout).get("Buckets")
+
+    def delete_bucket(self, bucket_name):
+        """
+        Delete Bucket
+
+        Args:
+            bucket_name (str): bucket name
+
+        """
+        cmd = f"ibmcloud cos bucket-location-get --bucket {bucket_name}"
+        proc = exec_cmd(cmd)
+        match = re.search(r"Region: (\w+)", proc.stdout.decode("utf-8"))
+        region = match.group(1)
+        cmd = f"ibmcloud cos objects --bucket {bucket_name} --region {region} --output json"
+        proc = exec_cmd(cmd)
+        objects = json.loads(proc.stdout).get("Contents")
+        for object in objects:
+            cmd = (
+                f"ibmcloud cos object-delete --bucket {bucket_name} "
+                f"--key {object.get('Key')} --region {region} --force"
+            )
+            exec_cmd(cmd)
+        cmd = f"ibmcloud cos bucket-delete --bucket {bucket_name} --region {region} --force"
+        exec_cmd(cmd)

setup.py

@@ -114,6 +114,7 @@
             "vsphere-cleanup=ocs_ci.cleanup.vsphere.cleanup:vsphere_cleanup",
             "ocs-build=ocs_ci.utility.ocs_build:main",
             "get-ssl-cert=ocs_ci.utility.ssl_certs:main",
+            "ibm-cleanup=ocs_ci.cleanup.ibm.cleanup:ibm_cleanup",
         ],
     },
     zip_safe=True,