AWSの利用料金をPythonからSlackに投稿する #Python

Pythonのお勉強のため、AWSの利用料金をSlackに投稿するようにしたメモ。

参考リンク

以下の先人の記事を参考に作成。

LambdaでAWSの料金を毎日Slackに通知する（Python3）

動作イメージ

Slackの準備

何かと使えそうだし個人用Slackを作ることにする。

個人 Slack のススメ

Webhook URLの取得

以下を参考にWebhook URLを取得する。

SlackのWebhook URL取得手順

curlでテスト

curlで投稿をテストする。

CHANNEL="#sandbox"
WEBHOOK_URL="https://hooks.slack.com/services/hogehoge"
MESSAGE="テストメッセージ"
curl -X POST ${WEBHOOK_URL} \
  --data-urlencode "payload=
  {\"channel\": \"${CHANNEL}\",
   \"text\": \"${MESSAGE}\"}"

以下のように投稿されればOK。

AWS CLI

Pythonの前にAWS CLIで料金を取得してみる。

AWS CLIをセットアップする。

pip3 install awscli
aws configure

料金を確認する。

コマンド

aws cloudwatch get-metric-statistics \
  --region us-east-1 \
  --namespace "AWS/Billing" \
  --metric-name "EstimatedCharges" \
  --start-time "2019-01-01T00:00:00Z" \
  --end-time "2019-01-02T00:00:00Z" \
  --period 86400 \
  --statistics Maximum \
  --dimensions Name=Currency,Value=USD

実行例

$ aws cloudwatch get-metric-statistics \
>   --region us-east-1 \
>   --namespace "AWS/Billing" \
>   --metric-name "EstimatedCharges" \
>   --start-time "2019-01-01T00:00:00Z" \
>   --end-time "2019-01-02T00:00:00Z" \
>   --period 86400 \
>   --statistics Maximum \
>   --dimensions Name=Currency,Value=USD
EstimatedCharges
DATAPOINTS  0.45    2019-01-01T00:00:00Z    None
$

Python + AWS Lambda

AWS Lambda上のPythonからSlackに投稿する。

IAM Roleの作成

IAMコンソールでRoleを作成し、AWSLambdaBasicExecutionRoleとCloudWatchReadOnlyAccessのポリシーをアタッチする。

作成したRoleのARNを確認しておく。

コード作成

作成するファイルは以下の3つ。→GitHub

.
├── lambda.json
├── lambda_function.py
└── requirements.txt

lambda.jsonはlambda-uploderが使用するJSON形式のLambda関数の定義ファイル。先ほど作成したRoleのARN、Slack Webhook URL、Slackチャンネル名を指定する。lambda-uploaderの使い方と、jsonの書き方は以下を参照。

lambda-uploader

lambda.json

{
  "name": "aws_cost_checker",
  "description": "My AWS cost checker.",
  "region": "ap-northeast-1",
  "runtime": "python3.7",
  "handler": "lambda_function.lambda_handler",
  "role": "arn:aws:iam::hogehoge",
  "timeout": 300,
  "memory": 128,
  "variables":
    {
      "SLACK_WEBHOOK_URL": "https://hooks.slack.com/services/hogehoge",
      "SLACK_CHANNEL": "#hogehoge"
    }
}

lambda_function.pyが関数本体。以下を参考に作成。Low-Level ClientではなくResourceクラスを使ってみる。

lambda_function.py

import datetime
import json
import os

import boto3
import pytz
import requests

SLACK_WEBHOOK_URL = os.environ['SLACK_WEBHOOK_URL']
SLACK_CHANNEL = os.environ['SLACK_CHANNEL']


def get_cost():
    """boto3を使って直近の1日の最大の予想請求額を取得する

    :return: response
    """

    now = datetime.datetime.now()

    # Low-Level Clientの場合
    #
    # client = boto3.client('cloudwatch', region_name='us-east-1')
    # response = client.get_metric_statistics(
    #     Namespace='AWS/Billing',
    #     MetricName='EstimatedCharges',
    #     Dimensions=[
    #         {
    #             'Name': 'Currency',
    #             'Value': 'USD'
    #         }
    #     ],
    #     StartTime = now - datetime.timedelta(days=1),
    #     EndTime = now,
    #     Period=86400,
    #     Statistics=['Maximum']
    # )

    cloudwatch = boto3.resource('cloudwatch', region_name='us-east-1')
    metric = cloudwatch.Metric('AWS/Billing', 'EstimatedCharges')
    response = metric.get_statistics(
        Dimensions=[
            {
                'Name': 'Currency',
                'Value': 'USD'
            },
        ],
        StartTime=now - datetime.timedelta(days=1),
        EndTime=now,
        Period=86400,
        Statistics=['Maximum']
    )

    return response


def build_message(response):
    """SlackにPOSTするメッセージボディを作成する

    :param response:
    :return: message
    """

    cost = response['Datapoints'][0]['Maximum']
    timestamp = (response['Datapoints'][0]['Timestamp'] + datetime.timedelta(days=1)).astimezone(
        pytz.timezone('Asia/Tokyo')).strftime(
        '%Y年%m月%d日%H時%M分')

    text = '{}までのAWSの料金'.format(timestamp)
    attachment_text = '${}'.format(cost)

    if float(cost) >= 100.0:
        color = 'danger'   # red
    elif float(cost) >= 10.0:
        color = 'warning'  # yellow
    else:
        color = 'good'     # green

    atachements = {'text': attachment_text, 'color': color}

    message = {
        'text': text,
        'channel': SLACK_CHANNEL,
        'attachments': [atachements],
    }

    return message


def post_message(message):
    """SlackにPOSTする

    :param message:
    :return:
    """

    response = requests.post(SLACK_WEBHOOK_URL, data=json.dumps(message))
    response.raise_for_status()


def lambda_handler(event, context):

    response = get_cost()

    message = build_message(response)

    post_message(message)

requirements.txtは使用するモジュールの定義。

requirements.txt

boto3
pytz
requests

アップロード

lambda-uploaderをインストールする。

pip3 install lambda-uploader

アップロードする。

$ lambda-uploader
Î» Building Package
Î» Uploading Package
Î» Fin
$

スケジュール実行

lambda-uploaderによってLambda関数が作成されるので、LambdaコンソールからトリガーとしてCloudWatch Eventsを追加する。

スケジュールを設定する。とりあえずcron(* * * * ? *)毎分実行するようにする。

Slackに投稿されることを確認する。

上手く投稿できなかったら、CloudWatch Logsでログを確認する。上手く投稿が確認できたら、スケジュールをcron(0 15 * * ? *)（日本時間の0時）に変更する。