Manage Unused AWS Lambda Functions

import boto3 creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] def get_regions_for_service(): """Retrieve all regions where a specific AWS service is available.""" ec2 = boto3.client('ec2', aws_access_key_id=access_key, aws_secret_access_key=secret_key, region_name="us-east-1") return [region['RegionName'] for region in ec2.describe_regions()['Regions']] def get_lambda_client(region_name=None): """Create a Lambda client for the specified region.""" return boto3.client('lambda', aws_access_key_id=access_key, aws_secret_access_key=secret_key, region_name=region_name) def list_lambda_functions_in_region(region_name): """List all Lambda functions in a specific region, including the region name with each function.""" lambda_client = get_lambda_client(region_name) functions = [] paginator = lambda_client.get_paginator('list_functions') try: for page in paginator.paginate(): functions.extend(page['Functions']) for function in functions: function['Region'] = region_name # Add region information to each function except Exception as e: print(f"An error occurred in {region_name}: {e}") return [] return functions def display_lambda_functions(all_functions): """Display all Lambda functions in a formatted table.""" # Initialize table with the desired structure and headers table = context.newtable() table.title = "AWS Lambda Function Details" table.num_cols = 5 # Number of columns according to headers table.num_rows = 1 # Starts with one row for headers table.has_header_row = True headers = ["Function Name", "Region", "Memory Size", "Runtime", "Last Modified"] # Set headers in the first row for col_num, header in enumerate(headers): table.setval(0, col_num, header) # Populate the table with Lambda function data for row_num, function in enumerate(all_functions, start=1): # Starting from the second row table.num_rows += 1 # Add a row for each function values = [ function['FunctionName'], function['Region'], f"{function['MemorySize']} MB", function['Runtime'], function['LastModified'] ] for col_num, value in enumerate(values): table.setval(row_num, col_num, value) regions = get_regions_for_service() all_functions = [] for region in regions: functions = list_lambda_functions_in_region(region) if functions: all_functions.extend(functions) if all_functions: display_lambda_functions(all_functions) else: print("No Lambda functions found across all regions.") #print(all_functions) # for debugging

import boto3 from datetime import datetime, timedelta creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] def get_cloudwatch_events(lambda_arn, region): """Check if a Lambda function is triggered by any CloudWatch event.""" events_client = boto3.client('events', aws_access_key_id=access_key, aws_secret_access_key=secret_key, region_name=region) try: rules = events_client.list_rules() for rule in rules['Rules']: targets = events_client.list_targets_by_rule(Rule=rule['Name']) for target in targets['Targets']: if target['Arn'] == lambda_arn: return True except Exception as e: print(f"Error checking CloudWatch events for {lambda_arn}: {e}") return False def get_lambda_invocations(func_name, region, days_threshold): """Check for Lambda invocations in the last specified number of days.""" cloudwatch_client = boto3.client('cloudwatch', aws_access_key_id=access_key, aws_secret_access_key=secret_key, region_name=region) end_time = datetime.now() start_time = end_time - timedelta(days=days_threshold) try: metrics = cloudwatch_client.get_metric_statistics( Namespace='AWS/Lambda', MetricName='Invocations', Dimensions=[{'Name': 'FunctionName', 'Value': func_name}], StartTime=start_time, EndTime=end_time, Period=86400, Statistics=['Sum'] ) total_invocations = sum(data_point['Sum'] for data_point in metrics['Datapoints']) return total_invocations == 0 except Exception as e: print(f"Error retrieving invocations for {func_name}: {e}") return False def filter_unused_lambda_functions(functions, days_threshold): unused_functions = [] for func in functions: region = func['Region'] if not get_cloudwatch_events(func['FunctionArn'], region) and get_lambda_invocations(func['FunctionName'], region, days_threshold): last_modified = datetime.strptime(func['LastModified'], '%Y-%m-%dT%H:%M:%S.%f%z') details = { 'FunctionName': func['FunctionName'], 'Region': region, 'LastModified': last_modified.strftime('%Y-%m-%d %H:%M:%S'), 'Runtime': func['Runtime'], 'MemorySize': func['MemorySize'] } unused_functions.append(details) return unused_functions def display_lambda_details(data): """Display all unused Lambda functions in a formatted table.""" table = context.newtable() table.title = "Unused AWS Lambda Functions" table.num_cols = 5 table.num_rows = 1 table.has_header_row = True headers = ["Function Name", "Region", "Memory Size", "Runtime", "Last Modified"] for col_num, header in enumerate(headers): table.setval(0, col_num, header) for row_num, func in enumerate(data, start=1): table.num_rows += 1 values = [ func['FunctionName'], func['Region'], f"{func['MemorySize']} MB", func['Runtime'], func['LastModified'] ] for col_num, value in enumerate(values): table.setval(row_num, col_num, value) # Example usage #last_accessed_threshold_days = 90 # Modify as needed #all_functions = [{'FunctionName': 'exampleFunction', 'Region': 'us-east-1', 'LastModified': '2023-01-01T00:00:00.000Z', 'Runtime': 'python3.8', 'MemorySize': 128}] # Placeholder data unused_functions = filter_unused_lambda_functions(all_functions, last_accessed_threshold_days) if unused_functions: display_lambda_details(unused_functions) else: print("No unused Lambda functions found.") context.skip_sub_tasks = True