aws_cost_notification_lambda_function.py

#!/usr/bin/env python
# vim:fileencoding=utf-8

import os
import json
import requests
import datetime
import time
import boto3
import calendar
from dateutil.relativedelta import relativedelta


TIMEOUT = 5
OPENEXCHANGERATES_API_URL = "https://openexchangerates.org/api/latest.json"
SLACK_COLOR   = "#008000"
SLACK_PRETEXT = "AWS cost notification"
SLACK_MESSAGE = """
RATE(USD/JPY): {jpy_rate:.2f}
USD Costs: Total: {total_cost:,.2f} Forecast: {forecast_cost:,.2f}
JPY Costs: Total: {total_cost_jpy:,.2f} Forecast: {forecast_cost_jpy:,.2f}
"""

def get_account_id():

    return boto3.client("sts").get_caller_identity().get("Account")

# 対象年月の初日にオブジェクトを変換する
def get_first_date(dt):
    return dt.replace(day=1)

# 1USDの為替レートを取得
def get_jpy_rate():

    payload = { "app_id": os.environ["OPENEXCHANGERATES_APP_ID"] }

    res = requests.get(OPENEXCHANGERATES_API_URL, params=payload, timeout=TIMEOUT)
    if res.status_code != 200:
        res.raise_for_status()

    json_data = res.json()
    return float(json_data["rates"]["JPY"])


# 月末日取得
def get_monthly_last_date(dt):

    _, lastday = calendar.monthrange(dt.year, dt.month)
    return lastday

# 明日
def get_tomorrow(dt):
    return dt + datetime.timedelta(days=1)

# 翌月1日
def get_next_month_first_date(dt):
    return dt + relativedelta(months=+1, day=1)

# 期間内の総額取得
def get_total_costs(ce, **kwargs):

    res = ce.get_cost_and_usage(
            TimePeriod={"Start": kwargs["start"], "End": kwargs["end"]},
            Granularity='MONTHLY',
            Metrics=["UnblendedCost"]
          )
    total_cost     = float(res["ResultsByTime"][0]["Total"]["UnblendedCost"]["Amount"])
    total_cost_jpy = total_cost * kwargs["jpy_rate"]
    return { "total_cost": total_cost, "total_cost_jpy": total_cost_jpy }


# 当月の予測額取得
def get_costs_forecast(ce, **kwargs):

    res = ce.get_cost_forecast(
            TimePeriod={"Start": kwargs["start"], "End": kwargs["end"]},
            Granularity='MONTHLY',
            Metric="UNBLENDED_COST"
        )
    forecast_cost     = float(res["Total"]["Amount"])
    forecast_cost_jpy = forecast_cost * kwargs["jpy_rate"]
    return { "forecast_cost": forecast_cost, "forecast_cost_jpy": forecast_cost_jpy }


def send_to_slack(costs, **kwargs):

    headers = { "Content-Type": "application/json; charset=UTF-8" }

    title   = "AccountID:{0} From:{1} To:{2}".format(kwargs["account_id"], kwargs["start"], kwargs["end"])
    value   = SLACK_MESSAGE.format(**costs)

    payload = {
        "unfurl_links": True,
        "attachments": [
                { 
                    "fallback": SLACK_PRETEXT,
                    "pretext": SLACK_PRETEXT,
                    "color":  SLACK_COLOR,
                    "fields": [
                            {
                            "title": title,
                            "value": value,
                            "short": False
                        }
                    ]
                }
            ]
        }

    res = requests.post(os.environ["SLACK_WEBHOOK_URL"], data=json.dumps(payload), headers=headers, timeout=TIMEOUT)
    if res.status_code == 200:
        return True
    else:
        res.raise_for_status()


def lambda_handler(event, context):

    dt = datetime.datetime.now()

    start                  = get_first_date(dt).strftime("%Y-%m-%d")
    end                    = dt.strftime("%Y-%m-%d")
    tomorrow               = get_tomorrow(dt).strftime("%Y-%m-%d")
    next_month_first_date  = get_next_month_first_date(dt).strftime("%Y-%m-%d")

    ce          = boto3.client("ce", region_name="us-east-1")
    jpy_rate    = get_jpy_rate()
    account_id  = get_account_id()

    total_costs    = get_total_costs(ce, start=start, end=end, jpy_rate=jpy_rate)
    forecast_costs = get_costs_forecast(ce, start=tomorrow, end=next_month_first_date, jpy_rate=jpy_rate)
    costs = dict(total_costs, **forecast_costs)
    costs["jpy_rate"] = jpy_rate

    send_to_slack(costs, start=start, end=end, account_id=account_id)

    return {
        'statusCode': 200,
        'body': costs
    }

if __name__ == '__main__':
    if os.path.exists("event.json"):
        event = json.load(open("event.json", "r"))
    else:
        event = {}
    context = {}
    res = lambda_handler(event, context)
    print(res)