Skip to content

Commit

Permalink
feat(iot): add Action to put records to a Firehose stream (aws#17466)
Browse files Browse the repository at this point in the history
I'm trying to implement aws-iot L2 Constructs.

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

----

*By submitting this pull request, I confirm that my contribution is made under the terms of the Apache-2.0 license*
  • Loading branch information
yamatatsu authored and TikiTDO committed Feb 21, 2022
1 parent a4b9198 commit 3b63975
Show file tree
Hide file tree
Showing 8 changed files with 612 additions and 1 deletion.
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,
});
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 {
/**
* 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

0 comments on commit 3b63975

Please sign in to comment.