Filter Out Unused AWS CloudWatch Dashboards

import boto3 from datetime import datetime, timedelta creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] def get_cloudtrail_client(region_name): """Create a CloudTrail client for the specified region.""" return boto3.client('cloudtrail', aws_access_key_id=access_key, aws_secret_access_key=secret_key, region_name=region_name) def check_dashboard_activity(dashboard, cloudtrail_client, last_accessed_days_threshold): """Check if there has been recent activity on a dashboard using CloudTrail.""" start_time = datetime.now() - timedelta(days=last_accessed_days_threshold) try: events = cloudtrail_client.lookup_events( LookupAttributes=[ { 'AttributeKey': 'ResourceName', 'AttributeValue': dashboard['DashboardArn'] } ], StartTime=start_time, EndTime=datetime.now() ) return len(events['Events']) == 0 except Exception as e: print(f"Error checking activity for dashboard {dashboard['DashboardName']} in {dashboard['Region']}: {e}") return True # Assume unused if error occurs def filter_unused_dashboards(dashboards, last_accessed_days_threshold=90): """Filter dashboards to find unused ones based on activity and add days_old information.""" unused_dashboards = [] current_date = datetime.now() for dashboard in dashboards: cloudtrail_client = get_cloudtrail_client(dashboard['Region']) if check_dashboard_activity(dashboard, cloudtrail_client, last_accessed_days_threshold): last_modified = datetime.strptime(dashboard['LastModified'], '%Y-%m-%d %H:%M:%S') days_old = (current_date - last_modified).days dashboard['DaysOld'] = days_old # Adding days_old to the dashboard dictionary unused_dashboards.append(dashboard) return unused_dashboards def display_unused_dashboards(unused_dashboards): """Display unused dashboards in a formatted table with an additional 'Days Old' column.""" # Initialize table with the desired structure and headers table = context.newtable() table.title = "AWS CloudWatch Dashboard Details - Unused Dashboards" table.num_cols = 4 # Number of columns according to headers table.num_rows = 1 # Starts with one row for headers table.has_header_row = True headers = ["Dashboard Name", "Region", "Last Modified", "Days Old"] # Set headers in the first row for col_num, header in enumerate(headers): table.setval(0, col_num, header) # Populate the table with dashboard data for row_num, dashboard in enumerate(unused_dashboards, start=1): # Starting from the second row table.num_rows += 1 # Add a row for each dashboard values = [ dashboard['DashboardName'], dashboard['Region'], dashboard.get('LastModified', 'N/A'), # Use 'N/A' if 'LastModified' isn't available str(dashboard.get('DaysOld', 'N/A')) # Convert days old to string; use 'N/A' if not available ] for col_num, value in enumerate(values): table.setval(row_num, col_num, value) # Assuming all_dashboards is passed from upstream task #last_accessed_days_threshold = 90 # Example usage unused_dashboards = filter_unused_dashboards(all_dashboards, last_accessed_days_threshold) display_unused_dashboards(unused_dashboards) context.skip_sub_tasks = True