feat(kinesis): support stream consumers

- introduce `StreamConsumer` construct to model `AWS::Kinesis::StreamConsumer` - introduce `addToResourcePolicy` to enable creating/configuring a resource policy for the consumer - introduce `grant` and `grantRead` for granting permissions - leverage `iam.Grant.addToPrincipalOrResource` in `grant` to be able to use `grant` methods cross environments to update the grantee's iam policy and the consumer's resource policy as needed - update `ResourcePolicy` to support both `Stream` and `StreamConsumer` - update `Stream`'s `grant` to leverage `iam.Grant.addToPrincipalOrResource` for cross-environment support closes aws#32050
humanzz · Nov 11, 2024 · a004785 · a004785
1 parent 5648199
commit a004785
Show file tree

Hide file tree

Showing 23 changed files with 1,689 additions and 21 deletions.
diff --git a/...st/aws-kinesis/test/integ.resource-policy.js.snapshot/kinesis-resource-policy.assets.json b/...st/aws-kinesis/test/integ.resource-policy.js.snapshot/kinesis-resource-policy.assets.json
diff --git a/.../aws-kinesis/test/integ.resource-policy.js.snapshot/kinesis-resource-policy.template.json b/.../aws-kinesis/test/integ.resource-policy.js.snapshot/kinesis-resource-policy.template.json
@@ -67,6 +67,65 @@
      "Version": "2012-10-17"
     }
    }
+  },
+  "StreamConsumer58240CBA": {
+   "Type": "AWS::Kinesis::StreamConsumer",
+   "Properties": {
+    "ConsumerName": "stream-consumer",
+    "StreamARN": {
+     "Fn::GetAtt": [
+      "MyStream5C050E93",
+      "Arn"
+     ]
+    }
+   }
+  },
+  "StreamConsumerPolicy925BAE36": {
+   "Type": "AWS::Kinesis::ResourcePolicy",
+   "Properties": {
+    "ResourceArn": {
+     "Fn::GetAtt": [
+      "StreamConsumer58240CBA",
+      "ConsumerARN"
+     ]
+    },
+    "ResourcePolicy": {
+     "Statement": [
+      {
+       "Action": [
+        "kinesis:DescribeStreamConsumer",
+        "kinesis:SubscribeToShard"
+       ],
+       "Effect": "Allow",
+       "Principal": {
+        "AWS": {
+         "Fn::Join": [
+          "",
+          [
+           "arn:",
+           {
+            "Ref": "AWS::Partition"
+           },
+           ":iam::",
+           {
+            "Ref": "AWS::AccountId"
+           },
+           ":root"
+          ]
+         ]
+        }
+       },
+       "Resource": {
+        "Fn::GetAtt": [
+         "StreamConsumer58240CBA",
+         "ConsumerARN"
+        ]
+       }
+      }
+     ],
+     "Version": "2012-10-17"
+    }
+   }
   }
  },
  "Conditions": {

diff --git a/...ing/framework-integ/test/aws-kinesis/test/integ.resource-policy.js.snapshot/manifest.json b/...ing/framework-integ/test/aws-kinesis/test/integ.resource-policy.js.snapshot/manifest.json
diff --git a/...testing/framework-integ/test/aws-kinesis/test/integ.resource-policy.js.snapshot/tree.json b/...testing/framework-integ/test/aws-kinesis/test/integ.resource-policy.js.snapshot/tree.json
diff --git a/packages/@aws-cdk-testing/framework-integ/test/aws-kinesis/test/integ.resource-policy.ts b/packages/@aws-cdk-testing/framework-integ/test/aws-kinesis/test/integ.resource-policy.ts
@@ -1,5 +1,5 @@
 import { App, Stack } from 'aws-cdk-lib';
-import { Stream } from 'aws-cdk-lib/aws-kinesis';
+import { Stream, StreamConsumer } from 'aws-cdk-lib/aws-kinesis';
 import { AccountPrincipal, PolicyStatement } from 'aws-cdk-lib/aws-iam';
 import { IntegTest } from '@aws-cdk/integ-tests-alpha';
 
@@ -8,6 +8,11 @@ const stack = new Stack(app, 'kinesis-resource-policy');
 
 const stream = new Stream(stack, 'MyStream');
 
+const streamConsumer = new StreamConsumer(stack, 'StreamConsumer', {
+  streamConsumerName: 'stream-consumer',
+  stream: stream,
+});
+
 stream.addToResourcePolicy(new PolicyStatement({
   resources: [stream.streamArn],
   actions: [
@@ -17,6 +22,15 @@ stream.addToResourcePolicy(new PolicyStatement({
   principals: [new AccountPrincipal(stack.account)],
 }));
 
+streamConsumer.addToResourcePolicy(new PolicyStatement({
+  resources: [streamConsumer.streamConsumerArn],
+  actions: [
+    'kinesis:DescribeStreamConsumer',
+    'kinesis:SubscribeToShard',
+  ],
+  principals: [new AccountPrincipal(stack.account)],
+}));
+
 new IntegTest(app, 'integ-kinesis-resource-policy', {
   testCases: [stack],
   stackUpdateWorkflow: false,

diff --git a/...k-testing/framework-integ/test/aws-kinesis/test/integ.stream-consumer.js.snapshot/cdk.out b/...k-testing/framework-integ/test/aws-kinesis/test/integ.stream-consumer.js.snapshot/cdk.out
diff --git a/...esting/framework-integ/test/aws-kinesis/test/integ.stream-consumer.js.snapshot/integ.json b/...esting/framework-integ/test/aws-kinesis/test/integ.stream-consumer.js.snapshot/integ.json
diff --git a/...onsumer.js.snapshot/integkinesisstreamconsumerDefaultTestDeployAssert5861EECB.assets.json b/...onsumer.js.snapshot/integkinesisstreamconsumerDefaultTestDeployAssert5861EECB.assets.json
diff --git a/...sumer.js.snapshot/integkinesisstreamconsumerDefaultTestDeployAssert5861EECB.template.json b/...sumer.js.snapshot/integkinesisstreamconsumerDefaultTestDeployAssert5861EECB.template.json
diff --git a/...st/aws-kinesis/test/integ.stream-consumer.js.snapshot/kinesis-stream-consumer.assets.json b/...st/aws-kinesis/test/integ.stream-consumer.js.snapshot/kinesis-stream-consumer.assets.json