diff --git a/lib/backend/dynamo/atomicwrite.go b/lib/backend/dynamo/atomicwrite.go index 0dad530fa0c..d8301090b84 100644 --- a/lib/backend/dynamo/atomicwrite.go +++ b/lib/backend/dynamo/atomicwrite.go @@ -42,19 +42,13 @@ const ( txnAttemptLogInterval = 8 ) -var ( - existsExpr = "attribute_exists(FullPath)" - notExistsExpr = "attribute_not_exists(FullPath)" - revisionExpr = "Revision = :rev AND attribute_exists(FullPath)" - missingRevisionExpr = "attribute_not_exists(Revision) AND attribute_exists(FullPath)" -) - func (b *Backend) AtomicWrite(ctx context.Context, condacts []backend.ConditionalAction) (revision string, err error) { if err := backend.ValidateAtomicWrite(condacts); err != nil { return "", trace.Wrap(err) } revision = backend.CreateRevision() + now := b.clock.Now() tableName := aws.String(b.TableName) @@ -69,9 +63,15 @@ func (b *Backend) AtomicWrite(ctx context.Context, condacts []backend.Conditiona case backend.KindWhatever: // no comparison to assert case backend.KindExists: - condExpr = &existsExpr + condExpr = aws.String("attribute_exists(FullPath) AND (attribute_not_exists(Expires) OR Expires >= :now)") + exprAttrValues = map[string]types.AttributeValue{ + ":now": timeToAttributeValue(now), + } case backend.KindNotExists: - condExpr = ¬ExistsExpr + condExpr = aws.String("attribute_not_exists(FullPath) OR Expires < :now") + exprAttrValues = map[string]types.AttributeValue{ + ":now": timeToAttributeValue(now), + } case backend.KindRevision: switch ca.Condition.Revision { case "": @@ -79,44 +79,39 @@ func (b *Backend) AtomicWrite(ctx context.Context, condacts []backend.Conditiona return "", trace.Wrap(backend.ErrConditionFailed) case backend.BlankRevision: // item has not been modified since the introduction of the revision attr - condExpr = &missingRevisionExpr + condExpr = aws.String("attribute_exists(FullPath) AND attribute_not_exists(Revision) AND (attribute_not_exists(Expires) OR Expires >= :now)") + exprAttrValues = map[string]types.AttributeValue{ + ":now": timeToAttributeValue(now), + } default: // revision is expected to be present and well-defined - condExpr = &revisionExpr + condExpr = aws.String("Revision = :rev AND (attribute_not_exists(Expires) OR Expires >= :now)") exprAttrValues = map[string]types.AttributeValue{ ":rev": &types.AttributeValueMemberS{Value: ca.Condition.Revision}, + ":now": timeToAttributeValue(now), } } default: - return "", trace.BadParameter("unexpected condition kind %v in conditional action against key %q", ca.Condition.Kind, ca.Key) + return "", trace.BadParameter("unexpected condition kind %v in conditional action against key %+q", ca.Condition.Kind, ca.Key.String()) } - fullPath := prependPrefix(ca.Key) - var txnItem types.TransactWriteItem switch ca.Action.Kind { case backend.KindNop: - av, err := attributevalue.MarshalMap(keyLookup{ - HashKey: hashKey, - FullPath: fullPath, - }) - if err != nil { - return "", trace.Wrap(err) - } - txnItem.ConditionCheck = &types.ConditionCheck{ + TableName: tableName, + Key: keyToAttributeValueMap(ca.Key), + ConditionExpression: condExpr, ExpressionAttributeValues: exprAttrValues, - Key: av, - TableName: tableName, } case backend.KindPut: includesPut = true r := record{ HashKey: hashKey, - FullPath: fullPath, + FullPath: prependPrefix(ca.Key), Value: ca.Action.Item.Value, Timestamp: time.Now().UTC().Unix(), Revision: revision, @@ -131,29 +126,24 @@ func (b *Backend) AtomicWrite(ctx context.Context, condacts []backend.Conditiona } txnItem.Put = &types.Put{ + TableName: tableName, + + Item: av, + ConditionExpression: condExpr, ExpressionAttributeValues: exprAttrValues, - Item: av, - TableName: tableName, } case backend.KindDelete: - av, err := attributevalue.MarshalMap(keyLookup{ - HashKey: hashKey, - FullPath: fullPath, - }) - if err != nil { - return "", trace.Wrap(err) - } - txnItem.Delete = &types.Delete{ + TableName: tableName, + Key: keyToAttributeValueMap(ca.Key), + ConditionExpression: condExpr, ExpressionAttributeValues: exprAttrValues, - Key: av, - TableName: tableName, } default: - return "", trace.BadParameter("unexpected action kind %v in conditional action against key %q", ca.Action.Kind, ca.Key) + return "", trace.BadParameter("unexpected action kind %v in conditional action against key %+q", ca.Action.Kind, ca.Key.String()) } txnItems = append(txnItems, txnItem) diff --git a/lib/backend/dynamo/dynamodbbk.go b/lib/backend/dynamo/dynamodbbk.go index e535f4de907..324d18e0178 100644 --- a/lib/backend/dynamo/dynamodbbk.go +++ b/lib/backend/dynamo/dynamodbbk.go @@ -25,6 +25,7 @@ import ( "log/slog" "net/http" "strconv" + "strings" "sync/atomic" "time" @@ -186,11 +187,6 @@ type record struct { Timestamp int64 } -type keyLookup struct { - HashKey string - FullPath string -} - const ( // hashKey is actually the name of the partition. This backend // places all objects in the same DynamoDB partition @@ -517,7 +513,7 @@ func (b *Backend) GetName() string { func (b *Backend) Create(ctx context.Context, item backend.Item) (*backend.Lease, error) { rev, err := b.create(ctx, item, modeCreate) if trace.IsCompareFailed(err) { - err = trace.AlreadyExists("%s", err) + err = trace.AlreadyExists("%+q already exists", item.Key.String()) } if err != nil { return nil, trace.Wrap(err) @@ -541,7 +537,7 @@ func (b *Backend) Put(ctx context.Context, item backend.Item) (*backend.Lease, e func (b *Backend) Update(ctx context.Context, item backend.Item) (*backend.Lease, error) { rev, err := b.create(ctx, item, modeUpdate) if trace.IsCompareFailed(err) { - err = trace.NotFound("%s", err) + err = trace.NotFound("%+q is not found", item.Key.String()) } if err != nil { return nil, trace.Wrap(err) @@ -602,13 +598,13 @@ func (b *Backend) Items(ctx context.Context, params backend.ItemsParams) iter.Se // filter out expired items, otherwise they might show up in the query // http://docs.aws.amazon.com/amazondynamodb/latest/developerguide/howitworks-ttl.html - filter = "attribute_not_exists(Expires) OR Expires >= :timestamp" + filter = "attribute_not_exists(Expires) OR Expires >= :now" ) av := map[string]types.AttributeValue{ ":rangeStart": &types.AttributeValueMemberS{Value: prependPrefix(params.StartKey)}, ":rangeEnd": &types.AttributeValueMemberS{Value: prependPrefix(params.EndKey)}, - ":timestamp": timeToAttributeValue(b.clock.Now().UTC()), + ":now": timeToAttributeValue(b.clock.Now()), ":hashKey": &types.AttributeValueMemberS{Value: hashKey}, } @@ -760,10 +756,7 @@ func (b *Backend) DeleteRange(ctx context.Context, startKey, endKey backend.Key) requests = append(requests, types.WriteRequest{ DeleteRequest: &types.DeleteRequest{ - Key: map[string]types.AttributeValue{ - hashKeyKey: &types.AttributeValueMemberS{Value: hashKey}, - fullPathKey: &types.AttributeValueMemberS{Value: prependPrefix(item.Key)}, - }, + Key: keyToAttributeValueMap(item.Key), }, }) @@ -806,7 +799,7 @@ func (b *Backend) Get(ctx context.Context, key backend.Key) (*backend.Item, erro Revision: r.Revision, } if r.Expires != nil { - item.Expires = time.Unix(*r.Expires, 0) + item.Expires = time.Unix(*r.Expires, 0).UTC() } if item.Revision == "" { @@ -844,24 +837,26 @@ func (b *Backend) CompareAndSwap(ctx context.Context, expected backend.Item, rep if err != nil { return nil, trace.Wrap(err) } - input := dynamodb.PutItemInput{ - Item: av, - TableName: aws.String(b.TableName), - ConditionExpression: aws.String("#v = :prev"), + input := &dynamodb.PutItemInput{ + TableName: aws.String(b.TableName), + + Item: av, + + ConditionExpression: aws.String("#v = :prev AND (attribute_not_exists(Expires) OR Expires >= :now)"), ExpressionAttributeNames: map[string]string{ "#v": "Value", }, ExpressionAttributeValues: map[string]types.AttributeValue{ ":prev": &types.AttributeValueMemberB{Value: expected.Value}, + ":now": timeToAttributeValue(b.clock.Now()), }, } - _, err = b.svc.PutItem(ctx, &input) + _, err = b.svc.PutItem(ctx, input) err = convertError(err) if err != nil { - // in this case let's use more specific compare failed error - if trace.IsAlreadyExists(err) { - return nil, trace.CompareFailed("%s", err) + if trace.IsCompareFailed(err) { + return nil, trace.CompareFailed("%+q not found or does not match expected", replaceWith.Key.String()) } return nil, trace.Wrap(err) } @@ -870,10 +865,36 @@ func (b *Backend) CompareAndSwap(ctx context.Context, expected backend.Item, rep // Delete deletes item by key func (b *Backend) Delete(ctx context.Context, key backend.Key) error { - if _, err := b.getKey(ctx, key); err != nil { - return err + // since dynamodb's built-in deletion of expired items can be very slow, + // instead of filtering the point delete here to only delete if the item is + // not expired we unconditionally delete the item, and we return the correct + // error by checking if the item that was just deleted was expired + now := b.clock.Now().Truncate(time.Second) + out, err := b.svc.DeleteItem(ctx, &dynamodb.DeleteItemInput{ + TableName: aws.String(b.TableName), + Key: keyToAttributeValueMap(key), + + ReturnValues: types.ReturnValueAllOld, + }) + if err != nil { + return trace.Wrap(err) } - return b.deleteKey(ctx, key) + + if len(out.Attributes) < 1 { + return trace.NotFound("%+q is not found", key.String()) + } + + var r struct { + Expires *attributevalue.UnixTime + } + if err := attributevalue.UnmarshalMap(out.Attributes, &r); err != nil { + return trace.Wrap(err, "checking expiry") + } + if r.Expires != nil && time.Time(*r.Expires).Before(now) { + return trace.NotFound("%+q is not found", key.String()) + } + + return nil } // ConditionalUpdate updates the matching item in Dynamo if the provided revision matches @@ -889,6 +910,9 @@ func (b *Backend) ConditionalUpdate(ctx context.Context, item backend.Item) (*ba rev, err := b.create(ctx, item, modeConditionalUpdate) if err != nil { + if trace.IsCompareFailed(err) { + err = trace.Wrap(backend.ErrIncorrectRevision) + } return nil, trace.Wrap(err) } @@ -903,27 +927,25 @@ func (b *Backend) ConditionalDelete(ctx context.Context, key backend.Key, rev st return trace.Wrap(backend.ErrIncorrectRevision) } - av, err := attributevalue.MarshalMap(keyLookup{ - HashKey: hashKey, - FullPath: prependPrefix(key), - }) - if err != nil { - return trace.Wrap(err) - } - - input := dynamodb.DeleteItemInput{ - Key: av, + input := &dynamodb.DeleteItemInput{ TableName: aws.String(b.TableName), + Key: keyToAttributeValueMap(key), } if rev == backend.BlankRevision { - input.ConditionExpression = aws.String("attribute_not_exists(Revision) AND attribute_exists(FullPath)") + input.ConditionExpression = aws.String("attribute_exists(FullPath) AND attribute_not_exists(Revision) AND (attribute_not_exists(Expires) OR Expires >= :now)") + input.ExpressionAttributeValues = map[string]types.AttributeValue{ + ":now": timeToAttributeValue(b.clock.Now()), + } } else { - input.ExpressionAttributeValues = map[string]types.AttributeValue{":rev": &types.AttributeValueMemberS{Value: rev}} - input.ConditionExpression = aws.String("Revision = :rev AND attribute_exists(FullPath)") + input.ConditionExpression = aws.String("Revision = :rev AND (attribute_not_exists(Expires) OR Expires >= :now)") + input.ExpressionAttributeValues = map[string]types.AttributeValue{ + ":rev": &types.AttributeValueMemberS{Value: rev}, + ":now": timeToAttributeValue(b.clock.Now()), + } } - if _, err = b.svc.DeleteItem(ctx, &input); err != nil { + if _, err := b.svc.DeleteItem(ctx, input); err != nil { err = convertError(err) if trace.IsCompareFailed(err) { return trace.Wrap(backend.ErrIncorrectRevision) @@ -947,22 +969,20 @@ func (b *Backend) KeepAlive(ctx context.Context, lease backend.Lease, expires ti return trace.BadParameter("lease is missing key") } input := &dynamodb.UpdateItemInput{ - ExpressionAttributeValues: map[string]types.AttributeValue{ - ":expires": &types.AttributeValueMemberN{Value: strconv.FormatInt(expires.UTC().Unix(), 10)}, - ":timestamp": &types.AttributeValueMemberN{Value: strconv.FormatInt(b.clock.Now().UTC().Unix(), 10)}, - }, TableName: aws.String(b.TableName), - Key: map[string]types.AttributeValue{ - hashKeyKey: &types.AttributeValueMemberS{Value: hashKey}, - fullPathKey: &types.AttributeValueMemberS{Value: prependPrefix(lease.Key)}, - }, + Key: keyToAttributeValueMap(lease.Key), + UpdateExpression: aws.String("SET Expires = :expires"), - ConditionExpression: aws.String("attribute_exists(FullPath) AND (attribute_not_exists(Expires) OR Expires >= :timestamp)"), + ConditionExpression: aws.String("attribute_exists(FullPath) AND (attribute_not_exists(Expires) OR Expires >= :now)"), + ExpressionAttributeValues: map[string]types.AttributeValue{ + ":expires": timeToAttributeValue(expires), + ":now": timeToAttributeValue(b.clock.Now()), + }, } _, err := b.svc.UpdateItem(ctx, input) err = convertError(err) if trace.IsCompareFailed(err) { - err = trace.NotFound("%s", err) + err = trace.NotFound("%+q is not found", lease.Key.String()) } return err } @@ -1096,16 +1116,6 @@ func (b *Backend) createTable(ctx context.Context, tableName *string, rangeKey s return trace.Wrap(err) } -// isExpired returns 'true' if the given object (record) has a TTL and -// it's due. -func (r *record) isExpired(now time.Time) bool { - if r.Expires == nil { - return false - } - expiryDateUTC := time.Unix(*r.Expires, 0).UTC() - return now.UTC().After(expiryDateUTC) -} - const ( modeCreate = iota modePut @@ -1121,7 +1131,7 @@ func prependPrefix(key backend.Key) string { // trimPrefix removes leading 'teleport' from the key func trimPrefix(key string) backend.Key { - return backend.KeyFromString(key).TrimPrefix(backend.KeyFromString(keyPrefix)) + return backend.KeyFromString(strings.TrimPrefix(key, keyPrefix)) } // create is a helper that writes a key/value pair in Dynamo with a given expiration. @@ -1144,105 +1154,77 @@ func (b *Backend) create(ctx context.Context, item backend.Item, mode int) (stri return "", trace.Wrap(err) } input := dynamodb.PutItemInput{ - Item: av, TableName: aws.String(b.TableName), + + Item: av, } switch mode { case modeCreate: - input.ConditionExpression = aws.String("attribute_not_exists(FullPath)") + input.ConditionExpression = aws.String("attribute_not_exists(FullPath) OR Expires < :now") + input.ExpressionAttributeValues = map[string]types.AttributeValue{ + ":now": timeToAttributeValue(b.clock.Now()), + } case modeUpdate: - input.ConditionExpression = aws.String("attribute_exists(FullPath)") + input.ConditionExpression = aws.String("attribute_exists(FullPath) AND (attribute_not_exists(Expires) OR Expires >= :now)") + input.ExpressionAttributeValues = map[string]types.AttributeValue{ + ":now": timeToAttributeValue(b.clock.Now()), + } case modePut: case modeConditionalUpdate: // If the revision is empty, then the resource existed prior to revision support. Instead of validating that // the revisions match, validate that the revision attribute does not exist. Otherwise, validate that the revision // attribute matches the item revision. if item.Revision == "" { - input.ConditionExpression = aws.String("attribute_not_exists(Revision) AND attribute_exists(FullPath)") + input.ConditionExpression = aws.String("attribute_exists(FullPath) AND attribute_not_exists(Revision) AND (attribute_not_exists(Expires) OR Expires >= :now)") + input.ExpressionAttributeValues = map[string]types.AttributeValue{ + ":now": timeToAttributeValue(b.clock.Now()), + } } else { - input.ExpressionAttributeValues = map[string]types.AttributeValue{":rev": &types.AttributeValueMemberS{Value: item.Revision}} - input.ConditionExpression = aws.String("Revision = :rev AND attribute_exists(FullPath)") + input.ConditionExpression = aws.String("Revision = :rev AND (attribute_not_exists(Expires) OR Expires >= :now)") + input.ExpressionAttributeValues = map[string]types.AttributeValue{ + ":rev": &types.AttributeValueMemberS{Value: item.Revision}, + ":now": timeToAttributeValue(b.clock.Now()), + } } default: - return "", trace.BadParameter("unrecognized mode") + return "", trace.BadParameter("unrecognized write mode %d (this is a bug)", mode) } _, err = b.svc.PutItem(ctx, &input) err = convertError(err) if err != nil { - if mode == modeConditionalUpdate && trace.IsCompareFailed(err) { - return "", trace.Wrap(backend.ErrIncorrectRevision) - } - return "", trace.Wrap(err) } return r.Revision, nil } -func (b *Backend) deleteKey(ctx context.Context, key backend.Key) error { - av, err := attributevalue.MarshalMap(keyLookup{ - HashKey: hashKey, - FullPath: prependPrefix(key), - }) - if err != nil { - return trace.Wrap(err) - } - input := dynamodb.DeleteItemInput{Key: av, TableName: aws.String(b.TableName)} - if _, err = b.svc.DeleteItem(ctx, &input); err != nil { - return trace.Wrap(err) - } - return nil -} - -func (b *Backend) deleteKeyIfExpired(ctx context.Context, key backend.Key) error { - _, err := b.svc.DeleteItem(ctx, &dynamodb.DeleteItemInput{ +func (b *Backend) getKey(ctx context.Context, key backend.Key) (*record, error) { + input := &dynamodb.GetItemInput{ TableName: aws.String(b.TableName), Key: keyToAttributeValueMap(key), - // succeed if the item no longer exists - ConditionExpression: aws.String( - "attribute_not_exists(FullPath) OR (attribute_exists(Expires) AND Expires <= :timestamp)", - ), - ExpressionAttributeValues: map[string]types.AttributeValue{ - ":timestamp": timeToAttributeValue(b.clock.Now()), - }, - }) - return trace.Wrap(err) -} - -func (b *Backend) getKey(ctx context.Context, key backend.Key) (*record, error) { - av, err := attributevalue.MarshalMap(keyLookup{ - HashKey: hashKey, - FullPath: prependPrefix(key), - }) - if err != nil { - return nil, trace.Wrap(err) - } - input := dynamodb.GetItemInput{ - Key: av, - TableName: aws.String(b.TableName), ConsistentRead: aws.Bool(true), } - out, err := b.svc.GetItem(ctx, &input) + now := b.clock.Now().Truncate(time.Second) + out, err := b.svc.GetItem(ctx, input) if err != nil { // we deliberately use a "generic" trace error here, since we don't want // callers to make assumptions about the nature of the failure. - return nil, trace.WrapWithMessage(err, "failed to get %q (dynamo error)", key.String()) + return nil, trace.WrapWithMessage(err, "failed to get %+q (dynamo error)", key.String()) } if len(out.Item) == 0 { - return nil, trace.NotFound("%q is not found", key.String()) + return nil, trace.NotFound("%+q is not found", key.String()) } var r record if err := attributevalue.UnmarshalMap(out.Item, &r); err != nil { - return nil, trace.WrapWithMessage(err, "failed to unmarshal dynamo item %q", key.String()) + return nil, trace.WrapWithMessage(err, "failed to unmarshal dynamo item %+q", key.String()) } - // Check if key expired, if expired delete it - if r.isExpired(b.clock.Now()) { - if err := b.deleteKeyIfExpired(ctx, key); err != nil { - b.logger.WarnContext(ctx, "Failed deleting expired key", "key", key, "error", err) - } - return nil, trace.NotFound("%q is not found", key.String()) + // item expiry is stored as integer seconds since the unix epoch and so are + // the time filters in queries, so we replicate the same behavior here by + // comparing the expiry against the current clock truncated to the second + if r.Expires != nil && time.Unix(*r.Expires, 0).Before(now) { + return nil, trace.NotFound("%+q is not found", key.String()) } return &r, nil } @@ -1299,17 +1281,13 @@ func convertError(err error) error { return err } -func fullPathToAttributeValueMap(fullPath string) map[string]types.AttributeValue { +func keyToAttributeValueMap(key backend.Key) map[string]types.AttributeValue { return map[string]types.AttributeValue{ hashKeyKey: &types.AttributeValueMemberS{Value: hashKey}, - fullPathKey: &types.AttributeValueMemberS{Value: fullPath}, + fullPathKey: &types.AttributeValueMemberS{Value: prependPrefix(key)}, } } -func keyToAttributeValueMap(key backend.Key) map[string]types.AttributeValue { - return fullPathToAttributeValueMap(prependPrefix(key)) -} - func timeToAttributeValue(t time.Time) types.AttributeValue { return &types.AttributeValueMemberN{ Value: strconv.FormatInt(t.Unix(), 10), diff --git a/lib/backend/dynamo/dynamodbbk_test.go b/lib/backend/dynamo/dynamodbbk_test.go index 1169ace2d28..75757b92bd0 100644 --- a/lib/backend/dynamo/dynamodbbk_test.go +++ b/lib/backend/dynamo/dynamodbbk_test.go @@ -415,9 +415,7 @@ func TestDynamoDB(t *testing.T) { if err != nil { return nil, nil, trace.Wrap(err) } - clock := clockwork.NewFakeClockAt(time.Now()) - uut.clock = clock - return uut, clock, nil + return uut, test.BlockingFakeClock{Clock: clockwork.NewRealClock()}, nil } test.RunBackendComplianceSuite(t, newBackend) diff --git a/lib/backend/dynamo/shards.go b/lib/backend/dynamo/shards.go index 8e63bbe71d6..4256c2bf327 100644 --- a/lib/backend/dynamo/shards.go +++ b/lib/backend/dynamo/shards.go @@ -394,7 +394,7 @@ func toEvent(rec streamtypes.Record) (*backend.Event, error) { } var expires time.Time if r.Expires != nil { - expires = time.Unix(*r.Expires, 0) + expires = time.Unix(*r.Expires, 0).UTC() } return &backend.Event{ Type: op,