Using AWS Lambda Function to Identify Unused AWS Resources

In this blog post, we’ll walk through the process of creating an AWS Lambda function that identifies unused AWS resources, specifically focusing on EBS volume snapshots older than one year. We’ll use various AWS services and tools, including Python with the Boto3 module, AWS Lambda, EventBridge, SAM (Serverless Application Model), and SNS (Simple Notification Service).

Tech Stack

Python: Code to fetch unused resources using the Boto3 module.
Lambda: To execute the function.
EventBridge: To trigger the Lambda function.
SAM: To deploy Lambda and EventBridge.
SNS: To notify the user.

Steps

Create Lambda using SAM
Write Python Code
Build and Test Lambda
Create EventBridge Rule
Create SNS to Notify

Creating Lambda using SAM

Connect to AWS account from CLI.
Run sam init with the options as shown:

This will create a folder with the name Resource-Check-Application, with these files.
Rename the hello_world folder to snapshot_age.
Replace app.py with the Python code below to fetch snapshots.
Invoke check_age() in the lambda_handler and print the list of snapshots it returns.

Writing the Python Code

Using the Boto3 module, we can fetch EBS volume snapshots that are older than one year. This is a generic use case, and you can modify the condition as needed.

Import the boto3 and datetime modules.
Create a client object for the Amazon EC2 (Elastic Compute Cloud) service using the Boto3 library.

Define the check_age() function:

Fetch all snapshots created by users in the AWS account, OwnerIds='self' ignores snapshot that AWS has internally created.
Loop through the snapshots and find those that are older than a year. Store these in a list called snapshot_date.
Add appropriate exceptions, here we just have InvalidSnapshot.
Refer below for complete code.

import json
import boto3
from datetime import datetime, timezone, timedelta
#from dateutil import parser
# import requests


session = boto3.Session() #profile_name='Certification' used to test from cli, used for SSO user
ec2=session.client('ec2')
present = datetime.now()


def check_age():
    snapshot_date=[]
    try:
        snapshots = ec2.describe_snapshots(OwnerIds=[
            'self'
        ])
        one_year_ago = datetime.now() - timedelta(days=365)
        print(one_year_ago)
        for snap in snapshots['Snapshots']:
        # print(snap['SnapshotId'])
        #snapshot_date[snap['SnapshotId']]=snap['StartTime']
        # print(snap['StartTime'])
            if (snap['StartTime'].replace(tzinfo=None)<one_year_ago):
                snapshot_date.append(snap['SnapshotId'])
                #print(snap['SnapshotId'])
        return snapshot_date
    except ec2.exceptions.from_code('InvalidSnapshotIDNotFound'):
        return f"Snapshot not found: InvalidSnapshotIDNotFound"
    except Exception as e:
        return f"An error occurred: {e}"

def lambda_handler(event, context):
    snapshot_age=check_age()
    print(snapshot_age)

Configuring the Lambda Function and EventBridge Rule

Edit template.yaml:

Under resources section rename the function name to CheckSnapshotAgeFunction.
Update CodeUri to snapshot_age/.
Add IAM policy to describe snapshots, i.e. 'ec2:DescribeSnapshots'
Add ResourceCheckEventBridgeRule resource that specifies the above lambda function as it’s target and has a schedule that will trigger it.
Add LambdaInvokePermission that will give the Event Bridge rule permission to run the lambda function.

Complete project is available here.

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: >
  Resource-Check-Application

  Sample SAM Template for Resource-Check-Application

# More info about Globals: https://github.com/awslabs/serverless-application-model/blob/master/docs/globals.rst
Globals: 
  Function:
    Timeout: 45 #increase timeout 
    MemorySize: 128

    # You can add LoggingConfig parameters such as the Logformat, Log Group, and SystemLogLevel or ApplicationLogLevel. Learn more here https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/sam-resource-function.html#sam-function-loggingconfig.
    LoggingConfig:
      LogFormat: JSON
Resources:
  CheckSnapshotAgeFunction:
    Type: AWS::Serverless::Function # More info about Function Resource: https://github.com/awslabs/serverless-application-model/blob/master/versions/2016-10-31.md#awsserverlessfunction
    Properties:
      CodeUri: snapshot_age/
      FunctionName: Check_Snapshot_Age
      Handler: app.lambda_handler
      Runtime: python3.10
      Architectures:
      - x86_64
      Policies:
        - Version: '2012-10-17'
          Statement:
            - Effect: Allow
              Action:
                - 'ec2:DescribeSnapshots'
              Resource: '*'

  ResourceCheckEventBridgeRule:
    Type: 'AWS::Events::Rule'
    Properties:
      Name: MyEventBridgeRule
      Description: 'Rule to trigger Lambda function'
      State: 'ENABLED'
      Targets:
        - Arn: !GetAtt CheckSnapshotAgeFunction.Arn
          Id: 'MyLambdaFunctionTarget'
      ScheduleExpression : cron(0/1 * * * ? *)

  LambdaInvokePermission:
    Type: 'AWS::Lambda::Permission'
    Properties:
      FunctionName: !Ref CheckSnapshotAgeFunction
      Action: 'lambda:InvokeFunction'
      Principal: 'events.amazonaws.com'
      SourceArn: !GetAtt ResourceCheckEventBridgeRule.Arn

Building and Deploying the Lambda Function

Run sam build in the terminal.
Run sam deploy --guided --profile <profile-name>

This will show the resources that will be created in AWS account. Once verified enter “y”

The Lambda function will be triggered every minute. We can view the output in the Cloud Watch logs.

Go to cloud formation stack.
Find the Stack we deployed, go to the lambda function.
Under the monitor tab click on ViewCloudwatch logs, find the most recent log group.
Here we see that every minute the list of Snapshots older than a year are printed.

Notifying users via email

We can also configure SNS to send an email to the respective users instead of checking logs in CloudWatch.

Update the template.yaml file to send email notifications.

Create an Environment variable SNS_TOPIC_ARN for the SNS topic in the lambda function, this will be used in the code to fetch the SNS topic.
Add IAM policy to allow the CheckSnapshotAgeFunction to publish to SNS, i.e. sns:Publish
Create the SNS Topic and Subscription resources.

 Environment:
   Variables:
     SNS_TOPIC_ARN: !Ref SNSTopic 


 SNSTopic:
    Type: "AWS::SNS::Topic"
    Properties:
      TopicName: "POC-AWS-Cost-Optimization"
      DisplayName: "POC-AWS-Cost-Optimization"

  SNSSubscription:
    Type: "AWS::SNS::Subscription"
    Properties:
      TopicArn: !Ref SNSTopic
      Protocol: "email"
      Endpoint: "heloise@abc.com"

You also need to update the Lambda function code to publish the list of old snapshots to the SNS topic:

sns_client = boto3.client('sns')
sns_topic_arn = os.environ['SNS_TOPIC_ARN'] #fetch sns topic from template.yaml

def lambda_handler(event, context):
    snapshot_age = check_age()
     response = sns_client.publish(
    TopicArn=sns_topic_arn,
    Message=f"{snapshot_age}",
    Subject='POC - AWS Cost Optimization'
)
    print(snapshot_age)

To deploy run sam build and sam deploy like before.

Confirm the email subscription:

Once confirmed, we will receive an email every time the lambda function is triggered:

Enhancements

This functionality can be extended to monitor S3, EBS volumes and EC2 etc. For example, here I have created code to,

Fetch S3 buckets with full access and public access.
Unused EBS volumes.

Cleanup

Delete the stack from Cloud Formation once verified and tested.

Conclusion

By following these steps, you can automate the process of identifying and managing unused AWS resources, helping to optimize costs and improve resource management. Additionally, you can extend this solution to check for other resources like S3 buckets with public access, unused EBS volumes, and more.

To learn more about AWS cost optimization, this is a great article.

Happy learning! 🚀