feat(iot): add Action to put records to a Firehose stream (#17466)

I'm trying to implement aws-iot L2 Constructs.

This PR is one of steps after following PR: 
- #16681 (comment)

----

*By submitting this pull request, I confirm that my contribution is made under the terms of the Apache-2.0 license*

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

@@ -24,6 +24,7 @@ Currently supported are:
 - Invoke a Lambda function
 - Put objects to a S3 bucket
 - Put logs to CloudWatch Logs
+- Put records to Kinesis Data Firehose stream
 
 ## Invoke a Lambda function
 
@@ -121,3 +122,34 @@ new iot.TopicRule(this, 'TopicRule', {
   actions: [new actions.CloudWatchLogsAction(logGroup)],
 });
 ```
+
+
+## Put records to Kinesis Data Firehose stream
+
+The code snippet below creates an AWS IoT Rule that put records to Put records
+to Kinesis Data Firehose 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 s3 from '@aws-cdk/aws-s3';
+import * as firehose from '@aws-cdk/aws-kinesisfirehose';
+import * as destinations from '@aws-cdk/aws-kinesisfirehose-destinations';
+
+const bucket = new s3.Bucket(this, 'MyBucket', {
+  removalPolicy: cdk.RemovalPolicy.DESTROY,
+});
+const stream = new firehose.DeliveryStream(this, 'MyStream', {
+  destinations: [new destinations.S3Bucket(bucket)],
+});
+
+const topicRule = new iot.TopicRule(this, 'TopicRule', {
+  sql: iot.IotSql.fromStringAsVer20160323("SELECT * FROM 'device/+/data'"),
+  actions: [
+    new actions.FirehoseStreamAction(stream, {
+      batchMode: true,
+      recordSeparator: actions.FirehoseStreamRecordSeparator.NEWLINE,
+    })
+  ],
+});
+```

packages/@aws-cdk/aws-iot-actions/lib/firehose-stream-action.ts

@@ -0,0 +1,88 @@
+import * as iam from '@aws-cdk/aws-iam';
+import * as iot from '@aws-cdk/aws-iot';
+import * as firehose from '@aws-cdk/aws-kinesisfirehose';
+import { CommonActionProps } from './common-action-props';
+import { singletonActionRole } from './private/role';
+
+/**
+ * Record Separator to be used to separate records.
+ */
+export enum FirehoseStreamRecordSeparator {
+  /**
+   * Separate by a new line
+   */
+  NEWLINE = '\n',
+
+  /**
+   * Separate by a tab
+   */
+  TAB = '\t',
+
+  /**
+   * Separate by a windows new line
+   */
+  WINDOWS_NEWLINE = '\r\n',
+
+  /**
+   * Separate by a commma
+   */
+  COMMA = ',',
+}
+
+/**
+ * Configuration properties of an action for the Kinesis Data Firehose stream.
+ */
+export interface FirehoseStreamProps extends CommonActionProps {
+  /**
+   * Whether to deliver the Kinesis Data Firehose stream as a batch by using `PutRecordBatch`.
+   * When batchMode is true and the rule's SQL statement evaluates to an Array, each Array
+   * element forms one record in the PutRecordBatch request. The resulting array can't have
+   * more than 500 records.
+   *
+   * @default false
+   */
+  readonly batchMode?: boolean;
+
+  /**
+   * A character separator that will be used to separate records written to the Kinesis Data Firehose stream.
+   *
+   * @default - none -- the stream does not use a separator
+   */
+  readonly recordSeparator?: FirehoseStreamRecordSeparator;
+}
+
+
+/**
+ * The action to put the record from an MQTT message to the Kinesis Data Firehose stream.
+ */
+export class FirehoseStreamAction implements iot.IAction {
+  private readonly batchMode?: boolean;
+  private readonly recordSeparator?: string;
+  private readonly role?: iam.IRole;
+
+  /**
+   * @param stream The Kinesis Data Firehose stream to which to put records.
+   * @param props Optional properties to not use default
+   */
+  constructor(private readonly stream: firehose.IDeliveryStream, props: FirehoseStreamProps = {}) {
+    this.batchMode = props.batchMode;
+    this.recordSeparator = props.recordSeparator;
+    this.role = props.role;
+  }
+
+  bind(rule: iot.ITopicRule): iot.ActionConfig {
+    const role = this.role ?? singletonActionRole(rule);
+    this.stream.grantPutRecords(role);
+
+    return {
+      configuration: {
+        firehose: {
+          batchMode: this.batchMode,
+          deliveryStreamName: this.stream.deliveryStreamName,
+          roleArn: role.roleArn,
+          separator: this.recordSeparator,
+        },
+      },
+    };
+  }
+}

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

@@ -1,4 +1,5 @@
 export * from './cloudwatch-logs-action';
 export * from './common-action-props';
+export * from './firehose-stream-action';
 export * from './lambda-function-action';
 export * from './s3-put-object-action';

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

@@ -71,6 +71,7 @@
   "license": "Apache-2.0",
   "devDependencies": {
     "@aws-cdk/assertions": "0.0.0",
+    "@aws-cdk/aws-kinesisfirehose-destinations": "0.0.0",
     "@aws-cdk/cdk-build-tools": "0.0.0",
     "@aws-cdk/cdk-integ-tools": "0.0.0",
     "@aws-cdk/pkglint": "0.0.0",
@@ -81,6 +82,7 @@
   "dependencies": {
     "@aws-cdk/aws-iam": "0.0.0",
     "@aws-cdk/aws-iot": "0.0.0",
+    "@aws-cdk/aws-kinesisfirehose": "0.0.0",
     "@aws-cdk/aws-lambda": "0.0.0",
     "@aws-cdk/aws-logs": "0.0.0",
     "@aws-cdk/aws-s3": "0.0.0",
@@ -92,6 +94,7 @@
   "peerDependencies": {
     "@aws-cdk/aws-iam": "0.0.0",
     "@aws-cdk/aws-iot": "0.0.0",
+    "@aws-cdk/aws-kinesisfirehose": "0.0.0",
     "@aws-cdk/aws-lambda": "0.0.0",
     "@aws-cdk/aws-logs": "0.0.0",
     "@aws-cdk/aws-s3": "0.0.0",

packages/@aws-cdk/aws-iot-actions/test/firehose-stream-action/firehose-stream-action.test.ts

@@ -0,0 +1,143 @@
+import { Template, Match } from '@aws-cdk/assertions';
+import * as iam from '@aws-cdk/aws-iam';
+import * as iot from '@aws-cdk/aws-iot';
+import * as firehose from '@aws-cdk/aws-kinesisfirehose';
+import * as cdk from '@aws-cdk/core';
+import * as actions from '../../lib';
+
+test('Default firehose stream action', () => {
+  // GIVEN
+  const stack = new cdk.Stack();
+  const topicRule = new iot.TopicRule(stack, 'MyTopicRule', {
+    sql: iot.IotSql.fromStringAsVer20160323("SELECT topic(2) as device_id FROM 'device/+/data'"),
+  });
+  const stream = firehose.DeliveryStream.fromDeliveryStreamArn(stack, 'MyStream', 'arn:aws:firehose:xx-west-1:111122223333:deliverystream/my-stream');
+
+  // WHEN
+  topicRule.addAction(
+    new actions.FirehoseStreamAction(stream),
+  );
+
+  // THEN
+  Template.fromStack(stack).hasResourceProperties('AWS::IoT::TopicRule', {
+    TopicRulePayload: {
+      Actions: [
+        {
+          Firehose: {
+            DeliveryStreamName: 'my-stream',
+            RoleArn: {
+              'Fn::GetAtt': ['MyTopicRuleTopicRuleActionRoleCE2D05DA', 'Arn'],
+            },
+          },
+        },
+      ],
+    },
+  });
+
+  Template.fromStack(stack).hasResourceProperties('AWS::IAM::Role', {
+    AssumeRolePolicyDocument: {
+      Statement: [
+        {
+          Action: 'sts:AssumeRole',
+          Effect: 'Allow',
+          Principal: {
+            Service: 'iot.amazonaws.com',
+          },
+        },
+      ],
+      Version: '2012-10-17',
+    },
+  });
+
+  Template.fromStack(stack).hasResourceProperties('AWS::IAM::Policy', {
+    PolicyDocument: {
+      Statement: [
+        {
+          Action: ['firehose:PutRecord', 'firehose:PutRecordBatch'],
+          Effect: 'Allow',
+          Resource: 'arn:aws:firehose:xx-west-1:111122223333:deliverystream/my-stream',
+        },
+      ],
+      Version: '2012-10-17',
+    },
+    PolicyName: 'MyTopicRuleTopicRuleActionRoleDefaultPolicy54A701F7',
+    Roles: [
+      { Ref: 'MyTopicRuleTopicRuleActionRoleCE2D05DA' },
+    ],
+  });
+});
+
+test('can set batchMode', () => {
+  // GIVEN
+  const stack = new cdk.Stack();
+  const topicRule = new iot.TopicRule(stack, 'MyTopicRule', {
+    sql: iot.IotSql.fromStringAsVer20160323("SELECT topic(2) as device_id FROM 'device/+/data'"),
+  });
+  const stream = firehose.DeliveryStream.fromDeliveryStreamArn(stack, 'MyStream', 'arn:aws:firehose:xx-west-1:111122223333:deliverystream/my-stream');
+
+  // WHEN
+  topicRule.addAction(
+    new actions.FirehoseStreamAction(stream, { batchMode: true }),
+  );
+
+  // THEN
+  Template.fromStack(stack).hasResourceProperties('AWS::IoT::TopicRule', {
+    TopicRulePayload: {
+      Actions: [
+        Match.objectLike({ Firehose: { BatchMode: true } }),
+      ],
+    },
+  });
+});
+
+test('can set separotor', () => {
+  // GIVEN
+  const stack = new cdk.Stack();
+  const topicRule = new iot.TopicRule(stack, 'MyTopicRule', {
+    sql: iot.IotSql.fromStringAsVer20160323("SELECT topic(2) as device_id FROM 'device/+/data'"),
+  });
+  const stream = firehose.DeliveryStream.fromDeliveryStreamArn(stack, 'MyStream', 'arn:aws:firehose:xx-west-1:111122223333:deliverystream/my-stream');
+
+  // WHEN
+  topicRule.addAction(
+    new actions.FirehoseStreamAction(stream, { recordSeparator: actions.FirehoseStreamRecordSeparator.NEWLINE }),
+  );
+
+  // THEN
+  Template.fromStack(stack).hasResourceProperties('AWS::IoT::TopicRule', {
+    TopicRulePayload: {
+      Actions: [
+        Match.objectLike({ Firehose: { Separator: '\n' } }),
+      ],
+    },
+  });
+});
+
+test('can set role', () => {
+  // GIVEN
+  const stack = new cdk.Stack();
+  const topicRule = new iot.TopicRule(stack, 'MyTopicRule', {
+    sql: iot.IotSql.fromStringAsVer20160323("SELECT topic(2) as device_id FROM 'device/+/data'"),
+  });
+  const stream = firehose.DeliveryStream.fromDeliveryStreamArn(stack, 'MyStream', 'arn:aws:firehose:xx-west-1:111122223333:deliverystream/my-stream');
+  const role = iam.Role.fromRoleArn(stack, 'MyRole', 'arn:aws:iam::123456789012:role/ForTest');
+
+  // WHEN
+  topicRule.addAction(
+    new actions.FirehoseStreamAction(stream, { role }),
+  );
+
+  // THEN
+  Template.fromStack(stack).hasResourceProperties('AWS::IoT::TopicRule', {
+    TopicRulePayload: {
+      Actions: [
+        Match.objectLike({ Firehose: { RoleArn: 'arn:aws:iam::123456789012:role/ForTest' } }),
+      ],
+    },
+  });
+
+  Template.fromStack(stack).hasResourceProperties('AWS::IAM::Policy', {
+    PolicyName: 'MyRolePolicy64AB00A5',
+    Roles: ['ForTest'],
+  });
+});