Delete Underutilized AWS RDS Instances

import boto3 from botocore.exceptions import BotoCoreError, ClientError reg=None # Runs for specific region if provided otherwise runs for all regions if None/nothing is provided. creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] def list_all_rds_instances(region=None): try: if region: regions = [region] else: # If region is None, list instances in all available regions ec2 = boto3.client('ec2', aws_access_key_id=access_key,aws_secret_access_key=secret_key,region_name='us-east-1') regions = [region['RegionName'] for region in ec2.describe_regions()['Regions']] all_instances = [] for region in regions: #print(f"Listing RDS instances in region {region}:") client = boto3.client('rds', aws_access_key_id=access_key,aws_secret_access_key=secret_key,region_name=region) try: db_instances = client.describe_db_instances() region_instances = [instance['DBInstanceIdentifier'] for instance in db_instances['DBInstances']] if region_instances: print(f"Found {len(region_instances)} RDS instances in region {region}:") for instance in region_instances: print(instance) all_instances.append({"region": region, "instance": instance}) else: #print(f"No RDS instances found in region {region}.") p=1 # dummy line to end the conditional block properly except ClientError as e: print(f"Client error in region {region}: {e}") except BotoCoreError as e: print(f"BotoCoreError in region {region}: {e}") except Exception as e: print(f"Unexpected error in region {region}: {e}") return all_instances except Exception as e: print(f"Unexpected error: {e}") instances = list_all_rds_instances(reg) # reg is initialized in input parameters. Runs for specific region if provided otherwise runs for all regions if None/nothing is provided. #print(instances)

import boto3 from datetime import datetime, timedelta from botocore.exceptions import BotoCoreError, ClientError creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] # Constants # Define the threshold for CPU utilization(Defined in input parameters). # Instances below this will be considered low utilization. LOW_CPU_THRESHOLD=20 # Hardcoded for One time Filter Result LOOKBACK_PERIOD_HOURS=24 # Hardcoded for One time Filter Result LOOKBACK_PERIOD = 3600 * int(LOOKBACK_PERIOD_HOURS) # Define the period to check. Here, it's set to 24 hours. def get_low_cpu_rds_instances(instances): low_cpu_instances = [] # List to store the IDs of RDS instances with low CPU utilization # Group instances by region instances_by_region = {} for instance_info in instances: region = instance_info['region'] if region not in instances_by_region: instances_by_region[region] = [] instances_by_region[region].append(instance_info['instance']) # Check each region for region, instance_list in instances_by_region.items(): cloudwatch = boto3.client('cloudwatch', aws_access_key_id=access_key,aws_secret_access_key=secret_key,region_name=region) # Loop through each RDS instance for instance_id in instance_list: try: end_time = datetime.utcnow() start_time = end_time - timedelta(seconds=LOOKBACK_PERIOD) metrics = cloudwatch.get_metric_data( MetricDataQueries=[ { 'Id': 'cpuUtilization', 'MetricStat': { 'Metric': { 'Namespace': 'AWS/RDS', 'MetricName': 'CPUUtilization', 'Dimensions': [{ 'Name': 'DBInstanceIdentifier', 'Value': instance_id }] }, 'Period': LOOKBACK_PERIOD, 'Stat': 'Average' }, 'ReturnData': True, }, ], StartTime=start_time, EndTime=end_time ) if metrics['MetricDataResults'][0]['Values']: cpu_utilization = metrics['MetricDataResults'][0]['Values'][0] if cpu_utilization < int(LOW_CPU_THRESHOLD): low_cpu_instances.append({ 'InstanceID': instance_id, 'AverageCPU': cpu_utilization, 'Region': region }) except Exception as e: print(f"Error fetching CloudWatch metrics for RDS instance {instance_id} in {region}: {e}") return low_cpu_instances def display_low_cpu_instances(data): # Initialize table with the desired structure and headers table = context.newtable() table.title = "Low Usage RDS Instances Overview" table.num_cols = 3 # Number of columns for Instance ID, Average CPU, and Region table.num_rows = 1 # Starts with one row for headers table.has_header_row = True # Define header names based on the structure of low_cpu_instances data headers = ["Instance ID", "Average CPU (%)", "Region"] # Set headers in the first row for col_num, header in enumerate(headers): table.setval(0, col_num, header) # Populate the table with instance data for row_num, instance in enumerate(data, start=1): # Starting from the second row table.num_rows += 1 # Add a row for each entry values = [ instance["InstanceID"], f"{instance['AverageCPU']:.2f}", # Format average CPU as a float with 2 decimal places instance["Region"] ] for col_num, value in enumerate(values): table.setval(row_num, col_num, value) low_cpu_rds = get_low_cpu_rds_instances(instances) # Print the results if low_cpu_rds: display_low_cpu_instances(low_cpu_rds) ''' print("RDS instances with low CPU utilization:") for instance_info in low_cpu_rds: print(f"Instance ID: {instance_info['InstanceID']}, Average CPU: {instance_info['AverageCPU']}% in Region: {instance_info['Region']}")''' else: print("No RDS instances with low CPU utilization found.") context.skip_sub_tasks=True