From e6dacedfa09176c8d717d89d890ebbab240f0e30 Mon Sep 17 00:00:00 2001 From: Vitor Enes Date: Wed, 17 Aug 2022 11:50:18 +0100 Subject: [PATCH] Add `TurnOnTimeToLive` and `TurnOnStreams` functions to `dynamo/configure.go` (#15557) --- lib/backend/dynamo/configure.go | 41 +++++++++++++++++++++++++ lib/backend/dynamo/dynamodbbk.go | 4 +-- lib/backend/dynamo/shards.go | 41 ------------------------- lib/events/dynamoevents/dynamoevents.go | 23 +------------- 4 files changed, 44 insertions(+), 65 deletions(-) diff --git a/lib/backend/dynamo/configure.go b/lib/backend/dynamo/configure.go index d7695cdc584..40a72134876 100644 --- a/lib/backend/dynamo/configure.go +++ b/lib/backend/dynamo/configure.go @@ -150,3 +150,44 @@ func getWriteScalingPolicyName(resourceID string) string { func getReadScalingPolicyName(resourceID string) string { return fmt.Sprintf("%s-read-target-tracking-scaling-policy", strings.TrimPrefix(resourceID, "table/")) } + +func TurnOnTimeToLive(ctx context.Context, svc dynamodbiface.DynamoDBAPI, tableName string, ttlKey string) error { + status, err := svc.DescribeTimeToLiveWithContext(ctx, &dynamodb.DescribeTimeToLiveInput{ + TableName: aws.String(tableName), + }) + if err != nil { + return convertError(err) + } + switch aws.StringValue(status.TimeToLiveDescription.TimeToLiveStatus) { + case dynamodb.TimeToLiveStatusEnabled, dynamodb.TimeToLiveStatusEnabling: + return nil + } + _, err = svc.UpdateTimeToLiveWithContext(ctx, &dynamodb.UpdateTimeToLiveInput{ + TableName: aws.String(tableName), + TimeToLiveSpecification: &dynamodb.TimeToLiveSpecification{ + AttributeName: aws.String(ttlKey), + Enabled: aws.Bool(true), + }, + }) + return convertError(err) +} + +func TurnOnStreams(ctx context.Context, svc dynamodbiface.DynamoDBAPI, tableName string) error { + status, err := svc.DescribeTableWithContext(ctx, &dynamodb.DescribeTableInput{ + TableName: aws.String(tableName), + }) + if err != nil { + return convertError(err) + } + if status.Table.StreamSpecification != nil && aws.BoolValue(status.Table.StreamSpecification.StreamEnabled) { + return nil + } + _, err = svc.UpdateTableWithContext(ctx, &dynamodb.UpdateTableInput{ + TableName: aws.String(tableName), + StreamSpecification: &dynamodb.StreamSpecification{ + StreamEnabled: aws.Bool(true), + StreamViewType: aws.String(dynamodb.StreamViewTypeNewImage), + }, + }) + return convertError(err) +} diff --git a/lib/backend/dynamo/dynamodbbk.go b/lib/backend/dynamo/dynamodbbk.go index b26beb28eb0..3f9e7a5a5e2 100644 --- a/lib/backend/dynamo/dynamodbbk.go +++ b/lib/backend/dynamo/dynamodbbk.go @@ -289,13 +289,13 @@ func New(ctx context.Context, params backend.Params) (*Backend, error) { } // Enable TTL on table. - err = b.turnOnTimeToLive(ctx) + err = TurnOnTimeToLive(ctx, b.svc, b.TableName, ttlKey) if err != nil { return nil, trace.Wrap(err) } // Turn on DynamoDB streams, needed to implement events. - err = b.turnOnStreams(ctx) + err = TurnOnStreams(ctx, b.svc, b.TableName) if err != nil { return nil, trace.Wrap(err) } diff --git a/lib/backend/dynamo/shards.go b/lib/backend/dynamo/shards.go index 15c56b3df77..40d8e914a70 100644 --- a/lib/backend/dynamo/shards.go +++ b/lib/backend/dynamo/shards.go @@ -357,44 +357,3 @@ func (b *Backend) asyncPollShard(ctx context.Context, streamArn *string, shard * }() err = b.pollShard(ctx, streamArn, shard, eventsC, initC) } - -func (b *Backend) turnOnTimeToLive(ctx context.Context) error { - status, err := b.svc.DescribeTimeToLiveWithContext(ctx, &dynamodb.DescribeTimeToLiveInput{ - TableName: aws.String(b.TableName), - }) - if err != nil { - return trace.Wrap(convertError(err)) - } - switch aws.StringValue(status.TimeToLiveDescription.TimeToLiveStatus) { - case dynamodb.TimeToLiveStatusEnabled, dynamodb.TimeToLiveStatusEnabling: - return nil - } - _, err = b.svc.UpdateTimeToLiveWithContext(ctx, &dynamodb.UpdateTimeToLiveInput{ - TableName: aws.String(b.TableName), - TimeToLiveSpecification: &dynamodb.TimeToLiveSpecification{ - AttributeName: aws.String(ttlKey), - Enabled: aws.Bool(true), - }, - }) - return convertError(err) -} - -func (b *Backend) turnOnStreams(ctx context.Context) error { - status, err := b.svc.DescribeTableWithContext(ctx, &dynamodb.DescribeTableInput{ - TableName: aws.String(b.TableName), - }) - if err != nil { - return trace.Wrap(convertError(err)) - } - if status.Table.StreamSpecification != nil && aws.BoolValue(status.Table.StreamSpecification.StreamEnabled) { - return nil - } - _, err = b.svc.UpdateTableWithContext(ctx, &dynamodb.UpdateTableInput{ - TableName: aws.String(b.TableName), - StreamSpecification: &dynamodb.StreamSpecification{ - StreamEnabled: aws.Bool(true), - StreamViewType: aws.String(dynamodb.StreamViewTypeNewImage), - }, - }) - return convertError(err) -} diff --git a/lib/events/dynamoevents/dynamoevents.go b/lib/events/dynamoevents/dynamoevents.go index 4036f5004ad..0c9976bc8e3 100644 --- a/lib/events/dynamoevents/dynamoevents.go +++ b/lib/events/dynamoevents/dynamoevents.go @@ -314,7 +314,7 @@ func New(ctx context.Context, cfg Config, backend backend.Backend) (*Log, error) if err != nil { return nil, trace.Wrap(err) } - err = b.turnOnTimeToLive(ctx) + err = dynamo.TurnOnTimeToLive(ctx, b.svc, b.Tablename, keyExpires) if err != nil { return nil, trace.Wrap(err) } @@ -859,27 +859,6 @@ func fromWhereExpr(cond *types.WhereExpr, params *condFilterParams) (string, err return "", trace.BadParameter("failed to convert WhereExpr %q to DynamoDB filter expression", cond) } -func (l *Log) turnOnTimeToLive(ctx context.Context) error { - status, err := l.svc.DescribeTimeToLiveWithContext(ctx, &dynamodb.DescribeTimeToLiveInput{ - TableName: aws.String(l.Tablename), - }) - if err != nil { - return trace.Wrap(convertError(err)) - } - switch aws.StringValue(status.TimeToLiveDescription.TimeToLiveStatus) { - case dynamodb.TimeToLiveStatusEnabled, dynamodb.TimeToLiveStatusEnabling: - return nil - } - _, err = l.svc.UpdateTimeToLiveWithContext(ctx, &dynamodb.UpdateTimeToLiveInput{ - TableName: aws.String(l.Tablename), - TimeToLiveSpecification: &dynamodb.TimeToLiveSpecification{ - AttributeName: aws.String(keyExpires), - Enabled: aws.Bool(true), - }, - }) - return convertError(err) -} - // getTableStatus checks if a given table exists func (l *Log) getTableStatus(ctx context.Context, tableName string) (tableStatus, error) { _, err := l.svc.DescribeTableWithContext(ctx, &dynamodb.DescribeTableInput{