Optimize AWS Costs by Managing Elastic IPs with Python and Boto3

The Need for Cost Optimization

As businesses increasingly rely on cloud infrastructure, managing costs becomes critical. AWS Elastic IPs (EIPs) are essential for maintaining static IP addresses, but they can also become a source of unwanted costs if not managed properly. AWS charges for EIPs that are not associated with running instances, which can quickly add up if left unchecked.

To address this, we can write a Python script using Boto3 to automate the management of EIPs, ensuring that we only pay for what we use. Let's walk through the process of creating this script step-by-step.

Step 1: Setting Up the Environment

First, we need to set up our environment. Install Boto3 if you haven't already:

pip install boto3

Make sure your AWS credentials are configured. You can set them up using the AWS CLI or by setting environment variables.

Step 2: Initializing Boto3 Client

We'll start by initializing the Boto3 EC2 client, which will allow us to interact with AWS EC2 services.

import boto3

# Initialize boto3 EC2 client
ec2_client = boto3.client('ec2')

With this code, we now have an ec2_client object that we can use to call various EC2-related functions.

Step 3: Retrieving All Elastic IPs

Next, we need a function to retrieve all Elastic IPs along with their associated instance ID and allocation ID.

def get_all_eips():
    """
    Retrieve all Elastic IPs along with their associated instance ID and allocation ID.
    """
    response = ec2_client.describe_addresses()
    eips = [(address['PublicIp'], address.get('InstanceId', 'None'), address['AllocationId']) for address in response['Addresses']]
    return eips

You can run this function to get a list of all EIPs in your AWS account:

print(get_all_eips())

The output will be a list of tuples, each containing the EIP, associated instance ID (or 'None' if unassociated), and allocation ID.

Step 4: Checking Instance States

We need another function to check the state of an instance. This helps us determine if an EIP is associated with a stopped instance.

def get_instance_state(instance_id):
    """
    Retrieve the state of an EC2 instance.
    """
    response = ec2_client.describe_instances(InstanceIds=[instance_id])
    state = response['Reservations'][0]['Instances'][0]['State']['Name']
    return state

You can test this function with an instance ID to see its state:

print(get_instance_state('i-1234567890abcdef0'))

This will output the state of the instance, such as 'running', 'stopped', etc.

Step 5: Categorizing Elastic IPs

Now, let's categorize the EIPs based on their association and instance states.

def categorize_eips():
    """
    Categorize Elastic IPs into various categories and provide cost-related insights.
    """
    eips = get_all_eips()
    eip_map = {}
    unassociated_eips = {}
    stopped_instance_eips = {}

    for eip, instance_id, allocation_id in eips:
        eip_map[eip] = allocation_id
        if instance_id == 'None':
            unassociated_eips[eip] = allocation_id
        else:
            instance_state = get_instance_state(instance_id)
            if instance_state == 'stopped':
                stopped_instance_eips[eip] = instance_id

    return {
        "all_eips": eip_map,
        "unassociated_eips": unassociated_eips,
        "stopped_instance_eips": stopped_instance_eips
    }

Run this function to get categorized EIPs:

categorized_eips = categorize_eips()
print(categorized_eips)

The output will be a dictionary with categorized EIPs.

Step 6: Printing the Results

To make our script user-friendly, we'll add functions to print the categorized EIPs and provide cost insights.

def print_eip_categories(eips):
    """
    Print categorized Elastic IPs and provide cost-related information.
    """
    print("All Elastic IPs:")
    if eips["all_eips"]:
        for eip in eips["all_eips"]:
            print(eip)
    else:
        print("None")

    print("\nUnassociated Elastic IPs:")
    if eips["unassociated_eips"]:
        for eip in eips["unassociated_eips"]:
            print(eip)
    else:
        print("None")

    print("\nElastic IPs associated with stopped instances:")
    if eips["stopped_instance_eips"]:
        for eip in eips["stopped_instance_eips"]:
            print(eip)
    else:
        print("None")

Test this function by passing the categorized_eips dictionary:

print_eip_categories(categorized_eips)

Step 7: Identifying Secondary EIPs

We should also check for instances that have multiple EIPs associated with them.

def find_secondary_eips():
    """
    Find secondary Elastic IPs (EIPs which are connected to instances already assigned to another EIP).
    """
    eips = get_all_eips()
    instance_eip_map = {}

    for eip, instance_id, allocation_id in eips:
        if instance_id != 'None':
            if instance_id in instance_eip_map:
                instance_eip_map[instance_id].append(eip)
            else:
                instance_eip_map[instance_id] = [eip]

    secondary_eips = {instance_id: eips for instance_id, eips in instance_eip_map.items() if len(eips) > 1}
    return secondary_eips

Run this function to find secondary EIPs:

secondary_eips = find_secondary_eips()
print(secondary_eips)

Step 8: Printing Secondary EIPs

Let's add a function to print the secondary EIPs.

def print_secondary_eips(secondary_eips):
    """
    Print secondary Elastic IPs.
    """
    print("\nInstances with multiple EIPs:")
    if secondary_eips:
        for instance_id, eips in secondary_eips.items():
            print(f"Instance ID: {instance_id}")
            for eip in eips:
                print(f"  - {eip}")
    else:
        print("None")

Test this function by passing the secondary_eips dictionary:

print_secondary_eips(secondary_eips)

Step 9: Providing Cost Insights

Finally, we add a function to provide cost insights based on our findings.

def print_cost_insights(eips):
    """
    Print cost insights for Elastic IPs.
    """
    unassociated_count = len(eips["unassociated_eips"])
    stopped_instance_count = len(eips["stopped_instance_eips"])
    total_eip_count = len(eips["all_eips"])

    print("\nCost Insights:")
    print(f"Total EIPs: {total_eip_count}")
    print(f"Unassociated EIPs (incurring cost): {unassociated_count}")
    print(f"EIPs associated with stopped instances (incurring cost): {stopped_instance_count}")
    print("Note: AWS charges for each hour that an EIP is not associated with a running instance.")

Run this function to get cost insights:

print_cost_insights(categorized_eips)

Full Code

Here's the complete script with all the functions we've discussed:

import boto3

# Initialize boto3 EC2 client
ec2_client = boto3.client('ec2')

def get_all_eips():
    """
    Retrieve all Elastic IPs along with their associated instance ID and allocation ID.
    """
    response = ec2_client.describe_addresses()
    eips = [(address['PublicIp'], address.get('InstanceId', 'None'), address['AllocationId']) for address in response['Addresses']]
    return eips

def get_instance_state(instance_id):
    """
    Retrieve the state of an EC2 instance.
    """
    response = ec2_client.describe_instances(InstanceIds=[instance_id])
    state = response['Reservations'][0]['Instances'][0]['State']['Name']
    return state

def categorize_eips():
    """
    Categorize Elastic IPs into various categories and provide cost-related insights.
    """
    eips = get_all_eips()
    eip_map = {}
    unassociated_eips = {}
    stopped_instance_eips = {}

    for eip, instance_id, allocation_id in eips:
        eip_map[eip] = allocation_id
        if instance_id == 'None':
            unassociated_eips[eip] = allocation_id
        else:
            instance_state = get_instance_state(instance_id)
            if instance_state == 'stopped':
                stopped_instance_eips[eip] = instance_id

    return {
        "all_eips": eip_map,
        "unassociated_eips": unassociated_eips,
        "stopped_instance_eips": stopped

_instance_eips
    }

def print_eip_categories(eips):
    """
    Print categorized Elastic IPs and provide cost-related information.
    """
    print("All Elastic IPs:")
    if eips["all_eips"]:
        for eip in eips["all_eips"]:
            print(eip)
    else:
        print("None")

    print("\nUnassociated Elastic IPs:")
    if eips["unassociated_eips"]:
        for eip in eips["unassociated_eips"]:
            print(eip)
    else:
        print("None")

    print("\nElastic IPs associated with stopped instances:")
    if eips["stopped_instance_eips"]:
        for eip in eips["stopped_instance_eips"]:
            print(eip)
    else:
        print("None")

def find_secondary_eips():
    """
    Find secondary Elastic IPs (EIPs which are connected to instances already assigned to another EIP).
    """
    eips = get_all_eips()
    instance_eip_map = {}

    for eip, instance_id, allocation_id in eips:
        if instance_id != 'None':
            if instance_id in instance_eip_map:
                instance_eip_map[instance_id].append(eip)
            else:
                instance_eip_map[instance_id] = [eip]

    secondary_eips = {instance_id: eips for instance_id, eips in instance_eip_map.items() if len(eips) > 1}
    return secondary_eips

def print_secondary_eips(secondary_eips):
    """
    Print secondary Elastic IPs.
    """
    print("\nInstances with multiple EIPs:")
    if secondary_eips:
        for instance_id, eips in secondary_eips.items():
            print(f"Instance ID: {instance_id}")
            for eip in eips:
                print(f"  - {eip}")
    else:
        print("None")

def print_cost_insights(eips):
    """
    Print cost insights for Elastic IPs.
    """
    unassociated_count = len(eips["unassociated_eips"])
    stopped_instance_count = len(eips["stopped_instance_eips"])
    total_eip_count = len(eips["all_eips"])

    print("\nCost Insights:")
    print(f"Total EIPs: {total_eip_count}")
    print(f"Unassociated EIPs (incurring cost): {unassociated_count}")
    print(f"EIPs associated with stopped instances (incurring cost): {stopped_instance_count}")
    print("Note: AWS charges for each hour that an EIP is not associated with a running instance.")

# Main execution
if __name__ == "__main__":
    categorized_eips = categorize_eips()
    print_eip_categories(categorized_eips)
    print_secondary_eips(find_secondary_eips())
    print_cost_insights(categorized_eips)

Pros and Cons

Pros:

Cost Savings: Identifies and helps eliminate unwanted costs associated with unused or mismanaged EIPs.
Automation: Automates the process of monitoring and categorizing EIPs.
Insights: Provides clear insights into EIP usage and cost-related information.
Easy to Use: Simple and straightforward script that can be run with minimal setup.

Cons:

AWS Limits: The script relies on AWS API calls, which may be subject to rate limits.
Manual Intervention: While it identifies cost-incurring EIPs, the script does not automatically release or reallocate them.
Resource Intensive: For large-scale environments with many EIPs and instances, the script may take longer to run and consume more resources.

How We Can Improve It

Automatic Remediation: Extend the script to automatically release unassociated EIPs or notify administrators for manual intervention.
Scheduled Execution: Use AWS Lambda or a cron job to run the script at regular intervals, ensuring continuous monitoring and cost management.
Enhanced Reporting: Integrate with a logging or monitoring service to provide detailed reports and historical data on EIP usage and costs.
Notification System: Implement a notification system using AWS SNS or similar services to alert administrators of cost-incurring EIPs in real time.

By incorporating these improvements, we can make the script even more robust and useful for managing AWS EIPs and reducing associated costs effectively.

Blog