Skip to content

Commit 5536f4e

Browse files
msvechlamsvechla
committed
add continuous backup support (#1790)
1 parent 148510a commit 5536f4e

File tree

3 files changed

+173
-28
lines changed

3 files changed

+173
-28
lines changed

‎generator.yaml

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,10 @@ resources:
2323
from:
2424
operation: UpdateTimeToLive
2525
path: TimeToLiveSpecification
26+
ContinuousBackups:
27+
from:
28+
operation: UpdateContinuousBackups
29+
path: PointInTimeRecoverySpecification
2630
AttributeDefinitions:
2731
compare:
2832
is_ignored: true
@@ -134,4 +138,4 @@ resources:
134138
type: string
135139
- name: STATUS
136140
json_path: .status.backupStatus
137-
type: string
141+
type: string

‎pkg/resource/table/hooks.go

Lines changed: 85 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -32,20 +32,30 @@ import (
3232
)
3333

3434
var (
35-
ErrTableDeleting = fmt.Errorf("Table in '%v' state, cannot be modified or deleted", svcsdk.TableStatusDeleting)
36-
ErrTableCreating = fmt.Errorf("Table in '%v' state, cannot be modified or deleted", svcsdk.TableStatusCreating)
37-
ErrTableUpdating = fmt.Errorf("Table in '%v' state, cannot be modified or deleted", svcsdk.TableStatusUpdating)
38-
ErrTableGSIsUpdating = fmt.Errorf("Table GSIs in '%v' state, cannot be modified or deleted", svcsdk.IndexStatusCreating)
35+
ErrTableDeleting = fmt.Errorf(
36+
"Table in '%v' state, cannot be modified or deleted",
37+
svcsdk.TableStatusDeleting,
38+
)
39+
ErrTableCreating = fmt.Errorf(
40+
"Table in '%v' state, cannot be modified or deleted",
41+
svcsdk.TableStatusCreating,
42+
)
43+
ErrTableUpdating = fmt.Errorf(
44+
"Table in '%v' state, cannot be modified or deleted",
45+
svcsdk.TableStatusUpdating,
46+
)
47+
ErrTableGSIsUpdating = fmt.Errorf(
48+
"Table GSIs in '%v' state, cannot be modified or deleted",
49+
svcsdk.IndexStatusCreating,
50+
)
3951
)
4052

41-
var (
42-
// TerminalStatuses are the status strings that are terminal states for a
43-
// DynamoDB table
44-
TerminalStatuses = []v1alpha1.TableStatus_SDK{
45-
v1alpha1.TableStatus_SDK_ARCHIVING,
46-
v1alpha1.TableStatus_SDK_DELETING,
47-
}
48-
)
53+
// TerminalStatuses are the status strings that are terminal states for a
54+
// DynamoDB table
55+
var TerminalStatuses = []v1alpha1.TableStatus_SDK{
56+
v1alpha1.TableStatus_SDK_ARCHIVING,
57+
v1alpha1.TableStatus_SDK_DELETING,
58+
}
4959

5060
var DefaultTTLEnabledValue = false
5161

@@ -124,7 +134,10 @@ func (rm *resourceManager) customUpdateTable(
124134
defer func(err error) { exit(err) }(err)
125135

126136
if immutableFieldChanges := rm.getImmutableFieldChanges(delta); len(immutableFieldChanges) > 0 {
127-
msg := fmt.Sprintf("Immutable Spec fields have been modified: %s", strings.Join(immutableFieldChanges, ","))
137+
msg := fmt.Sprintf(
138+
"Immutable Spec fields have been modified: %s",
139+
strings.Join(immutableFieldChanges, ","),
140+
)
128141
return nil, ackerr.NewTerminalError(fmt.Errorf(msg))
129142
}
130143

@@ -187,6 +200,13 @@ func (rm *resourceManager) customUpdateTable(
187200
}
188201
}
189202

203+
if delta.DifferentAt("Spec.ContinuousBackups") {
204+
err = rm.syncContinuousBackup(ctx, desired)
205+
if err != nil {
206+
return nil, fmt.Errorf("cannot update table %v", err)
207+
}
208+
}
209+
190210
// We want to update fast fields first
191211
// Then attributes
192212
// then GSI
@@ -202,7 +222,8 @@ func (rm *resourceManager) customUpdateTable(
202222
}
203223
case delta.DifferentAt("Spec.GlobalSecondaryIndexes") && delta.DifferentAt("Spec.AttributeDefinitions"):
204224
if err := rm.syncTableGlobalSecondaryIndexes(ctx, latest, desired); err != nil {
205-
if awsErr, ok := ackerr.AWSError(err); ok && awsErr.Code() == "LimitExceededException" {
225+
if awsErr, ok := ackerr.AWSError(err); ok &&
226+
awsErr.Code() == "LimitExceededException" {
206227
return nil, requeueWaitGSIReady
207228
}
208229
return nil, err
@@ -257,13 +278,17 @@ func (rm *resourceManager) newUpdateTablePayload(
257278
input.ProvisionedThroughput = &svcsdk.ProvisionedThroughput{}
258279
if r.ko.Spec.ProvisionedThroughput != nil {
259280
if r.ko.Spec.ProvisionedThroughput.ReadCapacityUnits != nil {
260-
input.ProvisionedThroughput.ReadCapacityUnits = aws.Int64(*r.ko.Spec.ProvisionedThroughput.ReadCapacityUnits)
281+
input.ProvisionedThroughput.ReadCapacityUnits = aws.Int64(
282+
*r.ko.Spec.ProvisionedThroughput.ReadCapacityUnits,
283+
)
261284
} else {
262285
input.ProvisionedThroughput.ReadCapacityUnits = aws.Int64(0)
263286
}
264287

265288
if r.ko.Spec.ProvisionedThroughput.WriteCapacityUnits != nil {
266-
input.ProvisionedThroughput.WriteCapacityUnits = aws.Int64(*r.ko.Spec.ProvisionedThroughput.WriteCapacityUnits)
289+
input.ProvisionedThroughput.WriteCapacityUnits = aws.Int64(
290+
*r.ko.Spec.ProvisionedThroughput.WriteCapacityUnits,
291+
)
267292
} else {
268293
input.ProvisionedThroughput.WriteCapacityUnits = aws.Int64(0)
269294
}
@@ -277,8 +302,11 @@ func (rm *resourceManager) newUpdateTablePayload(
277302
StreamEnabled: aws.Bool(*r.ko.Spec.StreamSpecification.StreamEnabled),
278303
}
279304
// Only set streamViewType when streamSpefication is enabled and streamViewType is non-nil.
280-
if *r.ko.Spec.StreamSpecification.StreamEnabled && r.ko.Spec.StreamSpecification.StreamViewType != nil {
281-
input.StreamSpecification.StreamViewType = aws.String(*r.ko.Spec.StreamSpecification.StreamViewType)
305+
if *r.ko.Spec.StreamSpecification.StreamEnabled &&
306+
r.ko.Spec.StreamSpecification.StreamViewType != nil {
307+
input.StreamSpecification.StreamViewType = aws.String(
308+
*r.ko.Spec.StreamSpecification.StreamViewType,
309+
)
282310
}
283311
} else {
284312
input.StreamSpecification = &svcsdk.StreamSpecification{
@@ -317,7 +345,9 @@ func (rm *resourceManager) syncTableSSESpecification(
317345
input.SSESpecification.SSEType = aws.String(*r.ko.Spec.SSESpecification.SSEType)
318346
}
319347
if r.ko.Spec.SSESpecification.KMSMasterKeyID != nil {
320-
input.SSESpecification.KMSMasterKeyId = aws.String(*r.ko.Spec.SSESpecification.KMSMasterKeyID)
348+
input.SSESpecification.KMSMasterKeyId = aws.String(
349+
*r.ko.Spec.SSESpecification.KMSMasterKeyID,
350+
)
321351
}
322352
}
323353
} else {
@@ -350,13 +380,17 @@ func (rm *resourceManager) syncTableProvisionedThroughput(
350380
}
351381
if r.ko.Spec.ProvisionedThroughput != nil {
352382
if r.ko.Spec.ProvisionedThroughput.ReadCapacityUnits != nil {
353-
input.ProvisionedThroughput.ReadCapacityUnits = aws.Int64(*r.ko.Spec.ProvisionedThroughput.ReadCapacityUnits)
383+
input.ProvisionedThroughput.ReadCapacityUnits = aws.Int64(
384+
*r.ko.Spec.ProvisionedThroughput.ReadCapacityUnits,
385+
)
354386
} else {
355387
input.ProvisionedThroughput.ReadCapacityUnits = aws.Int64(0)
356388
}
357389

358390
if r.ko.Spec.ProvisionedThroughput.WriteCapacityUnits != nil {
359-
input.ProvisionedThroughput.WriteCapacityUnits = aws.Int64(*r.ko.Spec.ProvisionedThroughput.WriteCapacityUnits)
391+
input.ProvisionedThroughput.WriteCapacityUnits = aws.Int64(
392+
*r.ko.Spec.ProvisionedThroughput.WriteCapacityUnits,
393+
)
360394
} else {
361395
input.ProvisionedThroughput.WriteCapacityUnits = aws.Int64(0)
362396
}
@@ -395,6 +429,12 @@ func (rm *resourceManager) setResourceAdditionalFields(
395429
ko.Spec.TimeToLive = ttlSpec
396430
}
397431

432+
if pitrSpec, err := rm.getResourcePointInTimeRecoveryWithContext(ctx, ko.Spec.TableName); err != nil {
433+
return err
434+
} else {
435+
ko.Spec.ContinuousBackups = pitrSpec
436+
}
437+
398438
return nil
399439
}
400440

@@ -403,11 +443,14 @@ func customPreCompare(
403443
a *resource,
404444
b *resource,
405445
) {
406-
407446
if ackcompare.HasNilDifference(a.ko.Spec.SSESpecification, b.ko.Spec.SSESpecification) {
408447
if a.ko.Spec.SSESpecification != nil && b.ko.Spec.SSESpecification == nil {
409448
if *a.ko.Spec.SSESpecification.Enabled {
410-
delta.Add("Spec.SSESpecification", a.ko.Spec.SSESpecification, b.ko.Spec.SSESpecification)
449+
delta.Add(
450+
"Spec.SSESpecification",
451+
a.ko.Spec.SSESpecification,
452+
b.ko.Spec.SSESpecification,
453+
)
411454
}
412455
} else {
413456
delta.Add("Spec.SSESpecification", a.ko.Spec.SSESpecification, b.ko.Spec.SSESpecification)
@@ -447,23 +490,35 @@ func customPreCompare(
447490
}
448491

449492
if len(a.ko.Spec.AttributeDefinitions) != len(b.ko.Spec.AttributeDefinitions) {
450-
delta.Add("Spec.AttributeDefinitions", a.ko.Spec.AttributeDefinitions, b.ko.Spec.AttributeDefinitions)
493+
delta.Add(
494+
"Spec.AttributeDefinitions",
495+
a.ko.Spec.AttributeDefinitions,
496+
b.ko.Spec.AttributeDefinitions,
497+
)
451498
} else if a.ko.Spec.AttributeDefinitions != nil && b.ko.Spec.AttributeDefinitions != nil {
452499
if !equalAttributeDefinitions(a.ko.Spec.AttributeDefinitions, b.ko.Spec.AttributeDefinitions) {
453500
delta.Add("Spec.AttributeDefinitions", a.ko.Spec.AttributeDefinitions, b.ko.Spec.AttributeDefinitions)
454501
}
455502
}
456503

457504
if len(a.ko.Spec.GlobalSecondaryIndexes) != len(b.ko.Spec.GlobalSecondaryIndexes) {
458-
delta.Add("Spec.GlobalSecondaryIndexes", a.ko.Spec.GlobalSecondaryIndexes, b.ko.Spec.GlobalSecondaryIndexes)
505+
delta.Add(
506+
"Spec.GlobalSecondaryIndexes",
507+
a.ko.Spec.GlobalSecondaryIndexes,
508+
b.ko.Spec.GlobalSecondaryIndexes,
509+
)
459510
} else if a.ko.Spec.GlobalSecondaryIndexes != nil && b.ko.Spec.GlobalSecondaryIndexes != nil {
460511
if !equalGlobalSecondaryIndexesArrays(a.ko.Spec.GlobalSecondaryIndexes, b.ko.Spec.GlobalSecondaryIndexes) {
461512
delta.Add("Spec.GlobalSecondaryIndexes", a.ko.Spec.GlobalSecondaryIndexes, b.ko.Spec.GlobalSecondaryIndexes)
462513
}
463514
}
464515

465516
if len(a.ko.Spec.LocalSecondaryIndexes) != len(b.ko.Spec.LocalSecondaryIndexes) {
466-
delta.Add("Spec.LocalSecondaryIndexes", a.ko.Spec.LocalSecondaryIndexes, b.ko.Spec.LocalSecondaryIndexes)
517+
delta.Add(
518+
"Spec.LocalSecondaryIndexes",
519+
a.ko.Spec.LocalSecondaryIndexes,
520+
b.ko.Spec.LocalSecondaryIndexes,
521+
)
467522
} else if a.ko.Spec.LocalSecondaryIndexes != nil && b.ko.Spec.LocalSecondaryIndexes != nil {
468523
if !equalLocalSecondaryIndexesArrays(a.ko.Spec.LocalSecondaryIndexes, b.ko.Spec.LocalSecondaryIndexes) {
469524
delta.Add("Spec.LocalSecondaryIndexes", a.ko.Spec.LocalSecondaryIndexes, b.ko.Spec.LocalSecondaryIndexes)
@@ -614,7 +669,10 @@ func equalLocalSecondaryIndexes(
614669
if !equalStrings(a.Projection.ProjectionType, b.Projection.ProjectionType) {
615670
return false
616671
}
617-
if !ackcompare.SliceStringPEqual(a.Projection.NonKeyAttributes, b.Projection.NonKeyAttributes) {
672+
if !ackcompare.SliceStringPEqual(
673+
a.Projection.NonKeyAttributes,
674+
b.Projection.NonKeyAttributes,
675+
) {
618676
return false
619677
}
620678
}

‎pkg/resource/table/hooks_continuous_backup.go

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,83 @@
1+
// Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License"). You may
4+
// not use this file except in compliance with the License. A copy of the
5+
// License is located at
6+
//
7+
// http://aws.amazon.com/apache2.0/
8+
//
9+
// or in the "license" file accompanying this file. This file is distributed
10+
// on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
11+
// express or implied. See the License for the specific language governing
12+
// permissions and limitations under the License.
13+
14+
package table
15+
16+
import (
17+
"context"
18+
19+
ackrtlog "github.com/aws-controllers-k8s/runtime/pkg/runtime/log"
20+
svcsdk "github.com/aws/aws-sdk-go/service/dynamodb"
21+
22+
"github.com/aws-controllers-k8s/dynamodb-controller/apis/v1alpha1"
23+
)
24+
25+
// syncContinuousBackup syncs the PointInTimeRecoverySpecification of the dynamodb table.
26+
func (rm *resourceManager) syncContinuousBackup(
27+
ctx context.Context,
28+
desired *resource,
29+
) (err error) {
30+
rlog := ackrtlog.FromContext(ctx)
31+
exit := rlog.Trace("rm.syncContinuousBackup")
32+
defer func(err error) { exit(err) }(err)
33+
34+
pitrSpec := &svcsdk.PointInTimeRecoverySpecification{}
35+
if desired.ko.Spec.ContinuousBackups != nil &&
36+
desired.ko.Spec.ContinuousBackups.PointInTimeRecoveryEnabled != nil {
37+
pitrSpec.SetPointInTimeRecoveryEnabled(
38+
*desired.ko.Spec.ContinuousBackups.PointInTimeRecoveryEnabled,
39+
)
40+
}
41+
42+
_, err = rm.sdkapi.UpdateContinuousBackupsWithContext(
43+
ctx,
44+
&svcsdk.UpdateContinuousBackupsInput{
45+
TableName: desired.ko.Spec.TableName,
46+
PointInTimeRecoverySpecification: pitrSpec,
47+
},
48+
)
49+
rm.metrics.RecordAPICall("UPDATE", "UpdateContinuousBackups", err)
50+
return err
51+
}
52+
53+
// getResourcePointInTimeRecoveryWithContext gets the PointInTimeRecoverySpecification of the dynamodb table.
54+
func (rm *resourceManager) getResourcePointInTimeRecoveryWithContext(
55+
ctx context.Context,
56+
tableName *string,
57+
) (*v1alpha1.PointInTimeRecoverySpecification, error) {
58+
var err error
59+
rlog := ackrtlog.FromContext(ctx)
60+
exit := rlog.Trace("rm.getResourcePointInTimeRecoveryWithContext")
61+
defer func(err error) { exit(err) }(err)
62+
63+
res, err := rm.sdkapi.DescribeContinuousBackupsWithContext(
64+
ctx,
65+
&svcsdk.DescribeContinuousBackupsInput{
66+
TableName: tableName,
67+
},
68+
)
69+
70+
rm.metrics.RecordAPICall("GET", "DescribeContinuousBackups", err)
71+
if err != nil {
72+
return nil, err
73+
}
74+
75+
isEnabled := false
76+
if res.ContinuousBackupsDescription != nil {
77+
isEnabled = *res.ContinuousBackupsDescription.ContinuousBackupsStatus == svcsdk.ContinuousBackupsStatusEnabled
78+
}
79+
80+
return &v1alpha1.PointInTimeRecoverySpecification{
81+
PointInTimeRecoveryEnabled: &isEnabled,
82+
}, nil
83+
}

0 commit comments

Comments
 (0)