Skip to content

Commit 4940ca5

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

File tree

3 files changed

+183
-28
lines changed

3 files changed

+183
-28
lines changed

‎generator.yaml‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,11 @@ resources:
2323
from:
2424
operation: UpdateTimeToLive
2525
path: TimeToLiveSpecification
26+
ContinuousBackups:
27+
is_required: false
28+
from:
29+
operation: UpdateContinuousBackups
30+
path: PointInTimeRecoverySpecification
2631
AttributeDefinitions:
2732
compare:
2833
is_ignored: true
@@ -134,4 +139,4 @@ resources:
134139
type: string
135140
- name: STATUS
136141
json_path: .status.backupStatus
137-
type: string
142+
type: string

‎pkg/resource/table/hooks.go‎

Lines changed: 94 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -32,23 +32,36 @@ 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

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+
}
59+
4160
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-
}
61+
DefaultTTLEnabledValue = false
62+
DefaultPITREnabledValue = false
4863
)
4964

50-
var DefaultTTLEnabledValue = false
51-
5265
var (
5366
requeueWaitWhileDeleting = ackrequeue.NeededAfter(
5467
ErrTableDeleting,
@@ -124,7 +137,10 @@ func (rm *resourceManager) customUpdateTable(
124137
defer func(err error) { exit(err) }(err)
125138

126139
if immutableFieldChanges := rm.getImmutableFieldChanges(delta); len(immutableFieldChanges) > 0 {
127-
msg := fmt.Sprintf("Immutable Spec fields have been modified: %s", strings.Join(immutableFieldChanges, ","))
140+
msg := fmt.Sprintf(
141+
"Immutable Spec fields have been modified: %s",
142+
strings.Join(immutableFieldChanges, ","),
143+
)
128144
return nil, ackerr.NewTerminalError(fmt.Errorf(msg))
129145
}
130146

@@ -187,6 +203,13 @@ func (rm *resourceManager) customUpdateTable(
187203
}
188204
}
189205

206+
if delta.DifferentAt("Spec.ContinuousBackups") {
207+
err = rm.syncContinuousBackup(ctx, desired)
208+
if err != nil {
209+
return nil, fmt.Errorf("cannot update table %v", err)
210+
}
211+
}
212+
190213
// We want to update fast fields first
191214
// Then attributes
192215
// then GSI
@@ -202,7 +225,8 @@ func (rm *resourceManager) customUpdateTable(
202225
}
203226
case delta.DifferentAt("Spec.GlobalSecondaryIndexes") && delta.DifferentAt("Spec.AttributeDefinitions"):
204227
if err := rm.syncTableGlobalSecondaryIndexes(ctx, latest, desired); err != nil {
205-
if awsErr, ok := ackerr.AWSError(err); ok && awsErr.Code() == "LimitExceededException" {
228+
if awsErr, ok := ackerr.AWSError(err); ok &&
229+
awsErr.Code() == "LimitExceededException" {
206230
return nil, requeueWaitGSIReady
207231
}
208232
return nil, err
@@ -257,13 +281,17 @@ func (rm *resourceManager) newUpdateTablePayload(
257281
input.ProvisionedThroughput = &svcsdk.ProvisionedThroughput{}
258282
if r.ko.Spec.ProvisionedThroughput != nil {
259283
if r.ko.Spec.ProvisionedThroughput.ReadCapacityUnits != nil {
260-
input.ProvisionedThroughput.ReadCapacityUnits = aws.Int64(*r.ko.Spec.ProvisionedThroughput.ReadCapacityUnits)
284+
input.ProvisionedThroughput.ReadCapacityUnits = aws.Int64(
285+
*r.ko.Spec.ProvisionedThroughput.ReadCapacityUnits,
286+
)
261287
} else {
262288
input.ProvisionedThroughput.ReadCapacityUnits = aws.Int64(0)
263289
}
264290

265291
if r.ko.Spec.ProvisionedThroughput.WriteCapacityUnits != nil {
266-
input.ProvisionedThroughput.WriteCapacityUnits = aws.Int64(*r.ko.Spec.ProvisionedThroughput.WriteCapacityUnits)
292+
input.ProvisionedThroughput.WriteCapacityUnits = aws.Int64(
293+
*r.ko.Spec.ProvisionedThroughput.WriteCapacityUnits,
294+
)
267295
} else {
268296
input.ProvisionedThroughput.WriteCapacityUnits = aws.Int64(0)
269297
}
@@ -277,8 +305,11 @@ func (rm *resourceManager) newUpdateTablePayload(
277305
StreamEnabled: aws.Bool(*r.ko.Spec.StreamSpecification.StreamEnabled),
278306
}
279307
// 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)
308+
if *r.ko.Spec.StreamSpecification.StreamEnabled &&
309+
r.ko.Spec.StreamSpecification.StreamViewType != nil {
310+
input.StreamSpecification.StreamViewType = aws.String(
311+
*r.ko.Spec.StreamSpecification.StreamViewType,
312+
)
282313
}
283314
} else {
284315
input.StreamSpecification = &svcsdk.StreamSpecification{
@@ -317,7 +348,9 @@ func (rm *resourceManager) syncTableSSESpecification(
317348
input.SSESpecification.SSEType = aws.String(*r.ko.Spec.SSESpecification.SSEType)
318349
}
319350
if r.ko.Spec.SSESpecification.KMSMasterKeyID != nil {
320-
input.SSESpecification.KMSMasterKeyId = aws.String(*r.ko.Spec.SSESpecification.KMSMasterKeyID)
351+
input.SSESpecification.KMSMasterKeyId = aws.String(
352+
*r.ko.Spec.SSESpecification.KMSMasterKeyID,
353+
)
321354
}
322355
}
323356
} else {
@@ -350,13 +383,17 @@ func (rm *resourceManager) syncTableProvisionedThroughput(
350383
}
351384
if r.ko.Spec.ProvisionedThroughput != nil {
352385
if r.ko.Spec.ProvisionedThroughput.ReadCapacityUnits != nil {
353-
input.ProvisionedThroughput.ReadCapacityUnits = aws.Int64(*r.ko.Spec.ProvisionedThroughput.ReadCapacityUnits)
386+
input.ProvisionedThroughput.ReadCapacityUnits = aws.Int64(
387+
*r.ko.Spec.ProvisionedThroughput.ReadCapacityUnits,
388+
)
354389
} else {
355390
input.ProvisionedThroughput.ReadCapacityUnits = aws.Int64(0)
356391
}
357392

358393
if r.ko.Spec.ProvisionedThroughput.WriteCapacityUnits != nil {
359-
input.ProvisionedThroughput.WriteCapacityUnits = aws.Int64(*r.ko.Spec.ProvisionedThroughput.WriteCapacityUnits)
394+
input.ProvisionedThroughput.WriteCapacityUnits = aws.Int64(
395+
*r.ko.Spec.ProvisionedThroughput.WriteCapacityUnits,
396+
)
360397
} else {
361398
input.ProvisionedThroughput.WriteCapacityUnits = aws.Int64(0)
362399
}
@@ -395,6 +432,12 @@ func (rm *resourceManager) setResourceAdditionalFields(
395432
ko.Spec.TimeToLive = ttlSpec
396433
}
397434

435+
if pitrSpec, err := rm.getResourcePointInTimeRecoveryWithContext(ctx, ko.Spec.TableName); err != nil {
436+
return err
437+
} else {
438+
ko.Spec.ContinuousBackups = pitrSpec
439+
}
440+
398441
return nil
399442
}
400443

@@ -403,11 +446,14 @@ func customPreCompare(
403446
a *resource,
404447
b *resource,
405448
) {
406-
407449
if ackcompare.HasNilDifference(a.ko.Spec.SSESpecification, b.ko.Spec.SSESpecification) {
408450
if a.ko.Spec.SSESpecification != nil && b.ko.Spec.SSESpecification == nil {
409451
if *a.ko.Spec.SSESpecification.Enabled {
410-
delta.Add("Spec.SSESpecification", a.ko.Spec.SSESpecification, b.ko.Spec.SSESpecification)
452+
delta.Add(
453+
"Spec.SSESpecification",
454+
a.ko.Spec.SSESpecification,
455+
b.ko.Spec.SSESpecification,
456+
)
411457
}
412458
} else {
413459
delta.Add("Spec.SSESpecification", a.ko.Spec.SSESpecification, b.ko.Spec.SSESpecification)
@@ -447,23 +493,35 @@ func customPreCompare(
447493
}
448494

449495
if len(a.ko.Spec.AttributeDefinitions) != len(b.ko.Spec.AttributeDefinitions) {
450-
delta.Add("Spec.AttributeDefinitions", a.ko.Spec.AttributeDefinitions, b.ko.Spec.AttributeDefinitions)
496+
delta.Add(
497+
"Spec.AttributeDefinitions",
498+
a.ko.Spec.AttributeDefinitions,
499+
b.ko.Spec.AttributeDefinitions,
500+
)
451501
} else if a.ko.Spec.AttributeDefinitions != nil && b.ko.Spec.AttributeDefinitions != nil {
452502
if !equalAttributeDefinitions(a.ko.Spec.AttributeDefinitions, b.ko.Spec.AttributeDefinitions) {
453503
delta.Add("Spec.AttributeDefinitions", a.ko.Spec.AttributeDefinitions, b.ko.Spec.AttributeDefinitions)
454504
}
455505
}
456506

457507
if len(a.ko.Spec.GlobalSecondaryIndexes) != len(b.ko.Spec.GlobalSecondaryIndexes) {
458-
delta.Add("Spec.GlobalSecondaryIndexes", a.ko.Spec.GlobalSecondaryIndexes, b.ko.Spec.GlobalSecondaryIndexes)
508+
delta.Add(
509+
"Spec.GlobalSecondaryIndexes",
510+
a.ko.Spec.GlobalSecondaryIndexes,
511+
b.ko.Spec.GlobalSecondaryIndexes,
512+
)
459513
} else if a.ko.Spec.GlobalSecondaryIndexes != nil && b.ko.Spec.GlobalSecondaryIndexes != nil {
460514
if !equalGlobalSecondaryIndexesArrays(a.ko.Spec.GlobalSecondaryIndexes, b.ko.Spec.GlobalSecondaryIndexes) {
461515
delta.Add("Spec.GlobalSecondaryIndexes", a.ko.Spec.GlobalSecondaryIndexes, b.ko.Spec.GlobalSecondaryIndexes)
462516
}
463517
}
464518

465519
if len(a.ko.Spec.LocalSecondaryIndexes) != len(b.ko.Spec.LocalSecondaryIndexes) {
466-
delta.Add("Spec.LocalSecondaryIndexes", a.ko.Spec.LocalSecondaryIndexes, b.ko.Spec.LocalSecondaryIndexes)
520+
delta.Add(
521+
"Spec.LocalSecondaryIndexes",
522+
a.ko.Spec.LocalSecondaryIndexes,
523+
b.ko.Spec.LocalSecondaryIndexes,
524+
)
467525
} else if a.ko.Spec.LocalSecondaryIndexes != nil && b.ko.Spec.LocalSecondaryIndexes != nil {
468526
if !equalLocalSecondaryIndexesArrays(a.ko.Spec.LocalSecondaryIndexes, b.ko.Spec.LocalSecondaryIndexes) {
469527
delta.Add("Spec.LocalSecondaryIndexes", a.ko.Spec.LocalSecondaryIndexes, b.ko.Spec.LocalSecondaryIndexes)
@@ -496,6 +554,12 @@ func customPreCompare(
496554
Enabled: &DefaultTTLEnabledValue,
497555
}
498556
}
557+
if a.ko.Spec.ContinuousBackups == nil && b.ko.Spec.ContinuousBackups != nil &&
558+
b.ko.Spec.ContinuousBackups.PointInTimeRecoveryEnabled != nil {
559+
a.ko.Spec.ContinuousBackups = &v1alpha1.PointInTimeRecoverySpecification{
560+
PointInTimeRecoveryEnabled: &DefaultPITREnabledValue,
561+
}
562+
}
499563
}
500564

501565
// equalAttributeDefinitions return whether two AttributeDefinition arrays are equal or not.
@@ -614,7 +678,10 @@ func equalLocalSecondaryIndexes(
614678
if !equalStrings(a.Projection.ProjectionType, b.Projection.ProjectionType) {
615679
return false
616680
}
617-
if !ackcompare.SliceStringPEqual(a.Projection.NonKeyAttributes, b.Projection.NonKeyAttributes) {
681+
if !ackcompare.SliceStringPEqual(
682+
a.Projection.NonKeyAttributes,
683+
b.Projection.NonKeyAttributes,
684+
) {
618685
return false
619686
}
620687
}

‎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.PointInTimeRecoveryDescription.PointInTimeRecoveryStatus == svcsdk.PointInTimeRecoveryStatusEnabled
78+
}
79+
80+
return &v1alpha1.PointInTimeRecoverySpecification{
81+
PointInTimeRecoveryEnabled: &isEnabled,
82+
}, nil
83+
}

0 commit comments

Comments
 (0)