Post AWS Cost to Slack / AWS の費用を Slack に投稿する

post-aws-cost-to-slack.py

from collections import defaultdict
import datetime
import json
import urllib.parse
import os

import boto3
import requests

ce = boto3.client('ce')

DAYS = 14
TOP_N = 5  # これ以下は省略する
TOP_N_DAYS = 3  # 直近 n 日で top-n を探す
SLACK_WEBHOOK_URL = os.environ['SLACK_WEBHOOK_URL']


def lambda_handler(event, context):
    start = datetime.date.today() - datetime.timedelta(days=DAYS)
    end = datetime.date.today()

    res = ce.get_cost_and_usage(
        TimePeriod={'Start': str(start), 'End': str(end)},
        Granularity='DAILY',
        Metrics=['UnblendedCost'],
        Filter={'Not': {'Dimensions': {'Key': 'SERVICE', 'Values': ['Tax']}}},
        GroupBy=[{'Type': 'DIMENSION', 'Key': 'SERVICE'}]
    )

    labels = [str(start + datetime.timedelta(days=d))[-5:]
              for d in range(len(res['ResultsByTime']))]

    data = defaultdict(list)
    for metrics in res['ResultsByTime']:
        for m in metrics['Groups']:
            amount = float(m['Metrics']['UnblendedCost']['Amount'])
            data[m['Keys'][0]].append(amount)

    top = [kv[0] for kv in sorted(data.items(), key=lambda d: sum(d[1][-TOP_N_DAYS:]))[-TOP_N:]]
    agg = defaultdict(lambda: 0)
    for k, metrics in data.items():
        if k not in top:
            for d, m in enumerate(metrics):
                agg[d] += m

    data_agg = {k: v for k, v in data.items() if k in top}
    data_agg['others'] = [kv[1] for kv in sorted(agg.items())]

    keys = ['others'] + top
    datasets = [{'label': key, 'data': data_agg[key]} for key in keys]

    config = {
        'type': 'bar',
        'data': {'labels': labels, 'datasets': datasets},
        'options': {
            'scales': {
                'xAxes': [{'stacked': True}],
                'yAxes': [{'stacked': True}]
            }
        }
    }
    url = 'https://quickchart.io/chart?' + \
        urllib.parse.urlencode({'bkg': 'white', 'c': json.dumps(config)})

    res = requests.post(SLACK_WEBHOOK_URL, json={
        'text': f'AWS cost, {start} to {end}',
        'attachments': [{'image_url': url}]
    })
    res.raise_for_status()