Filter out AWS ECS Clusters having low CPU Utilization

import boto3 import datetime from botocore.exceptions import ClientError creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] def get_ecs_clusters_with_low_cpu_utilization(regions, threshold): """ Identifies ECS clusters with low average CPU utilization across multiple AWS regions. Args: regions (List[str]): List of AWS regions to check. threshold (int): CPU utilization percentage below which a cluster is considered underutilized. Returns: List[dict]: List of dictionaries containing cluster and service/task details. """ low_cpu_clusters = [] # List to store details of low CPU utilization clusters # Loop through each region for region in regions: try: # Initialize ECS and CloudWatch clients for the region ecs = boto3.client('ecs', aws_access_key_id=access_key,aws_secret_access_key=secret_key,region_name=region) cloudwatch = boto3.client('cloudwatch', aws_access_key_id=access_key,aws_secret_access_key=secret_key,region_name=region) # Fetch all cluster ARNs in the region clusters = ecs.list_clusters()['clusterArns'] # Loop through each cluster for cluster in clusters: cluster_name = cluster.split('/')[-1] # Extract the cluster name from the ARN # Define the time range for CloudWatch metrics as the last 60 minutes end_time = datetime.datetime.utcnow() start_time = end_time - datetime.timedelta(hours=1) # Fetch service ARNs in the cluster services = ecs.list_services(cluster=cluster_name)['serviceArns'] # Loop through each service to fetch its average CPU utilization for service in services: service_name = service.split('/')[-1] # Extract the service name from the ARN # Get average CPU utilization from CloudWatch cpu_response = cloudwatch.get_metric_statistics( Namespace='AWS/ECS', MetricName='CPUUtilization', Dimensions=[ {'Name': 'ClusterName', 'Value': cluster_name}, {'Name': 'ServiceName', 'Value': service_name}, ], StartTime=start_time, EndTime=end_time, Period=300, Statistics=['Average'] ) # Calculate average CPU utilization avg_cpu_service = sum(datapoint['Average'] for datapoint in cpu_response['Datapoints']) / len(cpu_response['Datapoints']) if cpu_response['Datapoints'] else 0 # Check if the average CPU utilization is below the threshold if avg_cpu_service < threshold: low_cpu_clusters.append({ 'Region': region, 'ClusterName': cluster_name, 'ServiceName': service_name, 'AverageCPU': avg_cpu_service, 'Type': 'Service' }) # Fetch task ARNs in the cluster tasks = ecs.list_tasks(cluster=cluster_name)['taskArns'] # Loop through each task to fetch its average CPU utilization for task in tasks: task_name = task.split('/')[-1] # Extract the task name from the ARN # Get average CPU utilization from CloudWatch cpu_response = cloudwatch.get_metric_statistics( Namespace='AWS/ECS', MetricName='CPUUtilization', Dimensions=[ {'Name': 'ClusterName', 'Value': cluster_name}, {'Name': 'TaskId', 'Value': task_name}, ], StartTime=start_time, EndTime=end_time, Period=300, Statistics=['Average'] ) # Calculate average CPU utilization avg_cpu_task = sum(datapoint['Average'] for datapoint in cpu_response['Datapoints']) / len(cpu_response['Datapoints']) if cpu_response['Datapoints'] else 0 # Check if the average CPU utilization is below the threshold if avg_cpu_task < threshold: low_cpu_clusters.append({ 'Region': region, 'ClusterName': cluster_name, 'TaskName': task_name, 'AverageCPU': avg_cpu_task, 'Type': 'Task' }) except ClientError as ce: print(f"A botocore exception occurred in region {region}: {ce.response['Error']['Message']}") except Exception as e: print(f"An unknown error occurred in region {region}: {e}") # General exception handling return low_cpu_clusters # Return the list of low CPU utilization clusters/services/tasks def display_low_cpu_clusters(data): # Initialize table with the desired structure and headers table = context.newtable() table.title = "Low CPU Usage Clusters Overview" table.num_cols = 5 # Number of columns for Region, ClusterName, TaskName, AverageCPU, and Type 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_clusters data headers = ["Region", "Cluster Name", "Task/Service Name", "Average CPU (%)", "Type"] # Set headers in the first row for col_num, header in enumerate(headers): table.setval(0, col_num, header) # Populate the table with data for row_num, cluster_info in enumerate(data, start=1): # Starting from the second row table.num_rows += 1 # Add a row for each entry values = [ cluster_info["Region"], cluster_info["ClusterName"], cluster_info.get("TaskName", cluster_info.get("ServiceName", "N/A")), # Handle both Task and Service f"{cluster_info['AverageCPU']:.2f}", # Format average CPU as a float with 2 decimal places cluster_info["Type"] ] for col_num, value in enumerate(values): table.setval(row_num, col_num, value) # Define the AWS regions and CPU utilization threshold #regions_to_check = ['us-east-1', 'us-west-2'] #cpu_threshold = 20 # In percentage # Execute the function and get low CPU utilization clusters low_cpu_clusters_list = get_ecs_clusters_with_low_cpu_utilization(regions, threshold=int(cpu_threshold)) # Display the result if low_cpu_clusters_list: display_low_cpu_clusters(low_cpu_clusters_list) ''' #print(f"Found {len(low_cpu_clusters_list)} ECS clusters with low CPU Utilization") #for entry in low_cpu_clusters_list: #print(entry) # Print the header print(f"{'Region':<15} {'ClusterName':<20} {'Type':<10} {'Name':<25} {'Average CPU Usage':<20}") # Iterate through each entry in the list for item in low_cpu_clusters_list: region = item['Region'] cluster_name = item['ClusterName'] entity_type = item['Type'] name = item.get('ServiceName', item.get('TaskName', 'Unknown')) # Use ServiceName or TaskName, default to 'Unknown' cpu_usage = item['AverageCPU'] # Print each item's details in a formatted manner print(f"{region:<15} {cluster_name:<20} {entity_type:<10} {name:<25} {cpu_usage:<20.2f}")''' else: print(f"Found {len(low_cpu_clusters_list)} ECS clusters with low CPU Utilization") context.skip_sub_tasks=True