Skip to content

Commit 8440de9

Browse files
michaelhtmmichaelhtm
committed
feat: add settable retention period and encryption to Stream
Retention period and KMS encryption are not part of CreateStream, so they are exposed as settable Spec inputs (desiredRetentionPeriodHours, desiredEncryptionType, encryptionKeyARN) applied via Increase/Decrease StreamRetentionPeriod and Start/StopStreamEncryption in the update path. The observed values remain in Status (retentionPeriodHours, encryptionType, keyID). A KMS key ARN is required so the observed key matches the desired value and the stream is not re-encrypted on every reconcile. Resolves aws-controllers-k8s/community#2420
1 parent d99b82d commit 8440de9

22 files changed

Lines changed: 550 additions & 19 deletions

‎apis/v1alpha1/ack-generate-metadata.yaml‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,13 @@
11
ack_generate_info:
2-
build_date: "2026-06-22T22:54:02Z"
2+
build_date: "2026-06-24T20:57:37Z"
33
build_hash: 2ae5d2cfadaa2a10b2ccb9e73a111b2a91c36642
44
go_version: go1.26.4
55
version: v0.60.0
6-
api_directory_checksum: f0678a236e7f3f6ad06bc5de36a7678d0084aeac
6+
api_directory_checksum: cbad1524b0c12ec5aca9ebfd518ae60d32fec536
77
api_version: v1alpha1
88
aws_sdk_go_version: v1.32.6
99
generator_config_info:
10-
file_checksum: cbcc5b2cf42cc3de0ffa32047a95d76d5590cec6
10+
file_checksum: 0d92d9440780ca6d3d3e13fc0210abf31c55d0c1
1111
original_file_name: generator.yaml
1212
last_modification:
1313
reason: API generation

‎apis/v1alpha1/generator.yaml‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,8 @@ resources:
3535
output_fields:
3636
TargetShardCount: ShardCount
3737
hooks:
38+
delta_pre_compare:
39+
template_path: hooks/stream/delta_pre_compare.go.tpl
3840
sdk_read_one_post_set_output:
3941
template_path: hooks/stream/sdk_read_one_post_set_output.go.tpl
4042
sdk_update_post_set_output:
@@ -54,6 +56,21 @@ resources:
5456
path: Tags
5557
Name:
5658
is_primary_key: true
59+
# Settable Spec inputs. None of these are part of CreateStream; they are
60+
# applied via separate API calls in the update path (Increase/Decrease
61+
# StreamRetentionPeriod and Start/StopStreamEncryption). sdkFind mirrors
62+
# the observed AWS values back into these fields so that auto-generated
63+
# delta detection compares desired vs actual state correctly.
64+
DesiredRetentionPeriodHours:
65+
type: int64
66+
DesiredEncryptionType:
67+
type: string
68+
# EncryptionKeyARN is the customer-managed KMS key used when
69+
# DesiredEncryptionType is KMS. A key *ARN* is required (aliases are
70+
# rejected in the update hook) so the observed Status.KeyID matches the
71+
# desired value and the controller does not re-encrypt on every reconcile.
72+
EncryptionKeyARN:
73+
type: string
5774
# NOTE(jaypipes): It is necessary to hand-roll all these fields because
5875
# the CreateStream output shape has no members at all...
5976
ConsumerCount:

‎apis/v1alpha1/stream.go‎

Lines changed: 18 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎apis/v1alpha1/zz_generated.deepcopy.go‎

Lines changed: 15 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎config/crd/bases/kinesis.services.k8s.aws_streams.yaml‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,31 @@ spec:
3939
spec:
4040
description: StreamSpec defines the desired state of Stream.
4141
properties:
42+
desiredEncryptionType:
43+
description: |-
44+
The desired server-side encryption type for the stream. Valid values
45+
are NONE and KMS. When set to KMS, EncryptionKeyARN must also be
46+
provided. The controller calls StartStreamEncryption or
47+
StopStreamEncryption as needed. The observed encryption type is
48+
reported in Status.EncryptionType.
49+
type: string
50+
desiredRetentionPeriodHours:
51+
description: |-
52+
The desired retention period for the stream, in hours. Valid range is
53+
24 to 8760 hours (365 days). When set, the controller calls
54+
IncreaseStreamRetentionPeriod or DecreaseStreamRetentionPeriod as
55+
needed. The observed retention period is reported in
56+
Status.RetentionPeriodHours.
57+
format: int64
58+
type: integer
59+
encryptionKeyARN:
60+
description: |-
61+
The ARN of the customer-managed KMS key to use for server-side
62+
encryption. Required when DesiredEncryptionType is KMS. A full key
63+
ARN must be supplied (aliases and bare key IDs are rejected) so that
64+
the observed Status.KeyID matches the desired value and the
65+
controller does not re-encrypt on every reconcile.
66+
type: string
4267
name:
4368
description: |-
4469
A name to identify the stream. The stream name is scoped to the Amazon Web

‎documentation.yaml‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
resources:
2+
Stream:
3+
fields:
4+
DesiredRetentionPeriodHours:
5+
override: |
6+
The desired retention period for the stream, in hours. Valid range is
7+
24 to 8760 hours (365 days). When set, the controller calls
8+
IncreaseStreamRetentionPeriod or DecreaseStreamRetentionPeriod as
9+
needed. The observed retention period is reported in
10+
Status.RetentionPeriodHours.
11+
DesiredEncryptionType:
12+
override: |
13+
The desired server-side encryption type for the stream. Valid values
14+
are NONE and KMS. When set to KMS, EncryptionKeyARN must also be
15+
provided. The controller calls StartStreamEncryption or
16+
StopStreamEncryption as needed. The observed encryption type is
17+
reported in Status.EncryptionType.
18+
EncryptionKeyARN:
19+
override: |
20+
The ARN of the customer-managed KMS key to use for server-side
21+
encryption. Required when DesiredEncryptionType is KMS. A full key
22+
ARN must be supplied (aliases and bare key IDs are rejected) so that
23+
the observed Status.KeyID matches the desired value and the
24+
controller does not re-encrypt on every reconcile.

‎generator.yaml‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,8 @@ resources:
3535
output_fields:
3636
TargetShardCount: ShardCount
3737
hooks:
38+
delta_pre_compare:
39+
template_path: hooks/stream/delta_pre_compare.go.tpl
3840
sdk_read_one_post_set_output:
3941
template_path: hooks/stream/sdk_read_one_post_set_output.go.tpl
4042
sdk_update_post_set_output:
@@ -54,6 +56,21 @@ resources:
5456
path: Tags
5557
Name:
5658
is_primary_key: true
59+
# Settable Spec inputs. None of these are part of CreateStream; they are
60+
# applied via separate API calls in the update path (Increase/Decrease
61+
# StreamRetentionPeriod and Start/StopStreamEncryption). sdkFind mirrors
62+
# the observed AWS values back into these fields so that auto-generated
63+
# delta detection compares desired vs actual state correctly.
64+
DesiredRetentionPeriodHours:
65+
type: int64
66+
DesiredEncryptionType:
67+
type: string
68+
# EncryptionKeyARN is the customer-managed KMS key used when
69+
# DesiredEncryptionType is KMS. A key *ARN* is required (aliases are
70+
# rejected in the update hook) so the observed Status.KeyID matches the
71+
# desired value and the controller does not re-encrypt on every reconcile.
72+
EncryptionKeyARN:
73+
type: string
5774
# NOTE(jaypipes): It is necessary to hand-roll all these fields because
5875
# the CreateStream output shape has no members at all...
5976
ConsumerCount:

‎helm/crds/kinesis.services.k8s.aws_streams.yaml‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,31 @@ spec:
3939
spec:
4040
description: StreamSpec defines the desired state of Stream.
4141
properties:
42+
desiredEncryptionType:
43+
description: |-
44+
The desired server-side encryption type for the stream. Valid values
45+
are NONE and KMS. When set to KMS, EncryptionKeyARN must also be
46+
provided. The controller calls StartStreamEncryption or
47+
StopStreamEncryption as needed. The observed encryption type is
48+
reported in Status.EncryptionType.
49+
type: string
50+
desiredRetentionPeriodHours:
51+
description: |-
52+
The desired retention period for the stream, in hours. Valid range is
53+
24 to 8760 hours (365 days). When set, the controller calls
54+
IncreaseStreamRetentionPeriod or DecreaseStreamRetentionPeriod as
55+
needed. The observed retention period is reported in
56+
Status.RetentionPeriodHours.
57+
format: int64
58+
type: integer
59+
encryptionKeyARN:
60+
description: |-
61+
The ARN of the customer-managed KMS key to use for server-side
62+
encryption. Required when DesiredEncryptionType is KMS. A full key
63+
ARN must be supplied (aliases and bare key IDs are rejected) so that
64+
the observed Status.KeyID matches the desired value and the
65+
controller does not re-encrypt on every reconcile.
66+
type: string
4267
name:
4368
description: |-
4469
A name to identify the stream. The stream name is scoped to the Amazon Web

‎pkg/resource/stream/delta.go‎

Lines changed: 29 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎pkg/resource/stream/hooks.go‎

Lines changed: 105 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,12 @@ package stream
1515

1616
import (
1717
"context"
18+
"fmt"
19+
"strings"
1820

21+
ackerr "github.com/aws-controllers-k8s/runtime/pkg/errors"
22+
ackrtlog "github.com/aws-controllers-k8s/runtime/pkg/runtime/log"
23+
svcsdk "github.com/aws/aws-sdk-go-v2/service/kinesis"
1924
svcsdktypes "github.com/aws/aws-sdk-go-v2/service/kinesis/types"
2025

2126
util "github.com/aws-controllers-k8s/kinesis-controller/pkg/resource/tags"
@@ -40,3 +45,103 @@ func (rm *resourceManager) syncTags(
4045
) (err error) {
4146
return util.SyncResourceTags(ctx, rm.sdkapi, rm.metrics, latest.ko.Spec.Name, desired.ko.Spec.Tags, latest.ko.Spec.Tags)
4247
}
48+
49+
// syncRetentionPeriod brings the stream's retention period in line with the
50+
// desired Spec.DesiredRetentionPeriodHours value. The Kinesis API exposes two
51+
// directional operations, so we pick Increase or Decrease based on the observed
52+
// retention period.
53+
func (rm *resourceManager) syncRetentionPeriod(
54+
ctx context.Context,
55+
desired *resource,
56+
latest *resource,
57+
) (err error) {
58+
rlog := ackrtlog.FromContext(ctx)
59+
exit := rlog.Trace("rm.syncRetentionPeriod")
60+
defer func() { exit(err) }()
61+
62+
desiredHours := int32(*desired.ko.Spec.DesiredRetentionPeriodHours)
63+
var observedHours *int32
64+
if latest.ko.Status.RetentionPeriodHours != nil {
65+
v := int32(*latest.ko.Status.RetentionPeriodHours)
66+
observedHours = &v
67+
}
68+
69+
if observedHours != nil && desiredHours > *observedHours {
70+
_, err = rm.sdkapi.IncreaseStreamRetentionPeriod(
71+
ctx,
72+
&svcsdk.IncreaseStreamRetentionPeriodInput{
73+
StreamName: latest.ko.Spec.Name,
74+
RetentionPeriodHours: &desiredHours,
75+
},
76+
)
77+
rm.metrics.RecordAPICall("UPDATE", "IncreaseStreamRetentionPeriod", err)
78+
return err
79+
}
80+
if observedHours == nil || desiredHours < *observedHours {
81+
_, err = rm.sdkapi.DecreaseStreamRetentionPeriod(
82+
ctx,
83+
&svcsdk.DecreaseStreamRetentionPeriodInput{
84+
StreamName: latest.ko.Spec.Name,
85+
RetentionPeriodHours: &desiredHours,
86+
},
87+
)
88+
rm.metrics.RecordAPICall("UPDATE", "DecreaseStreamRetentionPeriod", err)
89+
return err
90+
}
91+
return nil
92+
}
93+
94+
// syncEncryption brings the stream's server-side encryption in line with the
95+
// desired Spec.DesiredEncryptionType and Spec.EncryptionKeyARN values. KMS
96+
// encryption requires a customer-managed key ARN; aliases are rejected so that
97+
// the observed Status.KeyID matches the desired value and the controller does
98+
// not re-encrypt on every reconcile.
99+
func (rm *resourceManager) syncEncryption(
100+
ctx context.Context,
101+
desired *resource,
102+
latest *resource,
103+
) (err error) {
104+
rlog := ackrtlog.FromContext(ctx)
105+
exit := rlog.Trace("rm.syncEncryption")
106+
defer func() { exit(err) }()
107+
108+
if desired.ko.Spec.DesiredEncryptionType != nil &&
109+
*desired.ko.Spec.DesiredEncryptionType == string(svcsdktypes.EncryptionTypeKms) {
110+
if desired.ko.Spec.EncryptionKeyARN == nil {
111+
return ackerr.NewTerminalError(fmt.Errorf(
112+
"encryptionKeyARN is required when desiredEncryptionType is KMS"))
113+
}
114+
if !strings.HasPrefix(*desired.ko.Spec.EncryptionKeyARN, "arn:") {
115+
return ackerr.NewTerminalError(fmt.Errorf(
116+
"encryptionKeyARN must be a KMS key ARN, not an alias or key ID: %q",
117+
*desired.ko.Spec.EncryptionKeyARN))
118+
}
119+
_, err = rm.sdkapi.StartStreamEncryption(
120+
ctx,
121+
&svcsdk.StartStreamEncryptionInput{
122+
StreamName: latest.ko.Spec.Name,
123+
EncryptionType: svcsdktypes.EncryptionTypeKms,
124+
KeyId: desired.ko.Spec.EncryptionKeyARN,
125+
},
126+
)
127+
rm.metrics.RecordAPICall("UPDATE", "StartStreamEncryption", err)
128+
return err
129+
}
130+
131+
// Desired encryption is NONE (or unset). Stop encryption only if the stream
132+
// is currently encrypted with KMS.
133+
if latest.ko.Status.EncryptionType != nil &&
134+
*latest.ko.Status.EncryptionType == string(svcsdktypes.EncryptionTypeKms) {
135+
_, err = rm.sdkapi.StopStreamEncryption(
136+
ctx,
137+
&svcsdk.StopStreamEncryptionInput{
138+
StreamName: latest.ko.Spec.Name,
139+
EncryptionType: svcsdktypes.EncryptionTypeKms,
140+
KeyId: latest.ko.Status.KeyID,
141+
},
142+
)
143+
rm.metrics.RecordAPICall("UPDATE", "StopStreamEncryption", err)
144+
return err
145+
}
146+
return nil
147+
}

0 commit comments

Comments
 (0)