Alert recipient email addresses for failing AWS EC2 Instances Health Checks

import boto3 from botocore.exceptions import ClientError # Load AWS credentials from a secure source creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] def get_sns_client(access_key, secret_key, aws_region="us-east-1"): """ Initialize an SNS client for the primary notification region. """ return boto3.client( 'sns', region_name=aws_region, aws_access_key_id=access_key, aws_secret_access_key=secret_key ) def find_sns_topic(sns_client, topic_name): """ Attempt to find an existing SNS topic by name. """ try: response = sns_client.list_topics() for topic in response['Topics']: if topic_name in topic['TopicArn']: return topic['TopicArn'] except ClientError as e: print(f"Error retrieving SNS topics: {e}") return None def setup_sns_topic(topic_name, sns_client): """ Setup or retrieve an SNS topic and return its ARN. """ topic_arn = find_sns_topic(sns_client, topic_name) if not topic_arn: try: topic = sns_client.create_topic(Name=topic_name) topic_arn = topic['TopicArn'] except ClientError as e: print(f"Failed to create SNS topic: {e}") return None return topic_arn def check_subscription(topic_arn, email_address, sns_client): """ Check if an email address is already subscribed to the topic. """ try: response = sns_client.list_subscriptions_by_topic(TopicArn=topic_arn) for subscription in response['Subscriptions']: if subscription['Protocol'] == 'email' and subscription['Endpoint'] == email_address: print(f"{email_address} is already subscribed to the topic.") return True except ClientError as e: print(f"Error checking subscriptions for {email_address}: {e}") return False def subscribe_to_topic(topic_arn, email_address, sns_client): """ Subscribe an email address to an SNS topic if not already subscribed. """ if not check_subscription(topic_arn, email_address, sns_client): try: sns_client.subscribe( TopicArn=topic_arn, Protocol='email', Endpoint=email_address ) print(f"Subscription request sent to {email_address}. They need to confirm the subscription.") except ClientError as e: print(f"Failed to subscribe {email_address}: {e}") def publish_to_topic(topic_arn, failed_instances, sns_client): """ Publish a message to an SNS topic about failing EC2 instances. """ message = ("EC2 Instance Health Check Failure\n\n" "The following instances are failing health checks:\n" + "\n".join([f"Instance ID: {inst['InstanceId']} in {inst['Region']} at {inst['AlertTime']}" for inst in failed_instances])) subject = "Alert: EC2 Instance Health Check Failure" try: response = sns_client.publish( TopicArn=topic_arn, Message=message, Subject=subject ) print(f"Message published to SNS topic: {response['MessageId']}") except ClientError as e: print(f"An error occurred: {e}") # Main function to process failed instances and alert via SNS in a primary region def alert_failed_instances(failed_instances, access_key, secret_key, recipient_email, sns_topic_name="ec2-health-alerts", aws_region="us-east-1"): """ Main function to process failed instances and alert via SNS. :param failed_instances: List of dictionaries containing instance details. :param access_key: AWS access key ID. :param secret_key: AWS secret access key. :param recipient_email: Email address to receive alerts. :param sns_topic_name: (Optional) SNS topic name for publishing alerts. :param aws_region: (Optional) AWS region for the SNS client and topic. """ sns_client = get_sns_client(access_key, secret_key, aws_region) topic_arn = setup_sns_topic(sns_topic_name, sns_client) if topic_arn: subscribe_to_topic(topic_arn, recipient_email, sns_client) publish_to_topic(topic_arn, failed_instances, sns_client) '''failed_instances = [ {'InstanceId': 'i-0123456789abcdef0', 'Region': 'us-west-2', 'AlertTime': '2024-05-17T12:33:21.757843'}, {'InstanceId': 'i-023456789abcdef01', 'Region': 'us-east-1', 'AlertTime': '2024-05-17T12:33:23.036687'} ]''' #recipient_email='xyz@example.com' sns_topic_name='ec2-health-alerts' alert_failed_instances(failed_instances, access_key, secret_key, recipient_email)