kawabatas技術ブログ

試してみたことを書いていきます

AWS/GCPのコストをGoogleデータポータルで可視化する

AWS/GCP の各プロジェクト/各サービスにいくらかかっているのか把握したく、システムを構築しました。

システムの概要は下記になります。

  • AWS/GCP の支払いレポートを BigQuery に集約する
  • Google データポータルで BigQuery のデータを可視化する
  • データポータルのメール配信機能で、毎週 Slack へ通知する

f:id:kawabatas:20201030154845p:plain

できるだけシンプルに、ドキュメントに沿った形で実現することを心がけました。

1つ Lambda 関数を作成しましたが、それ以外にコードは書いていません。

GCP 支払いレポート -> BigQuery

cloud.google.com

ドキュメント通りに設定するだけです。

BigQuery Data Transfer Service を使って、BigQuery に書き出してくれます。

AWS 支払いレポート -> BigQuery

docs.aws.amazon.com

まずはドキュメント通りにコストと使用状況レポートを作成します。

これで S3 に支払いレポートの CSV ファイルが作成されるようになります。図中の①です。

cloud.google.com

この CSV をそのまま BigQuery Data Transfer Service で BigQuery へ転送したかったのですが、

  1. BigQuery が CSV のカラム名の形式（lineItem/UsageStartDate スラッシュがダメ）に対応していない。さらに製品列は動的である
  2. CSV には 1ヶ月分のデータが入っており、BigQuery に重複して登録されてしまう

という問題があり、少し手を加える必要がありました。

  1. に関しては、BigQuery のテーブルを必要なスキーマをピックアップし、事前に作成することにしました。（lineItem/UsageStartDate -> lineItem_UsageStartDate のようにカラム名を変更し、スキーマを定義しました。パーティションも lineItem_UsageStartDate を設定しました）
[
  {
    "name": "lineItem_UsageStartDate",
    "type": "TIMESTAMP"
  },
  {
    "name": "lineItem_UsageEndDate",
    "type": "TIMESTAMP"
  },
  {
    "name": "lineItem_UnblendedCost",
    "type": "FLOAT"
  },
  {
    "name": "lineItem_ProductCode",
    "type": "STRING"
  },
  {
    "name": "resourceTags_userProject",
    "type": "STRING"
  },
  {
    "name": "resourceTags_userEnvironment",
    "type": "STRING"
  },
  {
    "name": "lineItem_CurrencyCode",
    "type": "STRING"
  },
  {
    "name": "lineItem_UsageType",
    "type": "STRING"
  },
  {
    "name": "lineItem_UsageAmount",
    "type": "FLOAT"
  },
  {
    "name": "lineItem_UnblendedRate",
    "type": "FLOAT"
  },
  {
    "name": "lineItem_LineItemDescription",
    "type": "STRING"
  }
]
  1. に関しては、図中にあるように S3 へのファイル更新をトリガーに、Lambda を実行し、特定の日のデータだけを抽出した CSV を YYYYMMDD.csv.gz として書き出すようにしました。

そして、図中② BigQuery Data Transfer Service で YYYYMMDD.csv.gz を BigQuery へ転送しました。（s3://バケット名/to-bigquery/{run_time-24h|"%F"}.csv.gz のように転送するファイル名を指定できます）

Lambda のコード

package main

import (
    "bytes"
    "compress/gzip"
    "context"
    "encoding/csv"
    "io"
    "io/ioutil"
    "log"
    "os"
    "time"

    "github.com/aws/aws-lambda-go/events"
    "github.com/aws/aws-lambda-go/lambda"
    "github.com/aws/aws-sdk-go/aws"
    "github.com/aws/aws-sdk-go/aws/session"
    "github.com/aws/aws-sdk-go/service/s3"
    "github.com/aws/aws-sdk-go/service/s3/s3manager"
    "github.com/jszwec/csvutil"
)

type Report struct {
    UsageStartDate      string `csv:"lineItem/UsageStartDate"`
    UsageEndDate        string `csv:"lineItem/UsageEndDate"`
    Cost                string `csv:"lineItem/UnblendedCost"`
    Service             string `csv:"lineItem/ProductCode"`
    Project             string `csv:"resourceTags/user:Project"`
    Env                 string `csv:"resourceTags/user:Environment"`
    CurrencyCode        string `csv:"lineItem/CurrencyCode"`
    UsageType           string `csv:"lineItem/UsageType"`
    UsageAmount         string `csv:"lineItem/UsageAmount"`
    UnblendedRate       string `csv:"lineItem/UnblendedRate"`
    LineItemDescription string `csv:"lineItem/LineItemDescription"`
}

func createSession() *session.Session {
    return session.Must(session.NewSession())
}

func s3Upload(ctx context.Context, buf bytes.Buffer, bucket, key string) (*s3manager.UploadOutput, error) {
    sess := createSession()

    uploader := s3manager.NewUploader(sess)
    res, err := uploader.UploadWithContext(ctx, &s3manager.UploadInput{
        Bucket: aws.String(bucket),
        Key:    aws.String(key),
        Body:   bytes.NewReader(buf.Bytes()),
    })
    if err != nil {
        return nil, err
    }
    return res, nil
}

func makeCsvGzip(w io.Writer, reports []Report) error {
    zw, err := gzip.NewWriterLevel(w, gzip.BestCompression)
    if err != nil {
        return err
    }
    defer zw.Close()

    cw := csv.NewWriter(zw)
    enc := csvutil.NewEncoder(cw)
    for _, report := range reports {
        if err := enc.Encode(report); err != nil {
            return err
        }
    }
    cw.Flush()
    return nil
}

func extractRecords(file *os.File, at time.Time) ([]Report, error) {
    gr, err := gzip.NewReader(file)
    if err != nil {
        return nil, err
    }
    defer gr.Close()

    cr := csv.NewReader(gr)
    dec, err := csvutil.NewDecoder(cr)
    if err != nil {
        return nil, err
    }

    validReports := []Report{}
    for {
        report := Report{}
        if err := dec.Decode(&report); err == io.EOF {
            break
        } else if err != nil {
            return nil, err
        }

        // lineItem/UsageStartDate カラム。例"2020-10-01T00:00:00Z"
        startTimeStr := report.UsageStartDate
        startTime, err := time.Parse(time.RFC3339, startTimeStr)
        if err != nil {
            return nil, err
        }

        // YYMMDDが等しいものを抽出
        if at.YearDay() == startTime.YearDay() && at.Year() == startTime.Year() {
            validReports = append(validReports, report)
        }
    }
    return validReports, nil
}

func s3Download(ctx context.Context, bucket, key string) (f *os.File, err error) {
    sess := createSession()

    tmpFile, err := ioutil.TempFile("", "tmp_")
    if err != nil {
        return nil, err
    }
    defer os.Remove(tmpFile.Name())

    downloader := s3manager.NewDownloader(sess)
    if _, err = downloader.DownloadWithContext(ctx, tmpFile, &s3.GetObjectInput{
        Bucket: aws.String(bucket),
        Key:    aws.String(key),
    }); err != nil {
        return nil, err
    }
    return tmpFile, nil
}

func HandleRequest(ctx context.Context, req events.S3Event) error {
    bucketName := req.Records[0].S3.Bucket.Name
    key := req.Records[0].S3.Object.Key
    // cost & usage report は UTC で書き出されている
    yesterday := time.Now().UTC().AddDate(0, 0, -1)
    today := time.Now().UTC()
    log.Printf("Object.Key: %v, Yesterday/Today: %v/%v\n", key, yesterday.Format("2006-01-02"), today.Format("2006-01-02"))

    // 2日分
    for _, day := range []time.Time{yesterday, today} {
        // s3から更新ファイル(gzip)をダウンロード
        file, err := s3Download(ctx, bucketName, key)
        if err != nil {
            log.Println("Error s3 download")
            return err
        }

        // csv.gzから対象レコードを抽出
        reports, err := extractRecords(file, day)
        if err != nil {
            log.Printf("Error extract on %v\n", day.Format("2006-01-02"))
            return err
        }
        if len(reports) < 1 {
            log.Printf("None billing data on %v\n", day.Format("2006-01-02"))
            return nil
        }

        // csv.gzを作成
        var buf bytes.Buffer
        if err = makeCsvGzip(&buf, reports); err != nil {
            log.Printf("Error make gzip on %v\n", day.Format("2006-01-02"))
            return err
        }

        // s3へupload
        upkey := "to-bigquery/" + day.Format("2006-01-02") + ".csv.gz"
        res, err := s3Upload(ctx, buf, bucketName, upkey)
        if err != nil {
            log.Printf("Error s3 upload on %v\n", day.Format("2006-01-02"))
            return err
        }
        log.Printf("Complete! %v\n", res)
    }
    return nil
}

func main() {
    lambda.Start(HandleRequest)
}

Google データポータルで可視化

cloud.google.com

データポータルでは、容易に BigQuery のデータを可視化できます。

見たいグラフに応じて Group By してカスタムクエリを作成します。

GCP 日別クエリ例

SELECT
  DATE(usage_start_time) AS date,
  project.name AS project_name,
  service.description AS service_name,
  (SUM(CAST(cost * 1000000 AS int64))
    + SUM(IFNULL((SELECT SUM(CAST(c.amount * 1000000 as int64))
                  FROM UNNEST(credits) c), 0))) / 1000000
    AS cost
FROM `テーブル名`
WHERE
  _PARTITIONTIME BETWEEN PARSE_TIMESTAMP('%Y%m%d', @DS_START_DATE) AND TIMESTAMP_ADD(PARSE_TIMESTAMP('%Y%m%d', @DS_END_DATE), INTERVAL 1 DAY)
AND
  DATE(usage_start_time) >= PARSE_DATE('%Y%m%d', @DS_START_DATE)
AND
  DATE(usage_start_time) <= PARSE_DATE('%Y%m%d', @DS_END_DATE)
GROUP BY 1, 2, 3
ORDER BY 1 ASC, 2 ASC, 3 ASC

AWS 日別クエリ例

SELECT
  DATE(lineItem_UsageStartDate) AS date,
  resourceTags_userProject AS project_name,
  lineItem_ProductCode AS service_name,
  resourceTags_userEnvironment AS env_name,
  SUM(lineItem_UnblendedCost) AS cost
FROM `テーブル名`
WHERE
  DATE(lineItem_UsageStartDate) >= PARSE_DATE('%Y%m%d', @DS_START_DATE)
AND
  DATE(lineItem_UsageStartDate) <= PARSE_DATE('%Y%m%d', @DS_END_DATE)
GROUP BY 1, 2, 3, 4
ORDER BY 1 ASC, 2 ASC, 3 ASC, 4 ASC

Google データポータルでメールをスケジュール配信

f:id:kawabatas:20201030163326p:plain

「共有」の項目にメール配信機能もあります。

そのメールを Slack へ転送するようにすると、データポータルを開かずとも、見たいグラフを見れるようにできました。

f:id:kawabatas:20201030163655p:plain