feat(iot): add Action to put record to Kinesis Data stream

packages/@aws-cdk/aws-iot-actions/README.md

@@ -26,6 +26,7 @@ Currently supported are:
 - Put logs to CloudWatch Logs
 - Capture CloudWatch metrics
 - Change state for a CloudWatch alarm
+- Put records to Kinesis Data stream
 - Put records to Kinesis Data Firehose stream
 - Send messages to SQS queues
 
@@ -183,6 +184,28 @@ const topicRule = new iot.TopicRule(this, 'TopicRule', {
 });
 ```
 
+## Put records to Kinesis Data stream
+
+The code snippet below creates an AWS IoT Rule that put records to Kinesis Data
+stream when it is triggered.
+
+```ts
+import * as iot from '@aws-cdk/aws-iot';
+import * as actions from '@aws-cdk/aws-iot-actions';
+import * as kinesis from '@aws-cdk/aws-kinesis';
+
+const stream = new kinesis.Stream(this, 'MyStream');
+
+const topicRule = new iot.TopicRule(this, 'TopicRule', {
+  sql: iot.IotSql.fromStringAsVer20160323("SELECT * FROM 'device/+/data'"),
+  actions: [
+    new actions.KinesisPutRecordAction(stream, {
+      partitionKey: '${timestamp()}', // optional property
+    }),
+  ],
+});
+```
+
 ## Put records to Kinesis Data Firehose stream
 
 The code snippet below creates an AWS IoT Rule that put records to Put records

packages/@aws-cdk/aws-iot-actions/lib/index.ts

@@ -3,7 +3,7 @@ export * from './cloudwatch-put-metric-action';
 export * from './cloudwatch-set-alarm-state-action';
 export * from './common-action-props';
 export * from './firehose-stream-action';
+export * from './kinesis-put-record-action';
 export * from './lambda-function-action';
 export * from './s3-put-object-action';
 export * from './sqs-queue-action';
-

packages/@aws-cdk/aws-iot-actions/lib/kinesis-put-record-action.ts

@@ -0,0 +1,54 @@
+import * as iam from '@aws-cdk/aws-iam';
+import * as iot from '@aws-cdk/aws-iot';
+import * as kinesis from '@aws-cdk/aws-kinesis';
+import { CommonActionProps } from './common-action-props';
+import { singletonActionRole } from './private/role';
+
+/**
+ * Configuration properties of an action for the Kinesis Data stream.
+ */
+export interface KinesisPutRecordActionProps extends CommonActionProps {
+  /**
+   * The partition key used to determine to which shard the data is written.
+   * The partition key is usually composed of an expression (for example, ${topic()} or ${timestamp()}).
+   * For more information @see https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html#API_PutRecord_RequestParameters
+   *
+   * @default - None
+   */
+  readonly partitionKey?: string;
+}
+
+/**
+ * The action to put the record from an MQTT message to the Kinesis Data stream.
+ */
+export class KinesisPutRecordAction implements iot.IAction {
+  private readonly partitionKey?: string;
+  private readonly role?: iam.IRole;
+
+  /**
+   * @param stream The Kinesis Data stream to which to put records.
+   * @param props Optional properties to not use default
+   */
+  constructor(private readonly stream: kinesis.IStream, props: KinesisPutRecordActionProps = {}) {
+    this.partitionKey = props.partitionKey;
+    this.role = props.role;
+  }
+
+  bind(rule: iot.ITopicRule): iot.ActionConfig {
+    const role = this.role ?? singletonActionRole(rule);
+    role.addToPrincipalPolicy(new iam.PolicyStatement({
+      actions: ['kinesis:PutRecord'],
+      resources: [this.stream.streamArn],
+    }));
+
+    return {
+      configuration: {
+        kinesis: {
+          streamName: this.stream.streamName,
+          partitionKey: this.partitionKey,
+          roleArn: role.roleArn,
+        },
+      },
+    };
+  }
+}

packages/@aws-cdk/aws-iot-actions/package.json

@@ -83,6 +83,7 @@
     "@aws-cdk/aws-cloudwatch": "0.0.0",
     "@aws-cdk/aws-iam": "0.0.0",
     "@aws-cdk/aws-iot": "0.0.0",
+    "@aws-cdk/aws-kinesis": "0.0.0",
     "@aws-cdk/aws-kinesisfirehose": "0.0.0",
     "@aws-cdk/aws-lambda": "0.0.0",
     "@aws-cdk/aws-logs": "0.0.0",
@@ -97,6 +98,7 @@
     "@aws-cdk/aws-cloudwatch": "0.0.0",
     "@aws-cdk/aws-iam": "0.0.0",
     "@aws-cdk/aws-iot": "0.0.0",
+    "@aws-cdk/aws-kinesis": "0.0.0",
     "@aws-cdk/aws-kinesisfirehose": "0.0.0",
     "@aws-cdk/aws-lambda": "0.0.0",
     "@aws-cdk/aws-logs": "0.0.0",

packages/@aws-cdk/aws-iot-actions/test/kinesis-stream/integ.kinesis-put-record-action.expected.json

@@ -0,0 +1,157 @@
+{
+  "Resources": {
+    "Logs6819BB44": {
+      "Type": "AWS::Logs::LogGroup",
+      "Properties": {
+        "RetentionInDays": 731
+      },
+      "UpdateReplacePolicy": "Delete",
+      "DeletionPolicy": "Delete"
+    },
+    "TopicRuleTopicRuleActionRole246C4F77": {
+      "Type": "AWS::IAM::Role",
+      "Properties": {
+        "AssumeRolePolicyDocument": {
+          "Statement": [
+            {
+              "Action": "sts:AssumeRole",
+              "Effect": "Allow",
+              "Principal": {
+                "Service": "iot.amazonaws.com"
+              }
+            }
+          ],
+          "Version": "2012-10-17"
+        }
+      }
+    },
+    "TopicRuleTopicRuleActionRoleDefaultPolicy99ADD687": {
+      "Type": "AWS::IAM::Policy",
+      "Properties": {
+        "PolicyDocument": {
+          "Statement": [
+            {
+              "Action": [
+                "logs:CreateLogStream",
+                "logs:PutLogEvents"
+              ],
+              "Effect": "Allow",
+              "Resource": {
+                "Fn::GetAtt": [
+                  "Logs6819BB44",
+                  "Arn"
+                ]
+              }
+            },
+            {
+              "Action": "logs:DescribeLogStreams",
+              "Effect": "Allow",
+              "Resource": {
+                "Fn::GetAtt": [
+                  "Logs6819BB44",
+                  "Arn"
+                ]
+              }
+            },
+            {
+              "Action": "kinesis:PutRecord",
+              "Effect": "Allow",
+              "Resource": {
+                "Fn::GetAtt": [
+                  "MyStream5C050E93",
+                  "Arn"
+                ]
+              }
+            }
+          ],
+          "Version": "2012-10-17"
+        },
+        "PolicyName": "TopicRuleTopicRuleActionRoleDefaultPolicy99ADD687",
+        "Roles": [
+          {
+            "Ref": "TopicRuleTopicRuleActionRole246C4F77"
+          }
+        ]
+      }
+    },
+    "TopicRule40A4EA44": {
+      "Type": "AWS::IoT::TopicRule",
+      "Properties": {
+        "TopicRulePayload": {
+          "Actions": [
+            {
+              "Kinesis": {
+                "PartitionKey": "${timestamp()}",
+                "RoleArn": {
+                  "Fn::GetAtt": [
+                    "TopicRuleTopicRuleActionRole246C4F77",
+                    "Arn"
+                  ]
+                },
+                "StreamName": {
+                  "Ref": "MyStream5C050E93"
+                }
+              }
+            }
+          ],
+          "AwsIotSqlVersion": "2016-03-23",
+          "ErrorAction": {
+            "CloudwatchLogs": {
+              "LogGroupName": {
+                "Ref": "Logs6819BB44"
+              },
+              "RoleArn": {
+                "Fn::GetAtt": [
+                  "TopicRuleTopicRuleActionRole246C4F77",
+                  "Arn"
+                ]
+              }
+            }
+          },
+          "Sql": "SELECT * FROM 'device/+/data'"
+        }
+      }
+    },
+    "MyStream5C050E93": {
+      "Type": "AWS::Kinesis::Stream",
+      "Properties": {
+        "ShardCount": 3,
+        "RetentionPeriodHours": 24,
+        "StreamEncryption": {
+          "Fn::If": [
+            "AwsCdkKinesisEncryptedStreamsUnsupportedRegions",
+            {
+              "Ref": "AWS::NoValue"
+            },
+            {
+              "EncryptionType": "KMS",
+              "KeyId": "alias/aws/kinesis"
+            }
+          ]
+        }
+      }
+    }
+  },
+  "Conditions": {
+    "AwsCdkKinesisEncryptedStreamsUnsupportedRegions": {
+      "Fn::Or": [
+        {
+          "Fn::Equals": [
+            {
+              "Ref": "AWS::Region"
+            },
+            "cn-north-1"
+          ]
+        },
+        {
+          "Fn::Equals": [
+            {
+              "Ref": "AWS::Region"
+            },
+            "cn-northwest-1"
+          ]
+        }
+      ]
+    }
+  }
+}

packages/@aws-cdk/aws-iot-actions/test/kinesis-stream/integ.kinesis-put-record-action.ts

@@ -0,0 +1,35 @@
+import * as iot from '@aws-cdk/aws-iot';
+import * as kinesis from '@aws-cdk/aws-kinesis';
+import * as logs from '@aws-cdk/aws-logs';
+import * as cdk from '@aws-cdk/core';
+import * as actions from '../../lib';
+
+
+const app = new cdk.App();
+
+class TestStack extends cdk.Stack {
+  constructor(scope: cdk.App, id: string, props?: cdk.StackProps) {
+    super(scope, id, props);
+
+    const topicRule = new iot.TopicRule(this, 'TopicRule', {
+      sql: iot.IotSql.fromStringAsVer20160323(
+        "SELECT * FROM 'device/+/data'",
+      ),
+      errorAction: new actions.CloudWatchLogsAction(
+        new logs.LogGroup(this, 'Logs', { removalPolicy: cdk.RemovalPolicy.DESTROY }),
+      ),
+    });
+
+    const stream = new kinesis.Stream(this, 'MyStream', {
+      shardCount: 3,
+    });
+    topicRule.addAction(
+      new actions.KinesisPutRecordAction(stream, {
+        partitionKey: '${timestamp()}',
+      }),
+    );
+  }
+}
+
+new TestStack(app, 'test-stack');
+app.synth();