Refractored the S3 inactive to new approach

cloud_governance/common/clouds/aws/s3/s3_operations.py

@@ -9,6 +9,7 @@
 from os import listdir
 from os.path import isfile, join
 
+from cloud_governance.common.logger.init_logger import logger
 from cloud_governance.common.logger.logger_time_stamp import logger_time_stamp
 
 
@@ -325,3 +326,64 @@ def get_last_s3_policy_content(self, policy: str = '', file_name: str = '', s3_f
             os.system(f"gzip -d {local_file}")
             with open(os.path.join(temp_local_directory, file_name)) as f:
                 return f.read()
+
+    def list_buckets(self):
+        """
+        This method list all buckets
+        :return:
+        """
+        try:
+            return self.__s3_client.list_buckets().get('Buckets', [])
+        except Exception as err:
+            return []
+
+    def get_bucket_tagging(self, bucket_name: str, **kwargs):
+        """
+        This method get tagging buckets
+        :param bucket_name:
+        :return:
+        """
+        tags = []
+        try:
+            bucket_tags = self.__s3_client.get_bucket_tagging(Bucket=bucket_name, **kwargs)
+            tags = bucket_tags.get('TagSet', [])
+        except Exception as err:
+            logger.error(err)
+        return tags
+
+    def list_objects_v2(self, bucket_name: str, prefix: str = '', **kwargs):
+        """
+        This method list all objects of a bucket
+        :param bucket_name:
+        :param prefix:
+        :return:
+        """
+        bucket_data = {}
+        try:
+            bucket_data = self.__s3_client.list_objects_v2(Bucket=bucket_name, Prefix=prefix, **kwargs)
+        except Exception as err:
+            logger.error(err)
+        return bucket_data
+
+    def get_bucket_contents(self, bucket_name: str, **kwargs):
+        """
+        This method get contents of a bucket
+        :param bucket_name:
+        :param kwargs:
+        :return:
+        """
+        bucket_data = self.list_objects_v2(bucket_name, **kwargs)
+        return bucket_data.get('Contents', [])
+
+    def get_bucket_location(self, bucket_name: str, **kwargs):
+        """
+        This method get location of a bucket
+        :param bucket_name:
+        :param kwargs:
+        :return:
+        """
+        try:
+            return self.__s3_client.get_bucket_location(Bucket=bucket_name, **kwargs).get('LocationConstraint', self.__region)
+        except Exception as err:
+            logger.error(err)
+            return self.__region

cloud_governance/main/main_oerations/main_operations.py

@@ -39,7 +39,8 @@ def run(self):
         for policy_type, policies in policies_list.items():
             # @Todo support for all the aws policies, currently supports ec2_run as urgent requirement
             if self._policy in policies and self._policy in ["instance_run", "unattached_volume", "cluster_run",
-                                                             "ip_unattached", "unused_nat_gateway", "instance_idle"]:
+                                                             "ip_unattached", "unused_nat_gateway", "instance_idle",
+                                                             "s3_inactive"]:
                 source = policy_type
                 if Utils.equal_ignore_case(policy_type, self._public_cloud_name):
                     source = ''

cloud_governance/policy/aws/s3_inactive.py

@@ -2,57 +2,57 @@
 from botocore.exceptions import ClientError
 
 from cloud_governance.common.logger.init_logger import logger
+from cloud_governance.common.utils.utils import Utils
+from cloud_governance.policy.helpers.aws.aws_policy_operations import AWSPolicyOperations
 from cloud_governance.policy.policy_operations.aws.zombie_non_cluster.run_zombie_non_cluster_policies import NonClusterZombiePolicy
 
 
-class S3Inactive(NonClusterZombiePolicy):
+class S3Inactive(AWSPolicyOperations):
     """
     This class sends an alert mail for empty bucket to the user after 4 days and delete after 7 days.
     """
 
+    RESOURCE_ACTION = 'Delete'
+
     def __init__(self):
         super().__init__()
+        self.__global_active_cluster_ids = self._get_global_active_cluster_ids()
 
-    def run(self):
-        """
-        This method returns all Empty buckets and delete if dry_run no
-        @return:
+    def run_policy_operations(self):
         """
-        return self.__delete_s3_inactive()
-
-    def __delete_s3_inactive(self):
-        """
-        This method delete the empty bucket more than 7 days
-        @return:
+        This method returns all Empty buckets
+        :return:
+        :rtype:
         """
         empty_buckets = []
-        buckets = self._s3_client.list_buckets()['Buckets']
-        for bucket in buckets:
-            bucket_empty = False
-            empty_days = 0
+        s3_buckets = self._s3operations.list_buckets()
+        for bucket in s3_buckets:
             bucket_name = bucket.get('Name')
-            try:
-                try:
-                    bucket_tags = self._s3_client.get_bucket_tagging(Bucket=bucket_name)
-                    tags = bucket_tags.get('TagSet')
-                except ClientError:
-                    tags = []
-                bucket_data = self._s3_client.list_objects_v2(Bucket=bucket_name)
-                if not bucket_data.get('Contents'):
-                    if not self._check_cluster_tag(tags=tags):
-                        if not self._get_tag_name_from_tags(tags=tags, tag_name='Name'):
-                            tags.append({'Key': 'Name', 'Value': bucket_name})
-                        empty_days = self._get_resource_last_used_days(tags=tags)
-                        bucket_empty = True
-                        if not self._get_tag_name_from_tags(tags=tags, tag_name='User'):
-                            region = self._s3_client.get_bucket_location(Bucket=bucket_name)['LocationConstraint']
-                            self._cloudtrail.set_cloudtrail(region_name=region)
-                        empty_bucket = self._check_resource_and_delete(resource_name='S3 Bucket', resource_id='Name', resource_type='CreateBucket', resource=bucket, empty_days=empty_days, days_to_delete_resource=self.DAYS_TO_DELETE_RESOURCE, tags=tags)
-                        if empty_bucket:
-                            empty_buckets.append({'ResourceId': bucket.get('Name'), 'Name': bucket.get('Name'), 'User': self._get_tag_name_from_tags(tags=tags, tag_name='User'), 'Date': str(bucket.get('CreationDate')), 'Days': str(empty_days), 'Skip': self._get_policy_value(tags=tags)})
-                else:
-                    empty_days = 0
-                self._update_resource_tags(resource_id=bucket_name, tags=tags, left_out_days=empty_days, resource_left_out=bucket_empty)
-            except Exception as err:
-                logger.info(f'{err}, {bucket.get("Name")}')
+            tags = self._s3operations.get_bucket_tagging(bucket_name)
+            cleanup_result = False
+            cluster_tag = self._get_cluster_tag(tags=tags)
+            cleanup_days = 0
+            s3_contents = self._s3operations.get_bucket_contents(bucket_name=bucket_name)
+            if cluster_tag not in self.__global_active_cluster_ids and len(s3_contents) == 0:
+                cleanup_days = self.get_clean_up_days_count(tags=tags)
+                cleanup_result = self.verify_and_delete_resource(resource_id=bucket_name, tags=tags,
+                                                                 clean_up_days=cleanup_days)
+                region = self._s3operations.get_bucket_location(bucket_name=bucket_name)
+                resource_data = self._get_es_schema(resource_id=bucket_name,
+                                                    user=self.get_tag_name_from_tags(tags=tags, tag_name='User'),
+                                                    skip_policy=self.get_skip_policy_value(tags=tags),
+                                                    cleanup_days=cleanup_days,
+                                                    dry_run=self._dry_run,
+                                                    name=bucket_name,
+                                                    region=region,
+                                                    cleanup_result=str(cleanup_result),
+                                                    resource_action=self.RESOURCE_ACTION,
+                                                    cloud_name=self._cloud_name,
+                                                    resource_type='EmptyBucket',
+                                                    resource_state="Empty",
+                                                    unit_price=0)
+                empty_buckets.append(resource_data)
+            if not cleanup_result:
+                self.update_resource_day_count_tag(resource_id=bucket_name, cleanup_days=cleanup_days, tags=tags)
+
         return empty_buckets

cloud_governance/policy/helpers/aws/aws_policy_operations.py

@@ -18,7 +18,7 @@ def __init__(self):
         super().__init__()
         self._region = self._environment_variables_dict.get('AWS_DEFAULT_REGION', 'us-east-2')
         self._cloud_name = 'AWS'
-        self.__s3operations = S3Operations(region_name=self._region)
+        self._s3operations = S3Operations(region_name=self._region)
         self._ec2_client = boto3.client('ec2', region_name=self._region)
         self._ec2_operations = EC2Operations(region=self._region)
         self._cloudwatch = CloudWatchOperations(region=self._region)
@@ -173,6 +173,23 @@ def _get_active_cluster_ids(self):
                     break
         return cluster_ids
 
+    def _get_global_active_cluster_ids(self):
+        """
+        This method returns the global active cluster ids
+        :return:
+        """
+        cluster_ids = []
+        active_regions = self._ec2_operations.get_active_regions()
+        for region in active_regions:
+            active_instances = self._ec2_operations.get_ec2_instance_list(ec2_client=boto3.client('ec2',
+                                                                                                  region_name=region))
+            for instance in active_instances:
+                for tag in instance.get('Tags', []):
+                    if tag.get('Key', '').startswith('kubernetes.io/cluster'):
+                        cluster_ids.append(tag.get('Key'))
+                        break
+        return cluster_ids
+
     def _get_cluster_tag(self, tags: list):
         """
         This method returns the cluster_tag