Automating SES Event Monitoring with AWS Lambda and CloudWatch Logs

Introduction 

Monitoring email notifications sent via Amazon Simple Email Service (SES) is critical for maintaining the health and reliability of your email system. By combining AWS Lambda with CloudWatch Logs, you can efficiently track and process SES event types such as Bounce, Complaint, and Delivery. 

This guide walks you through directly setting up an AWS Lambda function integrated with CloudWatch Logs and Amazon SNS to monitor and log SES events, without the need for a CloudFormation template. 

Overview 

This solution will create the following components: 

IAM Role: Provides the Lambda function the necessary privileges. 

SNS Subscription: Connects an SNS topic to the Lambda function for SES event triggers. 

Lambda Function: Processes SES notifications and logs filtered events to CloudWatch Logs. 

CloudWatch Logs: Stores SES event logs, enabling detailed monitoring and troubleshooting. 

Step-by-Step Implementation 

Step 1: Create the IAM Role for Lambda 

Create an IAM role with the following permissions for your Lambda function: 

{ 
  "Version": "2012-10-17", 
  "Statement": [ 
    { 
      "Effect": "Allow", 
      "Action": "sts:AssumeRole", 
      "Principal": {"Service": "lambda.amazonaws.com"} 
    }, 
    { 
      "Effect": "Allow", 
      "Action": [ 
        "logs:CreateLogGroup", 
        "logs:CreateLogStream", 
        "logs:PutLogEvents", 
        "logs:DescribeLogStreams" 
      ], 
      "Resource": "arn:aws:logs:*:*:log-group:/aws/ses/*" 
    } 
  ] 
} 

Attach this role to the Lambda function. 

Step 2: Create an SNS Topic for SES Notifications 

Navigate to the Amazon SNS console. 

Create a new topic (e.g., SESNotificationsTopic). 

Configure SES to publish Bounce, Complaint, or Delivery events to this topic. 

Open the SES console and go to Email Receiving Rules. 

Set the SNS topic under the Notification tab. 

Step 3: Write the Lambda Function 

Follow the below code to create a Python Lambda: 

import boto3 
import json 
import os 
import logging 
import secrets 
import time 
  
# Configure logging 
logger = logging.getLogger() 
logger.setLevel(logging.INFO) 
  
# Initialize CloudWatch Logs client 
logs_client = boto3.client('logs') 
  
# Environment variables 
log_group = os.getenv('GROUP_NAME', '/aws/ses/events') 
event_type = os.getenv('EVENT_TYPE', 'Bounce') 
  
  
def lambda_handler(event, context): 
    logger.info("Received event: %s", json.dumps(event)) 
  
    for record in event['Records']: 
        sns_message = json.loads(record['Sns']['Message']) 
        notification_type = sns_message.get('notificationType') 
  
        if notification_type != event_type: 
            logger.info("Skipping event of type: %s", notification_type) 
            continue 
  
        log_stream = f"{time.strftime('%Y/%m/%d')}[$LATEST]{secrets.token_hex(8)}" 
  
        try: 
            logs_client.create_log_group(logGroupName=log_group) 
        except logs_client.exceptions.ResourceAlreadyExistsException: 
            logger.info("Log group already exists: %s", log_group) 
  
        try: 
            logs_client.create_log_stream(logGroupName=log_group, logstreamName=log_stream) 
        except logs_client.exceptions.ResourceAlreadyExistsException: 
            logger.info("Log stream already exists: %s", log_stream) 
  
        log_event = { 
            'logGroupName': log_group, 
            'logStreamName': log_stream, 
            'logEvents': [ 
                { 
                    'timestamp': int(time.time() * 1000), 
                    'message': json.dumps(sns_message) 
                } 
            ] 
        } 
  
        response = logs_client.describe_log_streams( 
            logGroupName=log_group, logStreamNamePrefix=log_stream 
        ) 
  
        # Mistake: Missing check for 'logStreams' in response. 
        if 'uploadSequenceToken' in response['logStreams'][0]: 
            log_event['sequenceToken'] = response['logStreams'][0]['uploadSequenceToken'] 
  
        logs_client.put_log_events(**log_event) 
        logger.info("Log published successfully.") 

Step 4: Deploy the Lambda Function 

Environment Variables: 

GROUP_NAME: CloudWatch Log Group name (e.g., /aws/ses/bounce_logs). 

EVENT_TYPE: SES event type to monitor (Bounce, Complaint, or Delivery). 

Timeout and Memory: 

Set the timeout to 60 seconds and memory size to 128 MB. 

Attach the IAM Role  

Upload the Lambda Function Code: 

Zip the Python code and upload it to Lambda. 

Step 5: Subscribe the Lambda Function to the SNS Topic 

In the SNS console, select the topic you created earlier. 

Add a subscription with the following details: 

Protocol: AWS Lambda 

Endpoint: The ARN of your Lambda function. 

Grant SNS permission to invoke your Lambda function. 

aws lambda add-permission \ 
  --function-name <LambdaFunctionName> \ 
  --action lambda:InvokeFunction \ 
  --principal sns.amazonaws.com \ 
  --source-arn <SNSTopicARN> 

Testing and Monitoring 

Test with SES Events: 

Send test emails to trigger Bounce, Complaint, or Delivery events. 

Verify CloudWatch Logs: 

Navigate to the CloudWatch Logs console. 

Check the Log Group for entries corresponding to the processed SES events. 

Benefits of AWS SES Monitoring 

Centralized Logs: All SES events are logged in CloudWatch for easy access. 

Custom Filtering: Monitor specific event types to improve email deliverability. 

Scalability: Lambda automatically scales with incoming SES events. 

Cost-Efficient: Pay only for the resources you use. 

Conclusion 

This article helps how to directly use AWS Lambda to monitor SES events efficiently, without relying on a CloudFormation template. By integrating Lambda with SNS and CloudWatch Logs, you can ensure reliable email delivery monitoring, enhancing both system reliability and customer satisfaction.