List provisioned and consumed capacity for all DynamoDB tables

dynamodb_capacity_report.py

# Usage: python3 dynamodb_capacity_report.py 2020-04-01 2020-05-01

import csv
import sys
from datetime import datetime

import boto3

cloudwatch = boto3.client('cloudwatch')


def main(start_time, end_time):
    consumed_rcu = get_metric_data('ConsumedReadCapacityUnits', start_time, end_time)
    consumed_wcu = get_metric_data('ConsumedWriteCapacityUnits', start_time, end_time)
    provisioned_rcu = get_metric_data('ProvisionedReadCapacityUnits', start_time, end_time)
    provisioned_wcu = get_metric_data('ProvisionedWriteCapacityUnits', start_time, end_time)
    all_keys = set(consumed_rcu).union(consumed_wcu).union(provisioned_rcu).union(provisioned_wcu)
    writer = csv.writer(sys.stdout)
    writer.writerow(['table/index', 'consumed_rcu', 'consumed_wcu', 'provisioned_rcu', 'provisioned_wcu'])
    for key in all_keys:
        writer.writerow([
            '/'.join(key),
            consumed_rcu.get(key, 0),
            consumed_wcu.get(key, 0),
            provisioned_rcu.get(key, ''),
            provisioned_wcu.get(key, ''),
        ])


def get_metric_data(metric_name, start_time, end_time):
    is_weird_metric = metric_name.startswith('Consumed')
    period = 24 * 60 * 60
    data_by_dimensions = dict()
    paginator = cloudwatch.get_paginator('list_metrics')
    for page in paginator.paginate(Namespace='AWS/DynamoDB', MetricName=metric_name):
        data = cloudwatch.get_metric_data(
            MetricDataQueries=[
                {
                    'Id': f'm{i}',
                    'MetricStat': {
                        'Metric': {
                            'Namespace': metric['Namespace'],
                            'MetricName': metric['MetricName'],
                            'Dimensions': metric['Dimensions'],
                        },
                        'Stat': 'Sum' if is_weird_metric else 'Average',
                        'Period': period,
                    },
                }
                for i, metric in enumerate(page['Metrics'])
            ],
            StartTime=start_time,
            EndTime=end_time,
        )
        for result in data['MetricDataResults']:
            if not result['Values']:
                continue
            i = int(result['Id'][1:])
            metric = page['Metrics'][i]
            dimensions = tuple(dimension['Value'] for dimension in metric['Dimensions'])
            average_of_values = sum(result['Values']) / len(result['Values'])
            if is_weird_metric:
                average_of_values = average_of_values / period
            data_by_dimensions[dimensions] = average_of_values
    return data_by_dimensions


if __name__ == '__main__':
    start_time = datetime.strptime(sys.argv[1], '%Y-%m-%d')
    end_time = datetime.strptime(sys.argv[2], '%Y-%m-%d')
    main(start_time, end_time)