opensource/teleport
3
0
Fork 0
mirror of https://github.com/gravitational/teleport.git synced 2026-09-24 16:17:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add TurnOnTimeToLive and TurnOnStreams functions to dynamo/configure.go (#15557)

Browse Source
This commit is contained in:
Vitor Enes
2022-08-17 10:50:18 +00:00
committed by GitHub
parent 7629c635c1
commit e6dacedfa0
4 changed files with 44 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/backend/dynamo/configure.go
+41
View File
@@ -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)
}
lib/backend/dynamo/dynamodbbk.go
+2 -2
View File
@@ -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)
}
lib/backend/dynamo/shards.go
-41
View File
@@ -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)
}
lib/events/dynamoevents/dynamoevents.go
+1 -22
View File
@@ -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{