DynamoDBのCapacityUnitsの実際の使用量を割合で出そうとしても、直接APIなどの呼び出しができないため、Lambdaを経由して計算で出力した備忘録。

問題意識

CloudWatch Metricsは、実際の使用量は出せても（何ユニット使用した）、設定されているCapacityUnitsからの割合（何％）で表示することはできない
実際に予定しているCapacityUnitsの80%を継続して超過し続けていたら何かアクションを起こす、という流れが作り出せない

前提

DynamoDBのCloudWatch Metricsで実際の使用量を算出するようにする
Lambdaにて実際の使用量（ConsumedReadCapacityUnits/ConsumedWriteCapacityUnits）を取得し、割合を計算する

成果物

CloudWatch Metricsの設定を確認する

とりあえずテーブルを作成すればConsumedRead/WriteCapacityUnitsは用意されるようなので、これらがきちんと機能しているかを、ツールからデータを定期的に入れてみて確認します。

一旦ProvisionedRead/WriteCapacityUnitsを3に変更しておきます。

データ投入ツールの用意

こんな感じのデータ投入ツールを用意。2行のCSVファイルを読み込みます。同じデータを投入し続けると面白くないので、起動引数でゆらぎ（Fluctuation[%]）をつけてデータ投入します。

行	コンテンツ
1	カラム名
2	ベースデータ

$ python data-insert.py <FileName> <Region> <Table> <Timing> <Fluctuation(%)>

sample-data.csv

id,score
"100",150

data-insert.py

import sys

import csv
import math
from decimal import Decimal
from time import sleep
from datetime import datetime
import pandas
import random

import boto3
from boto3.dynamodb.conditions import Key, Attr
from boto3.session import Session


args = sys.argv

if len(args) < 6:
    print("Usage: python data-insert.py <FileName> <Region> <Table> <Timing> <Fluctuation(%)>")
    sys.exit()

dynamodb = boto3.resource('dynamodb', region_name=args[2])
table = dynamodb.Table(args[3])

if __name__ == "__main__":
    print("Data insert start. For stopping, Ctrl + C press.")
    target = pandas.read_csv(args[1])
    timing = Decimal(str(args[4]))
    fluctuation = float(str(args[5]))

    print("Timing: " + str(timing) + "/sec, Fluctuation: " + str(fluctuation) + "%")
    print(target)

    sleep_time = Decimal(1) / timing

    while True:
      for row_index, row in target.iterrows():
          item_dict = {}
          for col in target:
              if col == "id":
                item_dict[col] = str(row[col])
              else:
                value = Decimal(str(row[col]))
                difference = random.uniform(-1 * fluctuation / 100, fluctuation / 100)
                item_dict[col] = value + round(value * Decimal(difference), 2)
          item_dict["timestamp"] = datetime.now().isoformat()
          print(item_dict)

          table.put_item(Item=item_dict)
          sleep(sleep_time)

これでConsumedCapacityWriteUnitsのテスト用スクリプトは作成できました。

データ読み込みツールの用意

ConsumedCapacityReadUnitsのテスト用にデータ読み込みのスクリプトを用意します。このページのものを利用します。

AWS LambdaでDynamoDBから取得した値の最新レコードを取得する

$ python data-read.py <Region> <Table> <id> <key> <Timing>

data-read.py

import sys
import json

from decimal import Decimal
from time import sleep

import boto3
from boto3.dynamodb.conditions import Key, Attr


args = sys.argv

if len(args) < 6:
    print("Usage: python data-read.py <Region> <Table> <id> <key> <Timing>")
    sys.exit()

dynamodb = boto3.resource('dynamodb', region_name=args[1])
table = dynamodb.Table(args[2])

def decimal_default(obj):
    if isinstance(obj, Decimal):
        return float(obj)
    raise TypeError

if __name__ == "__main__":
    print("Data read start. For stopping, Ctrl + C press.")
    
    sleep_time = Decimal(1) / Decimal(args[5])
    while True:
      res = table.query(
          KeyConditionExpression=Key('id').eq(args[3])
      )

      return_response = max(res["Items"], key=(lambda x: x[args[4]]))

      print(json.dumps(return_response, default=decimal_default))
      sleep(sleep_time)

CloudWatchMetricsで確認する

実際に上記のツールをしばらく動かして、メトリクスをCloudWatch上で確認します。

毎秒書き込み／読み込みのはずですが、読み込みの実際のCapacity使用量がどんどんと上昇しています。書き込みのユニット消費は1書き込み1ユニットと単純なのに対して、読み込みはレコードのサイズとレコード数が起因するため、全scanをかけているから上昇するのは当然と言えば当然の動きですね（どこかでスループット超過のエラーが発生するはずです）。

読み込みを超過しないように、Timestampで期間を設ける

今のままのdata-read.pyだと、データ数が多ければエラーで止まってしまうため、期間を設けて取得するように修正して、一旦エラーを回避しておきます。

      res = table.query(
          KeyConditionExpression=Key('id').eq(args[3])
      )

この部分を

      res = table.query(
          KeyConditionExpression=Key('id').eq(args[3]) & Key('timestamp').between(str((now_iso - datetime.timedelta(minutes=5)).isoformat()),  str((now_iso.isoformat())))
      )

こうするだけ。（CloudWatchのログはUTCで出力されるため、期間もUTC変換をしています）

Lambdaで実際のユニット使用量を取得＆割合を算出する

書き込みと読み込みはシミュレーションできるようになったので、メトリクスにアクセスして読み／書きの使用量を取得します。

ConsumedCapacityReadUnitsとConsumedCapacityWriteUnitsの取得

キーとなる実装は以下の通り。boto3で取得が可能です。

handler.py

def get_metrics(parameters):

    metrics = cloud_watch.get_metric_statistics(
                            Namespace=parameters['namespace'],
                            MetricName=parameters['metric_name'],
                            Dimensions=[
                                {
                                    'Name': parameters['dimension_name'],
                                    'Value': parameters['dimension_value']
                                }
                            ],
        StartTime=parameters['start_time'],
        EndTime=parameters['end_time'],
        Period=parameters['period'],
        Statistics=parameters['statistics'],
        Unit=parameters['unit'])
    
    return metrics

def output_metrics(parameters, target_metric, provisioned_cap):
    sort_datapoints = sorted(target_metric['Datapoints'], key=lambda x: x['Timestamp'])   
    print(parameters['metric_name'] + " result:") 
    for data in sort_datapoints:
        print(str(data['Timestamp']) + "\t" + str(round(data['Average'], 2)) + "\t")

必要なパラメータを設定してメトリクスの取得が可能になりました。読み／書きのキャパシティユニットは以下の通りになります。（今回は直近1時間のデータを、5分間隔で取得するようにしました。）

ConsumedReadCapacityUnits result:
2018-04-14 08:57:00+00:00       1.5
2018-04-14 09:02:00+00:00       1.24
2018-04-14 09:07:00+00:00       1.5
2018-04-14 09:22:00+00:00       0.5
2018-04-14 09:27:00+00:00       1.25
2018-04-14 09:32:00+00:00       1.5
2018-04-14 09:37:00+00:00       1.5
2018-04-14 09:42:00+00:00       1.5
2018-04-14 09:47:00+00:00       1.5
2018-04-14 09:52:00+00:00       0.92
ConsumedWriteCapacityUnits result:
2018-04-14 08:57:00+00:00       1.0
2018-04-14 09:02:00+00:00       1.0
2018-04-14 09:07:00+00:00       1.0
2018-04-14 09:22:00+00:00       1.0
2018-04-14 09:27:00+00:00       1.0
2018-04-14 09:32:00+00:00       1.0
2018-04-14 09:37:00+00:00       1.0
2018-04-14 09:42:00+00:00       1.0
2018-04-14 09:47:00+00:00       1.0
2018-04-14 09:52:00+00:00       1.0

当たり前ですが、書き込みのキャパシティユニットはずっと1.0です（毎秒書き込んでいるため）

DynamoDBのProvisionedCapacityUnitsを取得する

ConsumedRead/WriteCapacityUnitsと同様に設定されたキャパシティユニットを取得しても良いのですが、区切りの時間が異なっていたりと何かと不便なので、手抜きをしてDynamoDBのAPIから取得するようにします。

client_dynamodb = boto3.client('dynamodb')
:
:
    dynamodb_table = client_dynamodb.describe_table(TableName='CapacityUnitsTest')

取得したdynamodb_tableにはDescribeAPIで取得した情報が入っているので、読み／書きのキャパシティユニットを取得するには、以下のような実装をします。

読み込みキャパシティユニット：dynamodb_table['Table']['ProvisionedThroughput']['ReadCapacityUnits']
書き込みキャパシティユニット：dynamodb_table['Table']['ProvisionedThroughput']['WriteCapacityUnits']

これで設定値は取得できるようになりました。（AutoScallingなどで設定したキャパシティユニットが変更される条件は、今回は無視します）出力部分をこんな感じで割合計算用に修正しています。（パラメータ受け取り）

       print(str(data['Timestamp']) + "\t" + str(round(data['Average'], 2)) + "\t" + str(round(data['Average'] / provisioned_cap, 2)))

ConsumedReadCapacityUnits result:
2018-04-14 08:57:00+00:00       1.5     0.5
2018-04-14 09:02:00+00:00       1.24    0.41
2018-04-14 09:07:00+00:00       1.5     0.5
2018-04-14 09:22:00+00:00       0.5     0.17
2018-04-14 09:27:00+00:00       1.25    0.42
2018-04-14 09:32:00+00:00       1.5     0.5
2018-04-14 09:37:00+00:00       1.5     0.5
2018-04-14 09:42:00+00:00       1.5     0.5
2018-04-14 09:47:00+00:00       1.5     0.5
2018-04-14 09:52:00+00:00       0.92    0.3
ConsumedWriteCapacityUnits result:
2018-04-14 08:57:00+00:00       1.0     0.33
2018-04-14 09:02:00+00:00       1.0     0.33
2018-04-14 09:07:00+00:00       1.0     0.33
2018-04-14 09:22:00+00:00       1.0     0.33
2018-04-14 09:27:00+00:00       1.0     0.33
2018-04-14 09:32:00+00:00       1.0     0.33
2018-04-14 09:37:00+00:00       1.0     0.33
2018-04-14 09:42:00+00:00       1.0     0.33
2018-04-14 09:47:00+00:00       1.0     0.33
2018-04-14 09:52:00+00:00       1.0     0.33

先ほどのキャパシティユニットの使用量に加え、実際にどの程度キャパシティユニットを使用したのかが割合で表示されました。これを出せれば、「設定しているキャパシティユニットの◯◯％を超えたら何かアクションをする」という制御が可能になります。

まとめ

DynamoDBのキャパシティユニットの使用量を割合で出せる様にしてみました。ロジックを使い回せば、この内容をベースに、色々な監視やらができそうな気がしています。