Alert on High Data Egress Costs from Cloudfront

Hey everyone, Darian Vance here from TechResolve. Let’s talk about something that gives every Ops person a little bit of anxiety: surprise cloud bills. Specifically, data egress from CloudFront. I used to have a recurring calendar reminder to manually check our CloudFront usage, which involved downloading gzipped logs from S3. It was a tedious process that I dreaded. After one too many Mondays spent grepping through log files, I automated it. This simple script has saved me hours and, more importantly, caught a couple of traffic spikes before they turned into budget problems.

This guide will walk you through setting up a Python script to do the heavy lifting for you. It’ll parse your CloudFront logs, calculate the data egress, and fire off an alert if it crosses a threshold you define. It’s a “set it and forget it” solution that gives you peace of mind.

Prerequisites

Before we dive in, make sure you have the following ready:

• An AWS account with IAM permissions to create policies and roles.
• Python 3 installed on the machine where you’ll run the script.
• A CloudFront distribution with access logging enabled and configured to send logs to an S3 bucket.
• The name of your S3 log bucket.
• An SNS Topic ARN that you’ll use for sending notifications.

The Guide: Step-by-Step

Step 1: The IAM Policy

First things first, our script needs permission to act on our behalf. We’ll create an IAM policy that grants it the minimum required access: reading logs from S3 and publishing a message to SNS. In my production setups, I attach this policy to a specific IAM user or role that is solely used for this automation.

Here’s the JSON for the policy. You’ll need to replace `YOUR-LOG-BUCKET-NAME` and `YOUR-SNS-TOPIC-ARN` with your actual resource names.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "AllowS3LogAccess",
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:ListBucket"
            ],
            "Resource": [
                "arn:aws:s3:::YOUR-LOG-BUCKET-NAME",
                "arn:aws:s3:::YOUR-LOG-BUCKET-NAME/*"
            ]
        },
        {
            "Sid": "AllowSNSPublish",
            "Effect": "Allow",
            "Action": "sns:Publish",
            "Resource": "YOUR-SNS-TOPIC-ARN"
        }
    ]
}

Pro Tip: The `s3:ListBucket` permission is for the bucket ARN itself, while `s3:GetObject` is for the objects inside it (the `/*`). This is a common point of confusion that can lead to “Access Denied” errors.

Step 2: The Python Script

Alright, let’s get to the code. This script will connect to AWS, find the logs for the previous day, calculate the total bytes transferred, and then decide if it needs to send an alert.

I’ll skip the standard virtual environment setup since you likely have your own workflow for that. Just make sure you have the AWS SDK for Python, Boto3, installed. You can typically install it by running `python3 -m pip install boto3` in your terminal.

Here’s the full script. I’ve added comments to explain what each part does.

import boto3
import gzip
from datetime import datetime, timedelta, timezone

# --- Configuration ---
# Your S3 bucket where CloudFront logs are stored
S3_BUCKET_NAME = 'your-cloudfront-log-bucket-name'
# The prefix for your logs, if you have one. Often it's just 'logs/' or empty.
S3_PREFIX = ''
# Your SNS topic ARN for sending alerts
SNS_TOPIC_ARN = 'arn:aws:sns:us-east-1:123456789012:MyAlertingTopic'
# Set your egress threshold in Gigabytes (GB)
EGRESS_THRESHOLD_GB = 100

# --- AWS Clients ---
s3_client = boto3.client('s3')
sns_client = boto3.client('sns')

def get_log_files_for_yesterday(bucket, prefix):
    """Finds all CloudFront log files for the previous full day."""
    yesterday = datetime.now(timezone.utc) - timedelta(days=1)
    # CloudFront log file format includes the date like: EEXAMPLE.2023-10-27-14.gz
    date_prefix = yesterday.strftime('%Y-%m-%d')
    
    print(f"Searching for log files with date prefix: {date_prefix} in bucket {bucket}")
    
    log_files = []
    paginator = s3_client.get_paginator('list_objects_v2')
    pages = paginator.paginate(Bucket=bucket, Prefix=prefix)
    
    for page in pages:
        if 'Contents' in page:
            for obj in page['Contents']:
                if date_prefix in obj['Key'] and obj['Key'].endswith('.gz'):
                    log_files.append(obj['Key'])
                    
    print(f"Found {len(log_files)} log files for yesterday.")
    return log_files

def calculate_egress_from_logs(bucket, log_files):
    """Downloads, unzips, and parses log files to sum up the data transfer."""
    total_bytes_sent = 0
    
    for log_file_key in log_files:
        try:
            # Get the gzipped log file from S3
            response = s3_client.get_object(Bucket=bucket, Key=log_file_key)
            
            # Decompress the content in memory
            log_content = gzip.decompress(response['Body'].read()).decode('utf-8')
            
            for line in log_content.splitlines():
                # Skip comment lines
                if line.startswith('#'):
                    continue
                
                fields = line.split('\t')
                # The 'sc-bytes' (server-to-client bytes) field is our egress data.
                # In standard CloudFront logs, it's typically the 3rd column (index 2).
                # This can vary if you have custom log formats!
                if len(fields) > 2:
                    try:
                        bytes_sent = int(fields[2])
                        total_bytes_sent += bytes_sent
                    except (ValueError, IndexError):
                        # Ignore malformed lines
                        continue
        except Exception as e:
            print(f"Error processing file {log_file_key}: {e}")
            
    return total_bytes_sent

def send_sns_alert(total_gb, threshold_gb):
    """Sends a formatted alert message to the configured SNS topic."""
    subject = f"ALERT: CloudFront Egress High - {total_gb:.2f} GB"
    message = (
        f"CloudFront data egress for the last 24 hours has exceeded the threshold.\n\n"
        f"Total Egress: {total_gb:.2f} GB\n"
        f"Threshold: {threshold_gb} GB\n\n"
        f"Please investigate the CloudFront distribution logs for potential issues."
    )
    
    try:
        response = sns_client.publish(
            TopicArn=SNS_TOPIC_ARN,
            Message=message,
            Subject=subject
        )
        print(f"Successfully sent alert to SNS. Message ID: {response['MessageId']}")
    except Exception as e:
        print(f"Failed to send SNS alert: {e}")

def main():
    """Main function to run the egress check."""
    log_files = get_log_files_for_yesterday(S3_BUCKET_NAME, S3_PREFIX)
    
    if not log_files:
        print("No log files found for yesterday. Exiting.")
        return
        
    total_bytes = calculate_egress_from_logs(S3_BUCKET_NAME, log_files)
    
    # Convert bytes to gigabytes for comparison
    total_gb = total_bytes / (1024**3)
    
    print(f"Total calculated egress: {total_gb:.2f} GB")
    
    if total_gb > EGRESS_THRESHOLD_GB:
        print(f"Threshold exceeded! Sending alert.")
        send_sns_alert(total_gb, EGRESS_THRESHOLD_GB)
    else:
        print("Egress is within the normal threshold. No action needed.")

if __name__ == "__main__":
    main()

Step 3: Scheduling the Script

This script is most useful when it runs automatically. A simple way to do this is with a cron job. For example, to run the script every day at 2 AM, you would set up a cron job like this. Note that CloudFront can take several hours to deliver logs, so running it for the *previous* day is the most reliable approach.

Just add this line to your crontab, assuming your script is named `check_egress.py`.

`0 2 * * * python3 check_egress.py`

Pro Tip: For a more cloud-native solution, you can package this script as an AWS Lambda function and trigger it daily using an Amazon EventBridge (formerly CloudWatch Events) schedule. This removes the need for a dedicated server to run the cron job.

Common Pitfalls

Here are a few places where I’ve messed up in the past, so you can avoid them:

• IAM Permissions: Forgetting `s3:ListBucket` is my classic mistake. The script can’t find any objects if it can’t list them first. Always start with the IAM policy.
• Log Delivery Lag: CloudFront logs are not real-time. Trying to analyze data from the last hour is a recipe for inaccurate results. That’s why the script is designed to look at yesterday’s complete data set.
• Log Format Changes: The script assumes the standard CloudFront log format where `sc-bytes` is the third column. If you’ve customized your log fields, you’ll need to adjust the index (`fields[2]`) in the `calculate_egress_from_logs` function.
• Timezones: CloudFront logs use UTC. My script accounts for this by using `datetime.now(timezone.utc)`. If your scheduling server is in a different timezone, relying on local time can cause you to miss a day or analyze the wrong one.

Conclusion

And that’s it! You now have a robust, automated way to monitor your CloudFront data egress. This small piece of automation frees you from a tedious manual task and acts as an early warning system for your cloud costs. It’s a perfect example of how a little bit of DevOps scripting can provide significant value and peace of mind.

Happy building,
-Darian Vance

Darian Vance

Lead Cloud Architect & DevOps Strategist

With over 12 years in system architecture and automation, Darian specializes in simplifying complex cloud infrastructures. An advocate for open-source solutions, he founded TechResolve to provide engineers with actionable, battle-tested troubleshooting guides and robust software alternatives.

🤖 Frequently Asked Questions