Make Unlock key delete conditional on being old leader's

physical/dynamodb/dynamodb.go

@@ -571,10 +571,27 @@ func (l *DynamoDBLock) Unlock() error {
 	}
 
 	l.held = false
-	if err := l.backend.Delete(context.Background(), l.key); err != nil {
-		return err
+
+	// Conditionally delete after check that the key is actually this Vault's and
+	// not been already claimed by another leader
+	condition := "#identity = :identity"
+	deleteMyLock := &dynamodb.DeleteItemInput{
+		TableName:           &l.backend.table,
+		ConditionExpression: &condition,
+		Key: map[string]*dynamodb.AttributeValue{
+			"Path": &dynamodb.AttributeValue{S: aws.String(recordPathForVaultKey(l.key))},
+			"Key":  &dynamodb.AttributeValue{S: aws.String(recordKeyForVaultKey(l.key))},
+		},
+		ExpressionAttributeNames: map[string]*string{
+			"#identity": aws.String("Identity"),
+		},
+		ExpressionAttributeValues: map[string]*dynamodb.AttributeValue{
+			":identity": &dynamodb.AttributeValue{B: []byte(l.identity)},
+		},
 	}
-	return nil
+
+	 _, err := l.backend.client.DeleteItem(deleteMyLock)
+	return filterError(err)
 }
 
 // Value checks whether or not the lock is held by any instance of DynamoDBLock,
@@ -609,11 +626,7 @@ func (l *DynamoDBLock) tryToLock(stop, success chan struct{}, errors chan error)
 			err := l.updateItem(true)
 			if err != nil {
 				if err, ok := err.(awserr.Error); ok {
-					// Don't report a condition check failure, this means that the lock
-					// is already being held.
-					if err.Code() != dynamodb.ErrCodeConditionalCheckFailedException {
-						errors <- err
-					}
+					errors <- err
 				} else {
 					// Its not an AWS error, and is probably not transient, bail out.
 					errors <- err
@@ -634,7 +647,10 @@ func (l *DynamoDBLock) periodicallyRenewLock(done chan struct{}) {
 		select {
 		case <-ticker.C:
 			// This should not renew the lock if the lock was deleted from under you.
-			l.updateItem(false)
+			err := l.updateItem(false)
+			if err != nil {
+				l.backend.logger.Error("error renewing leadership lock", "error", err)
+			}
 		case <-done:
 			ticker.Stop()
 			return
@@ -665,8 +681,8 @@ func (l *DynamoDBLock) updateItem(createIfMissing bool) error {
 	_, err := l.backend.client.UpdateItem(&dynamodb.UpdateItemInput{
 		TableName: aws.String(l.backend.table),
 		Key: map[string]*dynamodb.AttributeValue{
-			"Path": &dynamodb.AttributeValue{S: aws.String(recordPathForVaultKey(l.key))},
-			"Key":  &dynamodb.AttributeValue{S: aws.String(recordKeyForVaultKey(l.key))},
+			"Path": {S: aws.String(recordPathForVaultKey(l.key))},
+			"Key":  {S: aws.String(recordKeyForVaultKey(l.key))},
 		},
 		UpdateExpression: aws.String("SET #value=:value, #identity=:identity, #expires=:expires"),
 		// If both key and path already exist, we can only write if
@@ -682,13 +698,14 @@ func (l *DynamoDBLock) updateItem(createIfMissing bool) error {
 			"#value":    aws.String("Value"),
 		},
 		ExpressionAttributeValues: map[string]*dynamodb.AttributeValue{
-			":identity": &dynamodb.AttributeValue{B: []byte(l.identity)},
-			":value":    &dynamodb.AttributeValue{B: []byte(l.value)},
-			":now":      &dynamodb.AttributeValue{N: aws.String(strconv.FormatInt(now.UnixNano(), 10))},
-			":expires":  &dynamodb.AttributeValue{N: aws.String(strconv.FormatInt(now.Add(l.ttl).UnixNano(), 10))},
+			":identity": {B: []byte(l.identity)},
+			":value":    {B: []byte(l.value)},
+			":now":      {N: aws.String(strconv.FormatInt(now.UnixNano(), 10))},
+			":expires":  {N: aws.String(strconv.FormatInt(now.Add(l.ttl).UnixNano(), 10))},
 		},
 	})
-	return err
+
+	return filterError(err)
 }
 
 // watch checks whether the lock has changed in the
@@ -831,3 +848,16 @@ func unescapeEmptyPath(s string) string {
 	}
 	return s
 }
+
+// filterError filters out expected errors which are expected
+// under normal operation.
+func filterError(err error) error{
+	if err, ok := err.(awserr.Error); ok {
+		// Don't report a condition check failure, this means that the lock
+		// is already being held by another vault
+		if err.Code() == dynamodb.ErrCodeConditionalCheckFailedException {
+			return nil
+		}
+	}
+	return err
+}

vault/ha.go

@@ -6,6 +6,8 @@ import (
 	"crypto/x509"
 	"errors"
 	"fmt"
+	"github.com/aws/aws-sdk-go/aws/awserr"
+	"github.com/aws/aws-sdk-go/service/dynamodb"
 	"sync/atomic"
 	"time"
 
@@ -576,7 +578,9 @@ func (c *Core) waitForLeadership(newLeaderCh chan func(), manualStepDownCh, stop
 					c.logger.Error("clearing leader advertisement failed", "error", err)
 				}
 
-				c.heldHALock.Unlock()
+				if err := c.heldHALock.Unlock(); err != nil {
+					c.logger.Error("unlocking HA lock failed", "error", err)
+				}
 				c.heldHALock = nil
 			}