forked from aws/aws-cdk
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(iot): add Action to put records to a Firehose stream (aws#17466)
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
Showing
8 changed files
with
612 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
88 changes: 88 additions & 0 deletions
88
packages/@aws-cdk/aws-iot-actions/lib/firehose-stream-action.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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, | ||
}, | ||
}, | ||
}; | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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'; |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
143 changes: 143 additions & 0 deletions
143
packages/@aws-cdk/aws-iot-actions/test/firehose-stream-action/firehose-stream-action.test.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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'], | ||
}); | ||
}); |
Oops, something went wrong.