func transformMetrics(stats map[string]interface{}) map[string]interface{} {
	if consumedReadCapacitySum, ok := stats["ConsumedReadCapacityUnitsSum"].(float64); ok {
		stats["ConsumedReadCapacityUnitsNormalized"] = consumedReadCapacitySum / 60.0
	}
	if consumedWriteCapacitySum, ok := stats["ConsumedWriteCapacityUnitsSum"].(float64); ok {
		stats["ConsumedWriteCapacityUnitsNormalized"] = consumedWriteCapacitySum / 60.0
	}
	return stats
}
TransformMetrics converts some of datapoints to post differences of two metrics
package mpawsdynamodb
import (
	"flag"
	"log"
	"strings"
	"sync"
	"time"
	"golang.org/x/sync/errgroup"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/credentials"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/cloudwatch"
	"github.com/aws/aws-sdk-go/service/cloudwatch/cloudwatchiface"
	mp "github.com/mackerelio/go-mackerel-plugin-helper"
)
const (
	namespace              = "AWS/DynamoDB"
	metricsTypeAverage     = "Average"
	metricsTypeSum         = "Sum"
	metricsTypeMaximum     = "Maximum"
	metricsTypeMinimum     = "Minimum"
	metricsTypeSampleCount = "SampleCount"
)
type metricsGroup struct {
	CloudWatchName string
	Metrics        []metric
}
type metric struct {
	MackerelName string
	Type         string
	FillZero     bool
}
type DynamoDBPlugin struct {
	TableName string
	Prefix    string
	AccessKeyID     string
	SecretAccessKey string
	Region          string
	CloudWatch      *cloudwatch.CloudWatch
}
func (p DynamoDBPlugin) MetricKeyPrefix() string {
	return p.Prefix
}
func (p *DynamoDBPlugin) prepare() error {
	sess, err := session.NewSession()
	if err != nil {
		return err
	}
	config := aws.NewConfig()
	if p.AccessKeyID != "" && p.SecretAccessKey != "" {
		config = config.WithCredentials(credentials.NewStaticCredentials(p.AccessKeyID, p.SecretAccessKey, ""))
	}
	if p.Region != "" {
		config = config.WithRegion(p.Region)
	}
	p.CloudWatch = cloudwatch.New(sess, config)
	return nil
}
func transformAndAppendDatapoint(stats map[string]interface{}, dp *cloudwatch.Datapoint, dataType string, label string, fillZero bool) map[string]interface{} {
	if dp != nil {
		switch dataType {
		case metricsTypeAverage:
			stats[label] = *dp.Average
		case metricsTypeSum:
			stats[label] = *dp.Sum
		case metricsTypeMaximum:
			stats[label] = *dp.Maximum
		case metricsTypeMinimum:
			stats[label] = *dp.Minimum
		case metricsTypeSampleCount:
			stats[label] = *dp.SampleCount
		}
	} else if fillZero {
		stats[label] = 0.0
	}
	return stats
}
func fetchOperationWildcardMetrics(cw cloudwatchiface.CloudWatchAPI, mg metricsGroup, baseDimensions []*cloudwatch.Dimension) (map[string]interface{}, error) {
	dimensionFilters := make([]*cloudwatch.DimensionFilter, len(baseDimensions))
	for i, dimension := range baseDimensions {
		dimensionFilters[i] = &cloudwatch.DimensionFilter{
			Name:  dimension.Name,
			Value: dimension.Value,
		}
	}
	input := &cloudwatch.ListMetricsInput{
		Dimensions: dimensionFilters,
		Namespace:  aws.String(namespace),
		MetricName: aws.String(mg.CloudWatchName),
	}
	res, err := cw.ListMetrics(input)
	if err != nil {
		return nil, err
	}
	stats := make(map[string]interface{})
	for _, cwMetric := range res.Metrics {
		dimensions := cwMetric.Dimensions
		var operation *string
		for _, d := range dimensions {
			if *d.Name == "Operation" {
				operation = d.Value
				break
			}
		}
		if operation == nil {
			log.Printf("Unexpected dimension, skip: %s", dimensions)
			continue
		}
		dp, err := getLastPointFromCloudWatch(cw, mg, dimensions)
		if err != nil {
			return nil, nil
		}
		if dp != nil {
			for _, met := range mg.Metrics {
				label := strings.Replace(met.MackerelName, "#", *operation, 1)
				stats = transformAndAppendDatapoint(stats, dp, met.Type, label, met.FillZero)
			}
		}
	}
	return stats, nil
}
func getLastPointFromCloudWatch(cw cloudwatchiface.CloudWatchAPI, metric metricsGroup, dimensions []*cloudwatch.Dimension) (*cloudwatch.Datapoint, error) {
	now := time.Now()
	statsInput := make([]*string, len(metric.Metrics))
	for i, typ := range metric.Metrics {
		statsInput[i] = aws.String(typ.Type)
	}
	input := &cloudwatch.GetMetricStatisticsInput{
		StartTime:  aws.Time(now.Add(time.Duration(480) * time.Second * -1)),
		EndTime:    aws.Time(now),
		MetricName: aws.String(metric.CloudWatchName),
		Period:     aws.Int64(60),
		Statistics: statsInput,
		Namespace:  aws.String(namespace),
		Dimensions: dimensions,
	}
	response, err := cw.GetMetricStatistics(input)
	if err != nil {
		return nil, err
	}
	datapoints := response.Datapoints
	if len(datapoints) == 0 {
		return nil, nil
	}
	latest := new(time.Time)
	var latestDp *cloudwatch.Datapoint
	for _, dp := range datapoints {
		if dp.Timestamp.Before(*latest) {
			continue
		}
		latest = dp.Timestamp
		latestDp = dp
	}
	return latestDp, nil
}
var defaultMetricsGroup = []metricsGroup{
	{CloudWatchName: "ConditionalCheckFailedRequests", Metrics: []metric{
		{MackerelName: "ConditionalCheckFailedRequests", Type: metricsTypeSum},
	}},
	{CloudWatchName: "ConsumedReadCapacityUnits", Metrics: []metric{
		{MackerelName: "ConsumedReadCapacityUnitsSum", Type: metricsTypeSum},
		{MackerelName: "ConsumedReadCapacityUnitsAverage", Type: metricsTypeAverage},
	}},
	{CloudWatchName: "ConsumedWriteCapacityUnits", Metrics: []metric{
		{MackerelName: "ConsumedWriteCapacityUnitsSum", Type: metricsTypeSum},
		{MackerelName: "ConsumedWriteCapacityUnitsAverage", Type: metricsTypeAverage},
	}},
	{CloudWatchName: "ProvisionedReadCapacityUnits", Metrics: []metric{
		{MackerelName: "ProvisionedReadCapacityUnits", Type: metricsTypeMinimum},
	}},
	{CloudWatchName: "ProvisionedWriteCapacityUnits", Metrics: []metric{
		{MackerelName: "ProvisionedWriteCapacityUnits", Type: metricsTypeMinimum},
	}},
	{CloudWatchName: "SystemErrors", Metrics: []metric{
		{MackerelName: "SystemErrors", Type: metricsTypeSum},
	}},
	{CloudWatchName: "UserErrors", Metrics: []metric{
		{MackerelName: "UserErrors", Type: metricsTypeSum},
	}},
	{CloudWatchName: "ReadThrottleEvents", Metrics: []metric{
		{MackerelName: "ReadThrottleEvents", Type: metricsTypeSum, FillZero: true},
	}},
	{CloudWatchName: "WriteThrottleEvents", Metrics: []metric{
		{MackerelName: "WriteThrottleEvents", Type: metricsTypeSum, FillZero: true},
	}},
	{CloudWatchName: "TimeToLiveDeletedItemCount", Metrics: []metric{
		{MackerelName: "TimeToLiveDeletedItemCount", Type: metricsTypeSum},
	}},
}
var operationalMetricsGroup = []metricsGroup{
	{CloudWatchName: "SuccessfulRequestLatency", Metrics: []metric{
		{MackerelName: "SuccessfulRequests.#", Type: metricsTypeSampleCount},
		{MackerelName: "SuccessfulRequestLatency.#.Minimum", Type: metricsTypeMinimum},
		{MackerelName: "SuccessfulRequestLatency.#.Maximum", Type: metricsTypeMaximum},
		{MackerelName: "SuccessfulRequestLatency.#.Average", Type: metricsTypeAverage},
	}},
	{CloudWatchName: "ThrottledRequests", Metrics: []metric{
		{MackerelName: "ThrottledRequests.#", Type: metricsTypeSampleCount},
	}},
	{CloudWatchName: "SystemErrors", Metrics: []metric{
		{MackerelName: "SystemErrors.#", Type: metricsTypeSampleCount},
	}},
	{CloudWatchName: "UserErrors", Metrics: []metric{
		{MackerelName: "UserErrors.#", Type: metricsTypeSampleCount},
	}},
	{CloudWatchName: "ReturnedItemCount", Metrics: []metric{
		{MackerelName: "ReturnedItemCount.#", Type: metricsTypeAverage},
	}},
}
func (p DynamoDBPlugin) FetchMetrics() (map[string]interface{}, error) {
	stats := make(map[string]interface{})
	tableDimensions := []*cloudwatch.Dimension{{
		Name:  aws.String("TableName"),
		Value: aws.String(p.TableName),
	}}
	var mu sync.Mutex
	var eg errgroup.Group
	for _, met := range defaultMetricsGroup {
		met := met
		eg.Go(func() error {
			dp, err := getLastPointFromCloudWatch(p.CloudWatch, met, tableDimensions)
			if err != nil {
				return err
			}
			for _, m := range met.Metrics {
				mu.Lock()
				stats = transformAndAppendDatapoint(stats, dp, m.Type, m.MackerelName, m.FillZero)
				mu.Unlock()
			}
			return nil
		})
	}
	for _, met := range operationalMetricsGroup {
		met := met
		eg.Go(func() error {
			operationalStats, err := fetchOperationWildcardMetrics(p.CloudWatch, met, tableDimensions)
			if err != nil {
				return err
			}
			if operationalStats != nil {
				for name, s := range operationalStats {
					mu.Lock()
					stats[name] = s
					mu.Unlock()
				}
			}
			return nil
		})
	}
	if err := eg.Wait(); err != nil {
		return nil, err
	}
	return transformMetrics(stats), nil
}