Add recovery option to DynamoDB backend

When Vault is killed without the chance to clean up the lock
entry in DynamoDB, no further Vault nodes can become leaders after
that.

To recover from this situation, this commit adds an environment
variable and a configuration flag that when set to "1" causes Vault
to delete the lock entry from DynamoDB.

physical/dynamodb.go

@@ -60,8 +60,9 @@ const (
 // a DynamoDB table. It can be run in high-availability mode
 // as DynamoDB has locking capabilities.
 type DynamoDBBackend struct {
-	table  string
-	client *dynamodb.DynamoDB
+	table    string
+	client   *dynamodb.DynamoDB
+	recovery bool
 }
 
 // DynamoDBRecord is the representation of a vault entry in
@@ -79,6 +80,7 @@ type DynamoDBLock struct {
 	value, key string
 	held       bool
 	lock       sync.Mutex
+	recovery   bool
 }
 
 // newDynamoDBBackend constructs a DynamoDB backend. If the
@@ -167,9 +169,15 @@ func newDynamoDBBackend(conf map[string]string) (Backend, error) {
 		return nil, err
 	}
 
+	recoveryMode := os.Getenv("RECOVERY_MODE")
+	if recoveryMode == "" {
+		recoveryMode = conf["recovery_mode"]
+	}
+
 	return &DynamoDBBackend{
-		table:  table,
-		client: client,
+		table:    table,
+		client:   client,
+		recovery: recoveryMode == "1",
 	}, nil
 }
 
@@ -318,9 +326,10 @@ func (d *DynamoDBBackend) List(prefix string) ([]string, error) {
 // LockWith is used for mutual exclusion based on the given key.
 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,
+		backend:  d,
+		key:      filepath.Join(filepath.Dir(key), DynamoDBLockPrefix+filepath.Base(key)),
+		value:    value,
+		recovery: d.recovery,
 	}, nil
 }
 
@@ -460,6 +469,20 @@ func (l *DynamoDBLock) tryToLock(stop, success chan struct{}, errors chan error)
 				if err, ok := err.(awserr.Error); ok && err.Code() != "ConditionalCheckFailedException" {
 					errors <- err
 				}
+				if l.recovery {
+					_, err := l.backend.client.DeleteItem(&dynamodb.DeleteItemInput{
+						TableName: aws.String(l.backend.table),
+						Key: map[string]*dynamodb.AttributeValue{
+							"Path": {S: aws.String(record.Path)},
+							"Key":  {S: aws.String(record.Key)},
+						},
+					})
+					if err != nil {
+						errors <- fmt.Errorf("could not delete lock record: %s", err)
+					} else {
+						l.recovery = false
+					}
+				}
 			} else {
 				ticker.Stop()
 				close(success)