twitter-image

Introduction

Amazon provides the ability to create Billing Alarms that can be used to alert whenever your AWS bill exceeds a certain threshold. However, this approach has a few shortcomings:

  • You need to set a predefined threshold. When you are first starting to use AWS, it’s hard to know what your AWS bill will look like. A lot of people set this threshold pretty low to be safe.

  • An alert like this tends to be reactive instead of proactive. The alarm gets triggered once the threshold has already crossed. For example, I had forgotten to turn off an EC2 instance I was no longer using but I only found out about a week later once my billing threshold crossed the limit.

Cost Explorer does provide an easy-to-use interface that can help keep you on top of your billing data. However, this requires one to use the tool regularly to ensure we don’t miss anything.

In this article, we will look at how to build a simple pipeline to send us billing reports over email. The generated report will look like this:

billing-alert-email

The tools we will use are:

  • AWS Lambda
  • Simple Email Service
  • Cost Explorer API

Getting Started

The source code for the project is available at this repository.

Project Setup

Create a new directory for the application

mkdir serverless-cost-alerts
cd serverless-cost-alerts

Setup a virtualenv for the application

virtualenv env
source env/bin/activate

Create requirements.txt with all the dependencies

boto3

Install the dependencies

Run the following command in the shell to install all the dependencies.

pip install -r requirements.txt

Create scaffolding

We will create a new folder called app which will store the application logic for the Cost Explorer API as well as using SES.

mkdir app
touch app/__init__.py app/cost_explorer.py app/email.py

We will also create a file called handler.py in the root directory. This file will contain the logic used to generate billing reports.

touch handler.py

Cost Explorer API

We will be using the Cost And Usage endpoint from Cost Explorer to get the billing data we need.

Replace the contents of the cost_explorer.py file we created earlier with this:


import boto3
from datetime import date, datetime, timedelta
from dateutil.relativedelta import relativedelta


class CostExplorer:
    TODAY_DATE = datetime.utcnow().date()
    CUR_MONTH_DATE = TODAY_DATE.replace(day=1)
    PREV_MONTH_DATE: date = CUR_MONTH_DATE - relativedelta(months=+1)

    def __init__(self):
        self.client = boto3.client("ce")
        self.metrics = ["UNBLENDED_COST"]
        self.currency:str = "USD"

        self.daily_report_kwargs = {
            "TimePeriod": self._get_timeperiod(
                start=self.TODAY_DATE - timedelta(days=2), # start_dt is inclusive
                end=self.TODAY_DATE, # end_dt is exclusive
            ),
            "Metrics": self.metrics,
            "Granularity": "DAILY"
        }

        self.monthly_report_kwargs = {
            "TimePeriod": self._get_timeperiod(
                start=self.PREV_MONTH_DATE, # start_dt is inclusive
                end=self.TODAY_DATE, # end_dt is exclusive
            ),
            "Metrics": self.metrics,
            "Granularity": "MONTHLY"
        }

    def _get_timeperiod(self, start: date, end: date):
        return {
            "Start": start.isoformat(),
            "End": end.isoformat(),
        }

    def _get_data(self, results):
        """
        Retrieves the individual billing rows from cost explorer data.
        """
        rows = []
        for v in results:
            row = {"date":v["TimePeriod"]["Start"]}
            for i in v["Groups"]:
                key = i["Keys"][0]
                row.update({key:float(i["Metrics"]["UnblendedCost"]["Amount"])})
            row.update({"Total":float(v["Total"]["UnblendedCost"]["Amount"])})
            rows.append(row)

        return [f"{row['date']}: {round(row['Total'], 2)}" for row in rows]

    def generate_report(self, report_kwargs):
        """
        Get cost data based on the granularity, start date and end date.
        """
        response = self.client.get_cost_and_usage(**report_kwargs)
        return "\n".join(self._get_data(response["ResultsByTime"]))

Key Points

  • We want to generate two reports, one for the last 2 days (daily granularity) and one to compare the spend last month vs the current month (monthly granularity).

  • We used Unblended costs to represent our billing data. More information about this can be found here.

Simple Email Service (SES)

We will be using SES to send the billing reports to ourselves via email.

Prerequisites

SES requires a verified email address before it can be used. Verification can be done via the AWS console.

Using SES

Replace the contents of email.py with the code snippet below:


import boto3


class EmailClient:
    SENDER = "AWS Cost Alert <[email protected]>"
    SUBJECT = "Daily AWS Billing Report"
    # The email body for recipients with non-HTML email clients.
    BODY_TEXT = """AWS Billing Alerts \r\n
    Daily billing report\r\n
    {daily_billing_report}\r\n
    Monthly billing report\r\n
    {monthly_billing_report}\r\n
    """

    # The HTML body of the email.
    BODY_HTML = """<html>
    <head></head>
    <body>
    <h1>AWS Billing Alert</h1>
    <hr/>
    <h3>Daily Billing</h3>
    <p>{daily_billing_report}</p>
    <hr/>
    <h3>Monthly Billing</h3>
    <p>{monthly_billing_report}</p>
    </body>
    </html>
    """
    # The character encoding for the email.
    CHARSET = "UTF-8"

    # recipient email address
    RECIPIENT = "[email protected]"

    def __init__(self):
        self.client = boto3.client("ses")

    def send(self, daily_billing_report, monthly_billing_report):
        """Send an email which contains AWS billing data"""
        email_text = self.BODY_TEXT.format(
            daily_billing_report=daily_billing_report,
            monthly_billing_report=monthly_billing_report
        )

        email_html = self.BODY_HTML.format(
            daily_billing_report=daily_billing_report,
            monthly_billing_report=monthly_billing_report
        )
        response = self.client.send_email(
            Destination={
                "ToAddresses": [
                    self.RECIPIENT,
                ],
            },
            Message={
                "Body": {
                    "Html": {
                        "Charset": self.CHARSET,
                        "Data": email_html
                    },
                    "Text": {
                        "Charset": self.CHARSET,
                        "Data": email_text,
                    },
                },
                "Subject": {
                    "Charset": self.CHARSET,
                    "Data": self.SUBJECT,
                },
            },
            Source=self.SENDER,
        )

Key points:

  • Replace SENDER and RECEPIENT with the email address you verified with SES.

Generating a billing report

We will now generate a report using Cost Explorer and SES. Replace the contents of handler.py with the code snippet below:

from app.cost_explorer import CostExplorer
from app.email import EmailClient


def main():
    ce = CostExplorer()
    daily_report = ce.generate_report(ce.daily_report_kwargs)
    monthly_report = ce.generate_report(ce.monthly_report_kwargs)

    email_client = EmailClient()
    email_client.send(
        daily_billing_report=daily_report,
        monthly_billing_report=monthly_report
    )


if __name__ == "__main__":
    main()

Next, run the script from your terminal using the following command:

python handler.py

If your AWS profile is setup with the appropriate permissions, you should receive an email with your billing data. This what the email should look like:

billing-alert-email

Integrating AWS Lambda

Now that we have generated the billing report locally, we will make use of the Serverless Framework to create a serverless pipeline using AWS Lambda.

Prerequisites

  1. Setup serverless framework by following the instructions listed here.

  2. Run the following commands after installing the serverless framework.

    npm init -f
npm install --save-dev serverless-python-requirements

The serverless-python-requirements plugin automatically bundles dependencies from requirements.txt and makes them available to your Lambda function.

Setting up the serverless project

Next, create the file serverless.yml in your root directory and copy the following content:


service: serverless-cost-alerts

plugins:
  - serverless-python-requirements

package:
  excludeDevDependencies: true
  exclude:
    - node_modules/**

custom:
  pythonRequirements:
    slim: true
    strip: false
    slimPatternsAppendDefaults: true
    slimPatterns:
      - "**/*.egg-info*"
      - "**/*.dist-info*"
    dockerizePip: true

provider:
  name: aws
  runtime: python3.7
  stage: dev
  region: us-west-2
  iamRoleStatements:
    - Effect: Allow
      Action:
        - ses:SendEmail
      Resource:
        - "*"
    - Effect: Allow
      Action:
        - ce:GetCostAndUsage
      Resource:
        - "*"

functions:
  send_daily_cost_report:
    handler: handler.generate_report
    events:
      - http: GET hello

Key points

  • The lambda function will have the permission to query the CostAndUsage function in Cost Explorer and SendEmail in SES.
  • To test our Lambda function, we will create a temporary HTTP endpoint called hello.

Lambda Handler

Before we deploy the Lambda function to AWS, we need to update the Lambda function so that it can respond to an HTTP request. Update the contents of your handler.py with the following code snippet:

from app.cost_explorer import CostExplorer
from app.email import EmailClient


def generate_report(event, context):
    ce = CostExplorer()
    daily_report = ce.generate_report(ce.daily_report_kwargs)
    monthly_report = ce.generate_report(ce.monthly_report_kwargs)

    email_client = EmailClient()
    email_client.send(
        daily_billing_report=daily_report,
        monthly_billing_report=monthly_report
    )

    return {
        "statusCode": 200,
        "headers": {
            "Content-Type": "application/json",
        },
        "body": "success",
    }

Now we will deploy the Lambda function to AWS. Run the following command in your terminal:

sls deploy

If the deploy is successful, you should be able to see the HTTP endpoint for your Lambda like this:

sls-deploy

If you click on the HTTP endpoint, you should see output like this:

lambda-http

Scheduled Lambda

Lastly, we will update our Lambda function to run as a scheduled cron instead of an HTTP endpoint. We will setup our Lambda function to run once a day.

Update the functions section of serverless.yml:

functions:
  send_daily_cost_report:
    # handler: handler.generate_report
    # events:
    #   - http: GET hello
    handler: handler.generate_report
    events:
      - schedule: cron(0 9 * * ? *)

We have now scheduled our Lambda function to run every day at 9 am.

Conclusion

I have been using this setup for the last few weeks and it has been useful in keeping on top of my AWS billing without having to worry about checking the console frequently. The cost explorer API has a lot of different reporting options and you can generate a report that works for you.