func (d *DynamoDBBackend) batchWriteRequests(requests []*dynamodb.WriteRequest) error {
	for len(requests) > 0 {
		batchSize := int(math.Min(float64(len(requests)), 25))
		batch := requests[:batchSize]
		requests = requests[batchSize:]
		d.permitPool.Acquire()
		_, err := d.client.BatchWriteItem(&dynamodb.BatchWriteItemInput{
			RequestItems: map[string][]*dynamodb.WriteRequest{
				d.table: batch,
			},
		})
		d.permitPool.Release()
		if err != nil {
			return err
		}
	}
	return nil
}
batchWriteRequests takes a list of write requests and executes them in badges
with a maximum size of 25 (which is the limit of BatchWriteItem requests).
package physical
import (
	"fmt"
	"math"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
	log "github.com/mgutz/logxi/v1"
	"github.com/armon/go-metrics"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/awserr"
	"github.com/aws/aws-sdk-go/aws/credentials"
	"github.com/aws/aws-sdk-go/aws/credentials/ec2rolecreds"
	"github.com/aws/aws-sdk-go/aws/ec2metadata"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/dynamodb"
	"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbattribute"
	"github.com/hashicorp/errwrap"
)
const (
	DefaultDynamoDBRegion = "us-east-1"
	DefaultDynamoDBTableName = "vault-dynamodb-backend"
	DefaultDynamoDBReadCapacity = 5
	DefaultDynamoDBWriteCapacity = 5
	DynamoDBEmptyPath = " "
	DynamoDBLockPrefix = "_"
	DynamoDBLockRetryInterval = time.Second
	DynamoDBWatchRetryMax = 5
	DynamoDBWatchRetryInterval = 5 * time.Second
)
type DynamoDBBackend struct {
	table      string
	client     *dynamodb.DynamoDB
	recovery   bool
	logger     log.Logger
	haEnabled  bool
	permitPool *PermitPool
}
type DynamoDBRecord struct {
	Path  string
	Key   string
	Value []byte
}
type DynamoDBLock struct {
	backend    *DynamoDBBackend
	value, key string
	held       bool
	lock       sync.Mutex
	recovery   bool
}
func newDynamoDBBackend(conf map[string]string, logger log.Logger) (Backend, error) {
	table := os.Getenv("AWS_DYNAMODB_TABLE")
	if table == "" {
		table = conf["table"]
		if table == "" {
			table = DefaultDynamoDBTableName
		}
	}
	readCapacityString := os.Getenv("AWS_DYNAMODB_READ_CAPACITY")
	if readCapacityString == "" {
		readCapacityString = conf["read_capacity"]
		if readCapacityString == "" {
			readCapacityString = "0"
		}
	}
	readCapacity, err := strconv.Atoi(readCapacityString)
	if err != nil {
		return nil, fmt.Errorf("invalid read capacity: %s", readCapacityString)
	}
	if readCapacity == 0 {
		readCapacity = DefaultDynamoDBReadCapacity
	}
	writeCapacityString := os.Getenv("AWS_DYNAMODB_WRITE_CAPACITY")
	if writeCapacityString == "" {
		writeCapacityString = conf["write_capacity"]
		if writeCapacityString == "" {
			writeCapacityString = "0"
		}
	}
	writeCapacity, err := strconv.Atoi(writeCapacityString)
	if err != nil {
		return nil, fmt.Errorf("invalid write capacity: %s", writeCapacityString)
	}
	if writeCapacity == 0 {
		writeCapacity = DefaultDynamoDBWriteCapacity
	}
	accessKey := os.Getenv("AWS_ACCESS_KEY_ID")
	if accessKey == "" {
		accessKey = conf["access_key"]
	}
	secretKey := os.Getenv("AWS_SECRET_ACCESS_KEY")
	if secretKey == "" {
		secretKey = conf["secret_key"]
	}
	sessionToken := os.Getenv("AWS_SESSION_TOKEN")
	if sessionToken == "" {
		sessionToken = conf["session_token"]
	}
	endpoint := os.Getenv("AWS_DYNAMODB_ENDPOINT")
	if endpoint == "" {
		endpoint = conf["endpoint"]
	}
	region := os.Getenv("AWS_DEFAULT_REGION")
	if region == "" {
		region = conf["region"]
		if region == "" {
			region = DefaultDynamoDBRegion
		}
	}
	creds := credentials.NewChainCredentials([]credentials.Provider{
		&credentials.StaticProvider{Value: credentials.Value{
			AccessKeyID:     accessKey,
			SecretAccessKey: secretKey,
			SessionToken:    sessionToken,
		}},
		&credentials.EnvProvider{},
		&credentials.SharedCredentialsProvider{Filename: "", Profile: ""},
		&ec2rolecreds.EC2RoleProvider{Client: ec2metadata.New(session.New())},
	})
	awsConf := aws.NewConfig().
		WithCredentials(creds).
		WithRegion(region).
		WithEndpoint(endpoint)
	client := dynamodb.New(session.New(awsConf))
	if err := ensureTableExists(client, table, readCapacity, writeCapacity); err != nil {
		return nil, err
	}
	haEnabled := os.Getenv("DYNAMODB_HA_ENABLED")
	if haEnabled == "" {
		haEnabled = conf["ha_enabled"]
	}
	haEnabledBool, _ := strconv.ParseBool(haEnabled)
	recoveryMode := os.Getenv("RECOVERY_MODE")
	if recoveryMode == "" {
		recoveryMode = conf["recovery_mode"]
	}
	recoveryModeBool, _ := strconv.ParseBool(recoveryMode)
	maxParStr, ok := conf["max_parallel"]
	var maxParInt int
	if ok {
		maxParInt, err = strconv.Atoi(maxParStr)
		if err != nil {
			return nil, errwrap.Wrapf("failed parsing max_parallel parameter: {{err}}", err)
		}
		if logger.IsDebug() {
			logger.Debug("physical/dynamodb: max_parallel set", "max_parallel", maxParInt)
		}
	}
	return &DynamoDBBackend{
		table:      table,
		client:     client,
		permitPool: NewPermitPool(maxParInt),
		recovery:   recoveryModeBool,
		haEnabled:  haEnabledBool,
		logger:     logger,
	}, nil
}
func (d *DynamoDBBackend) Put(entry *Entry) error {
	defer metrics.MeasureSince([]string{"dynamodb", "put"}, time.Now())
	record := DynamoDBRecord{
		Path:  recordPathForVaultKey(entry.Key),
		Key:   recordKeyForVaultKey(entry.Key),
		Value: entry.Value,
	}
	item, err := dynamodbattribute.ConvertToMap(record)
	if err != nil {
		return fmt.Errorf("could not convert prefix record to DynamoDB item: %s", err)
	}
	requests := []*dynamodb.WriteRequest{{
		PutRequest: &dynamodb.PutRequest{
			Item: item,
		},
	}}
	for _, prefix := range prefixes(entry.Key) {
		record = DynamoDBRecord{
			Path: recordPathForVaultKey(prefix),
			Key:  fmt.Sprintf("%s/", recordKeyForVaultKey(prefix)),
		}
		item, err := dynamodbattribute.ConvertToMap(record)
		if err != nil {
			return fmt.Errorf("could not convert prefix record to DynamoDB item: %s", err)
		}
		requests = append(requests, &dynamodb.WriteRequest{
			PutRequest: &dynamodb.PutRequest{
				Item: item,
			},
		})
	}
	return d.batchWriteRequests(requests)
}
func (d *DynamoDBBackend) Get(key string) (*Entry, error) {
	defer metrics.MeasureSince([]string{"dynamodb", "get"}, time.Now())
	d.permitPool.Acquire()
	defer d.permitPool.Release()
	resp, err := d.client.GetItem(&dynamodb.GetItemInput{
		TableName:      aws.String(d.table),
		ConsistentRead: aws.Bool(true),
		Key: map[string]*dynamodb.AttributeValue{
			"Path": {S: aws.String(recordPathForVaultKey(key))},
			"Key":  {S: aws.String(recordKeyForVaultKey(key))},
		},
	})
	if err != nil {
		return nil, err
	}
	if resp.Item == nil {
		return nil, nil
	}
	record := &DynamoDBRecord{}
	if err := dynamodbattribute.ConvertFromMap(resp.Item, record); err != nil {
		return nil, err
	}
	return &Entry{
		Key:   vaultKey(record),
		Value: record.Value,
	}, nil
}
func (d *DynamoDBBackend) Delete(key string) error {
	defer metrics.MeasureSince([]string{"dynamodb", "delete"}, time.Now())
	requests := []*dynamodb.WriteRequest{{
		DeleteRequest: &dynamodb.DeleteRequest{
			Key: map[string]*dynamodb.AttributeValue{
				"Path": {S: aws.String(recordPathForVaultKey(key))},
				"Key":  {S: aws.String(recordKeyForVaultKey(key))},
			},
		},
	}}
	prefixes := prefixes(key)
	sort.Sort(sort.Reverse(sort.StringSlice(prefixes)))
	for _, prefix := range prefixes {
		items, err := d.List(prefix)
		if err != nil {
			return err
		}
		if len(items) == 1 {
			requests = append(requests, &dynamodb.WriteRequest{
				DeleteRequest: &dynamodb.DeleteRequest{
					Key: map[string]*dynamodb.AttributeValue{
						"Path": {S: aws.String(recordPathForVaultKey(prefix))},
						"Key":  {S: aws.String(fmt.Sprintf("%s/", recordKeyForVaultKey(prefix)))},
					},
				},
			})
		}
	}
	return d.batchWriteRequests(requests)
}
func (d *DynamoDBBackend) List(prefix string) ([]string, error) {
	defer metrics.MeasureSince([]string{"dynamodb", "list"}, time.Now())
	prefix = strings.TrimSuffix(prefix, "/")
	keys := []string{}
	prefix = escapeEmptyPath(prefix)
	queryInput := &dynamodb.QueryInput{
		TableName:      aws.String(d.table),
		ConsistentRead: aws.Bool(true),
		KeyConditions: map[string]*dynamodb.Condition{
			"Path": {
				ComparisonOperator: aws.String("EQ"),
				AttributeValueList: []*dynamodb.AttributeValue{{
					S: aws.String(prefix),
				}},
			},
		},
	}
	d.permitPool.Acquire()
	defer d.permitPool.Release()
	err := d.client.QueryPages(queryInput, func(out *dynamodb.QueryOutput, lastPage bool) bool {
		var record DynamoDBRecord
		for _, item := range out.Items {
			dynamodbattribute.ConvertFromMap(item, &record)
			if !strings.HasPrefix(record.Key, DynamoDBLockPrefix) {
				keys = append(keys, record.Key)
			}
		}
		return !lastPage
	})
	if err != nil {
		return nil, err
	}
	return keys, nil
}
func (d *DynamoDBBackend) LockWith(key, value string) (Lock, error) {
	return &DynamoDBLock{
		backend:  d,
		key:      filepath.Join(filepath.Dir(key), DynamoDBLockPrefix+filepath.Base(key)),
		value:    value,
		recovery: d.recovery,
	}, nil
}
func (d *DynamoDBBackend) HAEnabled() bool {
	return d.haEnabled
}