Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

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

Merged
merged 5 commits into from
Nov 12, 2021
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 32 additions & 0 deletions packages/@aws-cdk/aws-iot-actions/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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,
});
Comment on lines +139 to +141
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's kill the removalPolicy in the example, it doesn't add much:

Suggested change
const bucket = new s3.Bucket(this, 'MyBucket', {
removalPolicy: cdk.RemovalPolicy.DESTROY,
});
const bucket = new s3.Bucket(this, 'MyBucket');

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,
})
],
});
```
88 changes: 88 additions & 0 deletions packages/@aws-cdk/aws-iot-actions/lib/firehose-stream-action.ts
Original file line number Diff line number Diff line change
@@ -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 {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Missed this:

Suggested change
export interface FirehoseStreamProps extends CommonActionProps {
export interface FirehoseStreamActionProps 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,
},
},
};
}
}
1 change: 1 addition & 0 deletions packages/@aws-cdk/aws-iot-actions/lib/index.ts
Original file line number Diff line number Diff line change
@@ -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';
3 changes: 3 additions & 0 deletions packages/@aws-cdk/aws-iot-actions/package.json
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand All @@ -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",
Expand All @@ -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",
Expand Down
Original file line number Diff line number Diff line change
@@ -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'],
});
});
Loading